# WRONG: Uses today's close to generate today's signal
df['Signal'] = np.where(df['MACD'] > df['Signal_Line'], 1, 0)

# CORRECT: Uses yesterday's signals for today's position
df['Signal'] = np.where(df['MACD'].shift(1) > df['Signal_Line'].shift(1), 1, 0)
df['Position'] = df['Signal'].shift(1)  # Trade tomorrow using today's signal

def walk_forward_validation(df, window_size=504, step=126):
    """
    Walk-forward: in-sample optimization, out-of-sample testing
Window: 504 days (2 years) in-sample
    Step: 126 days (6 months) out-of-sample testing
    """
    results = []
for start_idx in range(0, len(df) - window_size - step, step):
        # In-sample: optimize parameters
        in_sample = df.iloc[start_idx:start_idx + window_size]
        best_params = optimize_macd_parameters(in_sample)
# Out-of-sample: test on fresh data
        out_sample = df.iloc[start_idx + window_size:start_idx + window_size + step]
        metrics = backtest_with_params(out_sample, best_params)
results.append({
            'Period': f"{start_idx} to {start_idx + window_size + step}",
            'In_Sample_Sharpe': metrics['in_sample_sharpe'],
            'Out_Sample_Sharpe': metrics['out_sample_sharpe'],
            'Degradation': (metrics['in_sample_sharpe'] - metrics['out_sample_sharpe']) / metrics['in_sample_sharpe'],
        })
return pd.DataFrame(results)
Check for overfitting
wf_results = walk_forward_validation(df)
print(f"Average degradation: {wf_results['Degradation'].mean():.2%}")
Healthy: 10-25% degradation
Suspicious: >50% degradation indicates overfitting

def check_data_quality(df):
    """Validate data quality and identify gaps"""
# Check for gaps (weekends/holidays normal)
    expected_days = len(df) / 252  # Trading days per year
# Check for missing data
    missing_pct = df['Close'].isnull().sum() / len(df) * 100
# Check for unrealistic movements (>10% in single day)
    daily_returns = df['Close'].pct_change()
    outliers = (abs(daily_returns) > 0.10).sum()
# Check for data duplicates
    duplicates = df.index.duplicated().sum()
quality_report = {
        'Missing_Data_Pct': missing_pct,
        'Outlier_Days': outliers,
        'Duplicates': duplicates,
        'Data_Quality': 'Good' if missing_pct < 2 and outliers == 0 else 'Poor',
    }
return quality_report

def apply_realistic_costs(df, broker_type='retail_forex'):
    """Apply realistic transaction costs by broker type"""
# Typical costs
    costs = {
        'retail_forex': {'spread': 0.0002, 'commission': 0.00005},  # 2-3 pips
        'professional_forex': {'spread': 0.00005, 'commission': 0.00001},  # 0.5-1 pip
        'crypto_exchange': {'spread': 0.001, 'commission': 0.001},  # 0.1% each side
        'stocks': {'spread': 0.0001, 'commission': 0.0005},  # 1 cent + $5 per $10k
    }
cost_data = costs[broker_type]
    total_cost = cost_data['spread'] + cost_data['commission']
# Apply to every position change
    df['Position_Change'] = df['Position'].diff().abs()
    df['Transaction_Cost'] = df['Position_Change'] * total_cost
# Reduce returns by transaction costs
    df['Net_Daily_Return'] = df['Daily_Return'] - df['Transaction_Cost']
    df['Strategy_Return'] = df['Position'].shift(1) * df['Net_Daily_Return']
return df

def reality_check(metrics):
    """Flag suspicious metrics"""
flags = []
if metrics['Sharpe_Ratio'] > 2.0:
        flags.append("⚠️ Sharpe > 2.0 (suspicious)")
if metrics['Win_Rate'] > 65:
        flags.append("⚠️ Win rate > 65% (suspicious)")
if metrics['Max_Drawdown'] > -5:
        flags.append("⚠️ Drawdown < -5% (suspicious)")
if len(flags) > 0:
        print("WARNING: Metrics suggest possible overfitting")
        for flag in flags:
            print(flag)
        return False
    else:
        print("✓ Metrics pass reality check")
        return True

def corrected_p_value(num_tests, uncorrected_p=0.05):
    """
    Bonferroni correction: adjust p-value for multiple tests
If you test 100 parameter combinations,
    the probability of finding ONE false positive is 1 - (1-0.05)^100 ≈ 99.4%
    """
    corrected = uncorrected_p / num_tests
    return corrected
Example
num_parameter_tests = 100  # Fast: 10 × Slow: 10 × Signal: 10
corrected_p = corrected_p_value(num_parameter_tests)
print(f"Corrected significance level: {corrected_p:.6f}")
Result: 0.0005 (much stricter than 0.05)

import pandas as pd
import numpy as np
from scipy import stats
import warnings
class SafeMACDBacktester:
    def __init__(self, symbol, fast=12, slow=26, signal=9):
        self.symbol = symbol
        self.fast = fast
        self.slow = slow
        self.signal = signal
        self.df = None
        self.validation_report = {}
def load_and_validate_data(self, start_date, end_date, min_days=1000):
        """
        Load data with strict validation
Minimum 1000 days (~4 years) to capture different market regimes
        """
        self.df = yf.download(self.symbol, start=start_date, end=end_date)
# Validation checks
        if len(self.df) < min_days:
            raise ValueError(f"Need {min_days} days minimum, got {len(self.df)}")
# Check data quality
        quality = self._check_quality()
        self.validation_report['Data_Quality'] = quality
if quality['Status'] != 'Pass':
            warnings.warn(f"Data quality issues detected: {quality['Issues']}")
return self.df
def _check_quality(self):
        """Detailed data quality checks"""
        issues = []
# Missing data
        missing_pct = self.df['Close'].isnull().sum() / len(self.df) * 100
        if missing_pct > 2:
            issues.append(f"Missing data: {missing_pct:.1f}%")
# Duplicates
        if self.df.index.duplicated().sum() > 0:
            issues.append("Duplicate timestamps found")
# Outliers
        returns = self.df['Close'].pct_change()
        outliers = (abs(returns) > 0.20).sum()
        if outliers > 0:
            issues.append(f"Extreme moves: {outliers} days > 20%")
status = 'Pass' if len(issues) == 0 else 'Fail'
return {
            'Status': status,
            'Issues': issues,
            'Missing_Pct': missing_pct,
            'Outlier_Count': outliers,
        }
def calculate_macd_strictly(self):
        """Calculate MACD with proper lag to prevent look-ahead"""
        # Calculate on proper period
        self.df['EMA_Fast'] = self.df['Close'].ewm(span=self.fast, adjust=False).mean()
        self.df['EMA_Slow'] = self.df['Close'].ewm(span=self.slow, adjust=False).mean()
        self.df['MACD'] = self.df['EMA_Fast'] - self.df['EMA_Slow']
        self.df['Signal'] = self.df['MACD'].ewm(span=self.signal, adjust=False).mean()
# Shift for next-day trading
        self.df['MACD_Tomorrow'] = self.df['MACD'].shift(-1)
        self.df['Signal_Tomorrow'] = self.df['Signal'].shift(-1)
return self.df
def generate_signals_strict(self):
        """Generate signals with explicit look-ahead prevention"""
        self.df['MACD_Prev'] = self.df['MACD'].shift(1)
        self.df['Signal_Prev'] = self.df['Signal'].shift(1)
# Check crossover BEFORE today's bar completes
        buy = (self.df['MACD_Prev'] <= self.df['Signal_Prev']) & (self.df['MACD'] > self.df['Signal'])
        sell = (self.df['MACD_Prev'] >= self.df['Signal_Prev']) & (self.df['MACD'] < self.df['Signal'])
self.df['Buy_Signal'] = buy.astype(int)
        self.df['Sell_Signal'] = sell.astype(int)
# Position filled next trading day
        self.df['Position'] = 0
        for i in range(1, len(self.df)):
            if self.df['Buy_Signal'].iloc[i]:
                self.df['Position'].iloc[i] = 1
            elif self.df['Sell_Signal'].iloc[i]:
                self.df['Position'].iloc[i] = 0
            else:
                self.df['Position'].iloc[i] = self.df['Position'].iloc[i-1]
return self.df
def calculate_returns_safe(self, transaction_cost=0.001):
        """Calculate returns with costs and strict accounting"""
        self.df['Daily_Return'] = self.df['Close'].pct_change()
# Transaction cost on every position change
        self.df['Position_Change'] = self.df['Position'].diff().abs()
        self.df['Transaction_Impact'] = self.df['Position_Change'] * transaction_cost
# Net return after costs
        self.df['Net_Return'] = self.df['Daily_Return'] - self.df['Transaction_Impact']
# Strategy return uses yesterday's position
        self.df['Strategy_Return'] = self.df['Position'].shift(1) * self.df['Net_Return']
# Cumulative
        self.df['Cumulative_Strategy'] = (1 + self.df['Strategy_Return']).cumprod()
        self.df['Cumulative_BH'] = (1 + self.df['Daily_Return']).cumprod()
return self.df
def calculate_safe_metrics(self):
        """Calculate metrics with conservative estimates"""
        valid_returns = self.df['Strategy_Return'].dropna()
total_ret = (self.df['Cumulative_Strategy'].iloc[-1] - 1) * 100
        bh_ret = (self.df['Cumulative_BH'].iloc[-1] - 1) * 100
sharpe = (valid_returns.mean() / valid_returns.std()) * np.sqrt(252) if valid_returns.std() > 0 else 0
# Win rate
        winning = len(valid_returns[valid_returns > 0])
        losing = len(valid_returns[valid_returns < 0])
        win_rate = (winning / (winning + losing) * 100) if (winning + losing) > 0 else 0
# Drawdown
        cum = self.df['Cumulative_Strategy'].fillna(method='ffill')
        running_max = cum.expanding().max()
        drawdown = ((cum - running_max) / running_max).min() * 100
metrics = {
            'Total_Return': total_ret,
            'BH_Return': bh_ret,
            'Excess_Return': total_ret - bh_ret,
            'Sharpe_Ratio': sharpe,
            'Win_Rate': win_rate,
            'Max_Drawdown': drawdown,
            'Profit_Factor': (valid_returns[valid_returns > 0].sum() / abs(valid_returns[valid_returns < 0].sum())),
        }
# Reality check
        reality_ok = self._reality_check(metrics)
        metrics['Reality_Check'] = reality_ok
return metrics
def _reality_check(self, metrics):
        """Flag unrealistic metrics"""
        if metrics['Sharpe_Ratio'] > 2.0:
            return False
        if metrics['Win_Rate'] > 70:
            return False
        if metrics['Max_Drawdown'] > -5:
            return False
        return True
def backtest_safe(self, start_date, end_date, transaction_cost=0.001):
        """Run complete safe backtest"""
        self.load_and_validate_data(start_date, end_date)
        self.calculate_macd_strictly()
        self.generate_signals_strict()
        self.calculate_returns_safe(transaction_cost)
        return self.calculate_safe_metrics()