trading bots, built to a task.

from decimal import Decimal, ROUND_DOWN

def position_size(equity: Decimal, entry: Decimal, stop: Decimal, *,
                  risk_per_trade: Decimal = Decimal("0.005"),
                  taker_fee: Decimal = Decimal("0.0006"),
                  slippage: Decimal = Decimal("0.0005"),
                  qty_step: Decimal = Decimal("0.001"),
                  max_notional: Decimal | None = None) -> Decimal:
    """Size so that a stop-out costs at most equity * risk_per_trade,
    counting fees on both legs and some slippage on the stop fill."""
    if entry <= 0 or stop <= 0 or entry == stop:
        return Decimal("0")
    price_risk = abs(entry - stop) + stop * slippage   # stops fill worse than the trigger
    fees = (entry + stop) * taker_fee                  # pay on entry and on exit
    qty = (equity * risk_per_trade) / (price_risk + fees)
    if max_notional is not None:
        qty = min(qty, max_notional / entry)           # leverage / exposure cap
    return (qty / qty_step).to_integral_value(rounding=ROUND_DOWN) * qty_step

from dataclasses import dataclass

@dataclass(frozen=True)
class OrderIntent:
    symbol: str
    side: str                  # "buy" or "sell"
    qty: float
    price: float
    stop: float | None
    reduce_only: bool = False

def refusals(o: OrderIntent, *, equity: float, open_notional: float,
             allowed: frozenset[str], max_leverage: float,
             min_notional: float, breaker_tripped: bool) -> list[str]:
    why, notional = [], o.qty * o.price
    if o.side not in ("buy", "sell"):
        why.append(f"unknown side {o.side!r}")
    if breaker_tripped and not o.reduce_only:
        why.append("breaker tripped: only reduce-only orders may pass")
    if o.symbol not in allowed:
        why.append(f"{o.symbol} is not on the allowlist")
    if o.qty <= 0 or o.price <= 0:
        why.append("quantity and price must be positive")
    elif notional < min_notional:
        why.append(f"notional {notional:.2f} is below the exchange minimum")
    if not o.reduce_only:
        if o.stop is None:
            why.append("directional entry without a stop")
        elif not (o.stop < o.price if o.side == "buy" else o.stop > o.price):
            why.append("stop is on the wrong side of the entry")
        if open_notional + notional > equity * max_leverage:
            why.append("would exceed the gross leverage cap")
    return why

def dca_ladder(entry: float, stop: float, risk_budget: float, *, rungs: int = 4,
               spacing: float = 0.02, mult: float = 1.6, fee_rate: float = 0.0006):
    """Long ladder: rung k rests spacing*k below entry; every rung shares ONE stop.
    Rung k gets a slice of the risk budget proportional to mult**k * (loss per unit
    from rung k to the stop). Sizes grow geometrically, yet if every rung fills and
    the stop hits, the total loss equals risk_budget, never more."""
    prices = [entry * (1 - spacing * k) for k in range(rungs)]
    if prices[-1] <= stop:
        raise ValueError("the deepest rung must sit above the stop")
    loss_per_unit = [p - stop + (p + stop) * fee_rate for p in prices]
    weights = [mult ** k * d for k, d in enumerate(loss_per_unit)]
    total = sum(weights)
    ladder = []
    for k, (p, d, w) in enumerate(zip(prices, loss_per_unit, weights)):
        risk_k = risk_budget * w / total
        ladder.append({"rung": k, "price": p, "qty": risk_k / d, "risk_at_stop": risk_k})
    return ladder

import time
import uuid

class OrderStateUnknown(Exception):
    """Could not prove whether the order exists. Caller must halt new entries."""

class SendTimeout(Exception): ...
class DuplicateClientId(Exception): ...

def send_once(client, order: dict, attempts: int = 3, backoff_s: float = 0.5) -> dict:
    """At most one live order per client_order_id (orderLinkId on Bybit V5).
    A timeout means 'unknown', not 'failed': look the id up before every retry."""
    cid = order.setdefault("client_order_id", "ord-" + uuid.uuid4().hex[:20])
    for attempt in range(attempts):
        if attempt:
            time.sleep(backoff_s * 2 ** (attempt - 1))
            found = client.find_order(order["symbol"], cid)   # open orders + recent history
            if found is not None:
                return found                                  # it landed after all
        try:
            return client.create_order(**order)
        except (SendTimeout, DuplicateClientId):
            continue              # unknown, or already exists: resolve it by lookup
    found = client.find_order(order["symbol"], cid)
    if found is not None:
        return found
    raise OrderStateUnknown(cid)

import logging, time
from enum import Enum

class State(Enum):
    RUNNING, PAUSED, HALTED = "running", "paused", "halted"

class CircuitBreaker:
    def __init__(self, daily_loss=0.02, drawdown=0.10, streak=6, api_errors=20, cooldown_s=3600):
        self.lim = dict(daily_loss=daily_loss, drawdown=drawdown, streak=streak, api_errors=api_errors)
        self.cooldown_s, self.state, self.reason, self.until = cooldown_s, State.RUNNING, "", 0.0

    def check(self, day_pnl: float, drawdown: float, streak: int, api_errors: int) -> State:
        if self.state is State.HALTED:
            return self.state                                     # never clears itself
        if drawdown >= self.lim["drawdown"] or api_errors >= self.lim["api_errors"]:
            return self._trip(State.HALTED, f"drawdown={drawdown:.3f} api_errors={api_errors}")
        if -day_pnl >= self.lim["daily_loss"] or streak >= self.lim["streak"]:
            return self._trip(State.PAUSED, f"day_pnl={day_pnl:.3f} streak={streak}")
        if self.state is State.PAUSED and time.monotonic() >= self.until:
            self.state, self.reason = State.RUNNING, ""           # condition gone + cooled down
        return self.state

    def _trip(self, state: State, reason: str) -> State:
        if self.state is not state:
            logging.warning("breaker -> %s: %s", state.value, reason)
            self.state, self.reason = state, reason
            self.until = time.monotonic() + self.cooldown_s
        return self.state

    def rearm(self, operator: str, note: str) -> None:
        if not operator or not note:
            raise ValueError("re-arming needs a named operator and a written reason")
        logging.warning("re-armed by %s (%s); was: %s", operator, note, self.reason)
        self.state, self.reason = State.RUNNING, ""

def backtest(bars, decide, cash: float, fee_rate=0.0006, slippage=0.0002):
    """bars: oldest-first dicts with 'open' and 'close'.
    decide(closed_bars) returns a target position in units.
    A decision taken at the close of bar i fills at the OPEN of bar i+1."""
    pos, target, curve = 0.0, None, []
    for i, bar in enumerate(bars):
        if target is not None and target != pos:
            delta = target - pos
            px = bar["open"] * ((1 + slippage) if delta > 0 else (1 - slippage))
            cash -= delta * px + abs(delta) * px * fee_rate
            pos = target
        target = decide(bars[: i + 1])        # sees closed bars only, never bar i+1
        curve.append(cash + pos * bar["close"])
    return curve                              # the final decision has no next open: dropped