from __future__ import annotations import asyncio from collections.abc import Callable from app.broker import build_broker from app.broker.base import Broker from app.config import Settings from app.engine.engine import Publish, TradingEngine from app.engine.strategies import get_strategy from app.market import build_market_data from app.market.base import MarketData from app.schemas import ( MultiStatusOut, StartManyRequest, StrategySpec, StrategyStatusOut, ) from app.store import Store FeedFactory = Callable[[Settings], MarketData] BrokerFactory = Callable[[Settings], Broker] def normalise_weights(specs: list[StrategySpec]) -> list[float]: """Turn relative weights into allocations that sum to exactly 1.0.""" total = sum(max(0.0, spec.weight) for spec in specs) if total <= 0: return [1 / len(specs)] * len(specs) return [max(0.0, spec.weight) / total for spec in specs] class StrategyRun: """One strategy's engine plus the allocation it was given.""" def __init__(self, spec: StrategySpec, engine: TradingEngine, allocation: float) -> None: self.spec = spec self.engine = engine self.allocation = allocation @property def key(self) -> str: return self.spec.strategy_key class MultiStrategyManager: """Runs several strategies concurrently, each on its own capital slice. Every strategy gets an independent :class:`TradingEngine` seeded with ``starting_balance * allocation``, so one strategy's losses cannot eat into another's capital. Risk limits apply per strategy (each engine owns its own RiskManager); the manager aggregates status and can halt everything at once. """ def __init__( self, *, store: Store, settings: Settings, feed_factory: FeedFactory = build_market_data, broker_factory: BrokerFactory = build_broker, ) -> None: self.store = store self.settings = settings self._feed_factory = feed_factory self._broker_factory = broker_factory self.runs: list[StrategyRun] = [] self.starting_balance = settings.starting_balance self._publish_cb: Publish | None = None self._task: asyncio.Task | None = None def set_publisher(self, publish: Publish | None) -> None: self._publish_cb = publish @property def running(self) -> bool: return bool(self.runs) and any(run.engine.running for run in self.runs) @property def last_price(self) -> float | None: for run in self.runs: if run.engine.last_price is not None: return run.engine.last_price return None async def start(self, request: StartManyRequest) -> MultiStatusOut: await self.stop() starting_balance = float(request.starting_balance or self.settings.starting_balance) allocations = normalise_weights(request.strategies) self.starting_balance = starting_balance self.runs = [] for spec, allocation in zip(request.strategies, allocations, strict=True): strategy_cls = get_strategy(spec.strategy_key) # Validate params early so a bad request fails before any orders. strategy_cls(spec.params) engine = TradingEngine( feed=self._feed_factory(self.settings), broker=self._broker_factory(self.settings), store=self.store, settings=self.settings, ) engine.set_publisher(self._publish_cb) await engine.start( spec.strategy_key, spec.params, symbol=spec.symbol, timeframe=spec.timeframe, starting_balance=starting_balance * allocation, ) self.runs.append(StrategyRun(spec, engine, allocation)) return self.status() async def stop(self) -> MultiStatusOut: for run in self.runs: await run.engine.stop() self.runs = [] return self.status() async def kill(self, reason: str = "manual kill switch") -> MultiStatusOut: for run in self.runs: await run.engine.kill(reason) return self.status() async def shutdown(self) -> None: for run in self.runs: await run.engine.shutdown() self.runs = [] def status(self) -> MultiStatusOut: total_equity = 0.0 per_strategy: list[StrategyStatusOut] = [] last_tick = None for run in self.runs: engine_status = run.engine.status() total_equity += engine_status.equity if engine_status.last_tick and ( last_tick is None or engine_status.last_tick > last_tick ): last_tick = engine_status.last_tick per_strategy.append( StrategyStatusOut( strategy_key=engine_status.strategy_key or run.key, strategy_name=engine_status.strategy_name, params=engine_status.params, symbol=engine_status.symbol, timeframe=engine_status.timeframe, allocation=run.allocation, cash=engine_status.cash, equity=engine_status.equity, realized_pnl=engine_status.realized_pnl, unrealized_pnl=engine_status.unrealized_pnl, total_pnl=engine_status.total_pnl, open_positions=engine_status.open_positions, halted=engine_status.halted, halt_reason=engine_status.halt_reason, trades_today=engine_status.trades_today, day_pnl=engine_status.day_pnl, drawdown_pct=engine_status.drawdown_pct, error=engine_status.error, ) ) return MultiStatusOut( running=self.running, mode=self.settings.mode, exchange=self.settings.exchange, price=self.last_price, starting_balance=self.starting_balance, equity=total_equity, total_pnl=total_equity - self.starting_balance, strategies=per_strategy, last_tick=last_tick, ) def engine_at(self, index: int) -> TradingEngine | None: if 0 <= index < len(self.runs): return self.runs[index].engine return None __all__ = ["MultiStrategyManager", "StrategyRun", "normalise_weights"]