Coverage for src/finbot/backtest/models.py: 98%

1from dataclasses import dataclass 

2from datetime import datetime, timedelta 

3from math import sqrt 

4from statistics import mean, median, stdev 

5from typing import Any 

6 

7from finbot.backtest.trades import Trade, trades_from_fills 

8from finbot.execution.fill import Fill 

9 

10 

11@dataclass(frozen=True) 

12class EquityPoint: 

13 timestamp: datetime 

14 value: float 

15 invested_value: float = 0.0 

16 

17 def serialize(self) -> dict[str, Any]: 

18 return { 

19 "timestamp": self.timestamp.isoformat(), 

20 "value": self.value, 

21 "invested_value": self.invested_value, 

22 } 

23 

24 @classmethod 

25 def deserialize(cls, data: dict[str, Any]) -> "EquityPoint": 

26 return cls( 

27 timestamp=datetime.fromisoformat(data["timestamp"]), 

28 value=data["value"], 

29 invested_value=data.get("invested_value", 0.0), 

30 ) 

31 

32 

33@dataclass(frozen=True) 

34class SignalRejection: 

35 timestamp: datetime 

36 symbol: str 

37 reason: str 

38 stage: str # Will be either "risk" or "sizing" for now 

39 

40 def serialize(self) -> dict[str, str]: # pragma: no cover 

41 return { 

42 "timestamp": self.timestamp.isoformat(), 

43 "symbol": self.symbol, 

44 "reason": self.reason, 

45 "stage": self.stage, 

46 } 

47 

48 @classmethod 

49 def deserialize(cls, data: dict[str, str]) -> "SignalRejection": 

50 return cls( 

51 timestamp=datetime.fromisoformat(data["timestamp"]), 

52 symbol=data["symbol"], 

53 reason=data["reason"], 

54 stage=data["stage"], 

55 ) 

56 

57 

58@dataclass(frozen=True) 

59class BacktestResult: 

60 starting_cash: float 

61 ending_value: float 

62 equity_curve: list[EquityPoint] 

63 fills: list[Fill] 

64 rejections: list[SignalRejection] 

65 periods_per_year: int = 252 

66 benchmark_return: float | None = None 

67 

68 @property 

69 def pnl(self) -> float: 

70 return self.ending_value - self.starting_cash 

71 

72 @property 

73 def total_return(self) -> float: 

74 if self.starting_cash == 0: 

75 raise ValueError("Starting cash must be greater than 0") 

76 

77 return (self.ending_value - self.starting_cash) / self.starting_cash 

78 

79 @property 

80 def max_drawdown(self) -> float: 

81 if not self.equity_curve: 

82 return 0.0 

83 

84 peak = self.starting_cash 

85 max_drawdown = 0.0 

86 

87 for point in self.equity_curve: 

88 peak = max(peak, point.value) 

89 if peak <= 0: continue 

90 drawdown = (peak - point.value) / peak 

91 max_drawdown = max(max_drawdown, drawdown) 

92 

93 return max_drawdown 

94 

95 @property 

96 def trades(self) -> list[Trade]: 

97 return trades_from_fills(self.fills) 

98 

99 @property 

100 def trade_count(self) -> int: 

101 return len(self.trades) 

102 

103 @property 

104 def winning_trades(self) -> int: 

105 return sum(trade.net_pnl > 0 for trade in self.trades) 

106 

107 @property 

108 def losing_trades(self) -> int: 

109 return sum(trade.net_pnl < 0 for trade in self.trades) 

110 

111 @property 

112 def win_rate(self) -> float: 

113 if not self.trades: 

114 return 0.0 

115 return self.winning_trades / self.trade_count 

116 

117 @property 

118 def profit_factor(self) -> float: 

119 gross_profit = sum(trade.net_pnl for trade in self.trades if trade.net_pnl > 0) 

120 gross_loss = abs(sum(trade.net_pnl for trade in self.trades if trade.net_pnl < 0)) 

121 

122 return gross_profit / gross_loss if gross_loss > 0 else float("inf") 

123 

124 @property 

125 def returns(self) -> list[float]: 

126 if len(self.equity_curve) < 2: 

127 return [] 

128 

129 return [ 

130 current.value / previous.value - 1 

131 for previous, current in zip(self.equity_curve, self.equity_curve[1:]) 

132 ] 

133 

134 @property 

135 def average_trade(self) -> float: 

136 if not self.trades: return 0.0 

137 return mean(trade.return_pct for trade in self.trades) 

138 

139 @property 

140 def median_trade(self) -> float: 

141 if not self.trades: return 0.0 

142 return median(trade.return_pct for trade in self.trades) 

143 

144 @property 

145 def best_trade(self) -> float: 

146 if not self.trades: return 0.0 

147 return max(trade.return_pct for trade in self.trades) 

148 

149 @property 

150 def worst_trade(self) -> float: 

151 if not self.trades: return 0.0 

152 return min(trade.return_pct for trade in self.trades) 

153 

154 @property 

155 def average_trade_duration(self) -> timedelta: 

156 if not self.trades: return timedelta(0) 

157 total = sum((trade.duration for trade in self.trades), start=timedelta(0)) 

158 return total / self.trade_count 

159 

160 @property 

161 def cagr(self) -> float: 

162 if len(self.equity_curve) < 2: 

163 return 0.0 

164 

165 if self.starting_cash <= 0: 

166 raise ValueError("Starting cash must be greater than 0") 

167 

168 if self.ending_value <= 0: 

169 return -1.0 

170 

171 duration = self.equity_curve[-1].timestamp - self.equity_curve[0].timestamp 

172 if duration.total_seconds() <= 0: 

173 return 0.0 

174 

175 years = duration.total_seconds() / (365.25 * 24 * 60 * 60) 

176 return (self.ending_value / self.starting_cash) ** (1 / years) - 1 

177 

178 @property 

179 def calmar_ratio(self) -> float: 

180 max_drawdown = self.max_drawdown 

181 cagr = self.cagr 

182 

183 if max_drawdown == 0: 

184 if cagr > 0: 

185 return float("inf") 

186 if cagr < 0: 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true

187 return float("-inf") 

188 return 0.0 

189 

190 return cagr / max_drawdown 

191 

192 @property 

193 def total_fees(self) -> float: 

194 return sum(fill.fee for fill in self.fills) 

195 

196 @property 

197 def total_slippage(self) -> float: 

198 return sum(fill.slippage_cost for fill in self.fills) 

199 

200 @property 

201 def turnover(self) -> float: 

202 traded_notional = sum(fill.price * fill.quantity for fill in self.fills) 

203 

204 if traded_notional == 0: return 0.0 

205 

206 if self.equity_curve: 

207 average_equity = mean(point.value for point in self.equity_curve) 

208 else: 

209 average_equity = self.starting_cash 

210 

211 if average_equity <= 0: return 0.0 211 ↛ exitline 211 didn't return from function 'turnover' because the return on line 211 wasn't executed

212 

213 return traded_notional / average_equity 

214 

215 @property 

216 def exposure(self) -> float: 

217 if not self.equity_curve: return 0.0 

218 

219 exposures = [ 

220 point.invested_value / point.value 

221 if point.value > 0 

222 else 0.0 

223 for point in self.equity_curve 

224 ] 

225 

226 return mean(exposures) 

227 

228 @property 

229 def excess_return(self) -> float | None: 

230 if self.benchmark_return is None: 

231 return None 

232 

233 return self.total_return - self.benchmark_return 

234 

235 @staticmethod 

236 def _risk_free_rate_per_period(risk_free_rate: float, periods_per_year: int) -> float: 

237 if periods_per_year <= 0: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true

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

239 

240 return (1 + risk_free_rate) ** (1 / periods_per_year) - 1 

241 

242 def annualized_volatility(self, periods_per_year: int | None = None) -> float: 

243 periods = self.periods_per_year if periods_per_year is None else periods_per_year 

244 

245 if periods <= 0: 

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

247 

248 returns = self.returns 

249 

250 if len(returns) < 2: 

251 return 0.0 

252 

253 return stdev(returns) * sqrt(periods) 

254 

255 def sharpe_ratio( 

256 self, 

257 risk_free_rate: float = 0.0, 

258 periods_per_year: int | None = None, 

259 ) -> float: 

260 periods = self.periods_per_year if periods_per_year is None else periods_per_year 

261 

262 if periods <= 0: 

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

264 

265 returns = self.returns 

266 

267 if len(returns) < 2: 

268 return 0.0 

269 

270 volatility = stdev(returns) 

271 

272 if volatility == 0: 

273 return 0.0 

274 

275 risk_free_rate_per_period = self._risk_free_rate_per_period( 

276 risk_free_rate, 

277 periods, 

278 ) 

279 excess_returns = [r - risk_free_rate_per_period for r in returns] 

280 return mean(excess_returns) / volatility * sqrt(periods) 

281 

282 def sortino_ratio( 

283 self, 

284 risk_free_rate: float = 0.0, 

285 periods_per_year: int | None = None, 

286 ) -> float: 

287 periods = self.periods_per_year if periods_per_year is None else periods_per_year 

288 

289 if periods <= 0: 

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

291 

292 returns = self.returns 

293 

294 if not returns: 

295 return 0.0 

296 

297 risk_free_rate_per_period = self._risk_free_rate_per_period( 

298 risk_free_rate, 

299 periods, 

300 ) 

301 excess_returns = [r - risk_free_rate_per_period for r in returns] 

302 

303 downside_squared = [min(excess_return, 0.0) ** 2 for excess_return in excess_returns] 

304 downside_deviation = sqrt(mean(downside_squared)) 

305 

306 if downside_deviation == 0: 

307 average_excess_return = mean(excess_returns) 

308 

309 if average_excess_return > 0: 

310 return float("inf") 

311 return 0.0 

312 

313 return mean(excess_returns) / downside_deviation * sqrt(periods)