run_strategy_backtests.py

"""
Strategy Backtest Runner

Backtests the 4 deduced strategies (3 Discord + 1 Telegram) against historical data
using BacktestingMCP engine, then compares results with original signal performance.
"""

import sys
import os
import json
import importlib
from datetime import datetime
from pathlib import Path

# Windows console encoding fix
if sys.platform == 'win32':
    import io
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

# Add BacktestingMCP to sys.path so its src package is importable
BACKTESTING_MCP_DIR = str(Path(r"C:\Users\danyw\Documents\Git\DanywayGit\BacktestingMCP"))
if BACKTESTING_MCP_DIR not in sys.path:
    sys.path.insert(0, BACKTESTING_MCP_DIR)

# Remove BackTestingSignals src from path to avoid conflicts
BACKTEST_SIGNALS_DIR = str(Path(__file__).parent)
sys.path = [p for p in sys.path if p != BACKTEST_SIGNALS_DIR]

from src.core.backtesting_engine import BacktestingEngine
from config.settings import TimeFrame
from src.strategies.generated.discord_v1_trend import DiscordV1TrendStrategy
from src.strategies.generated.discord_v2_meanreversion import DiscordV2MeanReversionStrategy
from src.strategies.generated.discord_v3_trendlong import DiscordV3TrendLongStrategy
from src.strategies.generated.telegram_trend_momentum import TelegramTrendMomentumStrategy


# ============================================================
# Configuration: which symbols/timeframes to test per strategy
# ============================================================

# Use 1h data for Discord strategies (closest available to 45M/2H)
# Use native timeframes for Telegram strategies where data exists

STRATEGY_TESTS = {
    "discord_v1_trend": {
        "class": DiscordV1TrendStrategy,
        "description": "MACD + ADX + SuperTrend trend-following (LONG & SHORT)",
        "tests": [
            # High-volume symbols from v1.x_2H (310 signals) and v1.x_45M (225 signals)
            {"symbol": "BTCUSDT", "timeframe": TimeFrame.H1, "label": "BTC 1H"},
            {"symbol": "ETHUSDT", "timeframe": TimeFrame.H1, "label": "ETH 1H"},
            {"symbol": "SOLUSDT", "timeframe": TimeFrame.H1, "label": "SOL 1H"},
            {"symbol": "XRPUSDT", "timeframe": TimeFrame.H1, "label": "XRP 1H"},
            {"symbol": "LINKUSDT", "timeframe": TimeFrame.H1, "label": "LINK 1H"},
            {"symbol": "ADAUSDT", "timeframe": TimeFrame.H1, "label": "ADA 1H"},
            {"symbol": "BNBUSDT", "timeframe": TimeFrame.H1, "label": "BNB 1H"},
            {"symbol": "DOGEUSDT", "timeframe": TimeFrame.H1, "label": "DOGE 1H"},
        ],
    },
    "discord_v2_meanreversion": {
        "class": DiscordV2MeanReversionStrategy,
        "description": "RSI + Bollinger Bands + ADX mean reversion (LONG & SHORT)",
        "tests": [
            # From v2.x_2H (265 signals)
            {"symbol": "BTCUSDT", "timeframe": TimeFrame.H1, "label": "BTC 1H"},
            {"symbol": "ETHUSDT", "timeframe": TimeFrame.H1, "label": "ETH 1H"},
            {"symbol": "SOLUSDT", "timeframe": TimeFrame.H1, "label": "SOL 1H"},
            {"symbol": "XRPUSDT", "timeframe": TimeFrame.H1, "label": "XRP 1H"},
            {"symbol": "LINKUSDT", "timeframe": TimeFrame.H1, "label": "LINK 1H"},
            {"symbol": "ADAUSDT", "timeframe": TimeFrame.H1, "label": "ADA 1H"},
            {"symbol": "BNBUSDT", "timeframe": TimeFrame.H1, "label": "BNB 1H"},
            {"symbol": "DOGEUSDT", "timeframe": TimeFrame.H1, "label": "DOGE 1H"},
        ],
    },
    "discord_v3_trendlong": {
        "class": DiscordV3TrendLongStrategy,
        "description": "EMA alignment + BB dip + ADX (LONG ONLY)",
        "tests": [
            # From v3.x_2H (132 signals) and v3.x_45M (134 signals)
            {"symbol": "BTCUSDT", "timeframe": TimeFrame.H1, "label": "BTC 1H"},
            {"symbol": "ETHUSDT", "timeframe": TimeFrame.H1, "label": "ETH 1H"},
            {"symbol": "SOLUSDT", "timeframe": TimeFrame.H1, "label": "SOL 1H"},
            {"symbol": "XRPUSDT", "timeframe": TimeFrame.H1, "label": "XRP 1H"},
            {"symbol": "LINKUSDT", "timeframe": TimeFrame.H1, "label": "LINK 1H"},
            {"symbol": "ADAUSDT", "timeframe": TimeFrame.H1, "label": "ADA 1H"},
            {"symbol": "BNBUSDT", "timeframe": TimeFrame.H1, "label": "BNB 1H"},
            {"symbol": "DOGEUSDT", "timeframe": TimeFrame.H1, "label": "DOGE 1H"},
        ],
    },
    "telegram_trend_momentum": {
        "class": TelegramTrendMomentumStrategy,
        "description": "Full EMA cascade + MACD + RSI momentum + McGinley (LONG & SHORT)",
        "tests": [
            # Telegram strategies are single-symbol - test each on native timeframe
            {"symbol": "ADAUSDT", "timeframe": TimeFrame.H1, "label": "ADA 1H (SuperF)"},
            {"symbol": "BTCUSDT", "timeframe": TimeFrame.H1, "label": "BTC 1H (Trendhoo)"},
            {"symbol": "DOGEUSDT", "timeframe": TimeFrame.H1, "label": "DOGE 1H (Liquidity Spec)"},
            {"symbol": "LINKUSDT", "timeframe": TimeFrame.H1, "label": "LINK 1H (McGinley)"},
            {"symbol": "NEARUSDT", "timeframe": TimeFrame.M30, "label": "NEAR 30M (Trendhoo)"},
            {"symbol": "SOLUSDT", "timeframe": TimeFrame.M30, "label": "SOL 30M (Precision Trend)"},
            {"symbol": "BTCUSDT", "timeframe": TimeFrame.M15, "label": "BTC 15M (Stiff Zone)"},
            {"symbol": "AGLDUSDT", "timeframe": TimeFrame.H1, "label": "AGLD 1H (Trigger Happy)"},
        ],
    },
}

# Date range covering the signal period (Jun 2024 - Mar 2026)
START_DATE = datetime(2024, 6, 1)
END_DATE = datetime(2026, 3, 31)


def run_all_backtests():
    """Run backtests for all strategies and collect results."""
    engine = BacktestingEngine()
    all_results = {}

    for strategy_name, config in STRATEGY_TESTS.items():
        strategy_class = config["class"]
        print(f"\n{'='*70}")
        print(f"Strategy: {strategy_name}")
        print(f"Description: {config['description']}")
        print(f"{'='*70}")

        strategy_results = []

        for test in config["tests"]:
            symbol = test["symbol"]
            timeframe = test["timeframe"]
            label = test["label"]

            print(f"\n  Testing {label}...", end=" ", flush=True)

            try:
                result = engine.run_backtest(
                    strategy_class=strategy_class,
                    symbol=symbol,
                    timeframe=timeframe,
                    start_date=START_DATE,
                    end_date=END_DATE,
                    cash=100_000,
                )

                stats = result.stats
                trades = result.trades

                # Extract key metrics (using BacktestingEngine stat keys)
                total_return = stats.get('total_return_pct', 0)
                win_rate = stats.get('win_rate_pct', 0)
                num_trades = stats.get('num_trades', 0)
                sharpe = stats.get('sharpe_ratio', 0)
                max_dd = stats.get('max_drawdown_pct', 0)
                avg_trade = stats.get('avg_trade_pct', 0)
                profit_factor = stats.get('profit_factor', 0)

                test_result = {
                    "label": label,
                    "symbol": symbol,
                    "timeframe": timeframe.value,
                    "total_return_pct": round(float(total_return), 2) if total_return else 0,
                    "win_rate_pct": round(float(win_rate), 2) if win_rate else 0,
                    "num_trades": int(num_trades) if num_trades else 0,
                    "sharpe_ratio": round(float(sharpe), 4) if sharpe else 0,
                    "max_drawdown_pct": round(float(max_dd), 2) if max_dd else 0,
                    "avg_trade_pct": round(float(avg_trade), 4) if avg_trade else 0,
                    "profit_factor": round(float(profit_factor), 4) if profit_factor else 0,
                }
                strategy_results.append(test_result)

                status = "WIN" if total_return and total_return > 0 else "LOSS"
                print(f"{status} | Return: {test_result['total_return_pct']}% | "
                      f"WR: {test_result['win_rate_pct']}% | "
                      f"Trades: {test_result['num_trades']} | "
                      f"Sharpe: {test_result['sharpe_ratio']}")

            except Exception as e:
                print(f"ERROR: {e}")
                strategy_results.append({
                    "label": label,
                    "symbol": symbol,
                    "timeframe": timeframe.value,
                    "error": str(e),
                })

        all_results[strategy_name] = {
            "description": config["description"],
            "results": strategy_results,
        }

    return all_results


def compute_strategy_summary(all_results: dict) -> dict:
    """Calculate aggregate stats per strategy."""
    summaries = {}

    for strategy_name, data in all_results.items():
        results = [r for r in data["results"] if "error" not in r]
        if not results:
            summaries[strategy_name] = {"error": "No successful backtests"}
            continue

        total_trades = sum(r["num_trades"] for r in results)
        avg_return = sum(r["total_return_pct"] for r in results) / len(results)
        avg_win_rate = sum(r["win_rate_pct"] for r in results if r["num_trades"] > 0) / max(
            len([r for r in results if r["num_trades"] > 0]), 1
        )
        avg_sharpe = sum(r["sharpe_ratio"] for r in results) / len(results)
        worst_dd = min(r["max_drawdown_pct"] for r in results)
        profitable_symbols = sum(1 for r in results if r["total_return_pct"] > 0)

        summaries[strategy_name] = {
            "description": data["description"],
            "symbols_tested": len(results),
            "symbols_profitable": profitable_symbols,
            "total_trades": total_trades,
            "avg_return_pct": round(avg_return, 2),
            "avg_win_rate_pct": round(avg_win_rate, 2),
            "avg_sharpe_ratio": round(avg_sharpe, 4),
            "worst_drawdown_pct": round(worst_dd, 2),
            "per_symbol": results,
        }

    return summaries


def print_comparison_report(summaries: dict):
    """Print a formatted comparison report."""
    print(f"\n\n{'='*80}")
    print("STRATEGY COMPARISON REPORT")
    print(f"{'='*80}")
    print(f"Backtest Period: {START_DATE.strftime('%Y-%m-%d')} to {END_DATE.strftime('%Y-%m-%d')}")
    print(f"Initial Capital: $100,000 per symbol")
    print()

    # Header
    header = f"{'Strategy':<30} {'Avg Return%':>12} {'Avg WR%':>10} {'Trades':>8} {'Sharpe':>8} {'Max DD%':>10} {'Profitable':>12}"
    print(header)
    print("-" * len(header))

    for name, summary in summaries.items():
        if "error" in summary:
            print(f"{name:<30} {'ERROR':>12}")
            continue

        prof_str = f"{summary['symbols_profitable']}/{summary['symbols_tested']}"
        print(
            f"{name:<30} "
            f"{summary['avg_return_pct']:>11.2f}% "
            f"{summary['avg_win_rate_pct']:>9.2f}% "
            f"{summary['total_trades']:>8d} "
            f"{summary['avg_sharpe_ratio']:>8.4f} "
            f"{summary['worst_drawdown_pct']:>9.2f}% "
            f"{prof_str:>12}"
        )

    print()

    # Per-symbol detail for each strategy
    for name, summary in summaries.items():
        if "error" in summary:
            continue

        print(f"\n--- {name}: {summary['description']} ---")
        detail_header = f"  {'Symbol/TF':<25} {'Return%':>10} {'WR%':>8} {'Trades':>8} {'Sharpe':>8} {'DD%':>8} {'PF':>8}"
        print(detail_header)
        print("  " + "-" * (len(detail_header) - 2))

        for r in summary["per_symbol"]:
            print(
                f"  {r['label']:<25} "
                f"{r['total_return_pct']:>9.2f}% "
                f"{r['win_rate_pct']:>7.2f}% "
                f"{r['num_trades']:>8d} "
                f"{r['sharpe_ratio']:>8.4f} "
                f"{r['max_drawdown_pct']:>7.2f}% "
                f"{r['profit_factor']:>8.4f}"
            )


def main():
    print("=" * 70)
    print("DEDUCED STRATEGY BACKTESTING")
    print("Testing 4 strategies across multiple symbols")
    print("=" * 70)

    # Run all backtests
    all_results = run_all_backtests()

    # Compute summaries
    summaries = compute_strategy_summary(all_results)

    # Print comparison
    print_comparison_report(summaries)

    # Save results
    output_dir = Path("data/analysis")
    output_dir.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_file = output_dir / f"strategy_backtest_comparison_{timestamp}.json"

    # Convert non-serializable values
    serializable = json.loads(json.dumps(summaries, default=str))
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(serializable, f, indent=2, ensure_ascii=False)

    print(f"\nResults saved to: {output_file}")


if __name__ == "__main__":
    main()