matitos
2ca5032e7e
- NOTIFY_LEVELS chooses which event levels are sent (default trade,error). - NOTIFY_LIFECYCLE sends explicit start/stop messages; the kill switch always notifies at error level. - SUMMARY_INTERVAL_MINUTES sends a periodic portfolio summary (equity, per strategy P&L, allocation, halted state) via a background task started with the run and stopped on stop/kill/shutdown. - Wires the new vars plus the go-live vars through both compose files and the deploy workflow. Adds tests.
300 lines
11 KiB
Python
300 lines
11 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.live_guard import check_live_ready
|
|
from app.engine.manager import MultiStrategyManager
|
|
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,
|
|
MultiStatusOut,
|
|
StartManyRequest,
|
|
StatusOut,
|
|
StrategySpec,
|
|
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 settings.notify_level_set or {"trade", "error"}
|
|
self.manager = MultiStrategyManager(
|
|
store=self.store,
|
|
settings=settings,
|
|
feed_factory=feed_factory,
|
|
broker_factory=broker_factory,
|
|
)
|
|
self.manager.set_publisher(self._broadcast)
|
|
self._subscribers: set[asyncio.Queue] = set()
|
|
self._notify_tasks: set[asyncio.Task] = set()
|
|
self._summary_task: asyncio.Task | None = None
|
|
|
|
# The first (usually only) engine, kept for the single-strategy API and
|
|
# backwards compatibility with existing callers and tests.
|
|
@property
|
|
def engine(self) -> TradingEngine | None:
|
|
return self.manager.runs[0].engine if self.manager.runs else None
|
|
|
|
# --- 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 event.get("level") not in self.notify_levels:
|
|
return
|
|
self.notify_now(event.get("level", "info"), event.get("message", ""))
|
|
|
|
def notify_now(self, level: str, message: str) -> None:
|
|
"""Send a notification, ignoring level filtering (best-effort)."""
|
|
if self.notifier is None:
|
|
return
|
|
try:
|
|
task = asyncio.create_task(self.notifier.send(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)
|
|
|
|
def _start_summary_loop(self) -> None:
|
|
minutes = self.settings.summary_interval_minutes
|
|
if minutes <= 0 or self.notifier is None:
|
|
return
|
|
self._stop_summary_loop()
|
|
self._summary_task = asyncio.create_task(self._summary_loop(minutes * 60))
|
|
|
|
def _stop_summary_loop(self) -> None:
|
|
if self._summary_task is not None:
|
|
self._summary_task.cancel()
|
|
self._summary_task = None
|
|
|
|
async def _summary_loop(self, interval_seconds: float) -> None:
|
|
try:
|
|
while True:
|
|
await asyncio.sleep(interval_seconds)
|
|
self.notify_now("summary", self.summary_message())
|
|
except asyncio.CancelledError:
|
|
raise
|
|
|
|
def summary_message(self) -> str:
|
|
status = self.manager.status()
|
|
lines = [
|
|
f"Portfolio: equity {status.equity:,.2f} "
|
|
f"({status.total_pnl:+,.2f}) across {len(status.strategies)} strategies"
|
|
]
|
|
for s in status.strategies:
|
|
state = "HALTED" if s.halted else ("running" if status.running else "stopped")
|
|
lines.append(
|
|
f"- {s.strategy_name or s.strategy_key} "
|
|
f"({s.allocation * 100:.0f}%): {s.equity:,.2f} ({s.total_pnl:+,.2f}) "
|
|
f"[{state}]"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
# --- 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:
|
|
await self.start_many(
|
|
StartManyRequest(
|
|
strategies=[
|
|
StrategySpec(
|
|
strategy_key=strategy_key,
|
|
params=params or {},
|
|
symbol=symbol,
|
|
timeframe=timeframe,
|
|
)
|
|
],
|
|
starting_balance=starting_balance,
|
|
)
|
|
)
|
|
engine = self.engine
|
|
return engine.status() if engine is not None else self._idle_status()
|
|
|
|
async def start_many(self, request: StartManyRequest) -> MultiStatusOut:
|
|
check = check_live_ready(
|
|
self.settings,
|
|
request.starting_balance or self.settings.starting_balance,
|
|
)
|
|
if not check.ok:
|
|
raise ValueError("live trading blocked: " + "; ".join(check.blockers))
|
|
status = await self.manager.start(request)
|
|
if self.settings.notify_lifecycle:
|
|
names = ", ".join(s.strategy_name or s.strategy_key for s in status.strategies)
|
|
self.notify_now(
|
|
"info",
|
|
f"Started {len(status.strategies)} strategy(ies) in {self.settings.mode} "
|
|
f"mode: {names}",
|
|
)
|
|
self._start_summary_loop()
|
|
return status
|
|
|
|
def live_check(self) -> dict:
|
|
return check_live_ready(self.settings).as_dict()
|
|
|
|
async def stop(self) -> StatusOut:
|
|
had_runs = bool(self.manager.runs)
|
|
await self.manager.stop()
|
|
self._stop_summary_loop()
|
|
if had_runs and self.settings.notify_lifecycle:
|
|
self.notify_now("info", "Stopped all strategies")
|
|
engine = self.engine
|
|
return engine.status() if engine is not None else self._idle_status()
|
|
|
|
async def kill(self, reason: str = "manual kill switch") -> StatusOut:
|
|
had_runs = bool(self.manager.runs)
|
|
await self.manager.kill(reason)
|
|
self._stop_summary_loop()
|
|
if had_runs:
|
|
self.notify_now("error", f"KILL SWITCH: {reason}")
|
|
engine = self.engine
|
|
return engine.status() if engine is not None else self._idle_status()
|
|
|
|
def multi_status(self) -> MultiStatusOut:
|
|
return self.manager.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:
|
|
self._stop_summary_loop()
|
|
await self.manager.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,
|
|
)
|