Coverage for src/finbot/cli/ml_experiment.py: 86%

1import argparse 

2import json 

3import subprocess 

4from pathlib import Path 

5 

6from finbot.logger import get_logger 

7from finbot.market.cache import DEFAULT_CACHE_DIR, HistoricalDataCache 

8from finbot.ml import ( 

9 MLExperimentConfig, 

10 MLExperimentRunner, 

11 MLExperimentStore, 

12 MLWalkForwardConfig, 

13 MLWalkForwardRunner, 

14 MLWalkForwardStore, 

15) 

16 

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

18logger = get_logger(__name__) 

19 

20 

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

22 parser.add_argument("--config", type=Path, required=True, help="Path to an ML experiment JSON configuration.") 

23 parser.add_argument("--cache-dir", type=Path, default=DEFAULT_CACHE_DIR, help="Historical dataset cache directory.") 

24 parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR, help="ML experiment artifact directory.") 

25 

26 

27def add_walk_forward_arguments(parser: argparse.ArgumentParser) -> None: 

28 parser.add_argument( 

29 "--config", 

30 type=Path, 

31 required=True, 

32 help="Path to an ML walk-forward JSON configuration.", 

33 ) 

34 parser.add_argument( 

35 "--cache-dir", 

36 type=Path, 

37 default=DEFAULT_CACHE_DIR, 

38 help="Historical dataset cache directory.", 

39 ) 

40 parser.add_argument( 

41 "--output-dir", 

42 type=Path, 

43 default=DEFAULT_OUTPUT_DIR, 

44 help="ML experiment artifact directory.", 

45 ) 

46 

47 

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

49 config = _load_config(args.config) 

50 store = MLExperimentStore(args.output_dir) 

51 experiment = store.execute( 

52 config, 

53 MLExperimentRunner(HistoricalDataCache(args.cache_dir)), 

54 git_commit=_git_revision(), 

55 ) 

56 logger.info("ML experiment completed: %s", experiment.run_id) 

57 logger.info("Artifacts: %s", args.output_dir / str(experiment.run_id)) 

58 return 0 

59 

60 

61def run_walk_forward(args: argparse.Namespace) -> int: 

62 config = _load_walk_forward_config(args.config) 

63 store = MLWalkForwardStore(args.output_dir) 

64 experiment = store.execute( 

65 config.experiment, 

66 config.splits, 

67 MLWalkForwardRunner(HistoricalDataCache(args.cache_dir)), 

68 git_commit=_git_revision(), 

69 ) 

70 logger.info("ML walk-forward experiment completed: %s", experiment.run_id) 

71 logger.info("Artifacts: %s", args.output_dir / str(experiment.run_id)) 

72 return 0 

73 

74 

75def _load_config(path: Path) -> MLExperimentConfig: 

76 return MLExperimentConfig.from_dict(_load_json_object(path)) 

77 

78 

79def _load_walk_forward_config(path: Path) -> MLWalkForwardConfig: 

80 return MLWalkForwardConfig.from_dict(_load_json_object(path)) 

81 

82 

83def _load_json_object(path: Path) -> dict[str, object]: 

84 try: 

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

86 except (OSError, json.JSONDecodeError) as exc: 

87 raise ValueError(f"Unable to read ML experiment configuration: {exc}") from exc 

88 if not isinstance(payload, dict): 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true

89 raise ValueError("ML experiment configuration must be a JSON object.") 

90 return payload 

91 

92 

93def _git_revision() -> str: 

94 try: 

95 completed = subprocess.run( 

96 ["git", "rev-parse", "HEAD"], 

97 check=True, 

98 capture_output=True, 

99 text=True, 

100 ) 

101 except (OSError, subprocess.CalledProcessError) as exc: 

102 raise RuntimeError("Unable to determine the current Git revision.") from exc 

103 revision = completed.stdout.strip() 

104 if not revision: 

105 raise RuntimeError("Unable to determine the current Git revision.") 

106 return revision