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

1from collections.abc import Sequence 

2from dataclasses import replace 

3from typing import Any, Protocol 

4 

5from finbot.backtest.experiment import ( 

6 ExperimentConfig, 

7 WalkForwardFoldResult, 

8 WalkForwardResult, 

9 WalkForwardSplit, 

10) 

11from finbot.backtest.models import BacktestResult 

12from finbot.backtest.walk_forward import ParameterEvaluation, select_best_parameters 

13from finbot.engine.backtest_engine import BacktestEngine 

14from finbot.execution.broker import SimulatedBroker 

15from finbot.market.data_provider import MarketDataProvider 

16from finbot.portfolio.portfolio import Portfolio 

17from finbot.risk import RiskLimits, RiskManager 

18from finbot.strategy.strategies import create_strategy 

19 

20 

21class ExperimentRunner: 

22 def __init__(self, market_data: MarketDataProvider): 

23 self.market_data = market_data 

24 

25 def run(self, config: ExperimentConfig) -> BacktestResult: 

26 strategy = create_strategy(config.strategy, config.strategy_params) 

27 

28 bars = self.market_data.get_bars( 

29 symbols=list(config.symbols), 

30 start=config.start, 

31 end=config.end, 

32 timeframe="1Day", 

33 ) 

34 

35 for symbol in config.symbols: 

36 if not bars.get(symbol): 

37 raise RuntimeError(f"No historical bars returned for {symbol!r}") 

38 

39 broker = SimulatedBroker( 

40 portfolio=Portfolio(cash=config.initial_cash), 

41 fee_rate=config.fee_rate, 

42 slippage_rate=config.slippage_rate, 

43 ) 

44 

45 risk_limits = RiskLimits( 

46 max_position_pct_nav=config.allocation, 

47 max_gross_exposure=config.allocation, 

48 max_symbol_exposure=config.allocation, 

49 ) 

50 

51 engine = BacktestEngine( 

52 broker=broker, 

53 strategy=strategy, 

54 risk_manager=RiskManager(risk_limits), 

55 ) 

56 

57 return engine.run(bars) 

58 

59 

60class BacktestRunner(Protocol): 

61 def run(self, config: ExperimentConfig) -> BacktestResult: 

62 ... 

63 

64 

65class WalkForwardRunner: 

66 """ 

67 Evaluate parameters on validation data and report results on test data. 

68 

69 The train period is reserved for strategies that require fitting; current 

70 rule-based strategies (like SMA) don't have a training step. 

71 """ 

72 

73 def __init__(self, runner: BacktestRunner): 

74 self.runner = runner 

75 

76 def run_fold( 

77 self, 

78 *, 

79 config: ExperimentConfig, 

80 split: WalkForwardSplit, 

81 parameter_candidates: Sequence[dict[str, Any]], 

82 ) -> WalkForwardFoldResult: 

83 evaluations: list[ParameterEvaluation] = [] 

84 

85 for params in parameter_candidates: 

86 validation_config = replace( 

87 config, 

88 start=split.validation.start, 

89 end=split.validation.end, 

90 strategy_params=dict(params), 

91 ) 

92 

93 validation_result = self.runner.run(validation_config) 

94 

95 evaluations.append( 

96 ParameterEvaluation( 

97 params=dict(params), 

98 validation_score=validation_result.total_return, 

99 ), 

100 ) 

101 

102 selected_params = select_best_parameters(evaluations) 

103 

104 test_config = replace( 

105 config, 

106 start=split.test.start, 

107 end=split.test.end, 

108 strategy_params=dict(selected_params), 

109 ) 

110 

111 test_result = self.runner.run(test_config) 

112 

113 return WalkForwardFoldResult( 

114 split=split, 

115 selected_params=dict(selected_params), 

116 test_result=test_result, 

117 ) 

118 

119 def run( 

120 self, 

121 *, 

122 config: ExperimentConfig, 

123 splits: Sequence[WalkForwardSplit], 

124 parameter_candidates: Sequence[dict[str, Any]], 

125 ) -> WalkForwardResult: 

126 if not splits: 

127 raise ValueError("At least one walk-forward split is required") 

128 

129 folds = tuple( 

130 (self.run_fold(config=config, split=split, parameter_candidates=parameter_candidates) 

131 for split in splits), 

132 ) 

133 

134 return WalkForwardResult(folds=folds)