Coverage for src / monte_neo / monte_carlo / scenarios.py: 98%

1"""Scenario builder module. 

2 

3Handles generation of various Monte Carlo scenarios. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING 

9 

10import pandas as pd 

11 

12from monte_neo.data.sampler import DataSampler 

13from monte_neo.monte_carlo.noise import NoiseInjector 

14from monte_neo.monte_carlo.sensitivity import SensitivityAnalyzer 

15from monte_neo.monte_carlo.shuffler import DataShuffler 

16from monte_neo.monte_carlo.walk_forward import WalkForwardAnalyzer 

17 

18if TYPE_CHECKING: 

19 from monte_neo.monte_carlo.engine import MCConfig 

20 

21 

22class ScenarioBuilder: 

23 """Builder for Monte Carlo scenarios.""" 

24 

25 def __init__(self, config: MCConfig): 

26 """Initialize builder. 

27 

28 Args: 

29 config: Monte Carlo configuration. 

30 """ 

31 self.config = config 

32 self.shuffler = DataShuffler(self.config.random_seed) 

33 self.noise_injector = NoiseInjector(self.config.random_seed) 

34 self.sensitivity = SensitivityAnalyzer() 

35 self.walk_forward = WalkForwardAnalyzer() 

36 self.sampler = DataSampler(self.config.random_seed) 

37 

38 def generate_shuffling(self, data: pd.DataFrame, iterations: int) -> list[pd.DataFrame]: 

39 """Generate shuffling scenarios.""" 

40 scenarios = [] 

41 scenarios.extend(self.shuffler.shuffle_returns(data, iterations // 2)) 

42 scenarios.extend(self.shuffler.shuffle_blocks(data, iterations // 2)) 

43 return scenarios 

44 

45 def generate_noise(self, data: pd.DataFrame, iterations: int) -> list[pd.DataFrame]: 

46 """Generate noise scenarios.""" 

47 return self.noise_injector.add_noise(data, iterations) 

48 

49 def generate_walk_forward(self, data: pd.DataFrame) -> list[pd.DataFrame]: 

50 """Generate walk-forward scenarios.""" 

51 return self.walk_forward.generate_scenarios( 

52 data, 

53 n_splits=self.config.walk_forward_splits, 

54 ) 

55 

56 def generate_block_bootstrap(self, data: pd.DataFrame, iterations: int) -> list[pd.DataFrame]: 

57 """Generate block bootstrap scenarios.""" 

58 return self.sampler.block_bootstrap(data, n_samples=iterations) 

59 

60 def generate(self, data: pd.DataFrame) -> list[pd.DataFrame]: 

61 """Generate all test scenarios. 

62 

63 Args: 

64 data: Base OHLCV data. 

65 

66 Returns: 

67 List of scenario DataFrames. 

68 """ 

69 scenarios = [data] # Original data 

70 

71 # Shuffled data 

72 if self.config.use_shuffling: 

73 scenarios.extend( 

74 self.shuffler.shuffle_returns(data, self.config.iterations // 4) 

75 ) 

76 scenarios.extend( 

77 self.shuffler.shuffle_blocks(data, self.config.iterations // 4) 

78 ) 

79 

80 # Noisy data 

81 if self.config.use_noise: 

82 scenarios.extend( 

83 self.noise_injector.add_noise(data, self.config.iterations // 4) 

84 ) 

85 

86 # Walk-forward scenarios 

87 if self.config.use_walk_forward: 

88 wf_scenarios = self.walk_forward.generate_scenarios( 

89 data, 

90 n_splits=self.config.walk_forward_splits, 

91 ) 

92 scenarios.extend(wf_scenarios) 

93 

94 # Block Bootstrap scenarios 

95 if self.config.use_block_bootstrap: 

96 # Distribute iterations among enabled methods 

97 n_methods = sum( 

98 [ 

99 self.config.use_shuffling, 

100 self.config.use_noise, 

101 self.config.use_sensitivity, 

102 self.config.use_block_bootstrap, 

103 ] 

104 ) 

105 n_per_method = self.config.iterations // max(1, n_methods) 

106 

107 bb_samples = self.sampler.block_bootstrap(data, n_samples=n_per_method) 

108 scenarios.extend(bb_samples) 

109 

110 # Limit total scenarios 

111 if len(scenarios) > self.config.iterations: 

112 scenarios = scenarios[: self.config.iterations] 

113 

114 return scenarios