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.
459 lines
16 KiB
Python
459 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
from collections.abc import Callable
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from app.broker.base import Broker
|
|
from app.config import Settings
|
|
from app.engine.portfolio import (
|
|
account_equity,
|
|
advance_trailing_stop,
|
|
apply_close,
|
|
apply_open,
|
|
order_qty,
|
|
realized_pnl,
|
|
risk_exit,
|
|
unrealized_pnl,
|
|
)
|
|
from app.engine.risk import RiskManager
|
|
from app.engine.strategies import Strategy, get_strategy
|
|
from app.engine.strategies.base import StrategyContext
|
|
from app.market.base import MarketData
|
|
from app.models import OrderSide, Position, PositionSide
|
|
from app.schemas import Candle, EventOut, StatusOut
|
|
from app.store import Store
|
|
|
|
Publish = Callable[[dict[str, Any]], None]
|
|
|
|
|
|
class TradingEngine:
|
|
"""Single-strategy trading loop.
|
|
|
|
The engine owns portfolio state (cash, open position), calls the strategy
|
|
once per candle, turns emitted intents into orders, and enforces stop-loss /
|
|
take-profit levels.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
feed: MarketData,
|
|
broker: Broker,
|
|
store: Store,
|
|
settings: Settings,
|
|
) -> None:
|
|
self.feed = feed
|
|
self.broker = broker
|
|
self.store = store
|
|
self.settings = settings
|
|
|
|
self.running = False
|
|
self.error: str | None = None
|
|
self.last_tick: datetime | None = None
|
|
self.last_price: float | None = None
|
|
self.cash: float = settings.starting_balance
|
|
self.starting_balance: float = settings.starting_balance
|
|
self.realized_pnl: float = 0.0
|
|
self.position: Position | None = None
|
|
self.strategy: Strategy | None = None
|
|
self.strategy_key: str | None = None
|
|
self.symbol: str = settings.symbol
|
|
self.timeframe: str = settings.timeframe
|
|
|
|
self.risk = RiskManager.from_settings(settings, settings.starting_balance)
|
|
self._entry_blocked_logged = False
|
|
self._task: asyncio.Task | None = None
|
|
self._publish_cb: Publish | None = None
|
|
|
|
# --- lifecycle ----------------------------------------------------
|
|
def set_publisher(self, publish: Publish | None) -> None:
|
|
self._publish_cb = publish
|
|
|
|
async def start(
|
|
self,
|
|
strategy_key: str,
|
|
params: dict[str, Any] | None = None,
|
|
*,
|
|
symbol: str | None = None,
|
|
timeframe: str | None = None,
|
|
starting_balance: float | None = None,
|
|
) -> None:
|
|
if self.running:
|
|
raise RuntimeError("engine already running")
|
|
|
|
strategy_cls = get_strategy(strategy_key)
|
|
self.strategy = strategy_cls(params)
|
|
self.strategy_key = strategy_key
|
|
self.symbol = symbol or self.settings.symbol
|
|
self.timeframe = timeframe or self.settings.timeframe
|
|
self.starting_balance = float(starting_balance or self.settings.starting_balance)
|
|
|
|
self.store.abandon_open_positions(strategy_key, reason="reset on engine start")
|
|
self.position = None
|
|
self.cash = self.starting_balance
|
|
self.realized_pnl = 0.0
|
|
self.last_price = None
|
|
self.error = None
|
|
self.last_tick = None
|
|
self.risk.reset(self.starting_balance)
|
|
self._entry_blocked_logged = False
|
|
|
|
self.running = True
|
|
self._emit("info", f"Started {strategy_cls.name or strategy_key} on {self.symbol}")
|
|
self._broadcast_status()
|
|
self._task = asyncio.create_task(self._run())
|
|
|
|
async def stop(self) -> None:
|
|
if not self.running:
|
|
return
|
|
self.running = False
|
|
if self._task:
|
|
self._task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._task
|
|
self._task = None
|
|
self._emit("info", "Engine stopped")
|
|
self._broadcast_status()
|
|
|
|
async def shutdown(self) -> None:
|
|
await self.stop()
|
|
await self.feed.close()
|
|
await self.broker.close()
|
|
|
|
# --- loop ---------------------------------------------------------
|
|
async def _run(self) -> None:
|
|
while self.running:
|
|
try:
|
|
await self.tick()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc: # keep the loop alive on transient errors
|
|
self.error = f"{type(exc).__name__}: {exc}"
|
|
self._emit("error", f"tick failed: {self.error}")
|
|
if not self.running:
|
|
break
|
|
await asyncio.sleep(self.settings.poll_interval)
|
|
|
|
async def tick(self) -> None:
|
|
candles = await self.feed.fetch_candles(
|
|
self.symbol, self.timeframe, self.settings.candle_limit
|
|
)
|
|
if not candles:
|
|
return
|
|
self.last_price = candles[-1].close
|
|
self.last_tick = datetime.now(UTC)
|
|
self.error = None
|
|
|
|
halt_reason = self.risk.update(self.equity)
|
|
if halt_reason is not None:
|
|
await self._halt(halt_reason)
|
|
return
|
|
|
|
if self.position is not None:
|
|
await self._check_risk(candles[-1])
|
|
|
|
if self.strategy is not None:
|
|
ctx = StrategyContext(
|
|
candles=candles,
|
|
position=self.position,
|
|
equity=self.equity,
|
|
cash=self.cash,
|
|
)
|
|
try:
|
|
self.strategy.on_candle(ctx)
|
|
except Exception as exc:
|
|
self.error = f"strategy error: {type(exc).__name__}: {exc}"
|
|
self._emit("error", self.error)
|
|
await self._process_intents(ctx)
|
|
|
|
self.store.record_equity(self.equity, self.cash, self.last_price)
|
|
self._broadcast_status()
|
|
|
|
# --- portfolio helpers -------------------------------------------
|
|
@property
|
|
def equity(self) -> float:
|
|
if self.position is None:
|
|
return self.cash
|
|
return account_equity(
|
|
self.cash, self.position.side, self.position.qty, self.last_price or 0.0
|
|
)
|
|
|
|
@property
|
|
def unrealized_pnl(self) -> float:
|
|
if self.position is None or self.last_price is None:
|
|
return 0.0
|
|
return unrealized_pnl(
|
|
self.position.side, self.position.entry_price, self.last_price, self.position.qty
|
|
)
|
|
|
|
async def _process_intents(self, ctx: StrategyContext) -> None:
|
|
for intent in ctx.take_intents():
|
|
if intent.action == "close":
|
|
if self.position is not None:
|
|
await self._close_position(intent.reason or "strategy close")
|
|
elif intent.action in {"enter_long", "enter_short"}:
|
|
if self.position is not None:
|
|
continue
|
|
side = PositionSide.LONG if intent.action == "enter_long" else PositionSide.SHORT
|
|
await self._open_position(
|
|
side,
|
|
size=intent.size if intent.size is not None else 1.0,
|
|
stop_loss=intent.stop_loss,
|
|
take_profit=intent.take_profit,
|
|
trailing_stop_pct=intent.trailing_stop_pct,
|
|
reason=intent.reason or intent.action,
|
|
)
|
|
|
|
async def _open_position(
|
|
self,
|
|
side: PositionSide,
|
|
*,
|
|
size: float,
|
|
stop_loss: float | None,
|
|
take_profit: float | None,
|
|
reason: str,
|
|
trailing_stop_pct: float = 0.0,
|
|
qty_override: float | None = None,
|
|
) -> None:
|
|
price = self.last_price
|
|
if not price or price <= 0:
|
|
return
|
|
|
|
if not self.risk.can_enter():
|
|
if not self._entry_blocked_logged:
|
|
self._entry_blocked_logged = True
|
|
reason = self.risk.entry_block_reason() or "entry blocked"
|
|
self._emit("warning", f"Entry blocked: {reason}")
|
|
return
|
|
|
|
if qty_override is not None:
|
|
qty = qty_override
|
|
else:
|
|
qty = order_qty(self.equity, self.risk.clamp_size(size), side, self.cash, price)
|
|
if qty <= 0:
|
|
return
|
|
|
|
fill = (
|
|
await self.broker.buy(self.symbol, qty, price)
|
|
if side == PositionSide.LONG
|
|
else await self.broker.sell(self.symbol, qty, price)
|
|
)
|
|
|
|
self.cash = apply_open(self.cash, side, fill.qty, fill.price, fill.fee)
|
|
|
|
self.position = self.store.create_position(
|
|
strategy_key=self.strategy_key or "manual",
|
|
symbol=self.symbol,
|
|
side=side,
|
|
qty=fill.qty,
|
|
entry_price=fill.price,
|
|
entry_fee=fill.fee,
|
|
stop_loss=stop_loss,
|
|
take_profit=take_profit,
|
|
trailing_stop_pct=trailing_stop_pct,
|
|
)
|
|
self.store.record_order(
|
|
position_id=self.position.id,
|
|
strategy_key=self.position.strategy_key,
|
|
symbol=self.symbol,
|
|
side=fill.side,
|
|
qty=fill.qty,
|
|
price=fill.price,
|
|
fee=fill.fee,
|
|
reason=reason,
|
|
)
|
|
self.risk.record_entry()
|
|
self._emit(
|
|
"trade",
|
|
f"Opened {side.value} {fill.qty:.6f} {self.symbol} @ {fill.price:.4f} ({reason})",
|
|
)
|
|
|
|
async def _close_position(self, reason: str) -> None:
|
|
position = self.position
|
|
price = self.last_price
|
|
if position is None or not price:
|
|
return
|
|
|
|
fill = (
|
|
await self.broker.sell(self.symbol, position.qty, price)
|
|
if position.side == PositionSide.LONG
|
|
else await self.broker.buy(self.symbol, position.qty, price)
|
|
)
|
|
|
|
self.cash = apply_close(self.cash, position.side, fill.qty, fill.price, fill.fee)
|
|
realized = realized_pnl(
|
|
position.side,
|
|
position.entry_price,
|
|
fill.price,
|
|
position.qty,
|
|
position.entry_fee,
|
|
fill.fee,
|
|
)
|
|
self.realized_pnl += realized
|
|
|
|
self.store.close_position(
|
|
position,
|
|
exit_price=fill.price,
|
|
exit_fee=fill.fee,
|
|
realized_pnl=realized,
|
|
)
|
|
self.store.record_order(
|
|
position_id=position.id,
|
|
strategy_key=position.strategy_key,
|
|
symbol=self.symbol,
|
|
side=fill.side,
|
|
qty=fill.qty,
|
|
price=fill.price,
|
|
fee=fill.fee,
|
|
reason=reason,
|
|
)
|
|
self.position = None
|
|
self._emit(
|
|
"trade",
|
|
f"Closed {position.side.value} {position.qty:.6f} {self.symbol} "
|
|
f"@ {fill.price:.4f} pnl={realized:.2f} ({reason})",
|
|
)
|
|
|
|
async def _check_risk(self, candle: Candle) -> None:
|
|
position = self.position
|
|
if position is None:
|
|
return
|
|
|
|
# Ratchet a trailing stop before testing levels, so a bar that both
|
|
# advances the trail and reverses is handled consistently.
|
|
if position.trailing_stop_pct and position.trailing_stop_pct > 0:
|
|
advanced = advance_trailing_stop(
|
|
position.side,
|
|
position.stop_loss,
|
|
position.trailing_stop_pct,
|
|
candle.high,
|
|
candle.low,
|
|
)
|
|
if advanced is not None and advanced != position.stop_loss:
|
|
position.stop_loss = advanced
|
|
self.store.update_stop_loss(position.id, advanced)
|
|
|
|
hit = risk_exit(
|
|
position.side,
|
|
position.stop_loss,
|
|
position.take_profit,
|
|
candle.high,
|
|
candle.low,
|
|
)
|
|
if hit is not None:
|
|
reason, level = hit
|
|
self.last_price = level
|
|
await self._close_position(reason)
|
|
|
|
# --- halt / kill switch -------------------------------------------
|
|
async def _halt(self, reason: str) -> None:
|
|
"""Flatten and stop trading. Sticky until the next start()."""
|
|
if self.risk.halted:
|
|
return
|
|
self.risk.halt(reason)
|
|
self._emit("error", f"HALTED: {reason}")
|
|
if self.position is not None:
|
|
if self.last_price is None:
|
|
self.last_price = await self.feed.fetch_price(self.symbol)
|
|
await self._close_position(f"halt: {reason}")
|
|
self.running = False
|
|
self._broadcast_status()
|
|
|
|
async def kill(self, reason: str = "manual kill switch") -> None:
|
|
"""Manual kill switch: halt, flatten, and cancel the loop task."""
|
|
await self._halt(reason)
|
|
if self._task is not None and not self._task.done():
|
|
self._task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._task
|
|
self._task = None
|
|
self._broadcast_status()
|
|
|
|
# --- manual controls ----------------------------------------------
|
|
async def manual_order(
|
|
self, side: OrderSide, qty: float | None, quote_amount: float | None, reason: str
|
|
) -> None:
|
|
price = self.last_price or await self.feed.fetch_price(self.symbol)
|
|
self.last_price = price
|
|
if qty is None:
|
|
notional = quote_amount if quote_amount is not None else self.cash * 0.99
|
|
qty = notional / price
|
|
if qty <= 0:
|
|
return
|
|
if side == OrderSide.BUY:
|
|
if self.position is not None:
|
|
return
|
|
await self._open_position(
|
|
PositionSide.LONG,
|
|
size=0.0,
|
|
stop_loss=None,
|
|
take_profit=None,
|
|
reason=reason,
|
|
qty_override=qty,
|
|
)
|
|
else:
|
|
if self.position is not None:
|
|
await self._close_position(reason or "manual close")
|
|
else:
|
|
await self._open_position(
|
|
PositionSide.SHORT,
|
|
size=0.0,
|
|
stop_loss=None,
|
|
take_profit=None,
|
|
reason=reason,
|
|
qty_override=qty,
|
|
)
|
|
|
|
async def close_current_position(self, reason: str = "manual") -> None:
|
|
if self.position is not None:
|
|
if self.last_price is None:
|
|
self.last_price = await self.feed.fetch_price(self.symbol)
|
|
await self._close_position(reason)
|
|
|
|
# --- status / events ----------------------------------------------
|
|
def status(self) -> StatusOut:
|
|
strategy_name = None
|
|
if self.strategy is not None:
|
|
strategy_name = self.strategy.name or self.strategy.key
|
|
return StatusOut(
|
|
running=self.running,
|
|
mode=self.settings.mode,
|
|
exchange=self.settings.exchange,
|
|
symbol=self.symbol,
|
|
timeframe=self.timeframe,
|
|
strategy_key=self.strategy_key,
|
|
strategy_name=strategy_name,
|
|
params=dict(self.strategy.params) if self.strategy else {},
|
|
price=self.last_price,
|
|
cash=self.cash,
|
|
equity=self.equity,
|
|
starting_balance=self.starting_balance,
|
|
realized_pnl=self.realized_pnl,
|
|
unrealized_pnl=self.unrealized_pnl,
|
|
total_pnl=self.equity - self.starting_balance,
|
|
open_positions=1 if self.position else 0,
|
|
last_tick=self.last_tick,
|
|
error=self.error,
|
|
halted=self.risk.halted,
|
|
halt_reason=self.risk.halt_reason,
|
|
trades_today=self.risk.trades_today,
|
|
day_pnl=self.risk.day_pnl,
|
|
drawdown_pct=self.risk.drawdown_pct,
|
|
)
|
|
|
|
def _emit(self, level: str, message: str) -> None:
|
|
event = self.store.record_event(level, message, self.strategy_key)
|
|
self._publish(
|
|
{"type": "event", "data": EventOut.model_validate(event).model_dump(mode="json")}
|
|
)
|
|
|
|
def _broadcast_status(self) -> None:
|
|
self._publish({"type": "status", "data": self.status().model_dump(mode="json")})
|
|
|
|
def _publish(self, payload: dict[str, Any]) -> None:
|
|
if self._publish_cb is not None:
|
|
self._publish_cb(payload)
|