from __future__ import annotations from dataclasses import dataclass from decimal import Decimal, ROUND_HALF_UP from src.execution.order_intent import OrderIntent @dataclass(frozen=True) class PaperCandle: open: Decimal high: Decimal low: Decimal close: Decimal @dataclass(frozen=True) class PaperLifecycleResult: strategy_id: str coin: str side: str entry_price: Decimal exit_price: Decimal exit_reason: str candles_held: int size: Decimal gross_pnl_usd: Decimal entry_fee_usd: Decimal exit_fee_usd: Decimal slippage_cost_usd: Decimal funding_cost_usd: Decimal net_pnl_usd: Decimal mfe_pct: Decimal mae_pct: Decimal requested_size: Decimal | None = None filled_size: Decimal | None = None unfilled_size: Decimal = Decimal("0") fill_ratio: Decimal = Decimal("1") entry_liquidity: str = "taker" exit_liquidity: str = "taker" entry_slippage_pct: Decimal = Decimal("0") exit_slippage_pct: Decimal = Decimal("0") stop_gap_slippage_pct: Decimal = Decimal("0") live_order_allowed: bool = False mainnet_signed_action: bool = False paper_trading: bool = True def to_journal_row(self) -> dict[str, str | bool | int]: return { "strategy_id": self.strategy_id, "coin": self.coin, "side": self.side, "entry_price": str(self.entry_price), "exit_price": str(self.exit_price), "exit_reason": self.exit_reason, "candles_held": self.candles_held, "size": str(self.size), "requested_size": str(self.requested_size if self.requested_size is not None else self.size), "filled_size": str(self.filled_size if self.filled_size is not None else self.size), "unfilled_size": str(self.unfilled_size), "fill_ratio": str(self.fill_ratio), "entry_liquidity": self.entry_liquidity, "exit_liquidity": self.exit_liquidity, "entry_slippage_pct": str(self.entry_slippage_pct), "exit_slippage_pct": str(self.exit_slippage_pct), "stop_gap_slippage_pct": str(self.stop_gap_slippage_pct), "gross_pnl_usd": str(self.gross_pnl_usd), "entry_fee_usd": str(self.entry_fee_usd), "exit_fee_usd": str(self.exit_fee_usd), "slippage_cost_usd": str(self.slippage_cost_usd), "funding_cost_usd": str(self.funding_cost_usd), "net_pnl_usd": str(self.net_pnl_usd), "mfe_pct": str(self.mfe_pct), "mae_pct": str(self.mae_pct), "live_order_allowed": self.live_order_allowed, "mainnet_signed_action": self.mainnet_signed_action, "paper_trading": self.paper_trading, } def _q(value: Decimal, places: str = "0.00000001") -> Decimal: return value.quantize(Decimal(places), rounding=ROUND_HALF_UP) def _pct(numerator: Decimal, denominator: Decimal) -> Decimal: if denominator <= 0: return Decimal("0") return (numerator / denominator * Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) def _fee_rate_for(liquidity: str, *, maker_fee_rate: Decimal, taker_fee_rate: Decimal) -> Decimal: return maker_fee_rate if liquidity == "maker" else taker_fee_rate def _slippage_for(liquidity: str, pct: Decimal) -> Decimal: # Maker fills are modeled at resting price; taker fills pay spread/depth slippage. return Decimal("0") if liquidity == "maker" else pct def simulate_paper_lifecycle( intent: OrderIntent, *, entry_price: Decimal, candles: list[PaperCandle], fee_rate: Decimal = Decimal("0.00045"), slippage_pct: Decimal = Decimal("0.02"), hold_hours_per_candle: Decimal = Decimal("1"), funding_rate_hourly_pct: Decimal = Decimal("0"), trailing_stop_pct: Decimal | None = None, max_fill_notional_usd: Decimal | None = None, maker_fee_rate: Decimal = Decimal("0.00020"), taker_fee_rate: Decimal | None = None, entry_liquidity: str = "taker", exit_liquidity: str = "taker", depth_slippage_pct: Decimal | None = None, stop_gap_slippage_pct: Decimal | None = None, ) -> PaperLifecycleResult: """Simulate a deterministic long-only Paper lifecycle. Conservative v1 lifecycle model for replacing expected-move-only Paper exits: entry fill, optional partial fill cap, maker/taker fee class, stop/TP/trailing exits, stop-gap slippage, depth-aware slippage, MFE/MAE, fees and funding. It never creates live side effects. """ if intent.side != "buy": raise ValueError("paper lifecycle currently supports long/buy intents only") if intent.stop_loss is None: raise ValueError("paper lifecycle requires a stop_loss") if entry_price <= 0: raise ValueError("entry_price must be positive") if not candles: raise ValueError("at least one candle is required") if entry_liquidity not in {"maker", "taker"} or exit_liquidity not in {"maker", "taker"}: raise ValueError("liquidity must be maker or taker") requested_size = intent.size fill_notional_cap = max_fill_notional_usd if max_fill_notional_usd is not None else entry_price * requested_size filled_size = min(requested_size, fill_notional_cap / entry_price) if filled_size <= 0: raise ValueError("paper lifecycle partial fill produced no filled size") unfilled_size = requested_size - filled_size fill_ratio = (filled_size / requested_size).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP) taker_rate = taker_fee_rate if taker_fee_rate is not None else fee_rate effective_slippage_pct = depth_slippage_pct if depth_slippage_pct is not None else slippage_pct entry_slip_pct = _slippage_for(entry_liquidity, effective_slippage_pct) stop_gap_pct = stop_gap_slippage_pct if stop_gap_slippage_pct is not None else effective_slippage_pct size = filled_size stop = intent.stop_loss take_profit = intent.take_profit highest = entry_price lowest = entry_price exit_price = candles[-1].close exit_reason = "time_exit" candles_held = len(candles) for idx, candle in enumerate(candles, start=1): highest = max(highest, candle.high) lowest = min(lowest, candle.low) if trailing_stop_pct is not None and highest > entry_price: trailing_stop = highest * (Decimal("1") - trailing_stop_pct / Decimal("100")) stop = max(stop, trailing_stop.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)) # Conservative ordering: protective stop first if both are inside one candle. if candle.low <= stop: if candle.open < stop and candle.high < stop: exit_price = candle.open * (Decimal("1") - stop_gap_pct / Decimal("100")) exit_reason = "stop_gap_slippage" else: exit_price = stop exit_reason = "trailing_stop" if stop > intent.stop_loss else "stop_loss" candles_held = idx break if take_profit is not None and candle.high >= take_profit: exit_price = take_profit exit_reason = "take_profit" candles_held = idx break exit_slip_pct = stop_gap_pct if exit_reason == "stop_gap_slippage" else _slippage_for(exit_liquidity, effective_slippage_pct) gross = (exit_price - entry_price) * size entry_notional = entry_price * size exit_notional = exit_price * size entry_fee = entry_notional * _fee_rate_for(entry_liquidity, maker_fee_rate=maker_fee_rate, taker_fee_rate=taker_rate) exit_fee = exit_notional * _fee_rate_for(exit_liquidity, maker_fee_rate=maker_fee_rate, taker_fee_rate=taker_rate) slippage_cost = (entry_notional * entry_slip_pct / Decimal("100")) + (exit_notional * exit_slip_pct / Decimal("100")) funding_cost = entry_notional * funding_rate_hourly_pct / Decimal("100") * hold_hours_per_candle * Decimal(candles_held) net = gross - entry_fee - exit_fee - slippage_cost - funding_cost mfe = _pct(highest - entry_price, entry_price) mae = _pct(lowest - entry_price, entry_price) return PaperLifecycleResult( strategy_id=intent.strategy_id, coin=intent.coin, side="long", entry_price=_q(entry_price), exit_price=_q(exit_price), exit_reason=exit_reason, candles_held=candles_held, size=_q(size), gross_pnl_usd=_q(gross), entry_fee_usd=_q(entry_fee), exit_fee_usd=_q(exit_fee), slippage_cost_usd=_q(slippage_cost), funding_cost_usd=_q(funding_cost), net_pnl_usd=_q(net), mfe_pct=mfe, mae_pct=mae, requested_size=_q(requested_size), filled_size=_q(filled_size), unfilled_size=_q(unfilled_size), fill_ratio=fill_ratio, entry_liquidity=entry_liquidity, exit_liquidity=exit_liquidity, entry_slippage_pct=entry_slip_pct, exit_slippage_pct=exit_slip_pct, stop_gap_slippage_pct=stop_gap_pct if exit_reason == "stop_gap_slippage" else Decimal("0"), live_order_allowed=False, mainnet_signed_action=False, paper_trading=True, )