Coverage for src/finbot/engine/backtest_engine.py: 92%

1from datetime import datetime 

2from enum import StrEnum 

3from itertools import groupby 

4 

5from finbot.backtest.models import BacktestResult, EquityPoint, SignalRejection 

6from finbot.engine.orchestrator import NoAction, TradingOrchestrator, TradingRejection 

7from finbot.execution.broker import BacktestBroker 

8from finbot.execution.models import Fill, Order 

9from finbot.market.models import Bar 

10from finbot.risk.models import RiskLimits 

11from finbot.risk.position_sizer import PositionSizer 

12from finbot.risk.risk_manager import RiskManager 

13from finbot.strategy.core import Signal, Strategy 

14 

15 

16class ExecutionPolicy(StrEnum): 

17 NEXT_OPEN = "next_open" 

18 SAME_CLOSE = "same_close" 

19 

20 

21class BacktestEngine: 

22 def __init__( 

23 self, 

24 broker: BacktestBroker, 

25 strategy: Strategy, 

26 execution_policy: ExecutionPolicy = ExecutionPolicy.NEXT_OPEN, 

27 risk_manager: RiskManager | None = None, 

28 position_sizer: PositionSizer | None = None, 

29 periods_per_year: int = 252, 

30 ): 

31 if periods_per_year <= 0: 

32 raise ValueError("Periods per year must be greater than 0") 

33 

34 self.broker = broker 

35 self.execution_policy = execution_policy 

36 self.orchestrator = TradingOrchestrator( 

37 strategy=strategy, 

38 risk_manager=risk_manager if risk_manager is not None else RiskManager(RiskLimits()), 

39 position_sizer=position_sizer, 

40 ) 

41 self.periods_per_year = periods_per_year 

42 self._has_run = False 

43 

44 def run(self, bars_by_symbol: dict[str, list[Bar]]) -> BacktestResult: 

45 if self._has_run: 

46 raise RuntimeError("BacktestEngine instances are one-shot and can only be run once.") 

47 

48 if self.broker.portfolio.positions: 

49 raise ValueError("Initial portfolio positions are not supported by BacktestEngine.") 

50 

51 self._has_run = True 

52 

53 bars = sorted( 

54 (bar for symbol_bars in bars_by_symbol.values() for bar in symbol_bars), 

55 key=lambda bar: (bar.timestamp, bar.symbol), 

56 ) 

57 

58 starting_cash = self.broker.portfolio.cash 

59 fills: list[Fill] = [] 

60 equity_curve: list[EquityPoint] = [] 

61 rejections: list[SignalRejection] = [] 

62 

63 current_prices: dict[str, float] = {} 

64 pending_signals: list[Signal] = [] 

65 

66 for timestamp, grouped_bars in groupby(bars, key=lambda bar: bar.timestamp): 

67 timestamp_bars = list(grouped_bars) 

68 bars_by_current_symbol = {bar.symbol: bar for bar in timestamp_bars} 

69 open_prices = dict(current_prices) 

70 open_prices.update({bar.symbol: bar.open for bar in timestamp_bars}) 

71 

72 if self.execution_policy == ExecutionPolicy.NEXT_OPEN: 

73 still_pending: list[Signal] = [] 

74 

75 for signal in pending_signals: 

76 if signal.symbol not in bars_by_current_symbol: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 still_pending.append(signal) 

78 continue 

79 

80 order = self._signal_to_order( 

81 signal=signal, 

82 market_prices=open_prices, 

83 timestamp=timestamp, 

84 rejections=rejections, 

85 ) 

86 if order is None: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 continue 

88 

89 fill = self.broker.execute(order, open_prices[order.symbol], timestamp) 

90 fills.append(fill) 

91 

92 pending_signals = still_pending 

93 

94 current_prices.update({bar.symbol: bar.close for bar in timestamp_bars}) 

95 

96 generated_signals: list[Signal] = [] 

97 for bar in timestamp_bars: 

98 generated_signals.extend(self.orchestrator.generate_signals(bar)) 

99 

100 if self.execution_policy == ExecutionPolicy.SAME_CLOSE: 

101 for signal in generated_signals: 

102 order = self._signal_to_order( 

103 signal=signal, 

104 market_prices=current_prices, 

105 timestamp=timestamp, 

106 rejections=rejections, 

107 ) 

108 

109 if order is None: 

110 continue 

111 

112 fill = self.broker.execute( 

113 order=order, 

114 market_price=current_prices[order.symbol], 

115 timestamp=timestamp, 

116 ) 

117 

118 fills.append(fill) 

119 

120 elif self.execution_policy == ExecutionPolicy.NEXT_OPEN: 120 ↛ 124line 120 didn't jump to line 124 because the condition on line 120 was always true

121 pending_signals.extend(generated_signals) 

122 

123 else: 

124 raise ValueError(f"Unsupported execution policy: {self.execution_policy}") 

125 

126 equity_curve.append( 

127 EquityPoint( 

128 timestamp=timestamp, 

129 value=self.broker.portfolio.total_value(current_prices), 

130 invested_value=self.broker.portfolio.positions_value(current_prices), 

131 ), 

132 ) 

133 

134 ending_value = ( 

135 self.broker.portfolio.total_value(current_prices) 

136 if current_prices 

137 else starting_cash 

138 ) 

139 

140 return BacktestResult( 

141 starting_cash=starting_cash, 

142 ending_value=ending_value, 

143 fills=fills, 

144 equity_curve=equity_curve, 

145 rejections=rejections, 

146 periods_per_year=self.periods_per_year, 

147 ) 

148 

149 def _signal_to_order( 

150 self, 

151 *, 

152 signal: Signal, 

153 market_prices: dict[str, float], 

154 timestamp: datetime, 

155 rejections: list[SignalRejection], 

156 ) -> Order | None: 

157 result = self.orchestrator.process_signal( 

158 signal=signal, 

159 portfolio=self.broker.portfolio, 

160 market_prices=market_prices, 

161 fee_rate=self.broker.fee_rate, 

162 slippage_rate=self.broker.slippage_rate, 

163 ) 

164 

165 if isinstance(result, TradingRejection): 

166 rejections.append( 

167 SignalRejection( 

168 timestamp=timestamp, 

169 symbol=result.signal.symbol, 

170 reason=result.reason, 

171 stage=result.stage.value, 

172 ), 

173 ) 

174 return None 

175 

176 if isinstance(result, NoAction): 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true

177 return None 

178 

179 return result