Coverage for src / monte_neo / monte_carlo / noise.py: 74%

1"""Noise injection module. 

2 

3Adds various types of noise to test indicator robustness. 

4""" 

5 

6from __future__ import annotations 

7 

8import numpy as np 

9import pandas as pd 

10 

11from monte_neo.utils.logger import get_logger 

12 

13logger = get_logger(__name__) 

14 

15 

16class NoiseInjector: 

17 """Noise injection for robustness testing.""" 

18 

19 def __init__(self, random_seed: int | None = None) -> None: 

20 """Initialize noise injector. 

21 

22 Args: 

23 random_seed: Random seed for reproducibility. 

24 """ 

25 self.rng = np.random.default_rng(random_seed) 

26 

27 def add_noise( 

28 self, 

29 data: pd.DataFrame, 

30 n_samples: int = 100, 

31 noise_level: float = 0.001, 

32 ) -> list[pd.DataFrame]: 

33 """Add Gaussian noise to price data. 

34 

35 Args: 

36 data: OHLCV DataFrame. 

37 n_samples: Number of noisy samples. 

38 noise_level: Standard deviation as fraction of price. 

39 

40 Returns: 

41 List of noisy DataFrames. 

42 """ 

43 samples = [] 

44 

45 for _ in range(n_samples): 

46 sample = data.copy() 

47 

48 for col in ["open", "high", "low", "close"]: 

49 noise = self.rng.normal(0, noise_level, len(data)) 

50 sample[col] = sample[col] * (1 + noise) 

51 

52 # Ensure OHLC consistency 

53 sample = self._fix_ohlc(sample) 

54 samples.append(sample) 

55 

56 logger.debug(f"Generated {n_samples} Gaussian noise samples") 

57 return samples 

58 

59 def add_slippage( 

60 self, 

61 data: pd.DataFrame, 

62 n_samples: int = 100, 

63 slippage_bps: float = 5.0, 

64 ) -> list[pd.DataFrame]: 

65 """Simulate price slippage. 

66 

67 Args: 

68 data: OHLCV DataFrame. 

69 n_samples: Number of samples. 

70 slippage_bps: Slippage in basis points. 

71 

72 Returns: 

73 List of DataFrames with slippage. 

74 """ 

75 samples = [] 

76 slippage_pct = slippage_bps / 10000 

77 

78 for _ in range(n_samples): 

79 sample = data.copy() 

80 

81 # Random slippage direction and magnitude 

82 slippage = self.rng.uniform(-slippage_pct, slippage_pct, len(data)) 

83 

84 # Apply to all prices 

85 for col in ["open", "high", "low", "close"]: 

86 sample[col] = sample[col] * (1 + slippage) 

87 

88 sample = self._fix_ohlc(sample) 

89 samples.append(sample) 

90 

91 logger.debug(f"Generated {n_samples} slippage samples ({slippage_bps} bps)") 

92 return samples 

93 

94 def add_spread_variation( 

95 self, 

96 data: pd.DataFrame, 

97 n_samples: int = 100, 

98 base_spread_bps: float = 2.0, 

99 volatility_multiplier: float = 2.0, 

100 ) -> list[pd.DataFrame]: 

101 """Add variable spread simulation. 

102 

103 Args: 

104 data: OHLCV DataFrame. 

105 n_samples: Number of samples. 

106 base_spread_bps: Base spread in basis points. 

107 volatility_multiplier: Spread increase during high volatility. 

108 

109 Returns: 

110 List of DataFrames with spread effects. 

111 """ 

112 samples = [] 

113 base_spread = base_spread_bps / 10000 

114 

115 # Estimate volatility 

116 returns = data["close"].pct_change().fillna(0) 

117 rolling_vol = returns.rolling(20, min_periods=1).std() 

118 normalized_vol = rolling_vol / rolling_vol.mean() 

119 

120 for _ in range(n_samples): 

121 sample = data.copy() 

122 

123 # Variable spread based on volatility 

124 spread = base_spread * (1 + (normalized_vol - 1) * volatility_multiplier) 

125 spread = spread.clip(base_spread, base_spread * 10) # Cap at 10x 

126 

127 # Apply to open/close (entry/exit simulation) 

128 direction = self.rng.choice([-1, 1], len(data)) 

129 sample["close"] = sample["close"] * (1 + spread.values * direction * 0.5) 

130 

131 sample = self._fix_ohlc(sample) 

132 samples.append(sample) 

133 

134 logger.debug(f"Generated {n_samples} spread variation samples") 

135 return samples 

136 

137 def add_gap_noise( 

138 self, 

139 data: pd.DataFrame, 

140 n_samples: int = 100, 

141 gap_probability: float = 0.05, 

142 max_gap_pct: float = 0.02, 

143 ) -> list[pd.DataFrame]: 

144 """Add random gaps between candles. 

145 

146 Args: 

147 data: OHLCV DataFrame. 

148 n_samples: Number of samples. 

149 gap_probability: Probability of gap per candle. 

150 max_gap_pct: Maximum gap as fraction of price. 

151 

152 Returns: 

153 List of DataFrames with gaps. 

154 """ 

155 samples = [] 

156 

157 for _ in range(n_samples): 

158 sample = data.copy() 

159 

160 # Generate random gaps 

161 has_gap = self.rng.random(len(data)) < gap_probability 

162 gap_size = self.rng.uniform(-max_gap_pct, max_gap_pct, len(data)) 

163 gap_size = gap_size * has_gap 

164 

165 # Apply cumulative gaps 

166 gap_factor = np.cumprod(1 + gap_size) 

167 

168 for col in ["open", "high", "low", "close"]: 

169 sample[col] = sample[col] * gap_factor 

170 

171 sample = self._fix_ohlc(sample) 

172 samples.append(sample) 

173 

174 logger.debug(f"Generated {n_samples} gap noise samples") 

175 return samples 

176 

177 def add_volume_noise( 

178 self, 

179 data: pd.DataFrame, 

180 n_samples: int = 100, 

181 noise_level: float = 0.3, 

182 ) -> list[pd.DataFrame]: 

183 """Add noise to volume data. 

184 

185 Args: 

186 data: OHLCV DataFrame. 

187 n_samples: Number of samples. 

188 noise_level: Standard deviation as fraction of volume. 

189 

190 Returns: 

191 List of DataFrames with volume noise. 

192 """ 

193 samples = [] 

194 

195 for _ in range(n_samples): 

196 sample = data.copy() 

197 

198 noise = self.rng.lognormal(0, noise_level, len(data)) 

199 sample["volume"] = sample["volume"] * noise 

200 sample["volume"] = sample["volume"].clip(lower=0) 

201 

202 samples.append(sample) 

203 

204 logger.debug(f"Generated {n_samples} volume noise samples") 

205 return samples 

206 

207 def add_latency_shift( 

208 self, 

209 data: pd.DataFrame, 

210 n_samples: int = 100, 

211 max_shift: int = 2, 

212 ) -> list[pd.DataFrame]: 

213 """Simulate execution latency by shifting data relative to itself. 

214  

215 Args: 

216 data: OHLCV DataFrame. 

217 n_samples: Number of samples. 

218 max_shift: Maximum number of candles to shift. 

219  

220 Returns: 

221 List of DataFrames with latency shifts. 

222 """ 

223 samples = [] 

224 

225 for _ in range(n_samples): 

226 shift = self.rng.integers(1, max_shift + 1) 

227 sample = data.copy() 

228 

229 # Shift prices forward (making signals appear late) 

230 # Actually, shifting prices backward has same effect as delaying signals 

231 sample = sample.shift(shift) 

232 sample = sample.bfill() 

233 

234 samples.append(sample) 

235 

236 logger.debug(f"Generated {n_samples} latency shift samples (max {max_shift} candles)") 

237 return samples 

238 

239 def _fix_ohlc(self, data: pd.DataFrame) -> pd.DataFrame: 

240 """Ensure OHLC consistency. 

241 

242 Args: 

243 data: OHLCV DataFrame. 

244 

245 Returns: 

246 Fixed DataFrame. 

247 """ 

248 data = data.copy() 

249 

250 # High must be >= max(open, close) 

251 data["high"] = data[["open", "high", "close"]].max(axis=1) 

252 

253 # Low must be <= min(open, close) 

254 data["low"] = data[["open", "low", "close"]].min(axis=1) 

255 

256 return data