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.execution.models import Fill 

8 

9 

10@dataclass(frozen=True) 

11class Trade: 

12 symbol: str 

13 

14 entry_time: datetime 

15 exit_time: datetime 

16 

17 quantity: float 

18 

19 entry_price: float 

20 exit_price: float 

21 

22 entry_fee: float = 0.0 

23 exit_fee: float = 0.0 

24 

25 @property 

26 def gross_pnl(self) -> float: 

27 return (self.exit_price - self.entry_price) * self.quantity 

28 

29 @property 

30 def fees(self) -> float: 

31 return self.entry_fee + self.exit_fee 

32 

33 @property 

34 def net_pnl(self) -> float: 

35 return self.gross_pnl - self.fees 

36 

37 @property 

38 def pnl(self) -> float: 

39 """net trade PnL, including entry and exit fees""" 

40 return self.net_pnl 

41 

42 @property 

43 def return_pct(self) -> float: 

44 return (self.exit_price - self.entry_price) / self.entry_price 

45 

46 @property 

47 def duration(self) -> timedelta: 

48 return self.exit_time - self.entry_time 

49 

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

51 return { 

52 "symbol": self.symbol, 

53 "entry_time": self.entry_time.isoformat(), 

54 "exit_time": self.exit_time.isoformat(), 

55 "quantity": self.quantity, 

56 "entry_price": self.entry_price, 

57 "exit_price": self.exit_price, 

58 "entry_fee": self.entry_fee, 

59 "exit_fee": self.exit_fee, 

60 "gross_pnl": self.gross_pnl, 

61 "fees": self.fees, 

62 "net_pnl": self.net_pnl, 

63 "return_pct": self.return_pct, 

64 "duration_seconds": self.duration.total_seconds(), 

65 } 

66 

67 

68def trades_from_fills(fills: list[Fill]) -> list[Trade]: 

69 trades: list[Trade] = [] 

70 open_fills: dict[str, Fill] = {} 

71 

72 for fill in fills: 

73 if fill.side == "buy": 

74 if fill.symbol in open_fills: 

75 raise ValueError(f"Already have an open trade for {fill.symbol}") 

76 open_fills[fill.symbol] = fill 

77 

78 if fill.side == "sell": 

79 if fill.symbol not in open_fills: 

80 raise ValueError(f"No open trade for {fill.symbol} to close") 

81 

82 entry_fill = open_fills[fill.symbol] 

83 trade = Trade( 

84 symbol=fill.symbol, 

85 entry_time=entry_fill.timestamp, 

86 exit_time=fill.timestamp, 

87 quantity=entry_fill.quantity, 

88 entry_price=entry_fill.price, 

89 exit_price=fill.price, 

90 entry_fee=entry_fill.fee, 

91 exit_fee=fill.fee, 

92 ) 

93 trades.append(trade) 

94 del open_fills[fill.symbol] 

95 

96 return trades 

97 

98 

99@dataclass(frozen=True) 

100class EquityPoint: 

101 timestamp: datetime 

102 value: float 

103 invested_value: float = 0.0 

104 

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

106 return { 

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

108 "value": self.value, 

109 "invested_value": self.invested_value, 

110 } 

111 

112 @classmethod 

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

114 return cls( 

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

116 value=data["value"], 

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

118 ) 

119 

120 

121@dataclass(frozen=True) 

122class SignalRejection: 

123 timestamp: datetime 

124 symbol: str 

125 reason: str 

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

127 

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

129 return { 

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

131 "symbol": self.symbol, 

132 "reason": self.reason, 

133 "stage": self.stage, 

134 } 

135 

136 @classmethod 

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

138 return cls( 

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

140 symbol=data["symbol"], 

141 reason=data["reason"], 

142 stage=data["stage"], 

143 ) 

144 

145 

146@dataclass(frozen=True) 

147class BacktestResult: 

148 starting_cash: float 

149 ending_value: float 

150 equity_curve: list[EquityPoint] 

151 fills: list[Fill] 

152 rejections: list[SignalRejection] 

153 periods_per_year: int = 252 

154 benchmark_return: float | None = None 

155 

156 @property 

157 def pnl(self) -> float: 

158 return self.ending_value - self.starting_cash 

159 

160 @property 

161 def total_return(self) -> float: 

162 if self.starting_cash == 0: 

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

164 

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

166 

167 @property 

168 def max_drawdown(self) -> float: 

169 if not self.equity_curve: 

170 return 0.0 

171 

172 peak = self.starting_cash 

173 max_drawdown = 0.0 

174 

175 for point in self.equity_curve: 

176 peak = max(peak, point.value) 

177 if peak <= 0: 

178 continue 

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

180 max_drawdown = max(max_drawdown, drawdown) 

181 

182 return max_drawdown 

183 

184 @property 

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

186 return trades_from_fills(self.fills) 

187 

188 @property 

189 def trade_count(self) -> int: 

190 return len(self.trades) 

191 

192 @property 

193 def winning_trades(self) -> int: 

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

195 

196 @property 

197 def losing_trades(self) -> int: 

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

199 

200 @property 

201 def win_rate(self) -> float: 

202 if not self.trades: 

203 return 0.0 

204 return self.winning_trades / self.trade_count 

205 

206 @property 

207 def profit_factor(self) -> float: 

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

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

210 

211 if gross_loss > 0: 

212 return gross_profit / gross_loss 

213 if gross_profit > 0: 

214 return float("inf") 

215 

216 return 0.0 

217 

218 @property 

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

220 if len(self.equity_curve) < 2: 

221 return [] 

222 

223 return [ 

224 current.value / previous.value - 1 

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

226 ] 

227 

228 @property 

229 def average_trade(self) -> float: 

230 if not self.trades: 

231 return 0.0 

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

233 

234 @property 

235 def median_trade(self) -> float: 

236 if not self.trades: 

237 return 0.0 

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

239 

240 @property 

241 def best_trade(self) -> float: 

242 if not self.trades: 

243 return 0.0 

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

245 

246 @property 

247 def worst_trade(self) -> float: 

248 if not self.trades: 

249 return 0.0 

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

251 

252 @property 

253 def average_trade_duration(self) -> timedelta: 

254 if not self.trades: 

255 return timedelta(0) 

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

257 return total / self.trade_count 

258 

259 @property 

260 def cagr(self) -> float: 

261 if len(self.equity_curve) < 2: 

262 return 0.0 

263 

264 if self.starting_cash <= 0: 

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

266 

267 if self.ending_value <= 0: 

268 return -1.0 

269 

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

271 if duration.total_seconds() <= 0: 

272 return 0.0 

273 

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

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

276 

277 @property 

278 def calmar_ratio(self) -> float: 

279 max_drawdown = self.max_drawdown 

280 cagr = self.cagr 

281 

282 if max_drawdown == 0: 

283 if cagr > 0: 

284 return float("inf") 

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

286 return float("-inf") 

287 return 0.0 

288 

289 return cagr / max_drawdown 

290 

291 @property 

292 def total_fees(self) -> float: 

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

294 

295 @property 

296 def total_slippage(self) -> float: 

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

298 

299 @property 

300 def turnover(self) -> float: 

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

302 

303 if traded_notional == 0: 

304 return 0.0 

305 

306 if self.equity_curve: 

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

308 else: 

309 average_equity = self.starting_cash 

310 

311 if average_equity <= 0: 311 ↛ 312line 311 didn't jump to line 312 because the condition on line 311 was never true

312 return 0.0 

313 

314 return traded_notional / average_equity 

315 

316 @property 

317 def exposure(self) -> float: 

318 if not self.equity_curve: 

319 return 0.0 

320 

321 exposures = [ 

322 point.invested_value / point.value 

323 if point.value > 0 

324 else 0.0 

325 for point in self.equity_curve 

326 ] 

327 

328 return mean(exposures) 

329 

330 @property 

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

332 if self.benchmark_return is None: 

333 return None 

334 

335 return self.total_return - self.benchmark_return 

336 

337 @staticmethod 

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

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

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

341 

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

343 

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

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

346 

347 if periods <= 0: 

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

349 

350 returns = self.returns 

351 

352 if len(returns) < 2: 

353 return 0.0 

354 

355 return stdev(returns) * sqrt(periods) 

356 

357 def sharpe_ratio( 

358 self, 

359 risk_free_rate: float = 0.0, 

360 periods_per_year: int | None = None, 

361 ) -> float: 

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

363 

364 if periods <= 0: 

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

366 

367 returns = self.returns 

368 

369 if len(returns) < 2: 

370 return 0.0 

371 

372 volatility = stdev(returns) 

373 

374 if volatility == 0: 

375 return 0.0 

376 

377 risk_free_rate_per_period = self._risk_free_rate_per_period( 

378 risk_free_rate, 

379 periods, 

380 ) 

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

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

383 

384 def sortino_ratio( 

385 self, 

386 risk_free_rate: float = 0.0, 

387 periods_per_year: int | None = None, 

388 ) -> float: 

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

390 

391 if periods <= 0: 

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

393 

394 returns = self.returns 

395 

396 if not returns: 

397 return 0.0 

398 

399 risk_free_rate_per_period = self._risk_free_rate_per_period( 

400 risk_free_rate, 

401 periods, 

402 ) 

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

404 

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

406 downside_deviation = sqrt(mean(downside_squared)) 

407 

408 if downside_deviation == 0: 

409 average_excess_return = mean(excess_returns) 

410 

411 if average_excess_return > 0: 

412 return float("inf") 

413 return 0.0 

414 

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