Coverage for src/finbot/cli/backtest.py: 97%

1import argparse 

2import json 

3import logging 

4from datetime import UTC, datetime 

5from pathlib import Path 

6from typing import Any 

7 

8from finbot.backtest.artifacts import ExperimentStore 

9from finbot.backtest.experiment import ExperimentConfig, ExperimentRun 

10from finbot.backtest.models import BacktestResult 

11from finbot.backtest.runner import ExperimentRunner 

12from finbot.config import load_settings 

13from finbot.logger import get_logger 

14from finbot.market.alpaca import AlpacaMarketDataProvider 

15from finbot.market.cache import HistoricalDataCache 

16 

17LINE_LENGTH = 40 

18 

19_DEFAULTS: dict[str, Any] = { 

20 "strategy": "sma_crossover", 

21 "strategy_params": {"short_window": 10, "long_window": 30}, 

22 "symbols": ["NVDA"], 

23 "dataset": "alpaca", 

24 "initial_cash": 10_000.0, 

25 "allocation": 0.995, 

26 "fee_rate": 0.0005, 

27 "slippage_rate": 0.001, 

28} 

29 

30DEFAULT_OUTPUT_DIR = Path(".finbot/experiments") 

31 

32logger = get_logger(__name__) 

33 

34 

35def _format_comparison(rows) -> list[str]: 

36 headers = ("RUN ID", "STRATEGY", "SYMBOLS", "STARTING CASH", "ENDING VALUE", "RETURN", "MAX DD") 

37 

38 values = [ 

39 ( 

40 str(row.run_id), 

41 row.strategy, 

42 ",".join(row.symbols), 

43 f"{row.starting_cash:,.2f}", 

44 f"{row.ending_value:,.2f}", 

45 f"{row.total_return:.2%}", 

46 f"{row.max_drawdown:.2%}", 

47 ) 

48 for row in rows 

49 ] 

50 

51 widths = [ 

52 max(len(headers[index]), *(len(row[index]) for row in values)) 

53 for index in range(len(headers)) 

54 ] 

55 

56 def format_row(row: tuple[str, ...]) -> str: 

57 return " ".join(value.ljust(widths[index]) for index, value in enumerate(row)) 

58 

59 separator = " ".join("-" * width for width in widths) 

60 lines = [format_row(headers), separator, *(format_row(row) for row in values)] 

61 

62 return lines 

63 

64 

65def add_arguments(parser: argparse.ArgumentParser) -> None: 

66 parser.add_argument("--config", type=Path, help="Path to a JSON experiment configuration.") 

67 parser.add_argument("--strategy") 

68 parser.add_argument( 

69 "--strategy-params", 

70 help='Strategy parameters as JSON, e.g. \'{"short_window":10,"long_window":30}\'', 

71 ) 

72 parser.add_argument( 

73 "--symbol", 

74 dest="symbols", 

75 action="append", 

76 help="Symbol to backtest. May be specified multiple times.", 

77 ) 

78 parser.add_argument("--start") 

79 parser.add_argument("--end") 

80 parser.add_argument("--dataset") 

81 parser.add_argument("--initial-cash", type=float) 

82 parser.add_argument("--allocation", type=float) 

83 parser.add_argument("--fee-rate", type=float) 

84 parser.add_argument("--slippage-rate", type=float) 

85 parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR) 

86 

87 

88def add_compare_arguments(parser: argparse.ArgumentParser) -> None: 

89 parser.add_argument("run_ids", nargs="+", help="Experiment run IDs to compare.") 

90 parser.add_argument( 

91 "--output-dir", 

92 type=Path, 

93 default=DEFAULT_OUTPUT_DIR, 

94 help="Directory containing stored experiment runs.", 

95 ) 

96 

97 

98def run_compare(args: argparse.Namespace) -> int: 

99 from finbot.backtest.artifacts import ExperimentStore 

100 from finbot.backtest.comparison import compare_experiments 

101 

102 logger.debug("Loading %d experiment runs from %s", len(args.run_ids), args.output_dir) 

103 

104 store = ExperimentStore(args.output_dir) 

105 runs = store.load_many(args.run_ids, expected_type=ExperimentRun) 

106 

107 logger.debug("Comparing %d experiment runs", len(runs)) 

108 

109 rows = compare_experiments(runs) 

110 

111 logger.info("Backtest comparison") 

112 

113 for row in _format_comparison(rows): 

114 logger.info("%s", row) 

115 

116 return 0 

117 

118 

119def _parse_datetime(value: str) -> datetime: 

120 parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) 

121 

122 if parsed.tzinfo is None: 122 ↛ 125line 122 didn't jump to line 125 because the condition on line 122 was always true

123 parsed = parsed.replace(tzinfo=UTC) 

124 

125 return parsed 

126 

127 

128def _load_config(path: Path) -> dict[str, Any]: 

129 payload = json.loads(path.read_text(encoding="utf-8")) 

130 

131 if not isinstance(payload, dict): 

132 raise ValueError("experiment config must be a JSON object") 

133 

134 return payload 

135 

136 

137def _build_config(args: argparse.Namespace) -> ExperimentConfig: 

138 values = dict(_DEFAULTS) 

139 config_values: dict[str, Any] = {} 

140 

141 if args.config is not None: 

142 config_values = _load_config(args.config) 

143 values.update(config_values) 

144 

145 if args.strategy is not None: 

146 values["strategy"] = args.strategy 

147 

148 if args.strategy_params is not None: 

149 strategy_params: dict[str, Any] = json.loads(args.strategy_params) 

150 

151 if not isinstance(strategy_params, dict): 

152 raise ValueError("strategy_params must be a JSON object") 

153 

154 values["strategy_params"] = strategy_params 

155 elif values["strategy"] != _DEFAULTS["strategy"] and "strategy_params" not in config_values: 

156 values["strategy_params"] = {} 

157 

158 if args.symbols is not None: 

159 values["symbols"] = args.symbols 

160 

161 for name in [ 

162 "start", 

163 "end", 

164 "dataset", 

165 "initial_cash", 

166 "allocation", 

167 "fee_rate", 

168 "slippage_rate", 

169 ]: 

170 value = getattr(args, name) 

171 if value is not None: 

172 values[name] = value 

173 

174 if "start" not in values: 

175 raise ValueError("start must be provided") 

176 

177 if "end" not in values: 

178 raise ValueError("end must be provided") 

179 

180 return ExperimentConfig( 

181 strategy=values["strategy"], 

182 strategy_params=dict(values["strategy_params"]), # type: ignore[arg-type] 

183 symbols=tuple(values["symbols"]), 

184 start=_parse_datetime(values["start"]), 

185 end=_parse_datetime(values["end"]), 

186 dataset=values["dataset"], 

187 initial_cash=float(values["initial_cash"]), 

188 allocation=float(values["allocation"]), 

189 fee_rate=float(values["fee_rate"]), 

190 slippage_rate=float(values["slippage_rate"]), 

191 ) 

192 

193 

194def _create_market_data_provider(config: ExperimentConfig) -> AlpacaMarketDataProvider: 

195 if config.dataset != "alpaca": 

196 raise ValueError(f"Unknown dataset: {config.dataset!r}") 

197 

198 settings = load_settings() 

199 

200 return AlpacaMarketDataProvider( 

201 api_key=settings.alpaca_api_key, 

202 api_secret=settings.alpaca_api_secret, 

203 feed=settings.alpaca_feed, 

204 cache=HistoricalDataCache(), 

205 ) 

206 

207 

208def _log_result(logger: logging.Logger, result: BacktestResult) -> None: 

209 logger.info(f"Starting cash:{f"{result.starting_cash:,.2f}":>{LINE_LENGTH - 14}}") 

210 logger.info(f"Ending value: {f"{result.ending_value:,.2f}":>{LINE_LENGTH - 14}}") 

211 logger.info(f"Return: {f"{result.total_return:.2%}":>{LINE_LENGTH - 14}}") 

212 logger.info(f"{"Stats":^{LINE_LENGTH}}") 

213 logger.info(f"Max drawdown: {f"{result.max_drawdown:.2%}":>{LINE_LENGTH - 14}}") 

214 logger.info(f"Volatility: {f"{result.annualized_volatility():.2%}":>{LINE_LENGTH - 14}}") 

215 logger.info(f"Sharpe ratio: {f"{result.sharpe_ratio():.2f}":>{LINE_LENGTH - 14}}") 

216 logger.info(f"CAGR: {f'{result.cagr:.2%}':>{LINE_LENGTH - 14}}") 

217 logger.info(f"Sortino: {f'{result.sortino_ratio():.2f}':>{LINE_LENGTH - 14}}") 

218 logger.info(f"Calmar: {f'{result.calmar_ratio:.2f}':>{LINE_LENGTH - 14}}") 

219 logger.info(f"Exposure: {f'{result.exposure:.2%}':>{LINE_LENGTH - 14}}") 

220 logger.info(f"Turnover: {f'{result.turnover:.2f}x':>{LINE_LENGTH - 14}}") 

221 logger.info(f"Fees paid: {f'{result.total_fees:,.2f}':>{LINE_LENGTH - 14}}") 

222 logger.info(f"Slippage: {f'{result.total_slippage:,.2f}':>{LINE_LENGTH - 14}}") 

223 

224 if result.trade_count > 0: 

225 logger.info(f"{"Trades":^{LINE_LENGTH}}") 

226 logger.info(f"Trades: {result.trade_count:>{LINE_LENGTH - 14}}") 

227 logger.info(f"Win rate: {f'{result.win_rate:.2%}':>{LINE_LENGTH - 14}}") 

228 logger.info(f"Avg trade: {f'{result.average_trade:.2%}':>{LINE_LENGTH - 14}}") 

229 logger.info(f"Median trade: {f'{result.median_trade:.2%}':>{LINE_LENGTH - 14}}") 

230 logger.info(f"Best trade: {f'{result.best_trade:.2%}':>{LINE_LENGTH - 14}}") 

231 logger.info(f"Worst trade: {f'{result.worst_trade:.2%}':>{LINE_LENGTH - 14}}") 

232 logger.info(f"Avg duration: {str(result.average_trade_duration):>{LINE_LENGTH - 14}}") 

233 

234 if result.benchmark_return is not None: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true

235 logger.info(f"{"Benchmark":^{LINE_LENGTH}}") 

236 logger.info(f"Benchmark: {f"{result.benchmark_return:.2%}":>{LINE_LENGTH - 14}}") 

237 logger.info(f"Excess return:{f"{result.excess_return:.2%}":>{LINE_LENGTH - 14}}") 

238 

239 

240def run(args: argparse.Namespace) -> int: 

241 config = _build_config(args) 

242 market_data = _create_market_data_provider(config) 

243 runner = ExperimentRunner(market_data) 

244 store = ExperimentStore(args.output_dir) 

245 

246 result = runner.run(config) 

247 experiment_run = store.create_run(config, result) 

248 

249 logger.info("Experiment completed: %s", experiment_run.metadata.run_id) 

250 logger.info( 

251 "Artifact: %s", 

252 args.output_dir / str(experiment_run.metadata.run_id) / "metadata.json", 

253 ) 

254 

255 _log_result(logger, result) 

256 

257 return 0