matitos
ac92dd5ac4
- GET /api/metrics renders Prometheus text (equity, P&L, per-strategy equity/pnl/allocation/halted, info) with no client library. - In-memory ring log handler (app/logs.py) exposed at GET /api/logs and a dashboard "Server logs" panel; Runtime mirrors bot events into the logger. - docs/runbook.md covers deploys, the kill switch, resetting paper state and going live. Adds tests for metrics rendering and the log ring.
316 lines
11 KiB
Python
316 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__)
|
|
|
|
_LOG_LEVELS = {
|
|
"debug": logging.DEBUG,
|
|
"info": logging.INFO,
|
|
"trade": logging.INFO,
|
|
"warning": logging.WARNING,
|
|
"error": logging.ERROR,
|
|
"critical": logging.CRITICAL,
|
|
}
|
|
|
|
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._log_event(payload["data"])
|
|
self._maybe_notify(payload["data"])
|
|
|
|
def _log_event(self, event: dict[str, Any]) -> None:
|
|
"""Mirror bot events into the logger so the dashboard log view is useful."""
|
|
level = str(event.get("level", "info")).lower()
|
|
message = event.get("message", "")
|
|
logger.log(_LOG_LEVELS.get(level, logging.INFO), "%s", message)
|
|
|
|
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,
|
|
)
|