diff --git a/backend/app/api/routes.py b/backend/app/api/routes.py index 221c66e..2a1f936 100644 --- a/backend/app/api/routes.py +++ b/backend/app/api/routes.py @@ -18,8 +18,10 @@ from app.schemas import ( EventOut, KillRequest, ManualOrderRequest, + MultiStatusOut, OrderOut, PositionOut, + StartManyRequest, StartRequest, StatusOut, StrategyConfigIn, @@ -96,6 +98,21 @@ async def start_bot(payload: StartRequest, runtime: GuardedRuntimeDep) -> Status raise HTTPException(status_code=400, detail=str(exc)) from exc +@router.get("/bot/status", response_model=MultiStatusOut) +def multi_status(runtime: GuardedRuntimeDep) -> MultiStatusOut: + return runtime.multi_status() + + +@router.post("/bot/start-many", response_model=MultiStatusOut) +async def start_many(payload: StartManyRequest, runtime: GuardedRuntimeDep) -> MultiStatusOut: + try: + return await runtime.start_many(payload) + except KeyError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except (ValueError, RuntimeError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + @router.post("/bot/stop", response_model=StatusOut) async def stop_bot(runtime: GuardedRuntimeDep) -> StatusOut: return await runtime.stop() diff --git a/backend/app/engine/manager.py b/backend/app/engine/manager.py new file mode 100644 index 0000000..33ea652 --- /dev/null +++ b/backend/app/engine/manager.py @@ -0,0 +1,183 @@ +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, + 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"] diff --git a/backend/app/runtime.py b/backend/app/runtime.py index 46c9215..55a1e08 100644 --- a/backend/app/runtime.py +++ b/backend/app/runtime.py @@ -11,6 +11,7 @@ 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 @@ -18,7 +19,10 @@ from app.notify import Notifier, build_notifier from app.schemas import ( BacktestRequest, BacktestResult, + MultiStatusOut, + StartManyRequest, StatusOut, + StrategySpec, SweepRequest, SweepResult, SweepRowOut, @@ -52,10 +56,22 @@ class Runtime: 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.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) @@ -96,36 +112,37 @@ class Runtime: 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, + await self.start_many( + StartManyRequest( + strategies=[ + StrategySpec( + strategy_key=strategy_key, + params=params or {}, + symbol=symbol, + timeframe=timeframe, + ) + ], + starting_balance=starting_balance, + ) ) - 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() + 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: - if self.engine is not None: - await self.engine.stop() - return self.engine.status() - return self._idle_status() + 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: - if self.engine is not None: - await self.engine.kill(reason) - return self.engine.status() - return self._idle_status() + 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 @@ -182,8 +199,7 @@ class Runtime: ) async def shutdown(self) -> None: - if self.engine is not None: - await self.engine.shutdown() + await self.manager.shutdown() if self.notifier is not None: await self.notifier.close() diff --git a/backend/app/schemas.py b/backend/app/schemas.py index d0d5c89..93f3a13 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -134,6 +134,18 @@ class StrategyConfigOut(BaseModel): created_at: datetime +class StrategySpec(BaseModel): + """One strategy in a multi-strategy run.""" + + strategy_key: str + params: dict[str, Any] = Field(default_factory=dict) + symbol: str | None = None + timeframe: str | None = None + # Fraction of the account allocated to this strategy (relative weights are + # normalised across the run, so [1, 3] gives 25% / 75%). + weight: float = Field(default=1.0, gt=0) + + class StartRequest(BaseModel): strategy_key: str symbol: str | None = None @@ -142,6 +154,44 @@ class StartRequest(BaseModel): starting_balance: float | None = None +class StartManyRequest(BaseModel): + strategies: list[StrategySpec] = Field(min_length=1) + starting_balance: float | None = None + + +class StrategyStatusOut(BaseModel): + """Per-strategy status inside a multi-strategy run.""" + + strategy_key: str + strategy_name: str | None + params: dict[str, Any] = Field(default_factory=dict) + symbol: str + timeframe: str + allocation: float + cash: float + equity: float + realized_pnl: float + unrealized_pnl: float + total_pnl: float + open_positions: int + halted: bool = False + halt_reason: str | None = None + trades_today: int = 0 + error: str | None = None + + +class MultiStatusOut(BaseModel): + running: bool + mode: str + exchange: str + price: float | None = None + starting_balance: float + equity: float + total_pnl: float + strategies: list[StrategyStatusOut] = Field(default_factory=list) + last_tick: datetime | None = None + + class ManualOrderRequest(BaseModel): side: Literal["buy", "sell"] qty: float | None = None diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index c3cd653..95b5686 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -111,6 +111,42 @@ def test_backtest_limit_validation(runtime: Runtime) -> None: ) +def test_start_many_and_multi_status(runtime: Runtime, market: FakeMarketData) -> None: + app = create_app(runtime.settings, runtime) + with TestClient(app) as client: + started = client.post( + "/api/bot/start-many", + json={ + "strategies": [ + {"strategy_key": "sma_cross", "params": {"fast": 2, "slow": 4}, "weight": 1}, + {"strategy_key": "ema_cross", "params": {"fast": 2, "slow": 4}, "weight": 3}, + ], + "starting_balance": 10000, + }, + ) + assert started.status_code == 200 + body = started.json() + assert body["running"] is True + assert len(body["strategies"]) == 2 + assert [s["allocation"] for s in body["strategies"]] == [0.25, 0.75] + + status = client.get("/api/bot/status").json() + assert status["running"] is True + assert len(status["strategies"]) == 2 + assert status["equity"] == 10000 + + client.post("/api/bot/stop") + + +def test_start_many_rejects_unknown_strategy(runtime: Runtime) -> None: + app = create_app(runtime.settings, runtime) + with TestClient(app) as client: + response = client.post( + "/api/bot/start-many", json={"strategies": [{"strategy_key": "nope"}]} + ) + assert response.status_code == 404 + + def test_kill_endpoint(runtime: Runtime) -> None: app = create_app(runtime.settings, runtime) with TestClient(app) as client: diff --git a/backend/tests/test_multi_strategy.py b/backend/tests/test_multi_strategy.py new file mode 100644 index 0000000..9d6fea0 --- /dev/null +++ b/backend/tests/test_multi_strategy.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import pytest + +from app.engine.manager import MultiStrategyManager, normalise_weights +from app.runtime import Runtime +from app.schemas import StartManyRequest, StrategySpec + + +def test_normalise_weights() -> None: + specs = [StrategySpec(strategy_key="a", weight=1), StrategySpec(strategy_key="b", weight=3)] + assert normalise_weights(specs) == [0.25, 0.75] + + +def test_normalise_weights_single() -> None: + assert normalise_weights([StrategySpec(strategy_key="a")]) == [1.0] + + +def test_weights_must_be_positive() -> None: + from pydantic import ValidationError + + with pytest.raises(ValidationError): + StrategySpec(strategy_key="a", weight=0) + + +def test_normalise_weights_guards_zero_total() -> None: + # The API rejects non-positive weights, but the helper still guards against + # a zero total (e.g. if it is ever called with raw values). + class Dummy: + def __init__(self, weight: float) -> None: + self.weight = weight + + assert normalise_weights([Dummy(0), Dummy(0)]) == [0.5, 0.5] + + +async def test_start_many_seeds_each_engine_with_its_allocation(settings, store, market) -> None: + runtime = Runtime( + settings, store, feed_factory=lambda _s: market, broker_factory=lambda _s: _paper() + ) + status = await runtime.start_many( + StartManyRequest( + strategies=[ + StrategySpec(strategy_key="sma_cross", params={"fast": 2, "slow": 4}, weight=1), + StrategySpec(strategy_key="ema_cross", params={"fast": 2, "slow": 4}, weight=3), + ], + starting_balance=10_000, + ) + ) + assert status.running is True + assert [s.allocation for s in status.strategies] == pytest.approx([0.25, 0.75]) + # Each engine starts with its slice of the balance as cash. + assert [s.cash for s in status.strategies] == pytest.approx([2_500, 7_500]) + assert status.starting_balance == pytest.approx(10_000) + assert status.equity == pytest.approx(10_000) + await runtime.shutdown() + + +async def test_strategies_are_isolated(settings, store, market) -> None: + """A loss in one strategy must not touch another's cash.""" + runtime = Runtime( + settings, store, feed_factory=lambda _s: market, broker_factory=lambda _s: _paper() + ) + await runtime.start_many( + StartManyRequest( + strategies=[ + StrategySpec(strategy_key="sma_cross", weight=1), + StrategySpec(strategy_key="ema_cross", weight=1), + ], + starting_balance=10_000, + ) + ) + first, second = runtime.manager.runs[0].engine, runtime.manager.runs[1].engine + first.last_price = 100.0 + second.last_price = 100.0 + + await first.manual_order(_buy(), qty=10, quote_amount=None, reason="lose") + first.last_price = 90.0 + await first.close_current_position("loss") + + assert first.cash == pytest.approx(4_900.0) + assert second.cash == pytest.approx(5_000.0) + await runtime.shutdown() + + +async def test_stop_clears_runs(settings, store, market) -> None: + runtime = Runtime( + settings, store, feed_factory=lambda _s: market, broker_factory=lambda _s: _paper() + ) + await runtime.start_many(StartManyRequest(strategies=[StrategySpec(strategy_key="sma_cross")])) + await runtime.stop() + assert runtime.multi_status().running is False + assert runtime.multi_status().strategies == [] + await runtime.shutdown() + + +async def test_duplicate_strategy_keys_are_allowed(settings, store, market) -> None: + runtime = Runtime( + settings, store, feed_factory=lambda _s: market, broker_factory=lambda _s: _paper() + ) + status = await runtime.start_many( + StartManyRequest( + strategies=[ + StrategySpec(strategy_key="sma_cross", params={"fast": 2, "slow": 10}), + StrategySpec(strategy_key="sma_cross", params={"fast": 5, "slow": 20}), + ] + ) + ) + assert len(status.strategies) == 2 + assert status.strategies[0].params["fast"] == 2 + assert status.strategies[1].params["fast"] == 5 + await runtime.shutdown() + + +async def test_bad_params_rejected_before_starting(settings, store, market) -> None: + manager = MultiStrategyManager( + store=store, + settings=settings, + feed_factory=lambda _s: market, + broker_factory=lambda _s: _paper(), + ) + with pytest.raises(ValueError): + await manager.start( + StartManyRequest( + strategies=[StrategySpec(strategy_key="sma_cross", params={"nope": 1})] + ) + ) + assert manager.runs == [] + + +def _paper(): + from app.broker.paper import PaperBroker + + return PaperBroker(fee_bps=0, slippage_bps=0) + + +def _buy(): + from app.models import OrderSide + + return OrderSide.BUY