ai-exchange/scripts/run_backtest.py

from pathlib import Path
import json
import os
import sys
from typing import Optional

from sqlalchemy.exc import SQLAlchemyError

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from src.services.backtest import (
    BacktestCostConfig,
    BacktestQualityGate,
    BacktestService,
    diagnose_backtest_quality,
    evaluate_backtest_quality,
    summarize_backtest_outcomes,
)


def build_cost_config_from_env(env=None) -> BacktestCostConfig:
    env = os.environ if env is None else env
    return BacktestCostConfig(
        fee_rate=_get_float(env, "BACKTEST_FEE_RATE", 0.0),
        slippage_rate=_get_float(env, "BACKTEST_SLIPPAGE_RATE", 0.0),
        fixed_cost_r=_get_float(env, "BACKTEST_FIXED_COST_R", 0.0),
    )


def build_require_quality_gate_from_env(env=None) -> bool:
    env = os.environ if env is None else env
    value = str(env.get("BACKTEST_REQUIRE_QUALITY_GATE") or "").strip().lower()
    return value in {"1", "true", "yes", "on"}


def build_quality_gate_from_env(env=None) -> BacktestQualityGate:
    env = os.environ if env is None else env
    defaults = BacktestQualityGate()
    return BacktestQualityGate(
        min_total_signals=_get_int(env, "BACKTEST_MIN_TOTAL_SIGNALS", defaults.min_total_signals),
        min_executed_signals=_get_int(env, "BACKTEST_MIN_EXECUTED_SIGNALS", defaults.min_executed_signals),
        min_expectancy_r=_get_float(env, "BACKTEST_MIN_EXPECTANCY_R", defaults.min_expectancy_r),
        min_profit_factor_r=_get_float(env, "BACKTEST_MIN_PROFIT_FACTOR_R", defaults.min_profit_factor_r),
        max_drawdown_r=_get_float(env, "BACKTEST_MAX_DRAWDOWN_R", defaults.max_drawdown_r),
        max_consecutive_losses=_get_int(env, "BACKTEST_MAX_CONSECUTIVE_LOSSES", defaults.max_consecutive_losses),
    )


def run_backtest_once(
    service: BacktestService,
    instrument_id: int = 1,
    limit: int = 100,
    cost_config: Optional[BacktestCostConfig] = None,
    quality_gate: Optional[BacktestQualityGate] = None,
    require_quality_gate: bool = False,
    output_json: bool = False,
) -> int:
    run_id, outcomes = service.run_backtest(
        instrument_id=instrument_id,
        limit=limit,
        cost_config=cost_config,
        quality_gate=quality_gate,
    )
    summary = summarize_backtest_outcomes(outcomes)
    quality_gate_result = evaluate_backtest_quality(summary, gate=quality_gate)
    quality_diagnostics = diagnose_backtest_quality(summary, quality_gate_result)
    summary["quality_gate"] = quality_gate_result
    summary["quality_diagnostics"] = quality_diagnostics
    payload = {
        "passed": not require_quality_gate or quality_gate_result["passed"],
        "backtest_run_id": run_id,
        "result_count": len(outcomes),
        "summary": summary,
        "quality_gate": quality_gate_result,
        "quality_diagnostics": quality_diagnostics,
        "quality_gate_required": require_quality_gate,
    }
    exit_code = 2 if require_quality_gate and not quality_gate_result["passed"] else 0

    if output_json:
        print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True))
    else:
        print(f"Backtest run id: {run_id}; results written: {len(outcomes)}")
        print(f"Backtest quality gate passed: {quality_gate_result['passed']}")
        if quality_gate_result["reasons"]:
            print(f"Backtest quality gate reasons: {', '.join(quality_gate_result['reasons'])}")
        print(f"Backtest quality actions: {', '.join(quality_diagnostics['recommended_actions'])}")
    return exit_code


def main(env=None) -> int:
    env = os.environ if env is None else env
    output_json = _get_bool(env, "BACKTEST_OUTPUT_JSON", False)
    try:
        service = BacktestService()
        cost_config = build_cost_config_from_env(env=env)
        return run_backtest_once(
            service=service,
            cost_config=cost_config,
            quality_gate=build_quality_gate_from_env(env=env),
            require_quality_gate=build_require_quality_gate_from_env(env=env),
            output_json=output_json,
        )
    except SQLAlchemyError as exc:
        summary = {
            "passed": False,
            "backtest_run_id": None,
            "result_count": 0,
            "summary": None,
            "quality_gate": None,
            "error_type": exc.__class__.__name__,
            "error": str(exc),
        }
        if output_json:
            print(json.dumps(summary, default=str, ensure_ascii=False, sort_keys=True))
        else:
            print(f"Database unavailable for backtest sample: {exc.__class__.__name__}")
        return 2


def _get_float(env, key: str, default: float) -> float:
    value = env.get(key)
    if value is None or value == "":
        return default
    return float(value)


def _get_int(env, key: str, default: int) -> int:
    value = env.get(key)
    if value is None or value == "":
        return default
    return int(value)


def _get_bool(env, key: str, default: bool) -> bool:
    value = env.get(key)
    if value is None or value == "":
        return default
    return str(value).strip().lower() in {"1", "true", "yes", "on"}


if __name__ == '__main__':
    raise SystemExit(main())