Coverage for src / monte_neo / monte_carlo / shuffler.py: 95%
79 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-28 16:27 +0200
1"""Data shuffling module.
3Implements various shuffling methods for Monte Carlo robustness testing.
4"""
6from __future__ import annotations
8import numpy as np
9import pandas as pd
11from monte_neo.utils.logger import get_logger
13logger = get_logger(__name__)
16class DataShuffler:
17 """Data shuffling for robustness testing."""
19 def __init__(self, random_seed: int | None = None) -> None:
20 """Initialize shuffler.
22 Args:
23 random_seed: Random seed for reproducibility.
24 """
25 self.rng = np.random.default_rng(random_seed)
27 def shuffle_returns(
28 self,
29 data: pd.DataFrame,
30 n_samples: int = 100,
31 ) -> list[pd.DataFrame]:
32 """Shuffle returns while preserving distribution.
34 This shuffles the order of returns, breaking time dependency.
36 Args:
37 data: OHLCV DataFrame.
38 n_samples: Number of shuffled samples.
40 Returns:
41 List of shuffled DataFrames.
42 """
43 samples = []
44 returns = data["close"].pct_change().fillna(0).values
46 for _ in range(n_samples):
47 # Shuffle returns
48 shuffled_returns = self.rng.permutation(returns)
50 # Reconstruct prices
51 initial_price = data["close"].iloc[0]
52 new_prices = initial_price * np.cumprod(1 + shuffled_returns)
54 # Create shuffled OHLCV
55 sample = self._reconstruct_ohlcv(data, new_prices)
56 samples.append(sample)
58 logger.debug(f"Generated {n_samples} return-shuffled samples")
59 return samples
61 def shuffle_blocks(
62 self,
63 data: pd.DataFrame,
64 n_samples: int = 100,
65 block_size: int | None = None,
66 ) -> list[pd.DataFrame]:
67 """Shuffle blocks of data.
69 Preserves short-term structure but breaks long-term patterns.
71 Args:
72 data: OHLCV DataFrame.
73 n_samples: Number of shuffled samples.
74 block_size: Size of each block.
76 Returns:
77 List of shuffled DataFrames.
78 """
79 n = len(data)
80 if block_size is None:
81 block_size = max(10, n // 20) # ~5% of data per block
83 n_blocks = n // block_size
84 samples = []
86 for _ in range(n_samples):
87 # Create block indices
88 block_indices = list(range(n_blocks))
89 self.rng.shuffle(block_indices)
91 # Collect shuffled blocks
92 parts = []
93 for block_idx in block_indices:
94 start = block_idx * block_size
95 end = start + block_size
96 parts.append(data.iloc[start:end])
98 # Handle remainder
99 remainder_start = n_blocks * block_size
100 if remainder_start < n:
101 parts.append(data.iloc[remainder_start:])
103 sample = pd.concat(parts, ignore_index=True)
104 samples.append(sample)
106 logger.debug(f"Generated {n_samples} block-shuffled samples")
107 return samples
109 def shuffle_within_session(
110 self,
111 data: pd.DataFrame,
112 n_samples: int = 100,
113 session_hours: int = 24,
114 ) -> list[pd.DataFrame]:
115 """Shuffle candles within trading sessions.
117 Useful for intraday data where sessions are independent.
119 Args:
120 data: OHLCV DataFrame with datetime index.
121 n_samples: Number of shuffled samples.
122 session_hours: Hours per session.
124 Returns:
125 List of shuffled DataFrames.
126 """
127 samples = []
129 # Group by session
130 if hasattr(data.index, "date"):
131 data = data.copy()
132 data["session"] = pd.to_datetime(data.index).date
133 else:
134 # Assume uniform spacing, create synthetic sessions
135 session_size = session_hours
136 data = data.copy()
137 data["session"] = np.arange(len(data)) // session_size
139 sessions = data.groupby("session")
141 for _ in range(n_samples):
142 shuffled_parts = []
144 for _, session_data in sessions:
145 # Shuffle within session
146 shuffled = session_data.sample(
147 frac=1, random_state=int(self.rng.integers(1_000_000_000))
148 )
149 shuffled_parts.append(shuffled)
151 sample = pd.concat(shuffled_parts)
152 sample = sample.drop(columns=["session"])
153 samples.append(sample)
155 logger.debug(f"Generated {n_samples} session-shuffled samples")
156 return samples
158 def shuffle_columns(
159 self,
160 data: pd.DataFrame,
161 n_samples: int = 100,
162 columns: list[str] | None = None,
163 ) -> list[pd.DataFrame]:
164 """Shuffle specific columns independently.
166 Tests sensitivity to inter-column relationships.
168 Args:
169 data: OHLCV DataFrame.
170 n_samples: Number of shuffled samples.
171 columns: Columns to shuffle (default: volume only).
173 Returns:
174 List of shuffled DataFrames.
175 """
176 if columns is None:
177 columns = ["volume"] # Safe default - don't break OHLC relationship
179 samples = []
181 for _ in range(n_samples):
182 sample = data.copy()
184 for col in columns:
185 if col in sample.columns:
186 sample[col] = self.rng.permutation(sample[col].values)
188 samples.append(sample)
190 logger.debug(f"Generated {n_samples} column-shuffled samples")
191 return samples
193 def _reconstruct_ohlcv(
194 self,
195 original: pd.DataFrame,
196 new_close: np.ndarray,
197 ) -> pd.DataFrame:
198 """Reconstruct OHLCV from new close prices.
200 Args:
201 original: Original OHLCV data.
202 new_close: New close prices.
204 Returns:
205 Reconstructed OHLCV DataFrame.
206 """
207 # Calculate original ratios
208 o_ratio = original["open"] / original["close"]
209 h_ratio = original["high"] / original["close"]
210 l_ratio = original["low"] / original["close"]
212 # Apply to new close
213 sample = pd.DataFrame(
214 {
215 "open": new_close * o_ratio.values,
216 "high": new_close * h_ratio.values,
217 "low": new_close * l_ratio.values,
218 "close": new_close,
219 "volume": original["volume"].values,
220 }
221 )
223 # Ensure high >= max(open, close) and low <= min(open, close)
224 sample["high"] = sample[["open", "high", "close"]].max(axis=1)
225 sample["low"] = sample[["open", "low", "close"]].min(axis=1)
227 return sample