bifitnex-trading/backtest/run_backtest.py

#!/usr/bin/env python3
"""CLI entry point for backtesting."""

import argparse
import csv
import json
import logging
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

import config
from backtest.data_loader import load_or_fetch, load_or_fetch_perp
from backtest.engine import BacktestEngine

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("backtest")

RESULTS_DIR = os.path.join(os.path.dirname(__file__), "results")


def print_report(results: dict):
    """Print formatted backtest report to stdout."""
    print("\n" + "=" * 60)
    print("  BACKTEST RESULTS")
    print("=" * 60)
    print(f"  Starting Capital:    {results['starting_capital']:>12,.2f} USDT")
    print(f"  Final Equity:        {results['final_equity']:>12,.2f} USDT")
    print(f"  Total Return:        {results['total_return_pct']:>+11.2f}%")
    print(f"  Annualized Return:   {results['annualized_return_pct']:>+11.2f}%")
    print(f"  Max Drawdown:        {results['max_drawdown_pct']:>11.2f}%")
    print(f"  Period:              {results['days']:>8d} days")
    print("-" * 60)
    print(f"  Total BUYs:          {results['total_buys']:>8d}")
    print(f"  Total SELLs:         {results['total_sells']:>8d}")
    print(f"    - Signal SELL:     {results['signal_sell_count']:>8d}")
    print(f"    - Take Profit:     {results['take_profit_count']:>8d}")
    print(f"    - Trailing Stop:   {results.get('trailing_stop_count', 0):>8d}")
    print(f"    - Stop Loss:       {results['stop_loss_count']:>8d}")
    print(f"    - Timeout:         {results.get('timeout_count', 0):>8d}")
    print("-" * 60)
    print(f"  Win Rate:            {results['win_rate_pct']:>11.2f}%")
    print(f"  Profit Factor:       {results['profit_factor']:>11.2f}")
    print(f"  Avg Win:             {results['avg_win_usdt']:>+11.2f} USDT")
    print(f"  Avg Loss:            {results['avg_loss_usdt']:>11.2f} USDT")
    print(f"  Gross Profit:        {results['gross_profit']:>+11.2f} USDT")
    print(f"  Gross Loss:          {results['gross_loss']:>11.2f} USDT")
    open_positions = results['total_buys'] - results['total_sells']
    print(f"  Open Positions:      {open_positions:>8d}")
    print("=" * 60 + "\n")


def save_results(results: dict):
    """Save trades CSV, summary JSON, and equity curve CSV."""
    os.makedirs(RESULTS_DIR, exist_ok=True)

    # Summary JSON (without large lists)
    summary = {k: v for k, v in results.items() if k not in ("trades", "equity_curve")}
    with open(os.path.join(RESULTS_DIR, "summary.json"), "w") as f:
        json.dump(summary, f, indent=2)

    # Trades CSV
    trades = results.get("trades", [])
    if trades:
        trades_path = os.path.join(RESULTS_DIR, "trades.csv")
        all_keys = set()
        for t in trades:
            all_keys.update(t.keys())
        keys = sorted(all_keys)
        with open(trades_path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
            writer.writeheader()
            writer.writerows(trades)
        logger.info("Saved %d trades to %s", len(trades), trades_path)

    # Equity curve CSV
    eq = results.get("equity_curve", [])
    if eq:
        eq_path = os.path.join(RESULTS_DIR, "equity_curve.csv")
        with open(eq_path, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["timestamp", "equity"])
            for ts, val in eq:
                writer.writerow([str(ts), f"{val:.2f}"])
        logger.info("Saved equity curve to %s", eq_path)


def main():
    parser = argparse.ArgumentParser(description="Backtest trading strategy")
    parser.add_argument("--start", default="2026-01-01", help="Start date (YYYY-MM-DD)")
    parser.add_argument("--end", default="2026-03-17", help="End date (YYYY-MM-DD)")
    parser.add_argument("--capital", type=float, default=10000.0, help="Starting capital (USDT)")
    parser.add_argument("--symbols", default=None, help="Comma-separated symbols (default: all 15)")
    parser.add_argument("--no-context", action="store_true", help="Disable non-price context filters (for A/B comparison)")
    args = parser.parse_args()

    symbols = args.symbols.split(",") if args.symbols else config.TOP_15_SYMBOLS

    logger.info("Backtest: %s to %s, capital=%.2f USDT, %d symbols",
                args.start, args.end, args.capital, len(symbols))

    # Step 1: Load data
    logger.info("Loading historical data...")
    data = load_or_fetch(symbols, args.start, args.end)

    # Filter symbols with sufficient data
    valid = {sym: d for sym, d in data.items() if not d["candles_5m"].empty}
    logger.info("Valid symbols: %d / %d", len(valid), len(symbols))

    if not valid:
        logger.error("No valid data found")
        return

    # Step 1b: Load BTC perpetual data for funding/basis context
    btc_perp_df = None
    use_context = not args.no_context
    if use_context:
        logger.info("Loading BTC perpetual data for funding context...")
        btc_perp_df = load_or_fetch_perp(args.start, args.end)
        if btc_perp_df is not None and not btc_perp_df.empty:
            logger.info("BTC perp data: %d candles", len(btc_perp_df))
        else:
            logger.warning("No BTC perp data available — funding context disabled")
    else:
        logger.info("Context filters disabled (--no-context)")

    # Step 2: Run backtest
    logger.info("Running simulation...")
    engine = BacktestEngine(starting_capital=args.capital)
    results = engine.run(valid, btc_perp_1h=btc_perp_df, use_context=use_context)

    # Step 3: Output
    print_report(results)
    save_results(results)


if __name__ == "__main__":
    main()