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

1from datetime import datetime 

2from itertools import groupby 

3 

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

5from finbot.execution.broker import SimulatedBroker 

6from finbot.execution.fill import Fill 

7from finbot.execution.order import Order, OrderSide 

8from finbot.execution.policy import ExecutionPolicy 

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.base import Strategy 

14from finbot.strategy.signal import Signal, SignalDirection 

15 

16 

17class BacktestEngine: 

18 def __init__( 

19 self, 

20 broker: SimulatedBroker, 

21 strategy: Strategy, 

22 execution_policy: ExecutionPolicy = ExecutionPolicy.NEXT_OPEN, 

23 risk_manager: RiskManager | None = None, 

24 position_sizer: PositionSizer | None = None, 

25 periods_per_year: int = 252, 

26 ): 

27 if periods_per_year <= 0: 

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

29 

30 self.broker = broker 

31 self.strategy = strategy 

32 self.execution_policy = execution_policy 

33 self.risk_manager = risk_manager or RiskManager(RiskLimits()) 

34 self.position_sizer = position_sizer or PositionSizer() 

35 self.periods_per_year = periods_per_year 

36 self._has_run = False 

37 

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

39 if self._has_run: 

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

41 

42 if self.broker.portfolio.positions: 

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

44 

45 self._has_run = True 

46 

47 bars = sorted( 

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

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

50 ) 

51 

52 starting_cash = self.broker.portfolio.cash 

53 fills: list[Fill] = [] 

54 equity_curve: list[EquityPoint] = [] 

55 rejections: list[SignalRejection] = [] 

56 

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

58 pending_signals: list[Signal] = [] 

59 

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

61 timestamp_bars = list(grouped_bars) 

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

63 open_prices = dict(current_prices) 

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

65 

66 if self.execution_policy == ExecutionPolicy.NEXT_OPEN: 

67 still_pending: list[Signal] = [] 

68 

69 for signal in pending_signals: 

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

71 still_pending.append(signal) 

72 continue 

73 

74 order = self._signal_to_order( 

75 signal=signal, 

76 market_prices=open_prices, 

77 timestamp=timestamp, 

78 rejections=rejections, 

79 ) 

80 if order is None: continue 80 ↛ 69line 80 didn't jump to line 69 because the continue on line 80 wasn't executed

81 

82 fill = self.broker.execute( 

83 order=order, 

84 market_price=open_prices[order.symbol], 

85 timestamp=timestamp, 

86 ) 

87 fills.append(fill) 

88 

89 pending_signals = still_pending 

90 

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

92 

93 generated_signals: list[Signal] = [] 

94 

95 for bar in timestamp_bars: 

96 result = self.strategy.on_bar(bar=bar) 

97 if result is None: continue 

98 if isinstance(result, Signal): 

99 generated_signals.append(result) 

100 else: 

101 generated_signals.extend(result) 

102 

103 if self.execution_policy == ExecutionPolicy.SAME_CLOSE: 

104 for signal in generated_signals: 

105 order = self._signal_to_order( 

106 signal=signal, 

107 market_prices=current_prices, 

108 timestamp=timestamp, 

109 rejections=rejections, 

110 ) 

111 

112 if order is None: 

113 continue 

114 

115 fill = self.broker.execute( 

116 order=order, 

117 market_price=current_prices[order.symbol], 

118 timestamp=timestamp, 

119 ) 

120 

121 fills.append(fill) 

122 

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

124 pending_signals.extend(generated_signals) 

125 

126 else: 

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

128 

129 equity_curve.append( 

130 EquityPoint( 

131 timestamp=timestamp, 

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

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

134 ), 

135 ) 

136 

137 ending_value = ( 

138 self.broker.portfolio.total_value(current_prices) 

139 if current_prices 

140 else starting_cash 

141 ) 

142 

143 return BacktestResult( 

144 starting_cash=starting_cash, 

145 ending_value=ending_value, 

146 fills=fills, 

147 equity_curve=equity_curve, 

148 rejections=rejections, 

149 periods_per_year=self.periods_per_year, 

150 ) 

151 

152 def _signal_to_order( 

153 self, 

154 *, 

155 signal: Signal, 

156 market_prices: dict[str, float], 

157 timestamp: datetime, 

158 rejections: list[SignalRejection], 

159 ) -> Order | None: 

160 risk_decision = self.risk_manager.assess( 

161 signal=signal, 

162 portfolio=self.broker.portfolio, 

163 market_prices=market_prices, 

164 ) 

165 

166 if not risk_decision.accepted: 

167 rejections.append( 

168 SignalRejection( 

169 timestamp=timestamp, 

170 symbol=signal.symbol, 

171 reason=risk_decision.reason or "Risk manager rejected signal.", 

172 stage="risk", 

173 ), 

174 ) 

175 return None 

176 

177 if risk_decision.order_value_budget is None: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true

178 raise RuntimeError("Accepted risk decision is missing an order value budget.") 

179 

180 if signal.direction == SignalDirection.LONG: 180 ↛ 184line 180 didn't jump to line 184 because the condition on line 180 was always true

181 order_side = OrderSide.BUY 

182 max_quantity = None 

183 

184 elif signal.direction == SignalDirection.SHORT: 

185 order_side = OrderSide.SELL 

186 position = self.broker.portfolio.positions.get(signal.symbol) 

187 max_quantity = None if position is None else position.quantity 

188 

189 elif signal.direction == SignalDirection.FLAT: 

190 return None 

191 

192 else: 

193 raise ValueError(f"Unsupported signal direction: {signal.direction}") 

194 

195 sizing_decision = self.position_sizer.size( 

196 order_side=order_side, 

197 market_price=market_prices[signal.symbol], 

198 order_value_budget=risk_decision.order_value_budget, 

199 fee_rate=self.broker.fee_rate, 

200 slippage_rate=self.broker.slippage_rate, 

201 max_quantity=max_quantity, 

202 ) 

203 

204 if not sizing_decision.accepted: 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true

205 rejections.append( 

206 SignalRejection( 

207 timestamp=timestamp, symbol=signal.symbol, 

208 reason=sizing_decision.reason or "Position sizer rejected signal.", 

209 stage="sizing", 

210 ), 

211 ) 

212 return None 

213 

214 if sizing_decision.quantity is None: 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true

215 raise RuntimeError("Accepted sizing decision is missing a quantity.") 

216 

217 return Order(symbol=signal.symbol, side=order_side, quantity=sizing_decision.quantity)