from __future__ import annotations from datetime import datetime, timezone from pathlib import Path from typing import Optional import json import os import sys 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 scripts.run_backtest import ( build_cost_config_from_env, build_quality_gate_from_env, build_replay_config_from_env, ) from src.services.ict_mainline import build_doctrine_backtest_cohort from src.services.backtest import ( BACKTEST_MODE_REPLAY_NO_LOOKAHEAD, BACKTEST_SUBJECT_PLAN_REPLAY, BACKTEST_SUBJECT_SIGNAL, BacktestService, evaluate_backtest_quality, summarize_backtest_outcomes, ) DEFAULT_SESSION_GROUPS = "ALL;LONDON;NY_AM;NY_PM;ASIA" def main(env=None) -> int: env = os.environ if env is None else env output_json = _get_bool(env, "BACKTEST_BATCH_OUTPUT_JSON", _get_bool(env, "BACKTEST_OUTPUT_JSON", False)) require_quality_gate = _get_bool(env, "BACKTEST_BATCH_REQUIRE_QUALITY_GATE", False) batch_id = str(env.get("BACKTEST_BATCH_ID") or _default_batch_id()).strip() try: payload = run_backtest_batch( service=BacktestService(), batch_id=batch_id, instrument_ids=build_instrument_ids_from_env(env), limit=_get_int(env, "BACKTEST_BATCH_LIMIT", _get_int(env, "BACKTEST_LIMIT", 1000)), subject_types=build_subject_types_from_env(env), modes=build_modes_from_env(env), timeframe_sets=build_timeframe_sets_from_env(env), session_groups=build_session_groups_from_env(env), candidate_model_codes=build_candidate_model_codes_from_env(env), setup_playbook_ids=build_setup_playbook_ids_from_env(env), require_quality_gate=require_quality_gate, quality_gate=build_quality_gate_from_env(env), cost_config=build_cost_config_from_env(env), replay_config=build_replay_config_from_env(env), ) except SQLAlchemyError as exc: payload = { "status": "blocked", "mode": "batch_historical_backtest", "batch_id": batch_id, "error_type": exc.__class__.__name__, "error": str(exc), "execution_boundary": "read_only_backtest_no_order_submission", } if output_json: print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True)) else: print(f"Database unavailable for batch backtest: {exc.__class__.__name__}") return 2 if output_json: print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True)) else: print(f"Batch backtest id: {payload['batch_id']}") print(f"Runs: {payload['run_count']} / quality gate passed: {payload['passed_quality_gate_count']}") print(f"Blocked quality gates: {payload['blocked_quality_gate_count']}") print(f"Total results: {payload['total_result_count']}; executed: {payload['total_executed_signals']}") if payload["quality_gate_reason_counts"]: print(f"Top quality gate reasons: {_format_counts(payload['quality_gate_reason_counts'])}") print(payload["operator_summary"]) return 2 if require_quality_gate and payload["blocked_quality_gate_count"] else 0 def run_backtest_batch( *, service: BacktestService, batch_id: str, instrument_ids: list[int], limit: int, subject_types: list[str], modes: list[str], timeframe_sets: list[dict], session_groups: list[list[str]], candidate_model_codes: list[Optional[str]], setup_playbook_ids: list[Optional[str]], require_quality_gate: bool, quality_gate, cost_config, replay_config, ) -> dict: results = [] quality_gate_reason_counts: dict[str, int] = {} failure_reason_counts: dict[str, int] = {} aggregate_by_subject: dict[str, dict] = {} aggregate_by_session_group: dict[str, dict] = {} aggregate_by_doctrine_cohort: dict[str, dict] = {} total_result_count = 0 total_executed_signals = 0 for instrument_id in instrument_ids: for subject_type in subject_types: for mode in modes: for timeframe_set in timeframe_sets: model_codes = candidate_model_codes if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY else [None] sessions_to_run = session_groups if subject_type == BACKTEST_SUBJECT_SIGNAL else [[]] for candidate_model_code in model_codes: for setup_playbook_id in setup_playbook_ids: for session_codes in sessions_to_run: run_metadata = { "batch_id": batch_id, "batch_mode": "bulk_historical_validation", "setup_playbook_id": setup_playbook_id, "execution_boundary": "read_only_backtest_no_order_submission", } doctrine_cohort = build_doctrine_backtest_cohort( { "setup_playbook_id": setup_playbook_id, "candidate_model_code": candidate_model_code, "timeframe_set": timeframe_set, "run_metadata": run_metadata, } ) run_id, outcomes = service.run_backtest( instrument_id=instrument_id, limit=limit, cost_config=cost_config, quality_gate=quality_gate, subject_type=subject_type, candidate_model_code=candidate_model_code, mode=mode, replay_config=replay_config, timeframe_set=timeframe_set, session_codes=session_codes, run_metadata=run_metadata, ) summary = summarize_backtest_outcomes(outcomes, subject_type=subject_type) sample_basis = "executed" if subject_type == BACKTEST_SUBJECT_PLAN_REPLAY else "total_plus_executed" gate_result = evaluate_backtest_quality(summary, gate=quality_gate, sample_basis=sample_basis) doctrine_cohort = build_doctrine_backtest_cohort( { "setup_playbook_id": setup_playbook_id, "candidate_model_code": candidate_model_code, "timeframe_set": timeframe_set, "summary": {"quality_gate": gate_result}, "run_metadata": run_metadata, }, quality_gate=gate_result, ) total_result_count += len(outcomes) total_executed_signals += int(summary.get("executed_signals") or 0) for reason in list(gate_result.get("reasons") or []): quality_gate_reason_counts[str(reason)] = quality_gate_reason_counts.get(str(reason), 0) + 1 for reason, count in dict(summary.get("failure_reasons") or {}).items(): failure_reason_counts[str(reason)] = failure_reason_counts.get(str(reason), 0) + int(count or 0) _accumulate_group(aggregate_by_subject, subject_type, summary, gate_result) session_key = _session_group_key(session_codes) _accumulate_group(aggregate_by_session_group, session_key, summary, gate_result) _accumulate_doctrine_cohort_group(aggregate_by_doctrine_cohort, doctrine_cohort, summary, gate_result) results.append( { "backtest_run_id": run_id, "instrument_id": instrument_id, "subject_type": subject_type, "mode": mode, "timeframe_set": dict(timeframe_set), "session_codes": list(session_codes), "candidate_model_code": candidate_model_code, "setup_playbook_id": setup_playbook_id, "doctrine_cohort_key": doctrine_cohort.get("cohort_key"), "doctrine_completeness": doctrine_cohort.get("doctrine_completeness"), "result_count": len(outcomes), "executed_signals": int(summary.get("executed_signals") or 0), "expectancy_r": float(summary.get("expectancy_r") or 0.0), "executed_expectancy_r": float(summary.get("executed_expectancy_r") or 0.0), "profit_factor_r": summary.get("profit_factor_r"), "max_drawdown_r": float(summary.get("max_drawdown_r") or 0.0), "quality_gate": gate_result, "doctrine_backtest_cohort": doctrine_cohort, "top_failure_reasons": _top_counts(dict(summary.get("failure_reasons") or {}), limit=5), } ) passed_count = sum(1 for item in results if item["quality_gate"].get("passed")) blocked_count = len(results) - passed_count payload = { "status": "blocked" if require_quality_gate and blocked_count else "completed", "mode": "batch_historical_backtest", "batch_id": batch_id, "generated_ts": datetime.now(timezone.utc).isoformat(), "run_count": len(results), "passed_quality_gate_count": passed_count, "blocked_quality_gate_count": blocked_count, "total_result_count": total_result_count, "total_executed_signals": total_executed_signals, "quality_gate_required": require_quality_gate, "quality_gate_reason_counts": quality_gate_reason_counts, "failure_reason_counts": failure_reason_counts, "aggregate_by_subject": aggregate_by_subject, "aggregate_by_session_group": aggregate_by_session_group, "aggregate_by_doctrine_cohort": aggregate_by_doctrine_cohort, "top_runs_by_expectancy": sorted( results, key=lambda item: (float(item.get("executed_expectancy_r") or 0.0), int(item.get("executed_signals") or 0)), reverse=True, )[:5], "results": results, "operator_summary": _build_operator_summary(results, quality_gate_reason_counts), "execution_boundary": "read_only_backtest_no_order_submission", } return payload def build_instrument_ids_from_env(env) -> list[int]: raw = str(env.get("BACKTEST_BATCH_INSTRUMENT_IDS") or env.get("BACKTEST_INSTRUMENT_IDS") or "1").strip() return [int(item.strip()) for item in raw.split(",") if item.strip()] def build_subject_types_from_env(env) -> list[str]: raw = str(env.get("BACKTEST_BATCH_SUBJECT_TYPES") or "plan_replay,signal").strip() return _csv_values(raw) def build_modes_from_env(env) -> list[str]: raw = str(env.get("BACKTEST_BATCH_MODES") or BACKTEST_MODE_REPLAY_NO_LOOKAHEAD).strip() return _csv_values(raw) def build_candidate_model_codes_from_env(env) -> list[Optional[str]]: raw = str(env.get("BACKTEST_BATCH_CANDIDATE_MODEL_CODES") or "").strip() values = _csv_values(raw) return [None] if not values else [None if item.upper() == "ALL" else item for item in values] def build_setup_playbook_ids_from_env(env) -> list[Optional[str]]: raw = str(env.get("BACKTEST_BATCH_SETUP_PLAYBOOK_IDS") or env.get("BACKTEST_SETUP_PLAYBOOK_ID") or "").strip() values = _csv_values(raw) return [None] if not values else [None if item.upper() == "ALL" else item for item in values] def build_timeframe_sets_from_env(env) -> list[dict]: raw = str(env.get("BACKTEST_BATCH_TIMEFRAME_SETS") or "").strip() if not raw: return [ { "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 "1m"), "trigger_tf": str(env.get("AI_ICT_TRIGGER_TIMEFRAME") or env.get("BACKTEST_TRIGGER_TIMEFRAME") or "1m"), } ] result = [] for group in raw.split(";"): values = [item.strip() for item in group.split("/") if item.strip()] if len(values) == 1: values = [values[0], values[0], values[0]] if len(values) != 3: raise ValueError(f"invalid_timeframe_set:{group}") result.append({"bias_tf": values[0], "setup_tf": values[1], "trigger_tf": values[2]}) return result def build_session_groups_from_env(env) -> list[list[str]]: raw = str(env.get("BACKTEST_BATCH_SESSION_GROUPS") or DEFAULT_SESSION_GROUPS).strip() groups = [] for group in raw.split(";"): values = _csv_values(group) groups.append([] if not values or values == ["ALL"] else values) return groups or [[]] def _accumulate_group(container: dict[str, dict], key: str, summary: dict, quality_gate: dict) -> None: bucket = container.setdefault( key, { "run_count": 0, "passed_quality_gate_count": 0, "total_result_count": 0, "total_executed_signals": 0, "total_r": 0.0, }, ) bucket["run_count"] += 1 if quality_gate.get("passed"): bucket["passed_quality_gate_count"] += 1 bucket["total_result_count"] += int(summary.get("total_signals") or 0) bucket["total_executed_signals"] += int(summary.get("executed_signals") or 0) bucket["total_r"] += float(summary.get("total_r") or 0.0) def _accumulate_doctrine_cohort_group(container: dict[str, dict], cohort: dict, summary: dict, quality_gate: dict) -> None: key = str(cohort.get("cohort_key") or "missing_playbook::incomplete") bucket = container.setdefault( key, { "cohort_key": key, "setup_playbook_id": cohort.get("setup_playbook_id"), "doctrine_completeness": cohort.get("doctrine_completeness"), "run_count": 0, "passed_quality_gate_count": 0, "promotable_count": 0, "total_result_count": 0, "total_executed_signals": 0, "total_r": 0.0, "promotion_blocker_counts": {}, }, ) bucket["run_count"] += 1 if quality_gate.get("passed"): bucket["passed_quality_gate_count"] += 1 if (cohort.get("promotion_gate") or {}).get("passed"): bucket["promotable_count"] += 1 bucket["total_result_count"] += int(summary.get("total_signals") or 0) bucket["total_executed_signals"] += int(summary.get("executed_signals") or 0) bucket["total_r"] += float(summary.get("total_r") or 0.0) for reason in list((cohort.get("promotion_gate") or {}).get("reasons") or []): reason_key = str(reason) bucket["promotion_blocker_counts"][reason_key] = int(bucket["promotion_blocker_counts"].get(reason_key) or 0) + 1 def _build_operator_summary(results: list[dict], reason_counts: dict[str, int]) -> str: if not results: return "No backtest runs were executed; check batch dimensions." passed = sum(1 for item in results if item["quality_gate"].get("passed")) if passed: return f"Batch backtest completed: {passed}/{len(results)} runs passed the quality gate. Review top expectancy runs before promoting any setup." top_reason = next(iter(_top_counts(reason_counts, limit=1)), {}).get("code", "quality_gate_not_passed") return f"Batch backtest completed: 0/{len(results)} runs passed the quality gate. Main blocker: {top_reason}." def _top_counts(values: dict[str, int], *, limit: int) -> list[dict]: return [ {"code": key, "count": int(value)} for key, value in sorted(values.items(), key=lambda item: (-int(item[1]), str(item[0])))[:limit] ] def _format_counts(values: dict[str, int]) -> str: return ", ".join(f"{item['code']}={item['count']}" for item in _top_counts(values, limit=5)) def _session_group_key(session_codes: list[str]) -> str: return "ALL" if not session_codes else ",".join(session_codes) def _csv_values(raw: str) -> list[str]: return [item.strip() for item in str(raw or "").split(",") if item.strip()] 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"} def _default_batch_id() -> str: return "bt-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") if __name__ == "__main__": raise SystemExit(main())