from __future__ import annotations from dataclasses import dataclass from app.models import PositionSide # All functions here are pure and shared by the live engine and the backtester # so both compute cash, P&L and risk exits identically. @dataclass class OpenPosition: """Lightweight position used by the backtester (the engine uses a DB row).""" side: PositionSide qty: float entry_price: float entry_fee: float = 0.0 stop_loss: float | None = None take_profit: float | None = None entry_ts: int = 0 def apply_open(cash: float, side: PositionSide, qty: float, price: float, fee: float) -> float: """Cash after opening a position (buying long / selling short).""" notional = qty * price if side == PositionSide.LONG: return cash - notional - fee return cash + notional - fee def apply_close(cash: float, side: PositionSide, qty: float, price: float, fee: float) -> float: """Cash after closing a position (selling long / buying back short).""" notional = qty * price if side == PositionSide.LONG: return cash + notional - fee return cash - notional - fee def realized_pnl( side: PositionSide, entry_price: float, exit_price: float, qty: float, entry_fee: float, exit_fee: float, ) -> float: gross = ( (exit_price - entry_price) * qty if side == PositionSide.LONG else (entry_price - exit_price) * qty ) return gross - entry_fee - exit_fee def unrealized_pnl(side: PositionSide, entry_price: float, mark: float, qty: float) -> float: if side == PositionSide.LONG: return (mark - entry_price) * qty return (entry_price - mark) * qty def position_value(side: PositionSide, qty: float, mark: float) -> float: """Mark-to-market value of the position as an asset (negative for shorts).""" return qty * mark if side == PositionSide.LONG else -qty * mark def account_equity(cash: float, side: PositionSide, qty: float, mark: float) -> float: return cash + position_value(side, qty, mark) def order_qty(equity: float, size: float, side: PositionSide, cash: float, price: float) -> float: """Position size as a fraction of equity, capped by available cash for longs.""" if price <= 0: return 0.0 size = max(0.0, min(size, 1.0)) notional = equity * size if side == PositionSide.LONG: notional = min(notional, cash * 0.99) return notional / price def risk_exit( side: PositionSide, stop_loss: float | None, take_profit: float | None, high: float, low: float, ) -> tuple[str, float] | None: """Return (reason, exit_price) if a stop/take level was hit within the bar.""" if side == PositionSide.LONG: if stop_loss is not None and low <= stop_loss: return "stop loss", stop_loss if take_profit is not None and high >= take_profit: return "take profit", take_profit else: if stop_loss is not None and high >= stop_loss: return "stop loss", stop_loss if take_profit is not None and low <= take_profit: return "take profit", take_profit return None