Add wavelet denoising engine with adaptive parameter selection

wavelet_denoising.py

@@ -0,0 +1,403 @@
+"""Wavelet Denoising for Financial Time Series
+
+Based on Lopez Gil et al. 2024 (xLSTM-TS paper, arxiv:2408.12408):
+Wavelet denoising improved prediction accuracy by 5-10% across ALL models tested.
+
+This is NOT optional. Without wavelet preprocessing, your models are 
+training on noise instead of signal.
+"""
+import numpy as np
+import pandas as pd
+from typing import Optional, Tuple, Dict, List
+import warnings
+warnings.filterwarnings('ignore')
+
+try:
+    import pywt
+    PYWT_AVAILABLE = True
+except ImportError:
+    PYWT_AVAILABLE = False
+    print("WARNING: pywt not available. Install with: pip install PyWavelets")
+
+
+class WaveletDenoiser:
+    """
+    Wavelet-based signal denoising for financial time series.
+    
+    The key insight: Financial time series contain:
+    1. Low-frequency signal (trend, regime changes)
+    2. High-frequency noise (intraday noise, microstructure effects)
+    
+    Wavelets separate these naturally. We keep the low frequencies
+    (signal) and discard/shrink the high frequencies (noise).
+    
+    Wavelet choice: db4 (Daubechies 4) is optimal for financial data
+    per Lopez Gil et al. because:
+    - Compact support (localizes in time)
+    - Orthogonal (no redundancy)
+    - 4 vanishing moments (captures polynomial trends well)
+    """
+    
+    def __init__(self, wavelet: str = 'db4', mode: str = 'symmetric',
+                 threshold_mode: str = 'soft', level: Optional[int] = None):
+        self.wavelet = wavelet
+        self.mode = mode
+        self.threshold_mode = threshold_mode
+        self.level = level
+    
+    def _get_max_level(self, signal_length: int) -> int:
+        """Maximum decomposition level for given signal length"""
+        if not PYWT_AVAILABLE:
+            return 1
+        return pywt.dwt_max_level(signal_length, pywt.Wavelet(self.wavelet).dec_len)
+    
+    def _estimate_noise_sigma(self, detail_coeffs: np.ndarray) -> float:
+        """
+        Estimate noise standard deviation from finest detail coefficients.
+        
+        Uses MAD (Median Absolute Deviation) estimator:
+        sigma = median(|coeffs|) / 0.6745
+        
+        This is robust to outliers — critical for financial data with fat tails.
+        """
+        mad = np.median(np.abs(detail_coeffs))
+        return mad / 0.6745 if mad > 0 else 1.0
+    
+    def _universal_threshold(self, coeffs: np.ndarray, 
+                             sigma: Optional[float] = None) -> float:
+        """
+        Donoho-Johnstone universal threshold:
+        lambda = sigma * sqrt(2 * log(N))
+        
+        Asymptotically optimal for Gaussian noise. For financial data,
+        we use the robust MAD estimate for sigma.
+        """
+        n = len(coeffs)
+        if sigma is None:
+            sigma = self._estimate_noise_sigma(coeffs)
+        
+        return sigma * np.sqrt(2 * np.log(n))
+    
+    def _sure_threshold(self, coeffs: np.ndarray) -> float:
+        """
+        Stein's Unbiased Risk Estimate (SURE) threshold.
+        
+        More adaptive than universal threshold — better when noise level
+        varies across the signal. Slightly more computationally expensive.
+        """
+        n = len(coeffs)
+        coeffs_sorted = np.sort(np.abs(coeffs)) ** 2
+        
+        risks = np.zeros(n)
+        for i in range(n):
+            risks[i] = (n - 2 * (i + 1) + (n - (i + 1)) * coeffs_sorted[i] + 
+                        np.sum(coeffs_sorted[:i+1]))
+        
+        min_risk_idx = np.argmin(risks)
+        return np.sqrt(coeffs_sorted[min_risk_idx])
+    
+    def denoise(self, signal: np.ndarray) -> np.ndarray:
+        """
+        Denoise a 1D signal using wavelet thresholding.
+        
+        Args:
+            signal: 1D numpy array
+        
+        Returns:
+            Denoised signal
+        """
+        if not PYWT_AVAILABLE:
+            # Fallback: simple moving average smoothing
+            return pd.Series(signal).rolling(5, center=True, min_periods=1).mean().values
+        
+        # Determine decomposition level
+        max_level = self._get_max_level(len(signal))
+        level = self.level or min(4, max_level)
+        level = max(1, min(level, max_level))
+        
+        # Wavelet decomposition
+        coeffs = pywt.wavedec(signal, self.wavelet, mode=self.mode, level=level)
+        
+        # Threshold detail coefficients (keep approximation)
+        denoised_coeffs = [coeffs[0]]  # approximation (low-frequency signal)
+        
+        for detail in coeffs[1:]:
+            # Estimate noise from this level's detail coefficients
+            sigma = self._estimate_noise_sigma(detail)
+            threshold = self._universal_threshold(detail, sigma)
+            
+            # Apply threshold
+            if self.threshold_mode == 'soft':
+                denoised_detail = pywt.threshold(detail, threshold, mode='soft')
+            elif self.threshold_mode == 'hard':
+                denoised_detail = pywt.threshold(detail, threshold, mode='hard')
+            elif self.threshold_mode == 'garotte':
+                # Firm threshold (compromise between soft and hard)
+                denoised_detail = pywt.threshold(detail, threshold, mode='greater')
+            else:
+                denoised_detail = pywt.threshold(detail, threshold, mode='soft')
+            
+            denoised_coeffs.append(denoised_detail)
+        
+        # Reconstruct signal
+        denoised = pywt.waverec(denoised_coeffs, self.wavelet, mode=self.mode)
+        
+        # wavrec may return slightly longer signal due to padding
+        return denoised[:len(signal)]
+    
+    def denoise_dataframe(self, df: pd.DataFrame, 
+                          columns: Optional[List[str]] = None) -> pd.DataFrame:
+        """
+        Denoise multiple columns of a DataFrame.
+        
+        Args:
+            df: DataFrame with time series data
+            columns: List of columns to denoise (None = all numeric)
+        
+        Returns:
+            DataFrame with denoised columns (original + _denoised suffix)
+        """
+        if columns is None:
+            columns = df.select_dtypes(include=[np.number]).columns.tolist()
+        
+        result = df.copy()
+        for col in columns:
+            if col in df.columns:
+                signal = df[col].values
+                # Handle NaN
+                nan_mask = np.isnan(signal)
+                if nan_mask.any():
+                    signal = pd.Series(signal).interpolate().fillna(method='bfill').fillna(method='ffill').values
+                
+                denoised = self.denoise(signal)
+                result[f'{col}_denoised'] = denoised
+        
+        return result
+    
+    def denoise_multivariate(self, signals: np.ndarray, 
+                             axis: int = 0) -> np.ndarray:
+        """
+        Denoise each column/row of a multivariate signal array independently.
+        
+        Args:
+            signals: 2D array (samples x features)
+            axis: 0 = denoise each column, 1 = denoise each row
+        
+        Returns:
+            Denoised array
+        """
+        if signals.ndim != 2:
+            raise ValueError("signals must be 2D")
+        
+        if axis == 1:
+            signals = signals.T
+        
+        denoised = np.zeros_like(signals)
+        for i in range(signals.shape[1]):
+            denoised[:, i] = self.denoise(signals[:, i])
+        
+        if axis == 1:
+            denoised = denoised.T
+        
+        return denoised
+    
+    def extract_features(self, signal: np.ndarray, level: Optional[int] = None) -> Dict[str, np.ndarray]:
+        """
+        Extract wavelet-based features from a signal.
+        
+        These features capture multi-scale information:
+        - Energy by frequency band
+        - Entropy (complexity measure)
+        - Coefficient statistics
+        
+        Useful as additional features for ML models.
+        """
+        if not PYWT_AVAILABLE:
+            return {}
+        
+        max_level = self._get_max_level(len(signal))
+        level = level or min(4, max_level)
+        level = max(1, min(level, max_level))
+        
+        coeffs = pywt.wavedec(signal, self.wavelet, mode=self.mode, level=level)
+        
+        features = {}
+        
+        # Approximation features
+        approx = coeffs[0]
+        features['approx_mean'] = np.mean(approx)
+        features['approx_std'] = np.std(approx)
+        features['approx_energy'] = np.sum(approx ** 2)
+        
+        # Detail features by level
+        for i, detail in enumerate(coeffs[1:], 1):
+            features[f'detail_{i}_mean'] = np.mean(detail)
+            features[f'detail_{i}_std'] = np.std(detail)
+            features[f'detail_{i}_energy'] = np.sum(detail ** 2)
+            features[f'detail_{i}_skew'] = self._skewness(detail)
+            features[f'detail_{i}_kurt'] = self._kurtosis(detail)
+        
+        # Multi-scale energy ratio
+        total_energy = sum(np.sum(c ** 2) for c in coeffs)
+        if total_energy > 0:
+            features['approx_energy_ratio'] = features['approx_energy'] / total_energy
+            for i in range(1, len(coeffs)):
+                features[f'detail_{i}_energy_ratio'] = np.sum(coeffs[i] ** 2) / total_energy
+        
+        # Wavelet entropy (measure of complexity)
+        energies = [np.sum(c ** 2) for c in coeffs]
+        energies = np.array(energies) / (sum(energies) + 1e-10)
+        features['wavelet_entropy'] = -np.sum(energies * np.log(energies + 1e-10))
+        
+        return features
+    
+    @staticmethod
+    def _skewness(x: np.ndarray) -> float:
+        x = x - np.mean(x)
+        n = len(x)
+        if n < 3 or np.std(x) == 0:
+            return 0.0
+        return np.sum(x ** 3) / (n * np.std(x) ** 3)
+    
+    @staticmethod
+    def _kurtosis(x: np.ndarray) -> float:
+        x = x - np.mean(x)
+        n = len(x)
+        if n < 4 or np.std(x) == 0:
+            return 0.0
+        return np.sum(x ** 4) / (n * np.std(x) ** 4) - 3
+
+
+class AdaptiveWaveletDenoiser:
+    """
+    Adaptive wavelet denoising that selects optimal parameters per signal.
+    
+    Instead of using fixed wavelet and level, this tries multiple combinations
+    and selects the one that best separates signal from noise.
+    
+    Selection criterion: SNR (Signal-to-Noise Ratio) maximization.
+    """
+    
+    def __init__(self, wavelets: Optional[List[str]] = None,
+                 levels: Optional[List[int]] = None,
+                 threshold_modes: Optional[List[str]] = None):
+        self.wavelets = wavelets or ['db2', 'db4', 'db6', 'sym4', 'coif2']
+        self.levels = levels or [2, 3, 4, 5]
+        self.threshold_modes = threshold_modes or ['soft', 'hard']
+        self.best_params = None
+    
+    def _snr(self, signal: np.ndarray, denoised: np.ndarray) -> float:
+        """Estimate signal-to-noise ratio"""
+        noise = signal - denoised
+        signal_power = np.sum(denoised ** 2)
+        noise_power = np.sum(noise ** 2) + 1e-10
+        return 10 * np.log10(signal_power / noise_power)
+    
+    def fit(self, signal: np.ndarray) -> Dict:
+        """
+        Find optimal denoising parameters for a signal.
+        
+        Uses grid search over wavelets, levels, and threshold modes.
+        Selects combination with highest estimated SNR.
+        """
+        best_snr = -np.inf
+        best_params = {}
+        best_denoised = signal.copy()
+        
+        if not PYWT_AVAILABLE:
+            return {'wavelet': 'none', 'level': 1, 'threshold_mode': 'none'}
+        
+        for wavelet in self.wavelets:
+            try:
+                max_level = pywt.dwt_max_level(len(signal), pywt.Wavelet(wavelet).dec_len)
+                valid_levels = [l for l in self.levels if l <= max_level]
+                if not valid_levels:
+                    valid_levels = [1]
+                
+                for level in valid_levels:
+                    for mode in self.threshold_modes:
+                        try:
+                            denoiser = WaveletDenoiser(
+                                wavelet=wavelet,
+                                level=level,
+                                threshold_mode=mode
+                            )
+                            denoised = denoiser.denoise(signal)
+                            snr = self._snr(signal, denoised)
+                            
+                            if snr > best_snr:
+                                best_snr = snr
+                                best_params = {
+                                    'wavelet': wavelet,
+                                    'level': level,
+                                    'threshold_mode': mode,
+                                    'snr': snr
+                                }
+                                best_denoised = denoised
+                        except:
+                            continue
+            except:
+                continue
+        
+        self.best_params = best_params
+        return best_params
+    
+    def denoise(self, signal: np.ndarray) -> np.ndarray:
+        """Denoise using best fitted parameters"""
+        if self.best_params is None:
+            self.fit(signal)
+        
+        denoiser = WaveletDenoiser(
+            wavelet=self.best_params.get('wavelet', 'db4'),
+            level=self.best_params.get('level', 4),
+            threshold_mode=self.best_params.get('threshold_mode', 'soft')
+        )
+        return denoiser.denoise(signal)
+
+
+def benchmark_denoising(signal: np.ndarray, noise_level: float = 0.1) -> Dict:
+    """
+    Benchmark denoising on a known signal with added noise.
+    
+    Returns MSE, SNR, correlation with true signal.
+    """
+    # Add noise
+    noisy = signal + np.random.randn(len(signal)) * noise_level * np.std(signal)
+    
+    # Denoise
+    denoiser = WaveletDenoiser(wavelet='db4', level=4, threshold_mode='soft')
+    denoised = denoiser.denoise(noisy)
+    
+    # Metrics
+    mse_noisy = np.mean((noisy - signal) ** 2)
+    mse_denoised = np.mean((denoised - signal) ** 2)
+    
+    corr_noisy = np.corrcoef(noisy, signal)[0, 1]
+    corr_denoised = np.corrcoef(denoised, signal)[0, 1]
+    
+    snr_noisy = 10 * np.log10(np.sum(signal ** 2) / np.sum((noisy - signal) ** 2))
+    snr_denoised = 10 * np.log10(np.sum(signal ** 2) / np.sum((denoised - signal) ** 2))
+    
+    return {
+        'mse_noisy': mse_noisy,
+        'mse_denoised': mse_denoised,
+        'improvement_factor': mse_noisy / (mse_denoised + 1e-10),
+        'corr_noisy': corr_noisy,
+        'corr_denoised': corr_denoised,
+        'snr_noisy': snr_noisy,
+        'snr_denoised': snr_denoised,
+        'snr_improvement': snr_denoised - snr_noisy
+    }
+
+
+if __name__ == '__main__':
+    # Test with synthetic signal
+    np.random.seed(42)
+    t = np.linspace(0, 4*np.pi, 1000)
+    true_signal = np.sin(t) + 0.5 * np.sin(3*t)
+    
+    results = benchmark_denoising(true_signal, noise_level=0.3)
+    print("Wavelet Denoising Benchmark")
+    print("=" * 50)
+    for k, v in results.items():
+        print(f"  {k}: {v:.4f}")