Coverage for src/finbot/ml/experiment.py: 88%
196 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-04 12:05 +0000
1import math
2from dataclasses import dataclass
3from datetime import datetime
5from finbot.backtest.experiment import ExperimentPeriod
6from finbot.backtest.models import BacktestResult
7from finbot.engine.backtest_engine import BacktestEngine, ExecutionPolicy
8from finbot.execution.broker import SimulatedBroker
9from finbot.features import (
10 FeatureDataset,
11 FeaturePipeline,
12 ForwardReturnTarget,
13 SupervisedDataset,
14 SupervisedDatasetBuilder,
15 initial_feature_set,
16)
17from finbot.market.cache import HistoricalDataCache
18from finbot.market.models import Bar
19from finbot.models import LOGISTIC_REGRESSION_MODEL_TYPE, LogisticRegressionModel, Prediction
20from finbot.portfolio.portfolio import Portfolio
21from finbot.risk import RiskLimits, RiskManager
22from finbot.strategy import ModelStrategy, ProbabilityThresholdPolicy, Strategy
23from finbot.strategy.strategies import BuyOnceStrategy, SmaCrossoverStrategy
26@dataclass(frozen=True)
27class MLExperimentConfig:
28 dataset_id: str
29 symbols: tuple[str, ...]
30 training_period: ExperimentPeriod
31 validation_period: ExperimentPeriod
32 test_period: ExperimentPeriod
33 initial_cash: float = 10_000.0
34 allocation: float = 0.995
35 fee_rate: float = 0.0005
36 slippage_rate: float = 0.001
37 execution_policy: ExecutionPolicy = ExecutionPolicy.NEXT_OPEN
38 feature_set: str = "initial"
39 target_horizon: int = 1
40 model_type: str = LOGISTIC_REGRESSION_MODEL_TYPE
41 random_seed: int = 0
42 model_c: float = 1.0
43 model_max_iter: int = 1_000
44 long_threshold: float = 0.55
45 short_threshold: float | None = 0.45
46 sma_short_window: int = 10
47 sma_long_window: int = 30
49 def __post_init__(self) -> None:
50 if not isinstance(self.dataset_id, str) or not self.dataset_id:
51 raise ValueError("dataset_id must not be empty")
52 if any(not isinstance(symbol, str) for symbol in self.symbols): 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 raise ValueError("symbols must contain strings")
54 normalized_symbols = tuple(dict.fromkeys(symbol.strip().upper() for symbol in self.symbols))
55 if not normalized_symbols or any(not symbol for symbol in normalized_symbols):
56 raise ValueError("symbols must contain non-empty values")
57 object.__setattr__(self, "symbols", normalized_symbols)
58 if self.training_period.end > self.validation_period.start: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true
59 raise ValueError("Training period must not overlap validation period.")
60 if self.validation_period.end > self.test_period.start: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true
61 raise ValueError("Validation period must not overlap test period.")
62 for period in (self.training_period, self.validation_period, self.test_period):
63 if period.start.tzinfo is None or period.end.tzinfo is None: 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true
64 raise ValueError("Experiment periods must be timezone-aware.")
65 if not math.isfinite(self.initial_cash) or self.initial_cash <= 0:
66 raise ValueError("initial_cash must be a positive finite number")
67 if not math.isfinite(self.allocation) or not 0 < self.allocation <= 1: 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true
68 raise ValueError("allocation must be greater than 0 and at most 1")
69 if not math.isfinite(self.fee_rate) or self.fee_rate < 0: 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true
70 raise ValueError("fee_rate must be a non-negative finite number")
71 if not math.isfinite(self.slippage_rate) or not 0 <= self.slippage_rate <= 1: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 raise ValueError("slippage_rate must be between 0 and 1")
73 if self.feature_set != "initial":
74 raise ValueError("Only the 'initial' feature set is currently supported.")
75 if self.model_type != LOGISTIC_REGRESSION_MODEL_TYPE:
76 raise ValueError("Only logistic_regression is currently supported.")
77 if ( 77 ↛ 82line 77 didn't jump to line 82 because the condition on line 77 was never true
78 isinstance(self.target_horizon, bool)
79 or not isinstance(self.target_horizon, int)
80 or self.target_horizon < 1
81 ):
82 raise ValueError("target_horizon must be a positive integer")
83 object.__setattr__(self, "execution_policy", ExecutionPolicy(self.execution_policy))
84 LogisticRegressionModel(
85 random_seed=self.random_seed,
86 c=self.model_c,
87 max_iter=self.model_max_iter,
88 )
89 ProbabilityThresholdPolicy(self.long_threshold, self.short_threshold)
90 SmaCrossoverStrategy(self.sma_short_window, self.sma_long_window)
92 def to_dict(self) -> dict[str, object]:
93 return {
94 "dataset_id": self.dataset_id,
95 "symbols": list(self.symbols),
96 "periods": {
97 "training": _period_to_dict(self.training_period),
98 "validation": _period_to_dict(self.validation_period),
99 "test": _period_to_dict(self.test_period),
100 },
101 "initial_cash": self.initial_cash,
102 "allocation": self.allocation,
103 "fee_rate": self.fee_rate,
104 "slippage_rate": self.slippage_rate,
105 "execution_policy": self.execution_policy.value,
106 "feature_set": self.feature_set,
107 "target_horizon": self.target_horizon,
108 "model": {
109 "type": self.model_type,
110 "random_seed": self.random_seed,
111 "C": self.model_c,
112 "max_iter": self.model_max_iter,
113 },
114 "signal_policy": {
115 "long_threshold": self.long_threshold,
116 "short_threshold": self.short_threshold,
117 },
118 "baselines": {
119 "sma_crossover": {
120 "short_window": self.sma_short_window,
121 "long_window": self.sma_long_window,
122 },
123 "buy_and_hold": {},
124 },
125 }
127 @classmethod
128 def from_dict(cls, payload: dict[str, object]) -> "MLExperimentConfig":
129 try:
130 periods = _object(payload["periods"], "periods")
131 model = _object(payload.get("model", {}), "model")
132 policy = _object(payload.get("signal_policy", {}), "signal_policy")
133 baselines = _object(payload.get("baselines", {}), "baselines")
134 sma = _object(baselines.get("sma_crossover", {}), "sma_crossover")
135 symbols = payload["symbols"]
136 if not isinstance(symbols, list) or not all(isinstance(item, str) for item in symbols): 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 raise ValueError("symbols must be a list of strings")
138 return cls(
139 dataset_id=str(payload["dataset_id"]),
140 symbols=tuple(symbols),
141 training_period=_period_from_dict(periods["training"]),
142 validation_period=_period_from_dict(periods["validation"]),
143 test_period=_period_from_dict(periods["test"]),
144 initial_cash=float(payload.get("initial_cash", 10_000.0)),
145 allocation=float(payload.get("allocation", 0.995)),
146 fee_rate=float(payload.get("fee_rate", 0.0005)),
147 slippage_rate=float(payload.get("slippage_rate", 0.001)),
148 execution_policy=ExecutionPolicy(str(payload.get("execution_policy", "next_open"))),
149 feature_set=str(payload.get("feature_set", "initial")),
150 target_horizon=int(payload.get("target_horizon", 1)),
151 model_type=str(model.get("type", LOGISTIC_REGRESSION_MODEL_TYPE)),
152 random_seed=int(model.get("random_seed", 0)),
153 model_c=float(model.get("C", 1.0)),
154 model_max_iter=int(model.get("max_iter", 1_000)),
155 long_threshold=float(policy.get("long_threshold", 0.55)),
156 short_threshold=(
157 None if policy.get("short_threshold", 0.45) is None
158 else float(policy.get("short_threshold", 0.45))
159 ),
160 sma_short_window=int(sma.get("short_window", 10)),
161 sma_long_window=int(sma.get("long_window", 30)),
162 )
163 except (KeyError, TypeError, ValueError) as exc:
164 raise ValueError(f"Invalid ML experiment configuration: {exc}") from exc
167@dataclass(frozen=True)
168class ClassificationMetrics:
169 sample_count: int
170 accuracy: float
171 log_loss: float
173 def to_dict(self) -> dict[str, object]:
174 return {
175 "sample_count": self.sample_count,
176 "accuracy": self.accuracy,
177 "log_loss": self.log_loss,
178 }
181@dataclass(frozen=True)
182class StrategyMetrics:
183 total_return: float
184 annualized_return: float
185 max_drawdown: float
186 sharpe_ratio: float
187 trade_count: int
188 win_rate: float | None
189 turnover: float
190 total_fees: float
191 total_slippage: float
193 @classmethod
194 def from_result(cls, result: BacktestResult) -> "StrategyMetrics":
195 return cls(
196 total_return=result.total_return,
197 annualized_return=result.cagr,
198 max_drawdown=result.max_drawdown,
199 sharpe_ratio=result.sharpe_ratio(),
200 trade_count=result.trade_count,
201 win_rate=result.win_rate if result.trade_count else None,
202 turnover=result.turnover,
203 total_fees=result.total_fees,
204 total_slippage=result.total_slippage,
205 )
207 def to_dict(self) -> dict[str, object]:
208 return {
209 name: None if isinstance(value, float) and not math.isfinite(value) else value
210 for name, value in self.__dict__.items()
211 }
214@dataclass(frozen=True)
215class MLExperimentResult:
216 feature_dataset: FeatureDataset
217 supervised_dataset: SupervisedDataset
218 model: LogisticRegressionModel
219 validation_predictions: tuple[Prediction, ...]
220 test_predictions: tuple[Prediction, ...]
221 validation_metrics: ClassificationMetrics
222 test_metrics: ClassificationMetrics
223 backtests: dict[str, BacktestResult]
224 comparison: dict[str, StrategyMetrics]
227class MLExperimentRunner:
228 def __init__(self, datasets: HistoricalDataCache) -> None:
229 self.datasets = datasets
231 def run(self, config: MLExperimentConfig) -> MLExperimentResult:
232 dataset_metadata, bars_by_symbol = self.datasets.load(config.dataset_id)
233 if not set(config.symbols).issubset(dataset_metadata.symbols): 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 raise ValueError("Experiment symbols are not present in the source dataset.")
235 if config.training_period.start < dataset_metadata.start or config.test_period.end > dataset_metadata.end:
236 raise ValueError("Experiment periods must be contained in the source dataset.")
238 bars = sorted(
239 (
240 bar
241 for symbol in config.symbols
242 for bar in bars_by_symbol[symbol]
243 ),
244 key=lambda bar: (bar.timestamp, bar.symbol),
245 )
246 feature_pipeline = FeaturePipeline(initial_feature_set())
247 feature_dataset = feature_pipeline.generate(bars, source_dataset_id=config.dataset_id)
248 supervised_dataset = SupervisedDatasetBuilder(
249 ForwardReturnTarget(config.target_horizon),
250 ).build(feature_dataset, bars)
252 training = _slice_supervised(supervised_dataset, config.training_period)
253 validation = _slice_supervised(supervised_dataset, config.validation_period)
254 test = _slice_supervised(supervised_dataset, config.test_period)
255 if not training.rows or not validation.rows or not test.rows:
256 raise ValueError("Training, validation, and test datasets must all contain samples.")
258 model = LogisticRegressionModel(
259 random_seed=config.random_seed,
260 c=config.model_c,
261 max_iter=config.model_max_iter,
262 ).fit(training)
263 validation_features = _features_for_supervised(feature_dataset, validation)
264 test_features = _features_in_period(feature_dataset, config.test_period)
265 validation_predictions = model.predict_proba(validation_features)
266 direct_test_predictions = model.predict_proba(test_features)
268 test_bars = bars_in_period(bars_by_symbol, config.symbols, config.test_period)
269 warmup_bars = [bar for bar in bars if bar.timestamp < config.test_period.start]
270 model_strategy = ModelStrategy(
271 model=model,
272 feature_pipeline=FeaturePipeline(initial_feature_set()),
273 policy=ProbabilityThresholdPolicy(config.long_threshold, config.short_threshold),
274 )
275 model_strategy.prime(warmup_bars)
276 model_backtest = run_ml_backtest(config, model_strategy, test_bars)
277 test_predictions = model_strategy.predictions
278 if test_predictions != direct_test_predictions: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true
279 raise RuntimeError("Backtest predictions differ from direct test-set inference.")
281 sma = SmaCrossoverStrategy(config.sma_short_window, config.sma_long_window)
282 sma.prime(warmup_bars)
283 backtests = {
284 "logistic_regression": model_backtest,
285 "sma_crossover": run_ml_backtest(config, sma, test_bars),
286 "buy_and_hold": run_ml_backtest(config, BuyOnceStrategy(), test_bars),
287 }
288 comparison = {
289 name: StrategyMetrics.from_result(result)
290 for name, result in backtests.items()
291 }
292 return MLExperimentResult(
293 feature_dataset=feature_dataset,
294 supervised_dataset=supervised_dataset,
295 model=model,
296 validation_predictions=validation_predictions,
297 test_predictions=test_predictions,
298 validation_metrics=_classification_metrics(validation_predictions, validation),
299 test_metrics=_classification_metrics(test_predictions, test),
300 backtests=backtests,
301 comparison=comparison,
302 )
305def run_ml_backtest(
306 config: MLExperimentConfig,
307 strategy: Strategy,
308 bars: dict[str, list[Bar]],
309) -> BacktestResult:
310 broker = SimulatedBroker(
311 Portfolio(cash=config.initial_cash),
312 fee_rate=config.fee_rate,
313 slippage_rate=config.slippage_rate,
314 )
315 limits = RiskLimits(
316 max_position_pct_nav=config.allocation,
317 max_gross_exposure=config.allocation,
318 max_symbol_exposure=config.allocation,
319 )
320 return BacktestEngine(
321 broker=broker,
322 strategy=strategy,
323 execution_policy=config.execution_policy,
324 risk_manager=RiskManager(limits),
325 ).run(bars)
328def _slice_supervised(dataset: SupervisedDataset, period: ExperimentPeriod) -> SupervisedDataset:
329 return SupervisedDataset(
330 feature_names=dataset.feature_names,
331 target_name=dataset.target_name,
332 rows=tuple(
333 row
334 for row in dataset.rows
335 if period.start <= row.timestamp and row.target_timestamp < period.end
336 ),
337 metadata=dataset.metadata,
338 )
341def _features_for_supervised(
342 features: FeatureDataset,
343 supervised: SupervisedDataset,
344) -> FeatureDataset:
345 keys = {(row.symbol, row.timestamp) for row in supervised.rows}
346 return FeatureDataset(
347 feature_names=features.feature_names,
348 rows=tuple(row for row in features.rows if (row.symbol, row.timestamp) in keys),
349 metadata=features.metadata,
350 )
353def _features_in_period(features: FeatureDataset, period: ExperimentPeriod) -> FeatureDataset:
354 return FeatureDataset(
355 feature_names=features.feature_names,
356 rows=tuple(
357 row
358 for row in features.rows
359 if period.start <= row.timestamp < period.end
360 and all(value is not None for value in row.values)
361 ),
362 metadata=features.metadata,
363 )
366def bars_in_period(
367 bars: dict[str, list[Bar]],
368 symbols: tuple[str, ...],
369 period: ExperimentPeriod,
370) -> dict[str, list[Bar]]:
371 result = {
372 symbol: [bar for bar in bars[symbol] if period.start <= bar.timestamp < period.end]
373 for symbol in symbols
374 }
375 if any(not symbol_bars for symbol_bars in result.values()): 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 raise ValueError("Every experiment symbol must have bars in the test period.")
377 return result
380def _classification_metrics(
381 predictions: tuple[Prediction, ...],
382 dataset: SupervisedDataset,
383) -> ClassificationMetrics:
384 targets = {(row.symbol, row.timestamp): row.target > 0 for row in dataset.rows}
385 predictions_by_key = {
386 (prediction.symbol, prediction.timestamp): prediction
387 for prediction in predictions
388 }
389 if len(predictions_by_key) != len(predictions) or not set(targets).issubset(predictions_by_key): 389 ↛ 390line 389 didn't jump to line 390 because the condition on line 389 was never true
390 raise ValueError("Predictions do not align with the evaluation dataset.")
391 correct = 0
392 losses: list[float] = []
393 epsilon = 1e-15
394 for key, target in targets.items():
395 prediction = predictions_by_key[key]
396 probability = min(max(prediction.probability_positive, epsilon), 1 - epsilon)
397 correct += (probability >= 0.5) == target
398 losses.append(-(math.log(probability) if target else math.log(1 - probability)))
399 return ClassificationMetrics(
400 sample_count=len(targets),
401 accuracy=correct / len(targets),
402 log_loss=sum(losses) / len(targets),
403 )
406def _period_to_dict(period: ExperimentPeriod) -> dict[str, str]:
407 return {"start": period.start.isoformat(), "end": period.end.isoformat()}
410def _period_from_dict(raw: object) -> ExperimentPeriod:
411 value = _object(raw, "period")
412 return ExperimentPeriod(
413 start=datetime.fromisoformat(str(value["start"])),
414 end=datetime.fromisoformat(str(value["end"])),
415 )
418def _object(raw: object, name: str) -> dict[str, object]:
419 if not isinstance(raw, dict): 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true
420 raise ValueError(f"{name} must be an object")
421 return raw