Coverage for src/finbot/ml/walk_forward.py: 96%
128 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-04 12:05 +0000
1from collections.abc import Sequence
2from dataclasses import dataclass, replace
3from datetime import datetime, timedelta
4from itertools import chain
6from finbot.backtest.experiment import ExperimentPeriod, WalkForwardSplit
7from finbot.backtest.models import BacktestResult
8from finbot.backtest.walk_forward import TrainingWindowMode, generate_walk_forward_splits
9from finbot.market.cache import HistoricalDataCache
10from finbot.ml.experiment import (
11 MLExperimentConfig,
12 MLExperimentResult,
13 MLExperimentRunner,
14 StrategyMetrics,
15 bars_in_period,
16 run_ml_backtest,
17)
18from finbot.models import Prediction
19from finbot.strategy import PredictionStrategy, ProbabilityThresholdPolicy
20from finbot.strategy.strategies import BuyOnceStrategy, SmaCrossoverStrategy
23@dataclass(frozen=True)
24class MLWalkForwardConfig:
25 experiment: MLExperimentConfig
26 evaluation_period: ExperimentPeriod
27 training_duration: timedelta
28 validation_duration: timedelta
29 test_duration: timedelta
30 training_window: TrainingWindowMode = TrainingWindowMode.ROLLING
32 def __post_init__(self) -> None:
33 object.__setattr__(self, "training_window", TrainingWindowMode(self.training_window))
34 for duration in (
35 self.training_duration,
36 self.validation_duration,
37 self.test_duration,
38 ):
39 _whole_days(duration)
40 first_split = self.splits[0]
41 object.__setattr__(
42 self,
43 "experiment",
44 replace(
45 self.experiment,
46 training_period=first_split.train,
47 validation_period=first_split.validation,
48 test_period=first_split.test,
49 ),
50 )
52 @property
53 def splits(self) -> tuple[WalkForwardSplit, ...]:
54 return tuple(
55 generate_walk_forward_splits(
56 self.evaluation_period,
57 train_duration=self.training_duration,
58 validation_duration=self.validation_duration,
59 test_duration=self.test_duration,
60 training_window=self.training_window,
61 ),
62 )
64 def to_dict(self) -> dict[str, object]:
65 payload = self.experiment.to_dict()
66 payload.pop("periods")
67 payload["walk_forward"] = {
68 "start": self.evaluation_period.start.isoformat(),
69 "end": self.evaluation_period.end.isoformat(),
70 "training_days": _whole_days(self.training_duration),
71 "validation_days": _whole_days(self.validation_duration),
72 "test_days": _whole_days(self.test_duration),
73 "training_window": self.training_window.value,
74 }
75 return payload
77 @classmethod
78 def from_dict(cls, payload: dict[str, object]) -> "MLWalkForwardConfig":
79 try:
80 raw_walk_forward = payload["walk_forward"]
81 if not isinstance(raw_walk_forward, dict): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 raise ValueError("walk_forward must be an object")
83 evaluation_period = ExperimentPeriod(
84 _datetime_value(raw_walk_forward["start"], "walk_forward.start"),
85 _datetime_value(raw_walk_forward["end"], "walk_forward.end"),
86 )
87 training_duration = _days_value(raw_walk_forward, "training_days")
88 validation_duration = _days_value(raw_walk_forward, "validation_days")
89 test_duration = _days_value(raw_walk_forward, "test_days")
90 raw_training_window = raw_walk_forward.get(
91 "training_window",
92 TrainingWindowMode.ROLLING,
93 )
94 try:
95 training_window = TrainingWindowMode(raw_training_window)
96 except (TypeError, ValueError) as exc:
97 raise ValueError(
98 f"Unsupported training window mode: {raw_training_window!r}",
99 ) from exc
100 splits = generate_walk_forward_splits(
101 evaluation_period,
102 train_duration=training_duration,
103 validation_duration=validation_duration,
104 test_duration=test_duration,
105 training_window=training_window,
106 )
107 experiment_payload = dict(payload)
108 experiment_payload.pop("walk_forward")
109 experiment_payload["periods"] = {
110 "training": _period_to_dict(splits[0].train),
111 "validation": _period_to_dict(splits[0].validation),
112 "test": _period_to_dict(splits[0].test),
113 }
114 return cls(
115 experiment=MLExperimentConfig.from_dict(experiment_payload),
116 evaluation_period=evaluation_period,
117 training_duration=training_duration,
118 validation_duration=validation_duration,
119 test_duration=test_duration,
120 training_window=training_window,
121 )
122 except (KeyError, TypeError, ValueError) as exc:
123 raise ValueError(f"Invalid ML walk-forward configuration: {exc}") from exc
126@dataclass(frozen=True)
127class MLWalkForwardFoldResult:
128 index: int
129 split: WalkForwardSplit
130 config: MLExperimentConfig
131 experiment: MLExperimentResult
134@dataclass(frozen=True)
135class MLWalkForwardResult:
136 folds: tuple[MLWalkForwardFoldResult, ...]
137 backtests: dict[str, BacktestResult]
138 comparison: dict[str, StrategyMetrics]
140 def __post_init__(self) -> None:
141 if not self.folds:
142 raise ValueError("Walk-forward ML evaluation requires at least one fold.")
143 if tuple(fold.index for fold in self.folds) != tuple(range(len(self.folds))):
144 raise ValueError("Walk-forward ML fold indexes must be consecutive.")
146 keys = [
147 (prediction.symbol, prediction.timestamp)
148 for prediction in self.oos_predictions
149 ]
150 if len(keys) != len(set(keys)):
151 raise ValueError("Walk-forward ML predictions must not overlap or appear twice.")
153 @property
154 def oos_predictions(self) -> tuple[Prediction, ...]:
155 return tuple(chain.from_iterable(
156 fold.experiment.test_predictions for fold in self.folds
157 ))
160class MLWalkForwardRunner:
161 def __init__(self, datasets: HistoricalDataCache) -> None:
162 self._datasets = datasets
163 self._runner = MLExperimentRunner(datasets)
165 def run(
166 self,
167 config: MLExperimentConfig,
168 splits: Sequence[WalkForwardSplit],
169 ) -> MLWalkForwardResult:
170 resolved_splits = tuple(splits)
171 _validate_test_periods(resolved_splits)
173 folds: list[MLWalkForwardFoldResult] = []
174 for index, split in enumerate(resolved_splits):
175 fold_config = replace(
176 config,
177 training_period=split.train,
178 validation_period=split.validation,
179 test_period=split.test,
180 )
181 try:
182 experiment = self._runner.run(fold_config)
183 except ValueError as exc:
184 raise ValueError(f"Walk-forward ML fold {index} failed: {exc}") from exc
185 folds.append(
186 MLWalkForwardFoldResult(
187 index=index,
188 split=split,
189 config=fold_config,
190 experiment=experiment,
191 ),
192 )
193 resolved_folds = tuple(folds)
194 predictions = tuple(chain.from_iterable(
195 fold.experiment.test_predictions for fold in resolved_folds
196 ))
197 evaluation_period = ExperimentPeriod(
198 resolved_splits[0].test.start,
199 resolved_splits[-1].test.end,
200 )
201 _, bars_by_symbol = self._datasets.load(config.dataset_id)
202 test_bars = bars_in_period(
203 bars_by_symbol,
204 config.symbols,
205 evaluation_period,
206 )
207 warmup_bars = sorted(
208 (
209 bar
210 for symbol in config.symbols
211 for bar in bars_by_symbol[symbol]
212 if bar.timestamp < evaluation_period.start
213 ),
214 key=lambda bar: (bar.timestamp, bar.symbol),
215 )
217 prediction_strategy = PredictionStrategy(
218 predictions,
219 ProbabilityThresholdPolicy(config.long_threshold, config.short_threshold),
220 )
221 sma = SmaCrossoverStrategy(config.sma_short_window, config.sma_long_window)
222 sma.prime(warmup_bars)
223 backtests = {
224 "logistic_regression": run_ml_backtest(config, prediction_strategy, test_bars),
225 "sma_crossover": run_ml_backtest(config, sma, test_bars),
226 "buy_and_hold": run_ml_backtest(config, BuyOnceStrategy(), test_bars),
227 }
228 return MLWalkForwardResult(
229 folds=resolved_folds,
230 backtests=backtests,
231 comparison={
232 name: StrategyMetrics.from_result(result)
233 for name, result in backtests.items()
234 },
235 )
238def _validate_test_periods(splits: tuple[WalkForwardSplit, ...]) -> None:
239 if not splits:
240 raise ValueError("Walk-forward ML evaluation requires at least one fold.")
241 for previous, current in zip(splits[:-1], splits[1:], strict=True):
242 if previous.test.end != current.test.start:
243 raise ValueError(
244 "Walk-forward ML test periods must be ordered and contiguous; "
245 "gaps and overlaps are not supported.",
246 )
249def _days_value(payload: dict[object, object], name: str) -> timedelta:
250 value = payload[name]
251 if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
252 raise ValueError(f"walk_forward.{name} must be a positive integer")
253 return timedelta(days=value)
256def _datetime_value(value: object, name: str) -> datetime:
257 if not isinstance(value, str): 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 raise ValueError(f"{name} must be an ISO-8601 string")
259 parsed = datetime.fromisoformat(value)
260 if parsed.tzinfo is None:
261 raise ValueError(f"{name} must include a timezone")
262 return parsed
265def _whole_days(value: timedelta) -> int:
266 if value.total_seconds() % timedelta(days=1).total_seconds(): 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true
267 raise ValueError("Walk-forward durations must contain whole days.")
268 return value.days
271def _period_to_dict(period: ExperimentPeriod) -> dict[str, str]:
272 return {"start": period.start.isoformat(), "end": period.end.isoformat()}