matitos
8b6cf8130c
Adds MultiStrategyManager: it runs several strategies concurrently, each with its own TradingEngine seeded with starting_balance * allocation, so one strategy's losses cannot eat into another's capital. Weights are normalised across the run ([1,3] -> 25%/75%). Risk limits stay per strategy (each engine owns a RiskManager). New API: POST /api/bot/start-many and GET /api/bot/status. The existing single-strategy endpoints delegate to the manager, so their behaviour is unchanged. Adds manager tests (allocation, isolation, duplicate keys) and API tests.
231 lines
8.0 KiB
Python
231 lines
8.0 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.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 {"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()
|
|
|
|
# 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 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:
|
|
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:
|
|
return await self.manager.start(request)
|
|
|
|
async def stop(self) -> StatusOut:
|
|
await self.manager.stop()
|
|
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:
|
|
await self.manager.kill(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:
|
|
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,
|
|
)
|