Coverage for src/finbot/ml/walk_forward_artifacts.py: 92%
165 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-04 12:05 +0000
1import csv
2import io
3from collections.abc import Sequence
4from dataclasses import dataclass
5from datetime import UTC, datetime
6from pathlib import Path
7from typing import Protocol
8from uuid import UUID, uuid4
10from finbot.backtest.experiment import ExperimentPeriod, WalkForwardSplit
11from finbot.ml.artifacts import (
12 MLArtifactError,
13 MLExperimentStore,
14 MLRunStatus,
15 _sha256,
16 _write_text_atomic,
17 write_backtest_trades,
18)
19from finbot.ml.experiment import MLExperimentConfig
20from finbot.ml.walk_forward import (
21 MLWalkForwardFoldResult,
22 MLWalkForwardResult,
23)
24from finbot.models import Model, load_model
26ML_WALK_FORWARD_SCHEMA_VERSION = 1
27ROOT_ARTIFACTS = (
28 "predictions.csv",
29 "trades.csv",
30 "metrics.json",
31 "comparison.json",
32)
33FOLD_ARTIFACTS = (
34 "config.json",
35 "model.json",
36 "feature_schema.json",
37 "predictions.csv",
38 "metrics.json",
39)
42@dataclass(frozen=True)
43class StoredMLWalkForwardExperiment:
44 run_id: UUID
45 metadata: dict[str, object]
46 models: tuple[Model, ...]
47 predictions: tuple[dict[str, str], ...]
48 metrics: dict[str, object]
49 comparison: dict[str, object]
52class WalkForwardRunner(Protocol):
53 def run(
54 self,
55 config: MLExperimentConfig,
56 splits: Sequence[WalkForwardSplit],
57 ) -> MLWalkForwardResult:
58 ...
61class MLWalkForwardStore(MLExperimentStore):
63 def execute(
64 self,
65 config: MLExperimentConfig,
66 splits: Sequence[WalkForwardSplit],
67 runner: WalkForwardRunner,
68 *,
69 git_commit: str,
70 run_id: UUID | None = None,
71 ) -> StoredMLWalkForwardExperiment:
72 resolved_run_id = run_id or uuid4()
73 resolved_splits = tuple(splits)
74 run_dir = self._begin(resolved_run_id, config, resolved_splits, git_commit)
75 try:
76 result = runner.run(config, resolved_splits)
77 self._complete(run_dir, resolved_run_id, config, git_commit, result)
78 except Exception as exc:
79 self._write_metadata(
80 run_dir,
81 self._metadata(
82 resolved_run_id,
83 config,
84 resolved_splits,
85 git_commit,
86 MLRunStatus.FAILED,
87 error={"type": type(exc).__name__, "message": str(exc)},
88 ),
89 )
90 raise
91 return self.load(resolved_run_id)
93 def load(self, run_id: UUID | str) -> StoredMLWalkForwardExperiment:
94 run_dir = self.root / str(run_id)
95 metadata = self._read_object(run_dir / "metadata.json")
96 if metadata.get("status") != MLRunStatus.COMPLETED:
97 raise MLArtifactError(
98 f"Walk-forward ML experiment {run_id} is "
99 f"{metadata.get('status') or 'incomplete'}.",
100 )
101 if ( 101 ↛ 106line 101 didn't jump to line 106 because the condition on line 101 was never true
102 metadata.get("schema_version") != ML_WALK_FORWARD_SCHEMA_VERSION
103 or metadata.get("run_type") != "ml_walk_forward"
104 or metadata.get("run_id") != str(run_id)
105 ):
106 raise MLArtifactError("Walk-forward ML experiment metadata is inconsistent.")
108 folds = metadata.get("folds")
109 if not isinstance(folds, list) or not folds: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 raise MLArtifactError("Walk-forward ML experiment has invalid fold metadata.")
111 expected_files = _required_artifacts(len(folds))
112 manifest = self._read_object(run_dir / "manifest.json")
113 files = manifest.get("files")
114 if (
115 manifest.get("schema_version") != ML_WALK_FORWARD_SCHEMA_VERSION
116 or not isinstance(files, dict)
117 or set(files) != expected_files
118 ):
119 raise MLArtifactError("Walk-forward ML experiment manifest is inconsistent.")
120 for name, expected_hash in files.items():
121 path = run_dir / name
122 if not path.is_file() or not isinstance(expected_hash, str): 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 raise MLArtifactError(f"Walk-forward ML experiment is missing artifact {name}.")
124 if _sha256(path.read_bytes()) != expected_hash:
125 raise MLArtifactError(f"Walk-forward ML artifact {name} failed integrity validation.")
127 try:
128 models = tuple(
129 load_model(run_dir / _fold_path(index) / "model.json")
130 for index in range(len(folds))
131 )
132 predictions = _read_csv(run_dir / "predictions.csv")
133 metrics = self._read_object(run_dir / "metrics.json")
134 comparison = self._read_object(run_dir / "comparison.json")
135 self._validate_folds(run_dir, folds, models)
136 self._validate_predictions(predictions, models)
137 except (OSError, TypeError, ValueError) as exc:
138 raise MLArtifactError(f"Walk-forward ML artifacts are invalid: {exc}") from exc
140 return StoredMLWalkForwardExperiment(
141 run_id=UUID(str(run_id)),
142 metadata=metadata,
143 models=models,
144 predictions=predictions,
145 metrics=metrics,
146 comparison=comparison,
147 )
149 def _begin(
150 self,
151 run_id: UUID,
152 config: MLExperimentConfig,
153 splits: tuple[WalkForwardSplit, ...],
154 git_commit: str,
155 ) -> Path:
156 if not git_commit.strip():
157 raise ValueError("git_commit must not be empty")
158 if not splits:
159 raise ValueError("Walk-forward ML evaluation requires at least one fold.")
160 self.root.mkdir(parents=True, exist_ok=True)
161 run_dir = self.root / str(run_id)
162 run_dir.mkdir(exist_ok=False)
163 self._write_metadata(
164 run_dir,
165 self._metadata(run_id, config, splits, git_commit, MLRunStatus.RUNNING),
166 )
167 return run_dir
169 def _complete(
170 self,
171 run_dir: Path,
172 run_id: UUID,
173 config: MLExperimentConfig,
174 git_commit: str,
175 result: MLWalkForwardResult,
176 ) -> None:
177 for fold in result.folds:
178 fold_dir = run_dir / _fold_path(fold.index)
179 fold_dir.mkdir(parents=True)
180 fold.experiment.model.save(fold_dir / "model.json")
181 self._write_json(fold_dir / "config.json", fold.config.to_dict())
182 self._write_json(
183 fold_dir / "feature_schema.json",
184 {
185 "feature_dataset_id": fold.experiment.feature_dataset.dataset_id,
186 "feature_names": list(fold.experiment.feature_dataset.feature_names),
187 "schema": [
188 entry.to_dict()
189 for entry in fold.experiment.feature_dataset.metadata.feature_schema
190 ],
191 },
192 )
193 _write_fold_predictions(fold_dir / "predictions.csv", fold)
194 self._write_json(
195 fold_dir / "metrics.json",
196 {
197 "validation": fold.experiment.validation_metrics.to_dict(),
198 "test": fold.experiment.test_metrics.to_dict(),
199 },
200 )
202 _write_oos_predictions(run_dir / "predictions.csv", result)
203 write_backtest_trades(run_dir / "trades.csv", result.backtests)
204 self._write_json(
205 run_dir / "metrics.json",
206 {
207 "fold_count": len(result.folds),
208 "aggregate": result.comparison["logistic_regression"].to_dict(),
209 },
210 )
211 self._write_json(
212 run_dir / "comparison.json",
213 {
214 "assumptions": {
215 "dataset_id": config.dataset_id,
216 "symbols": list(config.symbols),
217 "evaluation_period": {
218 "start": result.folds[0].split.test.start.isoformat(),
219 "end": result.folds[-1].split.test.end.isoformat(),
220 },
221 "initial_cash": config.initial_cash,
222 "fee_rate": config.fee_rate,
223 "slippage_rate": config.slippage_rate,
224 "execution_policy": config.execution_policy.value,
225 },
226 "parameters": {
227 "logistic_regression": {
228 "models": [
229 {
230 "fold_index": fold.index,
231 "model_id": fold.experiment.model.metadata.model_id,
232 "hyperparameters": dict(
233 fold.experiment.model.metadata.hyperparameters,
234 ),
235 }
236 for fold in result.folds
237 ],
238 "signal_policy": config.to_dict()["signal_policy"],
239 },
240 "sma_crossover": config.to_dict()["baselines"]["sma_crossover"],
241 "buy_and_hold": {},
242 },
243 "results": {
244 name: value.to_dict() for name, value in result.comparison.items()
245 },
246 },
247 )
248 expected_files = _required_artifacts(len(result.folds))
249 self._write_json(
250 run_dir / "manifest.json",
251 {
252 "schema_version": ML_WALK_FORWARD_SCHEMA_VERSION,
253 "files": {
254 name: _sha256((run_dir / name).read_bytes())
255 for name in sorted(expected_files)
256 },
257 },
258 )
259 self._write_metadata(
260 run_dir,
261 self._metadata(
262 run_id,
263 config,
264 tuple(fold.split for fold in result.folds),
265 git_commit,
266 MLRunStatus.COMPLETED,
267 result=result,
268 ),
269 )
271 @staticmethod
272 def _metadata(
273 run_id: UUID,
274 config: MLExperimentConfig,
275 splits: tuple[WalkForwardSplit, ...],
276 git_commit: str,
277 status: MLRunStatus,
278 *,
279 result: MLWalkForwardResult | None = None,
280 error: dict[str, str] | None = None,
281 ) -> dict[str, object]:
282 fold_results = result.folds if result is not None else ()
283 metadata: dict[str, object] = {
284 "schema_version": ML_WALK_FORWARD_SCHEMA_VERSION,
285 "run_type": "ml_walk_forward",
286 "run_id": str(run_id),
287 "status": status.value,
288 "updated_at": datetime.now(UTC).isoformat(),
289 "git_commit": git_commit,
290 "config": config.to_dict(),
291 "folds": [
292 {
293 "index": index,
294 "periods": _split_to_dict(split),
295 **(
296 {
297 "model_id": fold_results[index].experiment.model.metadata.model_id,
298 "model_type": fold_results[index].experiment.model.metadata.model_type,
299 "training_dataset_id": fold_results[index].experiment.model.metadata.training_configuration.dataset_id,
300 "feature_dataset_id": fold_results[index].experiment.feature_dataset.dataset_id,
301 "supervised_dataset_id": fold_results[index].experiment.supervised_dataset.dataset_id,
302 "artifact_directory": _fold_path(index),
303 }
304 if result is not None
305 else {}
306 ),
307 }
308 for index, split in enumerate(splits)
309 ],
310 }
311 if error is not None:
312 metadata["error"] = error
313 return metadata
315 @staticmethod
316 def _validate_folds(
317 run_dir: Path,
318 folds: list[object],
319 models: tuple[Model, ...],
320 ) -> None:
321 for index, (raw_fold, model) in enumerate(zip(folds, models, strict=True)):
322 if not isinstance(raw_fold, dict) or raw_fold.get("index") != index: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 raise ValueError("Fold metadata indexes are inconsistent.")
324 if raw_fold.get("model_id") != model.metadata.model_id:
325 raise ValueError("Fold model identity does not match metadata.")
326 fold_dir = run_dir / _fold_path(index)
327 feature_schema = MLWalkForwardStore._read_object(
328 fold_dir / "feature_schema.json",
329 )
330 if feature_schema.get("feature_names") != list(model.metadata.feature_schema):
331 raise ValueError("Fold feature schema does not match its model.")
332 fold_predictions = _read_csv(fold_dir / "predictions.csv")
333 if any(row.get("model_id") != model.metadata.model_id for row in fold_predictions): 333 ↛ 334line 333 didn't jump to line 334 because the condition on line 333 was never true
334 raise ValueError("Fold predictions do not match their model.")
336 @staticmethod
337 def _validate_predictions(
338 predictions: tuple[dict[str, str], ...],
339 models: tuple[Model, ...],
340 ) -> None:
341 keys: set[tuple[str, str]] = set()
342 previous_timestamp: datetime | None = None
343 for row in predictions:
344 index = int(row.get("fold_index", ""))
345 if index < 0 or index >= len(models): 345 ↛ 346line 345 didn't jump to line 346 because the condition on line 345 was never true
346 raise ValueError("Aggregate prediction has an invalid fold index.")
347 if row.get("model_id") != models[index].metadata.model_id:
348 raise ValueError("Aggregate prediction does not match its fold model.")
349 symbol = row.get("symbol")
350 raw_timestamp = row.get("timestamp")
351 if not symbol or not raw_timestamp: 351 ↛ 352line 351 didn't jump to line 352 because the condition on line 351 was never true
352 raise ValueError("Aggregate prediction identity is incomplete.")
353 timestamp = datetime.fromisoformat(raw_timestamp)
354 key = (symbol, raw_timestamp)
355 if key in keys: 355 ↛ 356line 355 didn't jump to line 356 because the condition on line 355 was never true
356 raise ValueError("Aggregate predictions contain duplicate rows.")
357 if previous_timestamp is not None and timestamp < previous_timestamp: 357 ↛ 358line 357 didn't jump to line 358 because the condition on line 357 was never true
358 raise ValueError("Aggregate predictions are not ordered.")
359 keys.add(key)
360 previous_timestamp = timestamp
363def _required_artifacts(fold_count: int) -> set[str]:
364 return set(ROOT_ARTIFACTS) | {
365 f"{_fold_path(index)}/{name}"
366 for index in range(fold_count)
367 for name in FOLD_ARTIFACTS
368 }
371def _fold_path(index: int) -> str:
372 return f"folds/{index:04d}"
375def _split_to_dict(split: WalkForwardSplit) -> dict[str, object]:
376 return {
377 "training": _period_to_dict(split.train),
378 "validation": _period_to_dict(split.validation),
379 "test": _period_to_dict(split.test),
380 }
383def _period_to_dict(period: ExperimentPeriod) -> dict[str, str]:
384 return {"start": period.start.isoformat(), "end": period.end.isoformat()}
387def _write_fold_predictions(path: Path, fold: MLWalkForwardFoldResult) -> None:
388 output = io.StringIO(newline="")
389 fields = ["split", "symbol", "timestamp", "probability_positive", "model_id", "model_type"]
390 writer = csv.DictWriter(output, fieldnames=fields, lineterminator="\n")
391 writer.writeheader()
392 for split, predictions in (
393 ("validation", fold.experiment.validation_predictions),
394 ("test", fold.experiment.test_predictions),
395 ):
396 for prediction in predictions:
397 writer.writerow({
398 "split": split,
399 "symbol": prediction.symbol,
400 "timestamp": prediction.timestamp.isoformat(),
401 "probability_positive": repr(prediction.probability_positive),
402 "model_id": prediction.model_id,
403 "model_type": prediction.model_type,
404 })
405 _write_text_atomic(path, output.getvalue())
408def _write_oos_predictions(path: Path, result: MLWalkForwardResult) -> None:
409 output = io.StringIO(newline="")
410 fields = ["fold_index", "symbol", "timestamp", "probability_positive", "model_id", "model_type"]
411 writer = csv.DictWriter(output, fieldnames=fields, lineterminator="\n")
412 writer.writeheader()
413 for fold in result.folds:
414 for prediction in fold.experiment.test_predictions:
415 writer.writerow({
416 "fold_index": fold.index,
417 "symbol": prediction.symbol,
418 "timestamp": prediction.timestamp.isoformat(),
419 "probability_positive": repr(prediction.probability_positive),
420 "model_id": prediction.model_id,
421 "model_type": prediction.model_type,
422 })
423 _write_text_atomic(path, output.getvalue())
426def _read_csv(path: Path) -> tuple[dict[str, str], ...]:
427 return tuple(csv.DictReader(io.StringIO(path.read_text(encoding="utf-8"))))