Coverage for src/finbot/backtest/artifacts.py: 98%
92 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-04 12:05 +0000
1import json
2from collections.abc import Iterable
3from datetime import datetime
4from pathlib import Path
5from typing import Any, TypeVar, cast
6from uuid import UUID, uuid4
8from finbot.backtest.experiment import (
9 ExperimentConfig,
10 ExperimentMetadata,
11 ExperimentPeriod,
12 ExperimentRun,
13 WalkForwardExperimentRun,
14 WalkForwardFoldResult,
15 WalkForwardResult,
16 WalkForwardSplit,
17)
18from finbot.backtest.models import BacktestResult, EquityPoint, SignalRejection
19from finbot.execution.models import Fill
21Run = ExperimentRun | WalkForwardExperimentRun
22RunT = TypeVar("RunT", ExperimentRun, WalkForwardExperimentRun)
25class ExperimentStore:
26 def __init__(self, root: Path):
27 self.root = root
29 @staticmethod
30 def _create_metadata(config: ExperimentConfig) -> ExperimentMetadata:
31 return ExperimentMetadata(
32 run_id=uuid4(),
33 created_at=datetime.now().astimezone(),
34 config=config,
35 )
37 def create_run(self, config: ExperimentConfig, result: BacktestResult) -> ExperimentRun:
38 metadata = self._create_metadata(config)
39 run = ExperimentRun(metadata=metadata, result=result)
41 self.save(run)
42 return run
44 def create_walk_forward_run(
45 self,
46 config: ExperimentConfig,
47 result: WalkForwardResult,
48 ) -> WalkForwardExperimentRun:
49 metadata = self._create_metadata(config)
50 run = WalkForwardExperimentRun(metadata=metadata, result=result)
51 self.save(run)
52 return run
54 def save(self, run: Run) -> None:
55 run_dir = self.root / str(run.metadata.run_id)
56 run_dir.mkdir(parents=True, exist_ok=False)
58 if isinstance(run, WalkForwardExperimentRun):
59 payload = self._serialize_walk_forward(run)
60 else:
61 payload = self._serialize_backtest(run)
63 (run_dir / "metadata.json").write_text(
64 json.dumps(payload, indent=2, sort_keys=True),
65 encoding="utf-8",
66 )
68 def load(self, run_id: UUID | str) -> Run:
69 path = self.root / str(run_id) / "metadata.json"
71 if not path.exists():
72 raise FileNotFoundError(f"Experiment run not found: {run_id}")
74 payload = json.loads(path.read_text(encoding="utf-8"))
75 return self._deserialize(payload)
77 def load_many(
78 self,
79 run_ids: Iterable[UUID | str],
80 *,
81 expected_type: type[RunT],
82 ) -> list[RunT]:
83 runs: list[RunT] = []
85 for run_id in run_ids:
86 run = self.load(run_id)
88 if not isinstance(run, expected_type): 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true
89 raise TypeError(
90 f"Expected {expected_type.__name__} for run {run_id}, "
91 f"got {type(run).__name__}.",
92 )
94 runs.append(cast(RunT, run))
96 return runs
98 @staticmethod
99 def _serialize_config(config: ExperimentConfig) -> dict[str, Any]:
100 return {
101 "strategy": config.strategy,
102 "strategy_params": config.strategy_params,
103 "symbols": list(config.symbols),
104 "start": config.start.isoformat(),
105 "end": config.end.isoformat(),
106 "dataset": config.dataset,
107 "initial_cash": config.initial_cash,
108 "allocation": config.allocation,
109 "fee_rate": config.fee_rate,
110 "slippage_rate": config.slippage_rate,
111 }
113 @staticmethod
114 def _deserialize_config(payload: dict[str, Any]) -> ExperimentConfig:
115 return ExperimentConfig(
116 strategy=payload["strategy"],
117 strategy_params=dict(payload["strategy_params"]), # type: ignore[arg-type]
118 symbols=tuple(payload["symbols"]),
119 start=datetime.fromisoformat(payload["start"]),
120 end=datetime.fromisoformat(payload["end"]),
121 dataset=payload["dataset"],
122 initial_cash=float(payload["initial_cash"]),
123 allocation=float(payload["allocation"]),
124 fee_rate=float(payload["fee_rate"]),
125 slippage_rate=float(payload["slippage_rate"]),
126 )
128 @staticmethod
129 def _serialize_backtest_result(result: BacktestResult) -> dict[str, Any]:
130 return {
131 "starting_cash": result.starting_cash,
132 "ending_value": result.ending_value,
133 "equity_curve": [point.serialize() for point in result.equity_curve],
134 "fills": [fill.serialize() for fill in result.fills],
135 "rejections": [rejection.serialize() for rejection in result.rejections],
136 "periods_per_year": result.periods_per_year,
137 "benchmark_return": result.benchmark_return,
138 "pnl": result.pnl,
139 "total_return": result.total_return,
140 "max_drawdown": result.max_drawdown,
141 "trades": [trade.serialize() for trade in result.trades],
142 "trade_count": result.trade_count,
143 "winning_trades": result.winning_trades,
144 "losing_trades": result.losing_trades,
145 "win_rate": result.win_rate,
146 "profit_factor": result.profit_factor,
147 "returns": result.returns,
148 "average_trade": result.average_trade,
149 "median_trade": result.median_trade,
150 "best_trade": result.best_trade,
151 "worst_trade": result.worst_trade,
152 "average_trade_duration": str(result.average_trade_duration),
153 "cagr": result.cagr,
154 "calmar_ratio": result.calmar_ratio,
155 "total_fees": result.total_fees,
156 "total_slippage": result.total_slippage,
157 "turnover": result.turnover,
158 "exposure": result.exposure,
159 "excess_return": result.excess_return,
160 }
162 @staticmethod
163 def _deserialize_backtest_result(payload: dict[str, Any]) -> BacktestResult:
164 return BacktestResult(
165 starting_cash=payload["starting_cash"],
166 ending_value=payload["ending_value"],
167 equity_curve=[EquityPoint.deserialize(p) for p in payload["equity_curve"]],
168 fills=[Fill.deserialize(f) for f in payload["fills"]],
169 rejections=[SignalRejection.deserialize(r) for r in payload["rejections"]],
170 periods_per_year=payload["periods_per_year"],
171 benchmark_return=payload.get("benchmark_return"),
172 )
174 @staticmethod
175 def _serialize_period(period: ExperimentPeriod) -> dict[str, Any]:
176 return {
177 "start": period.start.isoformat(),
178 "end": period.end.isoformat(),
179 }
181 @staticmethod
182 def _deserialize_period(payload: dict[str, Any]) -> ExperimentPeriod:
183 return ExperimentPeriod(
184 start=datetime.fromisoformat(payload["start"]),
185 end=datetime.fromisoformat(payload["end"]),
186 )
188 def _serialize_fold(self, fold: WalkForwardFoldResult) -> dict[str, Any]:
189 return {
190 "split": {
191 "train": self._serialize_period(fold.split.train),
192 "validation": self._serialize_period(fold.split.validation),
193 "test": self._serialize_period(fold.split.test),
194 },
195 "selected_params": fold.selected_params,
196 "test_result": self._serialize_backtest_result(fold.test_result),
197 }
199 def _deserialize_fold(self, payload: dict[str, Any]) -> WalkForwardFoldResult:
200 return WalkForwardFoldResult(
201 split=WalkForwardSplit(
202 train=self._deserialize_period(payload["split"]["train"]),
203 validation=self._deserialize_period(payload["split"]["validation"]),
204 test=self._deserialize_period(payload["split"]["test"]),
205 ),
206 selected_params=dict(payload["selected_params"]), # type: ignore[arg-type]
207 test_result=self._deserialize_backtest_result(payload["test_result"]),
208 )
210 def _serialize_metadata(self, metadata: ExperimentMetadata) -> dict[str, Any]:
211 return {
212 "run_id": str(metadata.run_id),
213 "created_at": metadata.created_at.isoformat(),
214 "config": self._serialize_config(metadata.config),
215 }
217 def _deserialize_metadata(self, payload: dict[str, Any]) -> ExperimentMetadata:
218 return ExperimentMetadata(
219 run_id=UUID(payload["run_id"]),
220 created_at=datetime.fromisoformat(payload["created_at"]),
221 config=self._deserialize_config(payload["config"]),
222 )
224 def _serialize_backtest(self, run: ExperimentRun) -> dict[str, Any]:
225 return {
226 "schema_version": 2,
227 "run_type": "backtest",
228 **self._serialize_metadata(run.metadata),
229 "result": self._serialize_backtest_result(run.result),
230 }
232 def _deserialize_backtest(self, payload: dict[str, Any]) -> ExperimentRun:
233 metadata = self._deserialize_metadata(payload)
235 return ExperimentRun(
236 metadata=metadata,
237 result=self._deserialize_backtest_result(payload["result"]),
238 )
240 def _serialize_walk_forward(self, run: WalkForwardExperimentRun) -> dict[str, Any]:
241 return {
242 "schema_version": 2,
243 "run_type": "walk_forward",
244 **self._serialize_metadata(run.metadata),
245 "result": {
246 "folds": [self._serialize_fold(fold) for fold in run.result.folds],
247 },
248 }
250 def _deserialize_walk_forward(self, payload: dict[str, Any]) -> WalkForwardExperimentRun:
251 metadata = self._deserialize_metadata(payload)
253 result = WalkForwardResult(
254 folds=tuple(
255 (self._deserialize_fold(fold)
256 for fold in payload["result"]["folds"]),
257 ),
258 )
260 return WalkForwardExperimentRun(metadata=metadata, result=result)
262 def _deserialize(self, payload: dict[str, Any]) -> Run:
263 schema_version = payload.get("schema_version")
264 if schema_version == 1 or (schema_version == 2 and payload["run_type"] == "backtest"):
265 return self._deserialize_backtest(payload)
267 elif schema_version == 2 and payload["run_type"] == "walk_forward":
268 return self._deserialize_walk_forward(payload)
270 else:
271 raise ValueError(f"Unsupported experiment schema version: {schema_version!r}")