import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
import talib
class MLMeanReversionBacktester:
    def __init__(self, symbol, period=20, zscore=2.0):
        self.symbol = symbol
        self.period = period
        self.zscore = zscore
        self.model = None
        self.scaler = StandardScaler()
def create_features(self, df):
        """Feature engineering for mean reversion"""
        df = df.copy()
# Mean reversion indicators
        df['SMA'] = df['Close'].rolling(self.period).mean()
        df['Std'] = df['Close'].rolling(self.period).std()
        df['Zscore'] = (df['Close'] - df['SMA']) / df['Std']
        df['Distance_to_Mean'] = abs(df['Zscore'])
# Momentum (mean reversion trades mean revert faster if losing momentum)
        df['Momentum'] = df['Close'].pct_change(5)
        df['Acceleration'] = df['Momentum'].diff()
# Volume profile
        df['Volume_MA'] = df['Volume'].rolling(20).mean()
        df['High_Volume'] = df['Volume'] > df['Volume_MA']
# Volatility
        df['Volatility'] = df['Close'].pct_change().rolling(20).std()
        df['High_Volatility'] = df['Volatility'] > df['Volatility'].quantile(0.75)
# Trend confirmation
        df['SMA_50'] = df['Close'].rolling(50).mean()
        df['In_Uptrend'] = (df['Close'] > df['SMA_50']).astype(int)
# Price action
        df['Recent_Range'] = (df['High'].rolling(10).max() - df['Low'].rolling(10).min()) / df['Close']
        df['Oversold_Count'] = (df['Zscore'] < -self.zscore).rolling(5).sum()
# Lagged features
        for lag in [1, 2, 3]:
            df[f'Zscore_lag{lag}'] = df['Zscore'].shift(lag)
# Target: 1 if price reverts within 5 days
        df['Future_Return'] = df['Close'].pct_change(5).shift(-5)
        df['Target'] = ((df['Future_Return'] > 0.001) & (df['Zscore'] < -self.zscore)).astype(int)
return df.dropna()
def train_model(self, df_train):
        """Train ensemble model"""
        feature_cols = [col for col in df_train.columns
                       if col not in ['Open', 'High', 'Low', 'Close', 'Volume', 'Target', 'Future_Return']]
X = df_train[feature_cols].values
        y = df_train['Target'].values
X_scaled = self.scaler.fit_transform(X)
# Random Forest
        rf = RandomForestClassifier(n_estimators=100, max_depth=10, random_state=42)
        rf.fit(X_scaled, y)
# Gradient Boosting
        gb = GradientBoostingClassifier(n_estimators=100, max_depth=5, learning_rate=0.1, random_state=42)
        gb.fit(X_scaled, y)
self.model = {'rf': rf, 'gb': gb, 'features': feature_cols}
        return self.model
def predict_reversion(self, df):
        """Predict probability of mean reversion"""
        feature_cols = self.model['features']
        X = df[feature_cols].values
        X_scaled = self.scaler.transform(X)
rf_prob = self.model['rf'].predict_proba(X_scaled)[:, 1]
        gb_prob = self.model['gb'].predict_proba(X_scaled)[:, 1]
# Ensemble average
        ensemble_prob = (rf_prob + gb_prob) / 2
        df['Reversion_Probability'] = ensemble_prob
return df
def generate_signals(self, df, confidence_threshold=0.60):
        """ML-filtered mean reversion signals"""
        df['Traditional_Signal'] = 0
        df.loc[df['Zscore'] < -self.zscore, 'Traditional_Signal'] = 1
        df.loc[df['Zscore'] > self.zscore, 'Traditional_Signal'] = -1
# ML filter: only trade if high confidence reversion
        df['ML_Signal'] = df['Traditional_Signal'] * (df['Reversion_Probability'] > confidence_threshold).astype(int)
# Position
        df['Position'] = 0
        df.loc[df['ML_Signal'] == 1, 'Position'] = 1
        df.loc[df['ML_Signal'] == -1, 'Position'] = 0
        df['Position'] = df['Position'].fillna(method='ffill').fillna(0)
return df
def backtest(self, df_full, train_ratio=0.7):
        """Train/test split backtest"""
        split_idx = int(len(df_full) * train_ratio)
df_train = self.create_features(df_full.iloc[:split_idx])
        df_test_raw = df_full.iloc[split_idx:]
        df_test = self.create_features(df_test_raw)
# Train
        self.train_model(df_train)
# Predict
        df_test = self.predict_reversion(df_test)
        df_test = self.generate_signals(df_test)
# Returns
        df_test['Daily_Return'] = df_test['Close'].pct_change()
        df_test['Transaction_Cost'] = df_test['Position'].diff().abs() * 0.001
        df_test['Strategy_Return'] = df_test['Position'].shift(1) * (df_test['Daily_Return'] - df_test['Transaction_Cost'])
df_test['Cumulative_Strategy'] = (1 + df_test['Strategy_Return']).cumprod()
        df_test['Cumulative_BH'] = (1 + df_test['Daily_Return']).cumprod()
return df_test
def calculate_metrics(self, df_test):
        """Performance metrics"""
        sr = df_test['Strategy_Return'].dropna()
return {
            'Total_Return': (df_test['Cumulative_Strategy'].iloc[-1] - 1) * 100,
            'BH_Return': (df_test['Cumulative_BH'].iloc[-1] - 1) * 100,
            'Sharpe': (sr.mean() / sr.std()) * np.sqrt(252) if sr.std() > 0 else 0,
            'Win_Rate': len(sr[sr > 0]) / len(sr) * 100,
            'Max_DD': ((df_test['Cumulative_Strategy'] / df_test['Cumulative_Strategy'].expanding().max() - 1).min() * 100),
            'Trades': len(df_test[df_test['ML_Signal'] != 0]),
        }

from sklearn.model_selection import TimeSeriesSplit
def robust_validation(df, n_splits=5):
    """Time series cross-validation"""
    tscv = TimeSeriesSplit(n_splits=n_splits)
    scores = []
for train_idx, test_idx in tscv.split(df):
        df_train = df.iloc[train_idx]
        df_test = df.iloc[test_idx]
# Train on past, test on future
        backtester.train_model(backtester.create_features(df_train))
        df_test = backtester.predict_reversion(backtester.create_features(df_test))
metrics = backtester.calculate_metrics(df_test)
        scores.append(metrics['Sharpe'])
# Check for overfitting
    avg_sharpe = np.mean(scores)
    std_sharpe = np.std(scores)
print(f"Mean Sharpe: {avg_sharpe:.2f} ± {std_sharpe:.2f}")
    print(f"Overfitting risk: {'High' if std_sharpe > avg_sharpe * 0.2 else 'Low'}")

# Retrain monthly
monthly_dates = df.resample('M').last_valid_index()
for month_end in monthly_dates:
    df_train = df[df.index < month_end].tail(504)  # Last 2 years
    df_test = df[(df.index >= month_end) & (df.index < month_end + pd.DateOffset(months=1))]
backtester.train_model(backtester.create_features(df_train))
    df_test = backtester.predict_reversion(backtester.create_features(df_test))