Backtesting Trading Strategies

# Backtesting Configuration
# Copy to settings.yaml and customize

# Data source settings
data:
  provider: yfinance    # Options: yfinance, coingecko
  cache_dir: ./data     # Where to cache downloaded data
  default_interval: 1d  # Default bar interval

# Backtest execution settings
backtest:
  default_capital: 10000   # Starting capital in USD
  commission: 0.001        # Commission per trade (0.1%)
  slippage: 0.0005         # Slippage per trade (0.05%)

# Risk management (optional)
risk:
  max_position_size: 0.95  # Max % of capital per position
  stop_loss: null          # Fixed stop loss % (null = disabled)
  take_profit: null        # Fixed take profit % (null = disabled)

# Report generation
reporting:
  output_dir: ./reports    # Where to save results
  save_trades: true        # Save trade log CSV
  save_equity: true        # Save equity curve CSV
  save_chart: true         # Generate equity chart PNG

# Strategy defaults (can be overridden via --params)
strategies:
  sma_crossover:
    fast_period: 20
    slow_period: 50

  ema_crossover:
    fast_period: 12
    slow_period: 26

  rsi_reversal:
    period: 14
    overbought: 70
    oversold: 30

  macd:
    fast: 12
    slow: 26
    signal: 9

  bollinger_bands:
    period: 20
    std_dev: 2.0

  breakout:
    lookback: 20
    threshold: 0.0

  mean_reversion:
    period: 20
    z_threshold: 2.0

  momentum:
    period: 14
    threshold: 5.0

#!/usr/bin/env python3
"""
Main Backtesting Engine
Run trading strategy backtests with performance analysis.

Usage:
    python backtest.py --strategy sma_crossover --symbol BTC-USD --period 1y
    python backtest.py --strategy rsi_reversal --symbol ETH-USD --start 2023-01-01 --end 2024-01-01
"""

import argparse
import json
import sys
from pathlib import Path
from datetime import datetime, timedelta
from typing import Dict, Any, Optional, List

import pandas as pd
import numpy as np

# Add script directory to path
sys.path.insert(0, str(Path(__file__).parent))

from strategies import get_strategy, list_strategies, Signal
from metrics import Trade, BacktestResult, calculate_all_metrics, format_results


def parse_period(period: str) -> timedelta:
    """Parse period string like '1y', '6m', '30d'."""
    unit = period[-1].lower()
    value = int(period[:-1])
    
    if unit == 'y':
        return timedelta(days=value * 365)
    elif unit == 'm':
        return timedelta(days=value * 30)
    elif unit == 'd':
        return timedelta(days=value)
    elif unit == 'w':
        return timedelta(weeks=value)
    else:
        raise ValueError(f"Unknown period unit: {unit}")


def load_data(symbol: str, start: datetime, end: datetime, data_dir: Path) -> pd.DataFrame:
    """Load price data from CSV or fetch if not cached."""
    
    # Try to load from cache
    cache_file = data_dir / f"{symbol.replace('/', '_').replace('-', '_')}_1d.csv"
    
    if cache_file.exists():
        df = pd.read_csv(cache_file, parse_dates=['date'], index_col='date')
        # Remove timezone info for comparison
        if df.index.tz is not None:
            df.index = df.index.tz_localize(None)
        df = df[(df.index >= pd.Timestamp(start)) & (df.index <= pd.Timestamp(end))]
        if len(df) > 0:
            return df

    # Fetch using yfinance
    try:
        import yfinance as yf
        ticker = yf.Ticker(symbol)
        df = ticker.history(start=start, end=end, interval='1d')
        df.columns = [c.lower() for c in df.columns]
        df.index.name = 'date'

        # Remove timezone for consistency
        if df.index.tz is not None:
            df.index = df.index.tz_localize(None)

        # Cache the data
        data_dir.mkdir(parents=True, exist_ok=True)
        df.to_csv(cache_file)

        return df
    except ImportError:
        print("yfinance not installed. Install with: pip install yfinance")
        sys.exit(1)
    except Exception as e:
        print(f"Error fetching data for {symbol}: {e}")
        sys.exit(1)


def run_backtest(
    strategy_name: str,
    data: pd.DataFrame,
    initial_capital: float = 10000,
    params: Dict[str, Any] = None,
    commission: float = 0.001,
    slippage: float = 0.0005,
) -> BacktestResult:
    """Run a backtest on historical data."""
    
    params = params or {}
    strategy = get_strategy(strategy_name)
    
    trades: List[Trade] = []
    equity = [initial_capital]
    cash = initial_capital
    position = None
    position_size = 0
    
    for i in range(strategy.lookback, len(data)):
        # Get data slice up to current bar
        slice_data = data.iloc[:i+1].copy()
        current_bar = data.iloc[i]
        current_price = current_bar['close']
        current_time = data.index[i]
        
        # Generate signals
        signal = strategy.generate_signals(slice_data, params)
        
        # Apply slippage
        buy_price = current_price * (1 + slippage)
        sell_price = current_price * (1 - slippage)
        
        # Execute trades
        if signal.entry and position is None:
            # Enter position
            position_value = cash * 0.95  # Use 95% of cash (keep some reserve)
            position_size = position_value / buy_price
            commission_cost = position_value * commission
            cash -= position_value + commission_cost
            
            position = {
                'entry_time': current_time,
                'entry_price': buy_price,
                'direction': signal.direction,
                'size': position_size,
            }
        
        elif signal.exit and position is not None:
            # Exit position
            exit_value = position_size * sell_price
            commission_cost = exit_value * commission
            cash += exit_value - commission_cost
            
            trade = Trade(
                entry_time=position['entry_time'],
                exit_time=current_time,
                entry_price=position['entry_price'],
                exit_price=sell_price,
                direction=position['direction'],
                size=position['size'],
            )
            trades.append(trade)
            position = None
            position_size = 0
        
        # Calculate equity (cash + position value)
        if position is not None:
            equity.append(cash + position_size * current_price)
        else:
            equity.append(cash)
    
    # Close any open position at end
    if position is not None:
        final_price = data.iloc[-1]['close'] * (1 - slippage)
        exit_value = position_size * final_price
        commission_cost = exit_value * commission
        cash += exit_value - commission_cost
        
        trade = Trade(
            entry_time=position['entry_time'],
            exit_time=data.index[-1],
            entry_price=position['entry_price'],
            exit_price=final_price,
            direction=position['direction'],
            size=position['size'],
        )
        trades.append(trade)
        equity[-1] = cash
    
    # Create equity curve
    equity_curve = pd.Series(equity, index=data.index[strategy.lookback-1:])
    
    # Build result
    result = BacktestResult(
        strategy=strategy_name,
        symbol=data.attrs.get('symbol', 'Unknown'),
        start_date=data.index[0],
        end_date=data.index[-1],
        initial_capital=initial_capital,
        final_capital=equity[-1],
        trades=trades,
        equity_curve=equity_curve,
        parameters=params,
    )
    
    # Calculate all metrics
    result = calculate_all_metrics(result)
    
    return result


def save_results(result: BacktestResult, output_dir: Path) -> None:
    """Save backtest results to files."""
    output_dir.mkdir(parents=True, exist_ok=True)
    
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    base_name = f"{result.strategy}_{result.symbol.replace('/', '_')}_{timestamp}"
    
    # Save summary
    summary_file = output_dir / f"{base_name}_summary.txt"
    with open(summary_file, 'w') as f:
        f.write(format_results(result))
    
    # Save trades to CSV
    if result.trades:
        trades_file = output_dir / f"{base_name}_trades.csv"
        trades_df = pd.DataFrame([
            {
                'entry_time': t.entry_time,
                'exit_time': t.exit_time,
                'entry_price': t.entry_price,
                'exit_price': t.exit_price,
                'direction': t.direction,
                'size': t.size,
                'pnl': t.pnl,
                'pnl_pct': t.pnl_pct,
                'duration': t.duration,
            }
            for t in result.trades
        ])
        trades_df.to_csv(trades_file, index=False)
    
    # Save equity curve
    equity_file = output_dir / f"{base_name}_equity.csv"
    result.equity_curve.to_csv(equity_file, header=['equity'])
    
    # Try to plot equity curve
    try:
        import matplotlib.pyplot as plt
        
        fig, axes = plt.subplots(2, 1, figsize=(12, 8))
        
        # Equity curve
        axes[0].plot(result.equity_curve, label='Portfolio Value', color='blue')
        axes[0].set_title(f'{result.strategy.upper()} - {result.symbol} Equity Curve')
        axes[0].set_ylabel('Portfolio Value ($)')
        axes[0].legend()
        axes[0].grid(True, alpha=0.3)
        
        # Drawdown
        rolling_max = result.equity_curve.expanding().max()
        drawdown = (result.equity_curve - rolling_max) / rolling_max * 100
        axes[1].fill_between(drawdown.index, drawdown, 0, alpha=0.5, color='red')
        axes[1].set_title('Drawdown')
        axes[1].set_ylabel('Drawdown (%)')
        axes[1].set_xlabel('Date')
        axes[1].grid(True, alpha=0.3)
        
        plt.tight_layout()
        chart_file = output_dir / f"{base_name}_chart.png"
        plt.savefig(chart_file, dpi=100)
        plt.close()
        
        print(f"Chart saved to: {chart_file}")
    except ImportError:
        pass  # matplotlib not available
    
    print(f"Results saved to: {output_dir}")


def main():
    parser = argparse.ArgumentParser(description='Backtest trading strategies')
    parser.add_argument('--strategy', '-s', required=True, help='Strategy name')
    parser.add_argument('--symbol', required=True, help='Trading symbol (e.g., BTC-USD)')
    parser.add_argument('--period', '-p', help='Lookback period (e.g., 1y, 6m, 30d)')
    parser.add_argument('--start', help='Start date (YYYY-MM-DD)')
    parser.add_argument('--end', help='End date (YYYY-MM-DD)')
    parser.add_argument('--capital', '-c', type=float, default=10000, help='Initial capital')
    parser.add_argument('--params', help='Strategy parameters as JSON')
    parser.add_argument('--commission', type=float, default=0.001, help='Commission per trade')
    parser.add_argument('--slippage', type=float, default=0.0005, help='Slippage per trade')
    parser.add_argument('--output', '-o', help='Output directory')
    parser.add_argument('--list', action='store_true', help='List available strategies')
    parser.add_argument('--quiet', '-q', action='store_true', help='Minimal output')
    
    args = parser.parse_args()
    
    # List strategies
    if args.list:
        print("Available strategies:")
        for name, desc in list_strategies().items():
            print(f"  {name}: {desc}")
        return
    
    # Determine date range
    if args.start and args.end:
        start = datetime.strptime(args.start, '%Y-%m-%d')
        end = datetime.strptime(args.end, '%Y-%m-%d')
    elif args.period:
        end = datetime.now()
        start = end - parse_period(args.period)
    else:
        end = datetime.now()
        start = end - timedelta(days=365)
    
    # Parse parameters
    params = json.loads(args.params) if args.params else {}
    
    # Set up directories
    script_dir = Path(__file__).parent.parent
    data_dir = script_dir / 'data'
    output_dir = Path(args.output) if args.output else script_dir / 'reports'
    
    # Load data
    if not args.quiet:
        print(f"Loading data for {args.symbol} from {start.date()} to {end.date()}...")
    
    data = load_data(args.symbol, start, end, data_dir)
    data.attrs['symbol'] = args.symbol
    
    if len(data) < 50:
        print(f"Error: Insufficient data. Got {len(data)} bars, need at least 50.")
        sys.exit(1)
    
    if not args.quiet:
        print(f"Loaded {len(data)} bars")
        print(f"Running backtest with {args.strategy} strategy...")
    
    # Run backtest
    result = run_backtest(
        strategy_name=args.strategy,
        data=data,
        initial_capital=args.capital,
        params=params,
        commission=args.commission,
        slippage=args.slippage,
    )
    
    # Print results
    print(format_results(result))
    
    # Save results
    save_results(result, output_dir)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Historical Data Fetcher
Fetch and cache price data from various sources.

Usage:
    python fetch_data.py --symbol BTC-USD --period 2y --interval 1d
    python fetch_data.py --symbol ETH-USD --start 2020-01-01 --end 2024-01-01
"""

import argparse
from datetime import datetime, timedelta
from pathlib import Path
import sys


def parse_period(period: str) -> timedelta:
    """Parse period string like '1y', '6m', '30d'."""
    unit = period[-1].lower()
    value = int(period[:-1])
    
    if unit == 'y':
        return timedelta(days=value * 365)
    elif unit == 'm':
        return timedelta(days=value * 30)
    elif unit == 'd':
        return timedelta(days=value)
    elif unit == 'w':
        return timedelta(weeks=value)
    else:
        raise ValueError(f"Unknown period unit: {unit}")


def fetch_yfinance(symbol: str, start: datetime, end: datetime, interval: str) -> 'pd.DataFrame':
    """Fetch data from Yahoo Finance."""
    try:
        import yfinance as yf
        import pandas as pd
    except ImportError:
        print("yfinance not installed. Install with: pip install yfinance pandas")
        sys.exit(1)
    
    print(f"Fetching {symbol} from Yahoo Finance...")
    ticker = yf.Ticker(symbol)
    df = ticker.history(start=start, end=end, interval=interval)
    
    if df.empty:
        raise ValueError(f"No data returned for {symbol}")
    
    df.columns = [c.lower() for c in df.columns]
    df.index.name = 'date'
    
    return df


def fetch_coingecko(symbol: str, days: int) -> 'pd.DataFrame':
    """Fetch data from CoinGecko (crypto only)."""
    try:
        import requests
        import pandas as pd
    except ImportError:
        print("requests/pandas not installed")
        sys.exit(1)
    
    # Map common symbols to CoinGecko IDs
    symbol_map = {
        'BTC': 'bitcoin',
        'ETH': 'ethereum',
        'SOL': 'solana',
        'AVAX': 'avalanche-2',
        'MATIC': 'matic-network',
        'DOT': 'polkadot',
        'LINK': 'chainlink',
        'UNI': 'uniswap',
        'AAVE': 'aave',
    }
    
    coin_id = symbol_map.get(symbol.split('-')[0].upper(), symbol.lower())
    
    print(f"Fetching {coin_id} from CoinGecko...")
    url = f"https://api.coingecko.com/api/v3/coins/{coin_id}/market_chart"
    params = {'vs_currency': 'usd', 'days': days}
    
    response = requests.get(url, params=params)
    response.raise_for_status()
    data = response.json()
    
    df = pd.DataFrame(data['prices'], columns=['timestamp', 'close'])
    df['date'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('date', inplace=True)
    df.drop('timestamp', axis=1, inplace=True)
    
    # Add OHLV columns (approximations for daily)
    df['open'] = df['close'].shift(1).fillna(df['close'])
    df['high'] = df['close'] * 1.01  # Rough estimate
    df['low'] = df['close'] * 0.99
    df['volume'] = 0  # Not available in basic API
    
    return df[['open', 'high', 'low', 'close', 'volume']]


def main():
    parser = argparse.ArgumentParser(description='Fetch historical price data')
    parser.add_argument('--symbol', '-s', required=True, help='Trading symbol')
    parser.add_argument('--period', '-p', help='Lookback period (e.g., 2y, 6m)')
    parser.add_argument('--start', help='Start date (YYYY-MM-DD)')
    parser.add_argument('--end', help='End date (YYYY-MM-DD)')
    parser.add_argument('--interval', '-i', default='1d', help='Data interval (1d, 1h, etc.)')
    parser.add_argument('--source', default='yfinance', choices=['yfinance', 'coingecko'])
    parser.add_argument('--output', '-o', help='Output directory')
    
    args = parser.parse_args()
    
    # Determine date range
    if args.start and args.end:
        start = datetime.strptime(args.start, '%Y-%m-%d')
        end = datetime.strptime(args.end, '%Y-%m-%d')
    elif args.period:
        end = datetime.now()
        start = end - parse_period(args.period)
    else:
        end = datetime.now()
        start = end - timedelta(days=730)  # 2 years default
    
    # Fetch data
    if args.source == 'yfinance':
        df = fetch_yfinance(args.symbol, start, end, args.interval)
    else:
        days = (end - start).days
        df = fetch_coingecko(args.symbol, days)
    
    # Save to file
    script_dir = Path(__file__).parent.parent
    output_dir = Path(args.output) if args.output else script_dir / 'data'
    output_dir.mkdir(parents=True, exist_ok=True)
    
    filename = f"{args.symbol.replace('/', '_').replace('-', '_')}_{args.interval}.csv"
    output_file = output_dir / filename
    
    df.to_csv(output_file)
    
    print(f"Data saved to: {output_file}")
    print(f"  Rows: {len(df)}")
    print(f"  Date range: {df.index[0]} to {df.index[-1]}")
    print(f"  Columns: {list(df.columns)}")


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Performance and Risk Metrics for Backtesting
"""

import numpy as np
import pandas as pd
from typing import List, Dict, Any
from dataclasses import dataclass, field


@dataclass
class Trade:
    """Represents a completed trade."""
    entry_time: pd.Timestamp
    exit_time: pd.Timestamp
    entry_price: float
    exit_price: float
    direction: str  # "long" or "short"
    size: float
    pnl: float = 0.0
    pnl_pct: float = 0.0
    duration: pd.Timedelta = None
    
    def __post_init__(self):
        if self.direction == "long":
            self.pnl = (self.exit_price - self.entry_price) * self.size
            self.pnl_pct = (self.exit_price - self.entry_price) / self.entry_price * 100
        else:
            self.pnl = (self.entry_price - self.exit_price) * self.size
            self.pnl_pct = (self.entry_price - self.exit_price) / self.entry_price * 100
        self.duration = self.exit_time - self.entry_time


@dataclass
class BacktestResult:
    """Complete backtest results."""
    strategy: str
    symbol: str
    start_date: pd.Timestamp
    end_date: pd.Timestamp
    initial_capital: float
    final_capital: float
    trades: List[Trade]
    equity_curve: pd.Series
    parameters: Dict[str, Any]
    
    # Performance metrics
    total_return: float = 0.0
    cagr: float = 0.0
    sharpe_ratio: float = 0.0
    sortino_ratio: float = 0.0
    calmar_ratio: float = 0.0
    
    # Risk metrics
    max_drawdown: float = 0.0
    max_drawdown_duration: int = 0
    volatility: float = 0.0
    var_95: float = 0.0
    cvar_95: float = 0.0
    ulcer_index: float = 0.0
    
    # Trade statistics
    total_trades: int = 0
    win_rate: float = 0.0
    profit_factor: float = 0.0
    avg_win: float = 0.0
    avg_loss: float = 0.0
    expectancy: float = 0.0
    max_consecutive_wins: int = 0
    max_consecutive_losses: int = 0
    avg_trade_duration: str = ""


def calculate_returns(equity_curve: pd.Series) -> pd.Series:
    """Calculate daily returns from equity curve."""
    return equity_curve.pct_change().dropna()


def calculate_total_return(initial: float, final: float) -> float:
    """Calculate total return percentage."""
    return ((final - initial) / initial) * 100


def calculate_cagr(initial: float, final: float, years: float) -> float:
    """Calculate Compound Annual Growth Rate."""
    if years <= 0 or initial <= 0:
        return 0.0
    return ((final / initial) ** (1 / years) - 1) * 100


def calculate_sharpe_ratio(returns: pd.Series, risk_free_rate: float = 0.02) -> float:
    """Calculate annualized Sharpe Ratio.
    
    Sharpe = (Return - Risk Free Rate) / Volatility
    """
    if len(returns) < 2 or returns.std() == 0:
        return 0.0
    
    # Annualize
    annual_return = returns.mean() * 252
    annual_vol = returns.std() * np.sqrt(252)
    
    return (annual_return - risk_free_rate) / annual_vol


def calculate_sortino_ratio(returns: pd.Series, risk_free_rate: float = 0.02) -> float:
    """Calculate Sortino Ratio (uses downside deviation only)."""
    if len(returns) < 2:
        return 0.0
    
    downside_returns = returns[returns < 0]
    if len(downside_returns) == 0 or downside_returns.std() == 0:
        return float('inf') if returns.mean() > 0 else 0.0
    
    annual_return = returns.mean() * 252
    downside_std = downside_returns.std() * np.sqrt(252)
    
    return (annual_return - risk_free_rate) / downside_std


def calculate_max_drawdown(equity_curve: pd.Series) -> tuple:
    """Calculate maximum drawdown and its duration.
    
    Returns: (max_drawdown_pct, max_drawdown_duration_days)
    """
    if len(equity_curve) < 2:
        return 0.0, 0
    
    rolling_max = equity_curve.expanding().max()
    drawdowns = (equity_curve - rolling_max) / rolling_max * 100
    
    max_dd = drawdowns.min()
    
    # Calculate duration
    in_drawdown = drawdowns < 0
    if not in_drawdown.any():
        return 0.0, 0
    
    # Find longest drawdown period
    drawdown_periods = []
    start = None
    for i, is_dd in enumerate(in_drawdown):
        if is_dd and start is None:
            start = i
        elif not is_dd and start is not None:
            drawdown_periods.append(i - start)
            start = None
    if start is not None:
        drawdown_periods.append(len(in_drawdown) - start)
    
    max_duration = max(drawdown_periods) if drawdown_periods else 0
    
    return max_dd, max_duration


def calculate_calmar_ratio(cagr: float, max_drawdown: float) -> float:
    """Calculate Calmar Ratio = CAGR / |Max Drawdown|."""
    if max_drawdown == 0:
        return 0.0
    return cagr / abs(max_drawdown)


def calculate_var(returns: pd.Series, confidence: float = 0.95) -> float:
    """Calculate Value at Risk at given confidence level."""
    if len(returns) < 10:
        return 0.0
    return np.percentile(returns, (1 - confidence) * 100)


def calculate_cvar(returns: pd.Series, confidence: float = 0.95) -> float:
    """Calculate Conditional VaR (Expected Shortfall)."""
    var = calculate_var(returns, confidence)
    return returns[returns <= var].mean() if len(returns[returns <= var]) > 0 else var


def calculate_volatility(returns: pd.Series) -> float:
    """Calculate annualized volatility."""
    return returns.std() * np.sqrt(252) * 100


def calculate_ulcer_index(equity_curve: pd.Series) -> float:
    """Calculate Ulcer Index (duration-weighted drawdown)."""
    if len(equity_curve) < 2:
        return 0.0
    
    rolling_max = equity_curve.expanding().max()
    drawdowns = ((equity_curve - rolling_max) / rolling_max * 100) ** 2
    
    return np.sqrt(drawdowns.mean())


def calculate_trade_stats(trades: List[Trade]) -> Dict[str, Any]:
    """Calculate trade statistics."""
    if not trades:
        return {
            "total_trades": 0,
            "win_rate": 0.0,
            "profit_factor": 0.0,
            "avg_win": 0.0,
            "avg_loss": 0.0,
            "expectancy": 0.0,
            "max_consecutive_wins": 0,
            "max_consecutive_losses": 0,
            "avg_trade_duration": "0d",
        }
    
    wins = [t for t in trades if t.pnl > 0]
    losses = [t for t in trades if t.pnl < 0]
    
    total_trades = len(trades)
    win_rate = len(wins) / total_trades * 100 if total_trades > 0 else 0
    
    gross_profit = sum(t.pnl for t in wins) if wins else 0
    gross_loss = abs(sum(t.pnl for t in losses)) if losses else 0
    profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf')
    
    avg_win = np.mean([t.pnl for t in wins]) if wins else 0
    avg_loss = np.mean([t.pnl for t in losses]) if losses else 0
    
    # Expectancy = (Win% * Avg Win) - (Loss% * |Avg Loss|)
    expectancy = (win_rate / 100 * avg_win) - ((1 - win_rate / 100) * abs(avg_loss))
    
    # Consecutive wins/losses
    max_consec_wins = 0
    max_consec_losses = 0
    current_wins = 0
    current_losses = 0
    
    for trade in trades:
        if trade.pnl > 0:
            current_wins += 1
            current_losses = 0
            max_consec_wins = max(max_consec_wins, current_wins)
        else:
            current_losses += 1
            current_wins = 0
            max_consec_losses = max(max_consec_losses, current_losses)
    
    # Average duration
    durations = [t.duration.days for t in trades if t.duration]
    avg_duration = f"{np.mean(durations):.1f}d" if durations else "0d"
    
    return {
        "total_trades": total_trades,
        "win_rate": win_rate,
        "profit_factor": profit_factor,
        "avg_win": avg_win,
        "avg_loss": avg_loss,
        "expectancy": expectancy,
        "max_consecutive_wins": max_consec_wins,
        "max_consecutive_losses": max_consec_losses,
        "avg_trade_duration": avg_duration,
    }


def calculate_all_metrics(result: BacktestResult) -> BacktestResult:
    """Calculate all performance and risk metrics for a backtest result."""
    
    returns = calculate_returns(result.equity_curve)
    years = (result.end_date - result.start_date).days / 365.25
    
    # Performance metrics
    result.total_return = calculate_total_return(result.initial_capital, result.final_capital)
    result.cagr = calculate_cagr(result.initial_capital, result.final_capital, years)
    result.sharpe_ratio = calculate_sharpe_ratio(returns)
    result.sortino_ratio = calculate_sortino_ratio(returns)
    
    # Risk metrics
    result.max_drawdown, result.max_drawdown_duration = calculate_max_drawdown(result.equity_curve)
    result.calmar_ratio = calculate_calmar_ratio(result.cagr, result.max_drawdown)
    result.volatility = calculate_volatility(returns)
    result.var_95 = calculate_var(returns, 0.95) * 100
    result.cvar_95 = calculate_cvar(returns, 0.95) * 100
    result.ulcer_index = calculate_ulcer_index(result.equity_curve)
    
    # Trade statistics
    trade_stats = calculate_trade_stats(result.trades)
    result.total_trades = trade_stats["total_trades"]
    result.win_rate = trade_stats["win_rate"]
    result.profit_factor = trade_stats["profit_factor"]
    result.avg_win = trade_stats["avg_win"]
    result.avg_loss = trade_stats["avg_loss"]
    result.expectancy = trade_stats["expectancy"]
    result.max_consecutive_wins = trade_stats["max_consecutive_wins"]
    result.max_consecutive_losses = trade_stats["max_consecutive_losses"]
    result.avg_trade_duration = trade_stats["avg_trade_duration"]
    
    return result


def format_results(result: BacktestResult) -> str:
    """Format backtest results as ASCII table."""
    
    params_str = ", ".join(f"{k}={v}" for k, v in result.parameters.items())
    
    return f"""
╔══════════════════════════════════════════════════════════════════════╗
║            BACKTEST RESULTS: {result.strategy.upper():^20}              ║
║            {result.symbol} | {result.start_date.strftime('%Y-%m-%d')} to {result.end_date.strftime('%Y-%m-%d')}             ║
╠══════════════════════════════════════════════════════════════════════╣
║ PERFORMANCE                          │ RISK                          ║
║ ─────────────────────────────────────┼─────────────────────────────  ║
║ Total Return:     {result.total_return:>+10.2f}%         │ Max Drawdown:   {result.max_drawdown:>+10.2f}%    ║
║ CAGR:             {result.cagr:>+10.2f}%         │ VaR (95%):      {result.var_95:>+10.2f}%    ║
║ Sharpe Ratio:     {result.sharpe_ratio:>10.2f}          │ Volatility:     {result.volatility:>10.2f}%    ║
║ Sortino Ratio:    {result.sortino_ratio:>10.2f}          │ Ulcer Index:    {result.ulcer_index:>10.2f}     ║
║ Calmar Ratio:     {result.calmar_ratio:>10.2f}          │ CVaR (95%):     {result.cvar_95:>+10.2f}%    ║
╠══════════════════════════════════════════════════════════════════════╣
║ TRADE STATISTICS                                                     ║
║ ─────────────────────────────────────────────────────────────────────║
║ Total Trades:     {result.total_trades:>10}          │ Profit Factor:  {result.profit_factor:>10.2f}     ║
║ Win Rate:         {result.win_rate:>10.1f}%         │ Expectancy:     ${result.expectancy:>10.2f}    ║
║ Avg Win:          ${result.avg_win:>10.2f}         │ Max Consec. Losses: {result.max_consecutive_losses:>5}    ║
║ Avg Loss:         ${result.avg_loss:>10.2f}         │ Avg Duration:   {result.avg_trade_duration:>10}     ║
╠══════════════════════════════════════════════════════════════════════╣
║ Capital: ${result.initial_capital:,.0f} → ${result.final_capital:,.0f}                                  ║
║ Parameters: {params_str:<56} ║
╚══════════════════════════════════════════════════════════════════════╝
"""

#!/usr/bin/env python3
"""
Strategy Parameter Optimizer
Grid search and optimization for trading strategy parameters.

Usage:
    python optimize.py --strategy sma_crossover --symbol BTC-USD \
        --param-grid '{"fast_period": [10,20,30], "slow_period": [50,100,200]}'
"""

import argparse
import json
import sys
from pathlib import Path
from datetime import datetime, timedelta
from itertools import product
from typing import Dict, Any, List

import pandas as pd

sys.path.insert(0, str(Path(__file__).parent))

from backtest import load_data, run_backtest, parse_period


def grid_search(
    strategy_name: str,
    data: pd.DataFrame,
    param_grid: Dict[str, List[Any]],
    initial_capital: float = 10000,
    metric: str = 'sharpe_ratio',
) -> pd.DataFrame:
    """Run grid search over parameter combinations."""
    
    # Generate all parameter combinations
    param_names = list(param_grid.keys())
    param_values = list(param_grid.values())
    combinations = list(product(*param_values))
    
    print(f"Testing {len(combinations)} parameter combinations...")
    
    results = []
    
    for i, combo in enumerate(combinations):
        params = dict(zip(param_names, combo))
        
        try:
            result = run_backtest(
                strategy_name=strategy_name,
                data=data.copy(),
                initial_capital=initial_capital,
                params=params,
            )
            
            results.append({
                **params,
                'total_return': result.total_return,
                'sharpe_ratio': result.sharpe_ratio,
                'sortino_ratio': result.sortino_ratio,
                'max_drawdown': result.max_drawdown,
                'win_rate': result.win_rate,
                'profit_factor': result.profit_factor,
                'total_trades': result.total_trades,
                'calmar_ratio': result.calmar_ratio,
            })
            
            # Progress indicator
            if (i + 1) % 10 == 0:
                print(f"  Completed {i + 1}/{len(combinations)}")
                
        except Exception as e:
            print(f"  Error with params {params}: {e}")
            continue
    
    df = pd.DataFrame(results)
    
    # Sort by target metric
    if metric in df.columns:
        df = df.sort_values(metric, ascending=False)
    
    return df


def format_optimization_results(df: pd.DataFrame, param_names: List[str]) -> str:
    """Format optimization results as table."""
    
    output = []
    output.append("=" * 80)
    output.append("PARAMETER OPTIMIZATION RESULTS")
    output.append("=" * 80)
    output.append("")
    
    # Top 10 results
    output.append("TOP 10 PARAMETER COMBINATIONS (by Sharpe Ratio):")
    output.append("-" * 80)
    
    header = param_names + ['Return%', 'Sharpe', 'MaxDD%', 'WinRate%', 'Trades']
    output.append("  ".join(f"{h:>10}" for h in header))
    output.append("-" * 80)
    
    for _, row in df.head(10).iterrows():
        values = [row[p] for p in param_names]
        values += [
            f"{row['total_return']:.1f}",
            f"{row['sharpe_ratio']:.2f}",
            f"{row['max_drawdown']:.1f}",
            f"{row['win_rate']:.1f}",
            f"{row['total_trades']:.0f}",
        ]
        output.append("  ".join(f"{v:>10}" for v in values))
    
    output.append("")
    output.append("=" * 80)
    
    # Best parameters
    best = df.iloc[0]
    output.append("BEST PARAMETERS:")
    for p in param_names:
        output.append(f"  {p}: {best[p]}")
    output.append("")
    output.append(f"Expected Performance:")
    output.append(f"  Total Return: {best['total_return']:.2f}%")
    output.append(f"  Sharpe Ratio: {best['sharpe_ratio']:.2f}")
    output.append(f"  Max Drawdown: {best['max_drawdown']:.2f}%")
    output.append(f"  Win Rate: {best['win_rate']:.1f}%")
    output.append("=" * 80)
    
    return "\n".join(output)


def main():
    parser = argparse.ArgumentParser(description='Optimize strategy parameters')
    parser.add_argument('--strategy', '-s', required=True, help='Strategy name')
    parser.add_argument('--symbol', required=True, help='Trading symbol')
    parser.add_argument('--param-grid', required=True, help='Parameter grid as JSON')
    parser.add_argument('--period', '-p', default='1y', help='Lookback period')
    parser.add_argument('--start', help='Start date (YYYY-MM-DD)')
    parser.add_argument('--end', help='End date (YYYY-MM-DD)')
    parser.add_argument('--capital', '-c', type=float, default=10000, help='Initial capital')
    parser.add_argument('--metric', '-m', default='sharpe_ratio', help='Optimization metric')
    parser.add_argument('--output', '-o', help='Output directory')
    
    args = parser.parse_args()
    
    # Parse parameter grid
    param_grid = json.loads(args.param_grid)
    
    # Determine date range
    if args.start and args.end:
        start = datetime.strptime(args.start, '%Y-%m-%d')
        end = datetime.strptime(args.end, '%Y-%m-%d')
    else:
        end = datetime.now()
        start = end - parse_period(args.period)
    
    # Load data
    script_dir = Path(__file__).parent.parent
    data_dir = script_dir / 'data'
    
    print(f"Loading data for {args.symbol}...")
    data = load_data(args.symbol, start, end, data_dir)
    data.attrs['symbol'] = args.symbol
    
    print(f"Loaded {len(data)} bars")
    
    # Run optimization
    results_df = grid_search(
        strategy_name=args.strategy,
        data=data,
        param_grid=param_grid,
        initial_capital=args.capital,
        metric=args.metric,
    )
    
    # Format and print results
    param_names = list(param_grid.keys())
    output = format_optimization_results(results_df, param_names)
    print(output)
    
    # Save results
    output_dir = Path(args.output) if args.output else script_dir / 'reports'
    output_dir.mkdir(parents=True, exist_ok=True)
    
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    
    csv_file = output_dir / f"optimization_{args.strategy}_{timestamp}.csv"
    results_df.to_csv(csv_file, index=False)
    print(f"\nFull results saved to: {csv_file}")
    
    txt_file = output_dir / f"optimization_{args.strategy}_{timestamp}.txt"
    with open(txt_file, 'w') as f:
        f.write(output)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Trading Strategy Definitions
Each strategy implements generate_signals() returning entry/exit signals.
"""

import numpy as np
import pandas as pd
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from dataclasses import dataclass


@dataclass
class Signal:
    """Trading signal with entry/exit information."""
    entry: bool = False
    exit: bool = False
    direction: str = "long"  # "long" or "short"
    strength: float = 1.0    # Signal strength 0-1


class Strategy(ABC):
    """Base class for all trading strategies."""
    
    name: str = "base"
    lookback: int = 1
    
    @abstractmethod
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        """Generate trading signals from price data."""
        pass
    
    def validate_params(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Validate and set default parameters."""
        return params


class SMAcrossover(Strategy):
    """Simple Moving Average Crossover Strategy.
    
    Buy when fast MA crosses above slow MA (golden cross).
    Sell when fast MA crosses below slow MA (death cross).
    """
    
    name = "sma_crossover"
    lookback = 200
    
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        params = self.validate_params(params)
        fast = params.get("fast_period", 20)
        slow = params.get("slow_period", 50)
        
        if len(data) < slow + 1:
            return Signal()
        
        close = data["close"]
        fast_ma = close.rolling(window=fast).mean()
        slow_ma = close.rolling(window=slow).mean()
        
        # Current and previous values
        curr_fast, prev_fast = fast_ma.iloc[-1], fast_ma.iloc[-2]
        curr_slow, prev_slow = slow_ma.iloc[-1], slow_ma.iloc[-2]
        
        # Golden cross: fast crosses above slow
        if prev_fast <= prev_slow and curr_fast > curr_slow:
            return Signal(entry=True, direction="long")
        
        # Death cross: fast crosses below slow
        if prev_fast >= prev_slow and curr_fast < curr_slow:
            return Signal(exit=True)
        
        return Signal()


class EMAcrossover(Strategy):
    """Exponential Moving Average Crossover Strategy."""
    
    name = "ema_crossover"
    lookback = 200
    
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        fast = params.get("fast_period", 12)
        slow = params.get("slow_period", 26)
        
        if len(data) < slow + 1:
            return Signal()
        
        close = data["close"]
        fast_ema = close.ewm(span=fast, adjust=False).mean()
        slow_ema = close.ewm(span=slow, adjust=False).mean()
        
        curr_fast, prev_fast = fast_ema.iloc[-1], fast_ema.iloc[-2]
        curr_slow, prev_slow = slow_ema.iloc[-1], slow_ema.iloc[-2]
        
        if prev_fast <= prev_slow and curr_fast > curr_slow:
            return Signal(entry=True, direction="long")
        
        if prev_fast >= prev_slow and curr_fast < curr_slow:
            return Signal(exit=True)
        
        return Signal()


class RSIreversal(Strategy):
    """RSI Overbought/Oversold Reversal Strategy.
    
    Buy when RSI crosses above oversold level.
    Sell when RSI crosses below overbought level.
    """
    
    name = "rsi_reversal"
    lookback = 14
    
    def _calculate_rsi(self, close: pd.Series, period: int) -> pd.Series:
        delta = close.diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
        rs = gain / loss
        return 100 - (100 / (1 + rs))
    
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        period = params.get("period", 14)
        overbought = params.get("overbought", 70)
        oversold = params.get("oversold", 30)
        
        if len(data) < period + 1:
            return Signal()
        
        rsi = self._calculate_rsi(data["close"], period)
        curr_rsi, prev_rsi = rsi.iloc[-1], rsi.iloc[-2]
        
        # Oversold reversal: buy signal
        if prev_rsi <= oversold and curr_rsi > oversold:
            return Signal(entry=True, direction="long", strength=min(1.0, (oversold - prev_rsi) / 10))
        
        # Overbought reversal: sell signal
        if prev_rsi >= overbought and curr_rsi < overbought:
            return Signal(exit=True)
        
        return Signal()


class MACD(Strategy):
    """MACD Signal Line Crossover Strategy."""
    
    name = "macd"
    lookback = 35
    
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        fast = params.get("fast", 12)
        slow = params.get("slow", 26)
        signal_period = params.get("signal", 9)
        
        if len(data) < slow + signal_period:
            return Signal()
        
        close = data["close"]
        fast_ema = close.ewm(span=fast, adjust=False).mean()
        slow_ema = close.ewm(span=slow, adjust=False).mean()
        macd_line = fast_ema - slow_ema
        signal_line = macd_line.ewm(span=signal_period, adjust=False).mean()
        
        curr_macd, prev_macd = macd_line.iloc[-1], macd_line.iloc[-2]
        curr_signal, prev_signal = signal_line.iloc[-1], signal_line.iloc[-2]
        
        # MACD crosses above signal: buy
        if prev_macd <= prev_signal and curr_macd > curr_signal:
            return Signal(entry=True, direction="long")
        
        # MACD crosses below signal: sell
        if prev_macd >= prev_signal and curr_macd < curr_signal:
            return Signal(exit=True)
        
        return Signal()


class BollingerBands(Strategy):
    """Bollinger Bands Mean Reversion Strategy.
    
    Buy when price touches lower band.
    Sell when price touches upper band.
    """
    
    name = "bollinger_bands"
    lookback = 20
    
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        period = params.get("period", 20)
        std_dev = params.get("std_dev", 2.0)
        
        if len(data) < period:
            return Signal()
        
        close = data["close"]
        sma = close.rolling(window=period).mean()
        std = close.rolling(window=period).std()
        
        upper_band = sma + (std * std_dev)
        lower_band = sma - (std * std_dev)
        
        curr_close = close.iloc[-1]
        prev_close = close.iloc[-2]
        
        # Price crosses below lower band: buy
        if prev_close >= lower_band.iloc[-2] and curr_close < lower_band.iloc[-1]:
            return Signal(entry=True, direction="long")
        
        # Price crosses above upper band: sell
        if prev_close <= upper_band.iloc[-2] and curr_close > upper_band.iloc[-1]:
            return Signal(exit=True)
        
        return Signal()


class Breakout(Strategy):
    """Price Breakout Strategy.
    
    Buy when price breaks above recent high.
    Sell when price breaks below recent low.
    """
    
    name = "breakout"
    lookback = 20
    
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        lookback = params.get("lookback", 20)
        threshold = params.get("threshold", 0.0)  # % above/below
        
        if len(data) < lookback + 1:
            return Signal()
        
        high = data["high"].iloc[-lookback-1:-1]
        low = data["low"].iloc[-lookback-1:-1]
        curr_close = data["close"].iloc[-1]
        
        resistance = high.max() * (1 + threshold / 100)
        support = low.min() * (1 - threshold / 100)
        
        # Breakout above resistance
        if curr_close > resistance:
            return Signal(entry=True, direction="long")
        
        # Breakdown below support
        if curr_close < support:
            return Signal(exit=True)
        
        return Signal()


class MeanReversion(Strategy):
    """Mean Reversion Strategy.
    
    Buy when price deviates significantly below moving average.
    Sell when price reverts to or exceeds moving average.
    """
    
    name = "mean_reversion"
    lookback = 20
    
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        period = params.get("period", 20)
        z_threshold = params.get("z_threshold", 2.0)
        
        if len(data) < period:
            return Signal()
        
        close = data["close"]
        sma = close.rolling(window=period).mean()
        std = close.rolling(window=period).std()
        
        z_score = (close.iloc[-1] - sma.iloc[-1]) / std.iloc[-1]
        prev_z_score = (close.iloc[-2] - sma.iloc[-2]) / std.iloc[-2]
        
        # Price significantly below mean: buy
        if z_score < -z_threshold and prev_z_score >= -z_threshold:
            return Signal(entry=True, direction="long", strength=min(1.0, abs(z_score) / 3))
        
        # Price reverts to mean: sell
        if z_score >= 0 and prev_z_score < 0:
            return Signal(exit=True)
        
        return Signal()


class Momentum(Strategy):
    """Rate of Change Momentum Strategy."""
    
    name = "momentum"
    lookback = 14
    
    def generate_signals(self, data: pd.DataFrame, params: Dict[str, Any]) -> Signal:
        period = params.get("period", 14)
        threshold = params.get("threshold", 5.0)  # % change threshold
        
        if len(data) < period + 1:
            return Signal()
        
        close = data["close"]
        roc = ((close.iloc[-1] - close.iloc[-period]) / close.iloc[-period]) * 100
        prev_roc = ((close.iloc[-2] - close.iloc[-period-1]) / close.iloc[-period-1]) * 100
        
        # Momentum turns positive and exceeds threshold
        if prev_roc <= threshold and roc > threshold:
            return Signal(entry=True, direction="long")
        
        # Momentum turns negative
        if prev_roc >= 0 and roc < 0:
            return Signal(exit=True)
        
        return Signal()


# Strategy registry
STRATEGIES = {
    "sma_crossover": SMAcrossover(),
    "ema_crossover": EMAcrossover(),
    "rsi_reversal": RSIreversal(),
    "macd": MACD(),
    "bollinger_bands": BollingerBands(),
    "breakout": Breakout(),
    "mean_reversion": MeanReversion(),
    "momentum": Momentum(),
}


def get_strategy(name: str) -> Strategy:
    """Get strategy by name."""
    if name not in STRATEGIES:
        raise ValueError(f"Unknown strategy: {name}. Available: {list(STRATEGIES.keys())}")
    return STRATEGIES[name]


def list_strategies() -> Dict[str, str]:
    """List all available strategies with descriptions."""
    return {name: strategy.__doc__.split('\n')[0] for name, strategy in STRATEGIES.items()}