Backtesting & Validation

import pandas as pd
import numpy as np

class Backtester:
    def __init__(self, initial_capital=100000,
                 commission=0.001, slippage=0.0005):
        self.initial_capital = initial_capital
        self.commission = commission  # 0.1% per trade
        self.slippage = slippage      # 0.05% slippage

    def run(self, strategy, data):
        """Run backtest simulation"""
        portfolio_value = [self.initial_capital]
        positions = {}
        cash = self.initial_capital
        equity_curve = []

        for timestamp, row in data.iterrows():
            # Generate signals
            signals = strategy.generate_signals(row)

            # Execute trades
            for asset, target in signals.items():
                current = positions.get(asset, 0)

                if target != current:
                    # Calculate trade cost
                    price = row[f'{asset}_price']
                    quantity = target - current
                    slippage_cost = abs(quantity) * price * self.slippage
                    commission_cost = abs(quantity) * price * self.commission
                    total_cost = abs(quantity) * price + slippage_cost + commission_cost

                    # Execute trade
                    if quantity > 0:  # Buy
                        if cash >= total_cost:
                            cash -= total_cost
                            positions[asset] = target
                    else:  # Sell
                        cash += abs(quantity) * price - slippage_cost - commission_cost
                        positions[asset] = target

            # Calculate portfolio value
            equity = sum(pos * row[f'{asset}_price']
                        for asset, pos in positions.items())
            total_value = cash + equity
            equity_curve.append({
                'timestamp': timestamp,
                'value': total_value,
                'cash': cash,
                'equity': equity
            })

        return pd.DataFrame(equity_curve)

def calculate_metrics(returns, risk_free_rate=0.02):
    """Calculate comprehensive performance metrics"""

    # Return metrics
    total_return = (1 + returns).prod() - 1
    cagr = (1 + total_return) ** (252 / len(returns)) - 1

    # Risk metrics
    volatility = returns.std() * np.sqrt(252)
    cumulative = (1 + returns).cumprod()
    running_max = cumulative.expanding().max()
    drawdown = (cumulative - running_max) / running_max
    max_drawdown = drawdown.min()

    # Risk-adjusted returns
    excess_returns = returns - risk_free_rate / 252
    sharpe = excess_returns.mean() / returns.std() * np.sqrt(252)

    downside_returns = returns[returns < 0]
    sortino = excess_returns.mean() / downside_returns.std() * np.sqrt(252)

    calmar = cagr / abs(max_drawdown) if max_drawdown != 0 else 0

    return {
        'Total Return': f'{total_return:.2%}',
        'CAGR': f'{cagr:.2%}',
        'Volatility': f'{volatility:.2%}',
        'Max Drawdown': f'{max_drawdown:.2%}',
        'Sharpe Ratio': f'{sharpe:.2f}',
        'Sortino Ratio': f'{sortino:.2f}',
        'Calmar Ratio': f'{calmar:.2f}'
    }

def walk_forward_analysis(data, strategy, train_period=252,
                         test_period=63, step=63):
    """Walk-forward optimization and testing"""

    results = []
    start = 0

    while start + train_period + test_period <= len(data):
        # In-sample period
        train_data = data.iloc[start:start + train_period]

        # Optimize strategy parameters
        optimized_params = optimize_strategy(strategy, train_data)

        # Out-of-sample period
        test_data = data.iloc[start + train_period:
                             start + train_period + test_period]

        # Test with optimized parameters
        strategy.set_params(optimized_params)
        oos_returns = backtest(strategy, test_data)

        results.append({
            'period': start,
            'returns': oos_returns,
            'params': optimized_params
        })

        start += step

    return pd.DataFrame(results)

def monte_carlo_backtest(returns, n_simulations=10000):
    """Monte Carlo simulation of strategy returns"""

    initial_capital = 100000
    final_values = []

    for _ in range(n_simulations):
        # Randomly shuffle returns (preserves distribution)
        shuffled_returns = returns.sample(frac=1, replace=True)

        # Calculate final portfolio value
        final_value = initial_capital * (1 + shuffled_returns).prod()
        final_values.append(final_value)

    final_values = np.array(final_values)

    return {
        'Mean': final_values.mean(),
        'Median': np.median(final_values),
        'Std': final_values.std(),
        '5th Percentile': np.percentile(final_values, 5),
        '95th Percentile': np.percentile(final_values, 95),
        'Prob of Loss': (final_values < initial_capital).mean()
    }