"""Bundles the simulator's collaborators so `run_once` can trigger it without
importing `simulation/`, `persistence/simulation.py`, or `SimulationSettings`
directly (issue #33) — the CLI builds one `SimulationRunner` and passes it in,
keeping the pipeline layer's import surface unchanged for a feature it merely
triggers.
"""
import logging
from collections.abc import Collection, Sequence
from dataclasses import dataclass
from datetime import datetime
from trader.config.schema import DiscoverySettings, PipelineConfig, SimulationSettings
from trader.persistence.bars import BarRepository
from trader.persistence.decisions import DecisionRepository
from trader.persistence.simulation import ABB_PORTFOLIO_SUFFIX, SimulationRepository
from trader.simulation.step import advance
__all__ = ["SimulationRunner"]
_logger = logging.getLogger("trader.simulation.runner")
[docs]
@dataclass(slots=True)
class SimulationRunner:
"""Steps every configured strategy's shadow portfolio forward one cycle.
Since issue #42, that is *two* portfolios per strategy id: the normal one
(its own sell decisions honoured) and an Always-be-Buying (AbB) variant
that reads the same decisions with sell signals ignored, so only the
trailing stop can close a position — isolating whether the sell side is
what costs performance, independent of whether the entries are any good.
"""
settings: SimulationSettings
pipeline_config: PipelineConfig
discovery_settings: DiscoverySettings
fixed_symbols: Collection[str]
bar_repository: BarRepository
decision_repository: DecisionRepository
simulation_repository: SimulationRepository
[docs]
def run(self, *, strategy_ids: Sequence[str], now: datetime) -> None:
"""Advance both of one strategy's portfolios per id, all four*
failure points isolated from each other.
(*) two per strategy, times as many strategies as `strategy_ids`
holds. One variant of one strategy raising (a malformed
`inputs_json`, a repository error) must not stop its own sibling
variant, another strategy's either variant, or the real cycle this
was called from — the same per-unit isolation discipline
`discover_symbols` uses per-theme.
"""
for strategy_id in strategy_ids:
self._advance_one(
strategy_id=strategy_id,
now=now,
portfolio_key=None,
disable_sell_signals=False,
)
self._advance_one(
strategy_id=strategy_id,
now=now,
portfolio_key=f"{strategy_id}{ABB_PORTFOLIO_SUFFIX}",
disable_sell_signals=True,
)
def _advance_one(
self,
*,
strategy_id: str,
now: datetime,
portfolio_key: str | None,
disable_sell_signals: bool,
) -> None:
variant = "AbB" if disable_sell_signals else "normal"
try:
advance(
strategy_id=strategy_id,
now=now,
bar_repository=self.bar_repository,
decision_repository=self.decision_repository,
simulation_repository=self.simulation_repository,
settings=self.settings,
pipeline_config=self.pipeline_config,
discovery_settings=self.discovery_settings,
fixed_symbols=self.fixed_symbols,
portfolio_key=portfolio_key,
disable_sell_signals=disable_sell_signals,
)
except Exception as exc: # noqa: BLE001 - isolate one variant's failure
_logger.error(
"simulation step failed for %s (%s): %s", strategy_id, variant, exc
)