Multi-strategy engine with per-strategy capital allocation

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.
This commit is contained in:
matitos committed 2026-09-21 20:10:28 +00:00
1 parent e07536542f
commit 8b6cf8130c
6 files changed
+466 -25

No files matched your search

+17
View File
@@ -18,8 +18,10 @@ from app.schemas import (
EventOut, EventOut,
KillRequest, KillRequest,
ManualOrderRequest, ManualOrderRequest,
MultiStatusOut,
OrderOut, OrderOut,
PositionOut, PositionOut,
StartManyRequest,
StartRequest, StartRequest,
StatusOut, StatusOut,
StrategyConfigIn, StrategyConfigIn,
@@ -96,6 +98,21 @@ async def start_bot(payload: StartRequest, runtime: GuardedRuntimeDep) -> Status
raise HTTPException(status_code=400, detail=str(exc)) from exc 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) @router.post("/bot/stop", response_model=StatusOut)
async def stop_bot(runtime: GuardedRuntimeDep) -> StatusOut: async def stop_bot(runtime: GuardedRuntimeDep) -> StatusOut:
return await runtime.stop() return await runtime.stop()
+183
View File
@@ -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"]
+41 -25
View File
@@ -11,6 +11,7 @@ from app.broker import build_broker
from app.broker.base import Broker from app.broker.base import Broker
from app.config import Settings from app.config import Settings
from app.engine.engine import TradingEngine from app.engine.engine import TradingEngine
from app.engine.manager import MultiStrategyManager
from app.engine.strategies import get_strategy from app.engine.strategies import get_strategy
from app.market import build_market_data from app.market import build_market_data
from app.market.base import MarketData from app.market.base import MarketData
@@ -18,7 +19,10 @@ from app.notify import Notifier, build_notifier
from app.schemas import ( from app.schemas import (
BacktestRequest, BacktestRequest,
BacktestResult, BacktestResult,
MultiStatusOut,
StartManyRequest,
StatusOut, StatusOut,
StrategySpec,
SweepRequest, SweepRequest,
SweepResult, SweepResult,
SweepRowOut, SweepRowOut,
@@ -52,10 +56,22 @@ class Runtime:
self._broker_factory = broker_factory self._broker_factory = broker_factory
self.notifier = notifier if notifier is not None else notifier_factory(settings) self.notifier = notifier if notifier is not None else notifier_factory(settings)
self.notify_levels = notify_levels or {"trade", "error"} 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._subscribers: set[asyncio.Queue] = set()
self._notify_tasks: set[asyncio.Task] = 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 ------------------------------------------------------ # --- pub/sub ------------------------------------------------------
def subscribe(self) -> asyncio.Queue: def subscribe(self) -> asyncio.Queue:
queue: asyncio.Queue = asyncio.Queue(maxsize=100) queue: asyncio.Queue = asyncio.Queue(maxsize=100)
@@ -96,36 +112,37 @@ class Runtime:
timeframe: str | None = None, timeframe: str | None = None,
starting_balance: float | None = None, starting_balance: float | None = None,
) -> StatusOut: ) -> StatusOut:
if self.engine is not None and self.engine.running: await self.start_many(
await self.stop() StartManyRequest(
engine = TradingEngine( strategies=[
feed=self._feed_factory(self.settings), StrategySpec(
broker=self._broker_factory(self.settings), strategy_key=strategy_key,
store=self.store, params=params or {},
settings=self.settings,
)
engine.set_publisher(self._broadcast)
self.engine = engine
await engine.start(
strategy_key,
params,
symbol=symbol, symbol=symbol,
timeframe=timeframe, timeframe=timeframe,
)
],
starting_balance=starting_balance, 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: async def stop(self) -> StatusOut:
if self.engine is not None: await self.manager.stop()
await self.engine.stop() engine = self.engine
return self.engine.status() return engine.status() if engine is not None else self._idle_status()
return self._idle_status()
async def kill(self, reason: str = "manual kill switch") -> StatusOut: async def kill(self, reason: str = "manual kill switch") -> StatusOut:
if self.engine is not None: await self.manager.kill(reason)
await self.engine.kill(reason) engine = self.engine
return self.engine.status() return engine.status() if engine is not None else self._idle_status()
return self._idle_status()
def multi_status(self) -> MultiStatusOut:
return self.manager.status()
async def run_backtest(self, request: BacktestRequest) -> BacktestResult: async def run_backtest(self, request: BacktestRequest) -> BacktestResult:
symbol = request.symbol or self.settings.symbol symbol = request.symbol or self.settings.symbol
@@ -182,8 +199,7 @@ class Runtime:
) )
async def shutdown(self) -> None: async def shutdown(self) -> None:
if self.engine is not None: await self.manager.shutdown()
await self.engine.shutdown()
if self.notifier is not None: if self.notifier is not None:
await self.notifier.close() await self.notifier.close()
+50
View File
@@ -134,6 +134,18 @@ class StrategyConfigOut(BaseModel):
created_at: datetime 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): class StartRequest(BaseModel):
strategy_key: str strategy_key: str
symbol: str | None = None symbol: str | None = None
@@ -142,6 +154,44 @@ class StartRequest(BaseModel):
starting_balance: float | None = None 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): class ManualOrderRequest(BaseModel):
side: Literal["buy", "sell"] side: Literal["buy", "sell"]
qty: float | None = None qty: float | None = None
+36
View File
@@ -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: def test_kill_endpoint(runtime: Runtime) -> None:
app = create_app(runtime.settings, runtime) app = create_app(runtime.settings, runtime)
with TestClient(app) as client: with TestClient(app) as client:
+139
View File
@@ -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