Upload regime_features.py

regime_features.py

@@ -0,0 +1,54 @@
+"""Regime Detection Features - Volatility and trend regimes"""
+import numpy as np
+import pandas as pd
+
+class RegimeFeatures:
+    """Market regime detection: bull/bear/high-vol"""
+    
+    @staticmethod
+    def volatility_regime(returns, windows=[5, 21, 63]):
+        features = pd.DataFrame(index=returns.index)
+        for w in windows:
+            vol = returns.rolling(w).std() * np.sqrt(252)
+            features[f'vol_regime_{w}d'] = pd.cut(vol, bins=[0, 0.10, 0.20, 0.35, 1.0], labels=[0, 1, 2, 3]).astype(float)
+            features[f'vol_zscore_{w}d'] = (vol - vol.rolling(252).mean()) / vol.rolling(252).std().replace(0, 1)
+        return features
+    
+    @staticmethod
+    def trend_regime(close):
+        features = pd.DataFrame(index=close.index)
+        for w in [21, 63, 126]:
+            ret = close.pct_change(w)
+            features[f'trend_{w}d'] = np.sign(ret)
+            features[f'trend_strength_{w}d'] = ret.abs()
+        # Composite regime
+        features['regime_composite'] = features['trend_21d'] + features['trend_63d']
+        return features
+    
+    @staticmethod
+    def liquidity_regime(volume, close):
+        features = pd.DataFrame(index=volume.index)
+        dollar_vol = volume * close
+        features['dollar_vol_rank'] = dollar_vol.rolling(63).rank(pct=True)
+        features['volume_trend'] = volume.rolling(21).mean() / volume.rolling(63).mean() - 1
+        return features
+    
+    @staticmethod
+    def detect_regime(returns, vol_window=21):
+        """Simple regime: bull/bear/high-vol/neutral"""
+        trend = returns.rolling(63).mean()
+        vol = returns.rolling(vol_window).std() * np.sqrt(252)
+        vol_median = vol.median()
+        regimes = pd.Series(index=returns.index, dtype='object')
+        for i, date in enumerate(returns.index):
+            if pd.isna(trend.loc[date]) or pd.isna(vol.loc[date]):
+                regimes.loc[date] = 'unknown'
+            elif vol.loc[date] > vol_median * 1.5:
+                regimes.loc[date] = 'high_vol'
+            elif trend.loc[date] > 0.001:
+                regimes.loc[date] = 'bull'
+            elif trend.loc[date] < -0.001:
+                regimes.loc[date] = 'bear'
+            else:
+                regimes.loc[date] = 'neutral'
+        return regimes