backtester.py

import pandas as pd
import numpy as np
import json
from datetime import datetime, timedelta

class Backtester:
    def __init__(self):
        pass

    def generate_mock_data(self, days=365, initial_price=100, volatility=0.02, drift=0.0005):
        """
        Generate synthetic OHLCV data using Geometric Brownian Motion.
        """
        np.random.seed(42)  # For reproducibility in demo
        dates = pd.date_range(end=datetime.now(), periods=days).tolist()
        
        # Geometric Brownian Motion for Close prices
        returns = np.random.normal(loc=drift, scale=volatility, size=days)
        price_paths = initial_price * np.exp(np.cumsum(returns))
        
        data = []
        for i, date in enumerate(dates):
            close_price = price_paths[i]
            # Simulate High, Low, Open based on Close
            daily_vol = close_price * volatility
            open_price = close_price + np.random.uniform(-daily_vol/2, daily_vol/2)
            high_price = max(open_price, close_price) + np.random.uniform(0, daily_vol)
            low_price = min(open_price, close_price) - np.random.uniform(0, daily_vol)
            volume = int(np.random.uniform(1000, 50000))
            
            data.append({
                'date': date.strftime('%Y-%m-%d'),
                'open': round(open_price, 2),
                'high': round(high_price, 2),
                'low': round(low_price, 2),
                'close': round(close_price, 2),
                'volume': volume
            })
            
        return pd.DataFrame(data)

    def run_strategy(self, short_window=20, long_window=50, initial_capital=10000, commission=0.001, custom_data=None):
        """
        Run a simple Dual Moving Average Crossover strategy.
        """
        if custom_data is not None:
            df = custom_data
            # Ensure required columns exist
            required_columns = ['close']
            if not all(col in df.columns for col in required_columns):
                 # Try to normalize columns if they are upper case
                df.columns = [c.lower() for c in df.columns]
                if not all(col in df.columns for col in required_columns):
                    raise ValueError(f"Data must contain columns: {required_columns}")
            
            # Ensure we have a date column or index
            if 'date' in df.columns:
                df['date'] = pd.to_datetime(df['date'])
                df['date'] = df['date'].dt.strftime('%Y-%m-%d')
            else:
                 # If no date, generate dummy dates
                 dates = pd.date_range(end=datetime.now(), periods=len(df)).tolist()
                 df['date'] = [d.strftime('%Y-%m-%d') for d in dates]

        else:
            df = self.generate_mock_data()
        
        # Calculate Indicators
        df['short_mavg'] = df['close'].rolling(window=short_window, min_periods=1).mean()
        df['long_mavg'] = df['close'].rolling(window=long_window, min_periods=1).mean()
        
        # Generate Signals
        df['signal'] = 0.0
        
        # Calculate signal using numpy to avoid SettingWithCopyWarning
        signals = np.zeros(len(df))
        if len(df) > short_window:
            signals[short_window:] = np.where(
                df['short_mavg'].iloc[short_window:] > df['long_mavg'].iloc[short_window:], 
                1.0, 
                0.0
            )
        df['signal'] = signals
        df['positions'] = df['signal'].diff()
        
        # Backtest Logic
        portfolio = pd.DataFrame(index=df.index)
        portfolio['holdings'] = df['signal'] * (initial_capital / df['close']) # Simplified: invest all capital
        # Adjust for fractional shares for simplicity in simulation or assume full investment
        # Better approach for simpler calculation:
        # Strategy: When signal is 1, hold stock. When 0, hold cash.
        
        df['pct_change'] = df['close'].pct_change()
        df['strategy_returns'] = df['signal'].shift(1) * df['pct_change']
        
        # Commission cost on trades
        trades = df['positions'].abs()
        # Approximate commission impact (simplified)
        # If position changed, we paid commission
        # We need a more robust equity curve calculation
        
        equity = [initial_capital]
        position = 0 # 0: Cash, 1: Invested
        cash = initial_capital
        shares = 0
        
        equity_curve = []
        signals_log = []
        
        for i, row in df.iterrows():
            price = row['close']
            signal = row['signal']
            
            # Simple execution logic
            if signal == 1 and position == 0:
                # Buy
                shares = (cash * (1 - commission)) / price
                cash = 0
                position = 1
                signals_log.append({'date': row['date'], 'type': 'buy', 'price': price})
            elif signal == 0 and position == 1:
                # Sell
                cash = shares * price * (1 - commission)
                shares = 0
                position = 0
                signals_log.append({'date': row['date'], 'type': 'sell', 'price': price})
                
            # Update Daily Equity
            current_equity = cash + (shares * price)
            equity_curve.append(current_equity)
            
        df['equity'] = equity_curve
        
        # Metrics
        total_return = (df['equity'].iloc[-1] - initial_capital) / initial_capital
        df['daily_ret'] = df['equity'].pct_change()
        volatility = df['daily_ret'].std() * np.sqrt(252)
        sharpe_ratio = (df['daily_ret'].mean() / df['daily_ret'].std()) * np.sqrt(252) if df['daily_ret'].std() != 0 else 0
        max_drawdown = (df['equity'] / df['equity'].cummax() - 1).min()
        win_rate = len(df[df['strategy_returns'] > 0]) / len(df[df['strategy_returns'] != 0]) if len(df[df['strategy_returns'] != 0]) > 0 else 0
        
        # Prepare Data for Frontend
        # Ensure OHLC columns exist for charting
        for col in ['open', 'high', 'low']:
            if col not in df.columns:
                df[col] = df['close']

        result_data = {
            'dates': df['date'].tolist(),
            'ohlc': df[['open', 'close', 'low', 'high']].values.tolist(), # ECharts candle format: [open, close, low, high]
            'equity_curve': df['equity'].round(2).tolist(),
            'short_mavg': df['short_mavg'].fillna(0).tolist(),
            'long_mavg': df['long_mavg'].fillna(0).tolist(),
            'signals': signals_log,
            'metrics': {
                'total_return': f"{total_return*100:.2f}%",
                'final_equity': f"${df['equity'].iloc[-1]:.2f}",
                'sharpe_ratio': f"{sharpe_ratio:.2f}",
                'max_drawdown': f"{max_drawdown*100:.2f}%",
                'volatility': f"{volatility*100:.2f}%"
            }
        }
        
        return result_data