backtest.py

#!/usr/bin/env python3
import argparse
import json
import sys
import os
from datetime import datetime, timedelta
from dotenv import load_dotenv


load_dotenv()


sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from core.backtesting import BacktestEngine
from core.metrics import generate_performance_report, calculate_buy_and_hold


def parse_arguments():

    parser = argparse.ArgumentParser(
        description='Backtest TradeGraph agents on historical market data',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python backtest.py SPY
  python backtest.py NVDA --start 2023-01-01 --end 2024-01-01
  python backtest.py AAPL --interval 1d --period 6mo
  python backtest.py TSLA --capital 20000 --output results.json
        """
    )
    
    parser.add_argument('ticker', type=str,
                        help='Stock ticker symbol (e.g., SPY, NVDA, AAPL)')
    

    date_group = parser.add_mutually_exclusive_group()
    date_group.add_argument('--start', type=str,
                           help='Start date (YYYY-MM-DD). Requires --end')
    date_group.add_argument('--period', type=str, default='1y',
                           help='Period (e.g., 1y, 6mo, 3mo, 1mo). Default: 1y')
    
    parser.add_argument('--end', type=str,
                        help='End date (YYYY-MM-DD). Requires --start. Default: today')
    
    parser.add_argument('--interval', type=str, default='1d',
                        choices=['1d', '1h', '5m', '15m', '30m', '1wk', '1mo'],
                        help='Data interval. Default: 1d (daily)')
    
    parser.add_argument('--capital', type=float, default=10000.0,
                        help='Initial capital. Default: $10,000')
    
    parser.add_argument('--output', type=str,
                        help='Output file path for detailed JSON results (optional)')
    
    parser.add_argument('--verbose', action='store_true',
                        help='Show detailed trade-by-trade output')
    
    return parser.parse_args()


def validate_dates(args):

    if args.start and args.end:

        try:
            start = datetime.strptime(args.start, '%Y-%m-%d')
            end = datetime.strptime(args.end, '%Y-%m-%d')
            
            if start >= end:
                print("Error: Start date must be before end date")
                sys.exit(1)
                
            return args.start, args.end
        except ValueError:
            print("Error: Invalid date format. Use YYYY-MM-DD")
            sys.exit(1)
    
    elif args.start and not args.end:
        print("Error: --start requires --end")
        sys.exit(1)
    
    elif args.end and not args.start:
        print("Error: --end requires --start")
        sys.exit(1)
    
    else:

        end = datetime.now()
        

        period_map = {
            '1y': 365, '2y': 730, '5y': 1825,
            '6mo': 180, '3mo': 90, '1mo': 30,
            '1w': 7, '1d': 1
        }
        
        period_lower = args.period.lower()
        if period_lower not in period_map:
            print(f"Error: Invalid period '{args.period}'. Use format like '1y', '6mo', '3mo', etc.")
            sys.exit(1)
        
        days = period_map[period_lower]
        start = end - timedelta(days=days)
        
        return start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d')


def main():

    args = parse_arguments()
    

    start_date, end_date = validate_dates(args)
    

    print("\n" + "="*70)
    print("  TradeGraph Backtesting Engine")
    print("="*70)
    print(f"\nTicker:          {args.ticker}")
    print(f"Period:          {start_date} to {end_date}")
    print(f"Interval:        {args.interval}")
    print(f"Initial Capital: ${args.capital:,.2f}")
    print()
    

    engine = BacktestEngine(
        ticker=args.ticker,
        start_date=start_date,
        end_date=end_date,
        interval=args.interval,
        initial_capital=args.capital
    )
    
    results = engine.run()
    

    if 'error' in results:
        print(f"\nError: {results['error']}")
        sys.exit(1)
    

    if engine.historical_data:
        bh_results = calculate_buy_and_hold(engine.historical_data, args.capital)
        print(f"\n{'='*70}")
        print("  BUY-AND-HOLD BENCHMARK")
        print(f"{'='*70}")
        print(f"Final Equity:    ${bh_results['final_equity']:,.2f}")
        print(f"Total Return:    {bh_results['total_return_pct']:+.2f}%")
        print(f"Shares Bought:   {bh_results['shares_purchased']:.2f}")
        

        results['benchmark'] = bh_results
    

    report = generate_performance_report(results)
    print(report)
    

    if args.verbose and results['trade_log']:
        print("\n" + "="*70)
        print("  TRADE LOG")
        print("="*70)
        print(f"{'Date':<12} {'Action':<30} {'Price':<10} {'Equity':<12}")
        print("-"*70)
        
        for trade in results['trade_log']:
            date_str = trade['timestamp'].strftime('%Y-%m-%d')
            action = trade['action'][:28]
            price = trade['price']
            equity = trade['total_equity']
            
            print(f"{date_str:<12} {action:<30} ${price:<9.2f} ${equity:<11,.2f}")
    

    if args.output:

        json_results = results.copy()
        

        json_results['equity_curve'] = [
            {**point, 'timestamp': point['timestamp'].isoformat()}
            for point in results['equity_curve']
        ]
        

        json_results['trade_log'] = [
            {**trade, 'timestamp': trade['timestamp'].isoformat()}
            for trade in results['trade_log']
        ]
        

        output_dir = os.path.dirname(args.output)
        if output_dir and not os.path.exists(output_dir):
            os.makedirs(output_dir)
        
        with open(args.output, 'w') as f:
            json.dump(json_results, f, indent=2)
        
        print(f"\n✓ Detailed results saved to: {args.output}")
    
    print()


if __name__ == "__main__":
    main()