Coverage for src/finbot/cli/main.py: 100%

1import argparse 

2import logging 

3import sys 

4from collections.abc import Sequence 

5 

6from finbot.cli.backtest import add_arguments as add_backtest_arguments 

7from finbot.cli.backtest import add_compare_arguments, run_compare 

8from finbot.cli.backtest import run as run_backtest 

9from finbot.cli.ml_experiment import add_arguments as add_ml_experiment_arguments 

10from finbot.cli.ml_experiment import add_walk_forward_arguments, run_walk_forward 

11from finbot.cli.ml_experiment import run as run_ml_experiment 

12from finbot.cli.paper import add_arguments as add_paper_arguments 

13from finbot.cli.paper import run as run_paper 

14from finbot.logger import configure_logging, get_logger 

15 

16logger = get_logger(__name__) 

17 

18 

19def _test_api(args: argparse.Namespace) -> int: 

20 from finbot.config import load_settings 

21 from finbot.market.alpaca import AlpacaMarketDataProvider 

22 

23 settings = load_settings(api_key=args.api_key, api_secret=args.api_secret) 

24 market_data = AlpacaMarketDataProvider( 

25 api_key=settings.alpaca_api_key, 

26 api_secret=settings.alpaca_api_secret, 

27 feed=settings.alpaca_feed, 

28 ) 

29 

30 bar = market_data.get_latest_bar("NVDA") 

31 logger.info(bar) 

32 

33 return 0 

34 

35 

36def _add_verbose_argument( 

37 parser: argparse.ArgumentParser, 

38 *, 

39 suppress_default: bool = False, 

40) -> None: 

41 kwargs = {} 

42 if suppress_default: 

43 kwargs["default"] = argparse.SUPPRESS 

44 

45 parser.add_argument( 

46 "-v", 

47 "--verbose", 

48 action="store_true", 

49 help="Enable debug logging", 

50 **kwargs, 

51 ) 

52 

53 

54def _build_parser() -> argparse.ArgumentParser: 

55 parser = argparse.ArgumentParser(description="FinBot CLI") 

56 _add_verbose_argument(parser) 

57 subparsers = parser.add_subparsers(dest="command", required=True) 

58 

59 test_api_parser = subparsers.add_parser( 

60 "test_api", 

61 help="Test the configured market data API connection", 

62 ) 

63 _add_verbose_argument(test_api_parser, suppress_default=True) 

64 test_api_parser.add_argument( 

65 "--api-key", 

66 "--api_key", 

67 dest="api_key", 

68 default=None, 

69 help="API key for the market data provider", 

70 ) 

71 test_api_parser.add_argument( 

72 "--api-secret", 

73 "--api_secret", 

74 dest="api_secret", 

75 default=None, 

76 help="API secret for the market data provider", 

77 ) 

78 

79 test_api_parser.set_defaults(func=_test_api) 

80 

81 backtest_parser = subparsers.add_parser( 

82 "backtest", 

83 help="Run or inspect reproducible backtest experiments", 

84 ) 

85 _add_verbose_argument(backtest_parser, suppress_default=True) 

86 add_backtest_arguments(backtest_parser) 

87 backtest_parser.set_defaults(func=run_backtest) 

88 

89 backtest_subparsers = backtest_parser.add_subparsers(dest="backtest_command") 

90 

91 compare_parser = backtest_subparsers.add_parser( 

92 "compare", 

93 help="Compare stored backtest experiments", 

94 ) 

95 _add_verbose_argument(compare_parser, suppress_default=True) 

96 add_compare_arguments(compare_parser) 

97 compare_parser.set_defaults(func=run_compare) 

98 

99 paper_parser = subparsers.add_parser("paper", help="Run the paper-trading runtime") 

100 _add_verbose_argument(paper_parser, suppress_default=True) 

101 add_paper_arguments(paper_parser) 

102 paper_parser.set_defaults(func=run_paper) 

103 

104 ml_parser = subparsers.add_parser( 

105 "ml-experiment", 

106 help="Run a reproducible offline ML experiment and baseline comparison", 

107 ) 

108 _add_verbose_argument(ml_parser, suppress_default=True) 

109 add_ml_experiment_arguments(ml_parser) 

110 ml_parser.set_defaults(func=run_ml_experiment) 

111 

112 ml_walk_forward_parser = subparsers.add_parser( 

113 "ml-walk-forward", 

114 help="Run reproducible fold-by-fold ML training and out-of-sample evaluation", 

115 ) 

116 _add_verbose_argument(ml_walk_forward_parser, suppress_default=True) 

117 add_walk_forward_arguments(ml_walk_forward_parser) 

118 ml_walk_forward_parser.set_defaults(func=run_walk_forward) 

119 

120 return parser 

121 

122 

123def main(argv: Sequence[str] | None = None) -> int: 

124 parser = _build_parser() 

125 args = parser.parse_args(argv) 

126 

127 level = logging.DEBUG if args.verbose else logging.INFO 

128 configure_logging(level) 

129 

130 try: 

131 return args.func(args) 

132 except (ValueError, RuntimeError) as exc: 

133 if args.verbose: 

134 logger.exception("%s", exc) 

135 else: 

136 logger.error("%s", exc) 

137 return 1 

138 

139 

140if __name__ == "__main__": 

141 sys.exit(main())