Coverage for src/finbot/backtest/walk_forward.py: 100%

1import math 

2from dataclasses import dataclass 

3from datetime import timedelta 

4from enum import StrEnum 

5from typing import Any 

6 

7from finbot.backtest.experiment import ExperimentPeriod, WalkForwardSplit 

8 

9 

10class TrainingWindowMode(StrEnum): 

11 ROLLING = "rolling" 

12 EXPANDING = "expanding" 

13 

14 

15def generate_walk_forward_splits( 

16 period: ExperimentPeriod, 

17 train_duration: timedelta, 

18 validation_duration: timedelta, 

19 test_duration: timedelta, 

20 training_window: TrainingWindowMode | str = TrainingWindowMode.ROLLING, 

21) -> list[WalkForwardSplit]: 

22 if train_duration.total_seconds() <= 0: 

23 raise ValueError("train_duration must be greater than zero.") 

24 if validation_duration.total_seconds() <= 0: 

25 raise ValueError("validation_duration must be greater than zero.") 

26 if test_duration.total_seconds() <= 0: 

27 raise ValueError("test_duration must be greater than zero.") 

28 try: 

29 resolved_training_window = TrainingWindowMode(training_window) 

30 except (TypeError, ValueError) as exc: 

31 raise ValueError( 

32 f"Unsupported training window mode: {training_window!r}.", 

33 ) from exc 

34 

35 fold_duration = train_duration + validation_duration + test_duration 

36 

37 if period.duration < fold_duration: 

38 raise ValueError("Period is too short for one complete walk-forward split.") 

39 

40 splits: list[WalkForwardSplit] = [] 

41 current_train_end = period.start + train_duration 

42 while current_train_end + validation_duration + test_duration <= period.end: 

43 train_start = ( 

44 period.start 

45 if resolved_training_window == TrainingWindowMode.EXPANDING 

46 else current_train_end - train_duration 

47 ) 

48 train_period = ExperimentPeriod(start=train_start, end=current_train_end) 

49 validation_period = ExperimentPeriod( 

50 start=train_period.end, 

51 end=train_period.end + validation_duration, 

52 ) 

53 test_period = ExperimentPeriod( 

54 start=validation_period.end, 

55 end=validation_period.end + test_duration, 

56 ) 

57 

58 splits.append( 

59 WalkForwardSplit(train=train_period, validation=validation_period, test=test_period), 

60 ) 

61 

62 current_train_end += test_duration 

63 

64 return splits 

65 

66 

67@dataclass(frozen=True) 

68class ParameterEvaluation: 

69 params: dict[str, Any] 

70 validation_score: float 

71 

72 def __post_init__(self): 

73 if not math.isfinite(self.validation_score): 

74 raise ValueError("validation_score must be finite") 

75 

76 

77def select_best_parameters(evaluations: list[ParameterEvaluation]) -> dict[str, Any]: 

78 if not evaluations: 

79 raise ValueError("At least one parameter evaluation is required.") 

80 

81 best_evaluation = max(evaluations, key=lambda e: e.validation_score) 

82 return best_evaluation.params