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.
215 lines
7.4 KiB
Python
215 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Callable
|
|
from contextlib import suppress
|
|
from typing import Any
|
|
|
|
from app.backtest import Backtester, run_sweep
|
|
from app.broker import build_broker
|
|
from app.broker.base import Broker
|
|
from app.config import Settings
|
|
from app.engine.engine import TradingEngine
|
|
from app.engine.strategies import get_strategy
|
|
from app.market import build_market_data
|
|
from app.market.base import MarketData
|
|
from app.notify import Notifier, build_notifier
|
|
from app.schemas import (
|
|
BacktestRequest,
|
|
BacktestResult,
|
|
StatusOut,
|
|
SweepRequest,
|
|
SweepResult,
|
|
SweepRowOut,
|
|
)
|
|
from app.store import Store
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
FeedFactory = Callable[[Settings], MarketData]
|
|
BrokerFactory = Callable[[Settings], Broker]
|
|
NotifierFactory = Callable[[Settings], Notifier | None]
|
|
|
|
|
|
class Runtime:
|
|
"""Owns the active engine and fans out live updates to WebSocket clients."""
|
|
|
|
def __init__(
|
|
self,
|
|
settings: Settings,
|
|
store: Store | None = None,
|
|
*,
|
|
feed_factory: FeedFactory = build_market_data,
|
|
broker_factory: BrokerFactory = build_broker,
|
|
notifier: Notifier | None = None,
|
|
notifier_factory: NotifierFactory = build_notifier,
|
|
notify_levels: set[str] | None = None,
|
|
) -> None:
|
|
self.settings = settings
|
|
self.store = store or Store()
|
|
self._feed_factory = feed_factory
|
|
self._broker_factory = broker_factory
|
|
self.notifier = notifier if notifier is not None else notifier_factory(settings)
|
|
self.notify_levels = notify_levels or {"trade", "error"}
|
|
self.engine: TradingEngine | None = None
|
|
self._subscribers: set[asyncio.Queue] = set()
|
|
self._notify_tasks: set[asyncio.Task] = set()
|
|
|
|
# --- pub/sub ------------------------------------------------------
|
|
def subscribe(self) -> asyncio.Queue:
|
|
queue: asyncio.Queue = asyncio.Queue(maxsize=100)
|
|
self._subscribers.add(queue)
|
|
return queue
|
|
|
|
def unsubscribe(self, queue: asyncio.Queue) -> None:
|
|
self._subscribers.discard(queue)
|
|
|
|
def _broadcast(self, payload: dict[str, Any]) -> None:
|
|
for queue in list(self._subscribers):
|
|
if queue.full():
|
|
with suppress(asyncio.QueueEmpty):
|
|
queue.get_nowait()
|
|
with suppress(asyncio.QueueFull):
|
|
queue.put_nowait(payload)
|
|
if payload.get("type") == "event":
|
|
self._maybe_notify(payload["data"])
|
|
|
|
def _maybe_notify(self, event: dict[str, Any]) -> None:
|
|
if self.notifier is None or event.get("level") not in self.notify_levels:
|
|
return
|
|
message = event.get("message", "")
|
|
try:
|
|
task = asyncio.create_task(self.notifier.send(event["level"], message))
|
|
except RuntimeError: # no running loop (e.g. sync test context)
|
|
return
|
|
self._notify_tasks.add(task)
|
|
task.add_done_callback(self._notify_tasks.discard)
|
|
|
|
# --- lifecycle ----------------------------------------------------
|
|
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,
|
|
) -> StatusOut:
|
|
if self.engine is not None and self.engine.running:
|
|
await self.stop()
|
|
engine = TradingEngine(
|
|
feed=self._feed_factory(self.settings),
|
|
broker=self._broker_factory(self.settings),
|
|
store=self.store,
|
|
settings=self.settings,
|
|
)
|
|
engine.set_publisher(self._broadcast)
|
|
self.engine = engine
|
|
await engine.start(
|
|
strategy_key,
|
|
params,
|
|
symbol=symbol,
|
|
timeframe=timeframe,
|
|
starting_balance=starting_balance,
|
|
)
|
|
return engine.status()
|
|
|
|
async def stop(self) -> StatusOut:
|
|
if self.engine is not None:
|
|
await self.engine.stop()
|
|
return self.engine.status()
|
|
return self._idle_status()
|
|
|
|
async def kill(self, reason: str = "manual kill switch") -> StatusOut:
|
|
if self.engine is not None:
|
|
await self.engine.kill(reason)
|
|
return self.engine.status()
|
|
return self._idle_status()
|
|
|
|
async def run_backtest(self, request: BacktestRequest) -> BacktestResult:
|
|
symbol = request.symbol or self.settings.symbol
|
|
timeframe = request.timeframe or self.settings.timeframe
|
|
strategy_cls = get_strategy(request.strategy_key)
|
|
strategy = strategy_cls(request.params)
|
|
|
|
feed = self._feed_factory(self.settings)
|
|
try:
|
|
candles = await feed.fetch_candles(symbol, timeframe, request.limit)
|
|
finally:
|
|
await feed.close()
|
|
|
|
backtester = Backtester(
|
|
fee_bps=self.settings.fee_bps,
|
|
slippage_bps=self.settings.slippage_bps,
|
|
starting_balance=request.starting_balance or self.settings.starting_balance,
|
|
)
|
|
return backtester.run(candles, strategy, symbol, timeframe)
|
|
|
|
async def run_sweep(self, request: SweepRequest) -> SweepResult:
|
|
symbol = request.symbol or self.settings.symbol
|
|
timeframe = request.timeframe or self.settings.timeframe
|
|
strategy_cls = get_strategy(request.strategy_key)
|
|
|
|
feed = self._feed_factory(self.settings)
|
|
try:
|
|
candles = await feed.fetch_candles(symbol, timeframe, request.limit)
|
|
finally:
|
|
await feed.close()
|
|
|
|
rows = run_sweep(
|
|
request.strategy_key,
|
|
request.base_params,
|
|
request.grid,
|
|
candles,
|
|
symbol,
|
|
timeframe,
|
|
fee_bps=self.settings.fee_bps,
|
|
slippage_bps=self.settings.slippage_bps,
|
|
starting_balance=request.starting_balance or self.settings.starting_balance,
|
|
top_n=request.top_n,
|
|
)
|
|
combinations = 1
|
|
for values in request.grid.values():
|
|
combinations *= max(1, len(values))
|
|
return SweepResult(
|
|
strategy_key=request.strategy_key,
|
|
strategy_name=strategy_cls.name or request.strategy_key,
|
|
symbol=symbol,
|
|
timeframe=timeframe,
|
|
combinations=combinations,
|
|
rows=[SweepRowOut(params=row.params, metrics=row.metrics) for row in rows],
|
|
)
|
|
|
|
async def shutdown(self) -> None:
|
|
if self.engine is not None:
|
|
await self.engine.shutdown()
|
|
if self.notifier is not None:
|
|
await self.notifier.close()
|
|
|
|
def status(self) -> StatusOut:
|
|
if self.engine is not None:
|
|
return self.engine.status()
|
|
return self._idle_status()
|
|
|
|
def _idle_status(self) -> StatusOut:
|
|
return StatusOut(
|
|
running=False,
|
|
mode=self.settings.mode,
|
|
exchange=self.settings.exchange,
|
|
symbol=self.settings.symbol,
|
|
timeframe=self.settings.timeframe,
|
|
strategy_key=None,
|
|
strategy_name=None,
|
|
params={},
|
|
price=None,
|
|
cash=self.settings.starting_balance,
|
|
equity=self.settings.starting_balance,
|
|
starting_balance=self.settings.starting_balance,
|
|
realized_pnl=0.0,
|
|
unrealized_pnl=0.0,
|
|
total_pnl=0.0,
|
|
open_positions=0,
|
|
last_tick=None,
|
|
)
|