matitos
9dc00eba95
Trailing stops: strategies can pass trailing_stop_pct to enter_long/enter_short. advance_trailing_stop in app/engine/portfolio.py ratchets the stop with the price (up for longs, down for shorts) and never loosens; the engine persists the ratcheted level and the backtester applies the same logic. Position gains a trailing_stop_pct column (additive migration). momentum exposes it as a param. Parameter sweeps: app/backtest/sweep.py runs a cartesian grid through the same backtester and ranks by return minus max drawdown. POST /api/sweep plus a dashboard Parameter sweep panel.
125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
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
|
|
# Distance to trail the stop below the high (long) / above the low (short),
|
|
# as a fraction of price (0.02 = 2%). 0 disables trailing.
|
|
trailing_stop_pct: float = 0.0
|
|
entry_ts: int = 0
|
|
|
|
|
|
def advance_trailing_stop(
|
|
side: PositionSide,
|
|
stop_loss: float | None,
|
|
trailing_stop_pct: float,
|
|
high: float,
|
|
low: float,
|
|
) -> float | None:
|
|
"""Ratchet a trailing stop toward the price; never loosens.
|
|
|
|
For a long the stop trails ``pct`` below the running high; for a short it
|
|
trails ``pct`` above the running low. Returns the (possibly unchanged) stop.
|
|
"""
|
|
if trailing_stop_pct <= 0:
|
|
return stop_loss
|
|
if side == PositionSide.LONG:
|
|
candidate = high * (1 - trailing_stop_pct)
|
|
return candidate if stop_loss is None else max(stop_loss, candidate)
|
|
candidate = low * (1 + trailing_stop_pct)
|
|
return candidate if stop_loss is None else min(stop_loss, candidate)
|
|
|
|
|
|
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
|