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 (
    BACKTEST_MODE_ENTRY_THEN_OUTCOME,
    BACKTEST_SUBJECT_PLAN_REPLAY,
    BacktestCostConfig,
    BacktestQualityGate,
    BacktestReplayConfig,
    BacktestService,
    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 build_backtest_mode_from_env(env=None) -> str:
    env = os.environ if env is None else env
    value = str(env.get("BACKTEST_MODE") or "").strip()
    return value or BACKTEST_MODE_ENTRY_THEN_OUTCOME


def build_backtest_subject_type_from_env(env=None) -> str:
    env = os.environ if env is None else env
    if build_backtest_session_codes_from_env(env=env):
        return "signal"
    value = str(env.get("BACKTEST_SUBJECT_TYPE") or "").strip()
    return value or BACKTEST_SUBJECT_PLAN_REPLAY


def build_backtest_trading_plan_id_from_env(env=None) -> Optional[int]:
    env = os.environ if env is None else env
    value = env.get("BACKTEST_TRADING_PLAN_ID")
    if value is None or value == "":
        return None
    return int(value)


def build_backtest_candidate_model_code_from_env(env=None) -> Optional[str]:
    env = os.environ if env is None else env
    value = str(env.get("BACKTEST_CANDIDATE_MODEL_CODE") or "").strip()
    return value or None


def build_replay_config_from_env(env=None) -> BacktestReplayConfig:
    env = os.environ if env is None else env
    defaults = BacktestReplayConfig()
    return BacktestReplayConfig(
        left_bars=_get_int(env, "BACKTEST_REPLAY_LEFT_BARS", defaults.left_bars),
        right_bars=_get_int(env, "BACKTEST_REPLAY_RIGHT_BARS", defaults.right_bars),
    )


def build_backtest_session_codes_from_env(env=None) -> list[str]:
    env = os.environ if env is None else env
    raw = str(env.get("BACKTEST_SESSION_CODES") or "").strip()
    if not raw:
        return []
    return [item.strip() for item in raw.split(",") if item.strip()]


def build_backtest_timeframe_set_from_env(env=None) -> dict:
    env = os.environ if env is None else env
    bias_tf = str(env.get("AI_ICT_BIAS_TIMEFRAME") or env.get("BACKTEST_BIAS_TIMEFRAME") or "1m")
    setup_tf = str(env.get("AI_ICT_SETUP_TIMEFRAME") or env.get("BACKTEST_SETUP_TIMEFRAME") or bias_tf)
    trigger_tf = str(env.get("AI_ICT_TRIGGER_TIMEFRAME") or env.get("BACKTEST_TRIGGER_TIMEFRAME") or setup_tf)
    return {
        "bias_tf": bias_tf,
        "setup_tf": setup_tf,
        "trigger_tf": trigger_tf,
    }


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,
    backtest_mode: str = BACKTEST_MODE_ENTRY_THEN_OUTCOME,
    subject_type: str = BACKTEST_SUBJECT_PLAN_REPLAY,
    trading_plan_id: Optional[int] = None,
    candidate_model_code: Optional[str] = None,
    replay_config: Optional[BacktestReplayConfig] = None,
    timeframe_set: Optional[dict] = None,
    session_codes: Optional[list[str]] = None,
) -> int:
    run_id, outcomes = service.run_backtest(
        instrument_id=instrument_id,
        limit=limit,
        cost_config=cost_config,
        quality_gate=quality_gate,
        subject_type=subject_type,
        trading_plan_id=trading_plan_id,
        candidate_model_code=candidate_model_code,
        mode=backtest_mode,
        replay_config=replay_config,
        timeframe_set=timeframe_set,
        session_codes=session_codes,
    )
    summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type)
    quality_gate_result = evaluate_backtest_quality(
        summary,
        gate=quality_gate,
        sample_basis="executed" if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY else "total_plus_executed",
    )
    payload = {
        "passed": not require_quality_gate or quality_gate_result["passed"],
        "backtest_run_id": run_id,
        "result_count": len(outcomes),
        "subject_type": subject_type,
        "trading_plan_id": trading_plan_id,
        "candidate_model_code": candidate_model_code,
        "timeframe_set": timeframe_set,
        "session_codes": session_codes,
        "summary": summary,
        "quality_gate": quality_gate_result,
        "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'])}")
    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,
            backtest_mode=build_backtest_mode_from_env(env=env),
            subject_type=build_backtest_subject_type_from_env(env=env),
            trading_plan_id=build_backtest_trading_plan_id_from_env(env=env),
            candidate_model_code=build_backtest_candidate_model_code_from_env(env=env),
            replay_config=build_replay_config_from_env(env=env),
            timeframe_set=build_backtest_timeframe_set_from_env(env=env),
            session_codes=build_backtest_session_codes_from_env(env=env),
        )
    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())