"""Paper trading engine: simulated maker fills against live book top.""" from __future__ import annotations import json import time import uuid from collections import deque from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional from .config import PaperCfg from .market_data import BookTop from .quoter import Quote @dataclass class Fill: id: str ts: float side: str # buy | sell price: float size_btc: float fee_usd: float mid_at_fill: float @dataclass class OpenLot: id: str side: str # long | short price: float size_btc: float notional_usd: float ts: float @dataclass class ClosedTrade: id: str ts: float side: str # long | short (the opened side that closed) entry_price: float exit_price: float size_btc: float pnl_usd: float fee_usd: float profitable: bool @dataclass class PaperState: equity0: float cash_usd: float inventory_btc: float = 0.0 realized_pnl: float = 0.0 fills: List[Fill] = field(default_factory=list) open_lots: List[OpenLot] = field(default_factory=list) closed_trades: List[ClosedTrade] = field(default_factory=list) day_start_ts: float = field(default_factory=time.time) day_start_equity: float = 0.0 quote: Optional[Quote] = None last_mid: float = 0.0 prev_mid: float = 0.0 # VWAP buys / sells (volume-weighted) buy_notional: float = 0.0 buy_qty: float = 0.0 sell_notional: float = 0.0 sell_qty: float = 0.0 # last ~120 mid samples for chart price_history: deque = field(default_factory=lambda: deque(maxlen=600)) candles: deque = field(default_factory=lambda: deque(maxlen=288)) def __post_init__(self) -> None: if not self.day_start_equity: self.day_start_equity = self.equity0 class PaperEngine: MAX_CLOSE_EVENTS = 30 def __init__(self, equity_usd: float, paper: PaperCfg): self.cfg = paper self.state = PaperState(equity0=equity_usd, cash_usd=equity_usd) def set_quote(self, q: Optional[Quote]) -> None: self.state.quote = q def on_book(self, top: BookTop) -> List[Fill]: st = self.state prev = st.prev_mid or top.mid st.prev_mid = st.last_mid or top.mid st.last_mid = top.mid if top.ok() and top.mid > 0: now = time.time() st.price_history.append({"ts": now, "mid": top.mid}) self._update_candle(top.mid, now) # accept both fill_on_touch (config) and legacy fills_on_touch fill_ok = bool(getattr(self.cfg, "fill_on_touch", True) or getattr(self.cfg, "fills_on_touch", False)) if not fill_ok or not st.quote or not top.ok(): return [] q = st.quote new_fills: List[Fill] = [] # Classic maker cross: opposite side of book trades through our price if top.ask > 0 and top.ask <= q.bid_px: new_fills.append(self._fill("buy", q.bid_px, q.size_btc, top.mid)) elif prev > q.bid_px >= top.mid: new_fills.append(self._fill("buy", q.bid_px, q.size_btc, top.mid)) if top.bid > 0 and top.bid >= q.ask_px: new_fills.append(self._fill("sell", q.ask_px, q.size_btc, top.mid)) elif prev < q.ask_px <= top.mid: new_fills.append(self._fill("sell", q.ask_px, q.size_btc, top.mid)) # Optional demo: fill when mid is within near_fill_bps of our quote near = float(getattr(self.cfg, "near_fill_bps", 0.0) or 0.0) if not new_fills and near > 0 and q.size_btc > 0: thr = top.mid * near / 10_000.0 if abs(top.mid - q.bid_px) <= thr: new_fills.append(self._fill("buy", q.bid_px, q.size_btc, top.mid)) elif abs(q.ask_px - top.mid) <= thr: new_fills.append(self._fill("sell", q.ask_px, q.size_btc, top.mid)) if new_fills: st.quote = None # avoid double-counting same quote each tick return new_fills def _fill(self, side: str, price: float, size: float, mid: float) -> Fill: fee_bps = self.cfg.maker_fee_bps notional = price * size fee = notional * fee_bps / 10_000.0 st = self.state if side == "buy": st.cash_usd -= notional + fee st.inventory_btc += size st.buy_notional += notional st.buy_qty += size else: st.cash_usd += notional - fee st.inventory_btc -= size st.sell_notional += notional st.sell_qty += size f = Fill( id=uuid.uuid4().hex[:12], ts=time.time(), side=side, price=price, size_btc=size, fee_usd=fee, mid_at_fill=mid, ) st.fills.append(f) self._apply_fifo(f) try: log = Path(__file__).resolve().parents[1] / "data" / "fills.jsonl" log.parent.mkdir(parents=True, exist_ok=True) with log.open("a") as fh: fh.write( json.dumps( { "id": f.id, "ts": f.ts, "side": f.side, "price": f.price, "size_btc": f.size_btc, "notional_usd": round(f.price * f.size_btc, 4), "fee_usd": f.fee_usd, "mid_at_fill": f.mid_at_fill, } ) + "\n" ) except Exception: pass return f def _apply_fifo(self, f: Fill) -> None: """Match fill against opposite open lots FIFO; leftover becomes a new open lot.""" st = self.state remaining = f.size_btc fee_per_btc = (f.fee_usd / f.size_btc) if f.size_btc else 0.0 if f.side == "buy": # Close shorts first while remaining > 1e-12 and st.open_lots and st.open_lots[0].side == "short": lot = st.open_lots[0] closed_sz = min(remaining, lot.size_btc) # short PnL: entry - exit entry_fee_share = 0.0 # entry fees already realized at open; keep simple exit_fee = fee_per_btc * closed_sz pnl = (lot.price - f.price) * closed_sz - exit_fee - entry_fee_share self._record_close(lot, f.price, closed_sz, pnl, exit_fee) lot.size_btc -= closed_sz lot.notional_usd = lot.price * lot.size_btc remaining -= closed_sz if lot.size_btc <= 1e-12: st.open_lots.pop(0) if remaining > 1e-12: st.open_lots.append( OpenLot( id=uuid.uuid4().hex[:10], side="long", price=f.price, size_btc=remaining, notional_usd=f.price * remaining, ts=f.ts, ) ) else: # sell → close longs first while remaining > 1e-12 and st.open_lots and st.open_lots[0].side == "long": lot = st.open_lots[0] closed_sz = min(remaining, lot.size_btc) exit_fee = fee_per_btc * closed_sz pnl = (f.price - lot.price) * closed_sz - exit_fee self._record_close(lot, f.price, closed_sz, pnl, exit_fee) lot.size_btc -= closed_sz lot.notional_usd = lot.price * lot.size_btc remaining -= closed_sz if lot.size_btc <= 1e-12: st.open_lots.pop(0) if remaining > 1e-12: st.open_lots.append( OpenLot( id=uuid.uuid4().hex[:10], side="short", price=f.price, size_btc=remaining, notional_usd=f.price * remaining, ts=f.ts, ) ) def _record_close( self, lot: OpenLot, exit_price: float, size: float, pnl: float, fee: float ) -> None: st = self.state ct = ClosedTrade( id=uuid.uuid4().hex[:10], ts=time.time(), side=lot.side, entry_price=lot.price, exit_price=exit_price, size_btc=size, pnl_usd=pnl, fee_usd=fee, profitable=pnl > 0, ) st.closed_trades.append(ct) if len(st.closed_trades) > 200: st.closed_trades = st.closed_trades[-200:] st.realized_pnl += pnl def open_positions(self) -> List[dict]: out = [] for lot in self.state.open_lots: out.append( { "id": lot.id, "side": lot.side, "price": lot.price, "size_btc": lot.size_btc, "notional_usd": round(lot.notional_usd, 4), "ts": lot.ts, } ) return out def close_events(self, n: int | None = None) -> List[dict]: n = n if n is not None else self.MAX_CLOSE_EVENTS recent = self.state.closed_trades[-n:] return [ { "id": c.id, "pnl_usd": round(c.pnl_usd, 6), "size_btc": c.size_btc, "ts": c.ts, "profitable": c.profitable, "side": c.side, "entry_price": c.entry_price, "exit_price": c.exit_price, } for c in recent ] def equity(self, mid: Optional[float] = None) -> float: m = mid if mid is not None else self.state.last_mid return self.state.cash_usd + self.state.inventory_btc * m def unrealized(self, mid: Optional[float] = None) -> float: return self.equity(mid) - self.state.equity0 def day_pnl(self, mid: Optional[float] = None) -> float: return self.equity(mid) - self.state.day_start_equity def day_pnl_pct(self, mid: Optional[float] = None) -> float: base = self.state.day_start_equity or self.state.equity0 if base <= 0: return 0.0 return 100.0 * self.day_pnl(mid) / base def avg_buy_px(self) -> Optional[float]: st = self.state if st.buy_qty <= 1e-12: return None return st.buy_notional / st.buy_qty def avg_sell_px(self) -> Optional[float]: st = self.state if st.sell_qty <= 1e-12: return None return st.sell_notional / st.sell_qty def price_history_list(self) -> List[dict]: return list(self.state.price_history) CANDLE_SEC = 300 def _update_candle(self, mid: float, ts: float) -> None: bucket = int(ts // self.CANDLE_SEC) * self.CANDLE_SEC candles = self.state.candles if not candles or int(candles[-1]["t"]) != bucket: candles.append({"t": float(bucket), "o": mid, "h": mid, "l": mid, "c": mid}) else: c = candles[-1] if mid > c["h"]: c["h"] = mid if mid < c["l"]: c["l"] = mid c["c"] = mid def candles_list(self) -> List[dict]: return list(self.state.candles) def fills_today(self) -> int: return len(self.state.fills)