ai-exchange/scripts/run_trading_delivery_rehearsal.py

from pathlib import Path
from datetime import datetime, timezone
import json
import os
import shlex
import subprocess
import sys
from typing import Optional


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_DOTENV_PATH = ROOT / ".env"
DEFAULT_DEPLOY_DIR = Path("/root/deploy/ai-ict")
DEFAULT_DEPLOY_PROJECT = "ai-ict"
DEFAULT_DEPLOY_CONTAINER = "ai-ict"
DEFAULT_DEPLOY_PORT = 8000
DEFAULT_HEALTH_MAX_ATTEMPTS = 24
DEFAULT_HEALTH_SLEEP_SECONDS = 5
SAFE_REPAIR_COUNT_KEYS = (
    "signal_invalidations",
    "mss_meta",
    "fvg_meta",
    "liquidity_safe_auto_merge_groups",
    "liquidity_safe_duplicate_sweep_merge_groups",
)


def build_trading_rehearsal_env(env=None) -> dict:
    env = dict(os.environ if env is None else env)
    if _get_bool(env, "TRADING_REHEARSAL_SKIP_DOTENV", False):
        return env
    dotenv_path = Path(env.get("TRADING_REHEARSAL_DOTENV_PATH") or DEFAULT_DOTENV_PATH)
    if not dotenv_path.is_file():
        return env
    for key, value in read_dotenv_pairs(dotenv_path).items():
        if key not in env or env[key] == "":
            env[key] = value
    return env


def read_dotenv_pairs(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("export "):
            line = line[len("export ") :].strip()
        if "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip()
        if not key:
            continue
        values[key] = _parse_dotenv_value(value)
    return values


def run_trading_delivery_rehearsal(env=None, runner=subprocess.run, cwd: Path = ROOT, sleeper=None) -> dict:
    env = build_trading_rehearsal_env(env=env)
    sleeper = sleeper or time_sleep
    run_deploy = _get_bool(env, "TRADING_REHEARSAL_RUN_DEPLOY", True)
    auto_apply_safe_repairs = _get_bool(env, "TRADING_REHEARSAL_AUTO_APPLY_SAFE_REPAIRS", False)
    allow_reference_migration = _get_bool(env, "TRADING_REHEARSAL_ALLOW_REFERENCE_MIGRATION", False)
    auto_apply_reference_migrations = _get_bool(env, "TRADING_REHEARSAL_AUTO_APPLY_REFERENCE_MIGRATIONS", False)
    deploy_dir = Path(env.get("TRADING_REHEARSAL_DEPLOY_DIR") or DEFAULT_DEPLOY_DIR)
    deploy_project = env.get("TRADING_REHEARSAL_DEPLOY_PROJECT") or DEFAULT_DEPLOY_PROJECT
    deploy_container = env.get("TRADING_REHEARSAL_DEPLOY_CONTAINER") or DEFAULT_DEPLOY_CONTAINER
    deploy_port = _get_int(env, "TRADING_REHEARSAL_DEPLOY_PORT", DEFAULT_DEPLOY_PORT)
    health_max_attempts = max(_get_int(env, "TRADING_REHEARSAL_HEALTH_MAX_ATTEMPTS", DEFAULT_HEALTH_MAX_ATTEMPTS), 1)
    health_sleep_seconds = max(_get_int(env, "TRADING_REHEARSAL_HEALTH_SLEEP_SECONDS", DEFAULT_HEALTH_SLEEP_SECONDS), 0)

    summary = {
        "mode": "starting",
        "passed": False,
        "started_ts": datetime.now(timezone.utc).isoformat(),
        "steps": [],
        "blockers": [],
        "manual_decisions": [],
        "promotion_decision": "hold",
        "continuation_decision": "continue",
        "policies": {
            "reference_migration": (
                "auto_apply"
                if allow_reference_migration and auto_apply_reference_migrations
                else "manual_apply" if allow_reference_migration else "defer"
            ),
            "safe_repairs": "auto_apply" if auto_apply_safe_repairs else "manual_apply",
        },
        "historical_repair": None,
        "autonomous_cycle": None,
        "deploy": {"requested": run_deploy, "dir": str(deploy_dir), "project": deploy_project, "container": deploy_container},
        "post_deploy": None,
        "finished_ts": None,
    }

    readiness_check = _run_python_step(
        name="trading_delivery_readiness",
        script="scripts/check_trading_delivery_readiness.py",
        env=env,
        runner=runner,
        cwd=cwd,
    )
    summary["steps"].append(readiness_check)
    if readiness_check["exit_code"] != 0:
        summary["mode"] = "checklist_drift_failed"
        summary["blockers"].append("checklist_drift_detected")
        return _finalize_trading_rehearsal(summary)

    repair_scan = _run_python_step(
        name="historical_repair_scan_initial",
        script="scripts/run_historical_compatibility_repair.py",
        env={**env, "HISTORICAL_REPAIR_OUTPUT_JSON": "1"},
        runner=runner,
        cwd=cwd,
    )
    summary["steps"].append(repair_scan)
    summary["historical_repair"] = repair_scan.get("summary")
    if repair_scan["exit_code"] != 0:
        summary["mode"] = "historical_repair_scan_failed"
        summary["blockers"].append("historical_repair_scan_failed")
        return _finalize_trading_rehearsal(summary)

    repair_summary = repair_scan.get("summary") or {}
    _append_historical_repair_decisions(summary, repair_summary, allow_reference_migration=allow_reference_migration)

    safe_candidate_count = count_safe_repair_candidates(repair_summary)
    if safe_candidate_count > 0:
        if auto_apply_safe_repairs:
            repair_apply = _run_python_step(
                name="historical_repair_apply_safe",
                script="scripts/run_historical_compatibility_repair.py",
                args=["--apply-safe-repairs"],
                env={**env, "HISTORICAL_REPAIR_OUTPUT_JSON": "1"},
                runner=runner,
                cwd=cwd,
            )
            summary["steps"].append(repair_apply)
            if repair_apply["exit_code"] != 0:
                summary["mode"] = "historical_repair_apply_failed"
                summary["blockers"].append("safe_repairs_apply_failed")
                return _finalize_trading_rehearsal(summary)

            repair_rescan = _run_python_step(
                name="historical_repair_scan_post_apply",
                script="scripts/run_historical_compatibility_repair.py",
                env={**env, "HISTORICAL_REPAIR_OUTPUT_JSON": "1"},
                runner=runner,
                cwd=cwd,
            )
            summary["steps"].append(repair_rescan)
            summary["historical_repair"] = repair_rescan.get("summary")
            if repair_rescan["exit_code"] != 0:
                summary["mode"] = "historical_repair_rescan_failed"
                summary["blockers"].append("historical_repair_rescan_failed")
                return _finalize_trading_rehearsal(summary)
            repair_summary = repair_rescan.get("summary") or {}
            _append_historical_repair_decisions(summary, repair_summary, allow_reference_migration=allow_reference_migration)
            if count_safe_repair_candidates(repair_summary) > 0:
                summary["mode"] = "historical_repair_safe_candidates_remain"
                summary["blockers"].append("safe_repairs_still_pending_after_apply")
                return _finalize_trading_rehearsal(summary)
        else:
            summary["mode"] = "historical_repair_manual_apply_required"
            summary["blockers"].append("safe_repairs_pending_manual_apply")
            return _finalize_trading_rehearsal(summary)

    reference_migration_group_count = int((repair_summary.get("candidate_repairs") or {}).get("liquidity_reference_migration_groups", 0))
    reference_migration_executable_group_count = int((repair_summary.get("candidate_repairs") or {}).get("liquidity_reference_migration_executable_groups", 0))
    reference_migration_blocked_group_count = int((repair_summary.get("candidate_repairs") or {}).get("liquidity_reference_migration_blocked_groups", 0))

    if reference_migration_group_count:
        if not allow_reference_migration:
            summary["mode"] = "reference_migration_deferred"
            summary["blockers"].append("liquidity_reference_migration_deferred")
            return _finalize_trading_rehearsal(summary)
        if reference_migration_blocked_group_count > 0:
            summary["mode"] = "reference_migration_conflicts_blocked"
            summary["blockers"].append("liquidity_reference_migration_conflicts_present")
            return _finalize_trading_rehearsal(summary)
        if reference_migration_executable_group_count > 0:
            if auto_apply_reference_migrations:
                repair_apply_reference = _run_python_step(
                    name="historical_repair_apply_reference_migrations",
                    script="scripts/run_historical_compatibility_repair.py",
                    args=["--apply-liquidity-reference-migrations"],
                    env={**env, "HISTORICAL_REPAIR_OUTPUT_JSON": "1"},
                    runner=runner,
                    cwd=cwd,
                )
                summary["steps"].append(repair_apply_reference)
                if repair_apply_reference["exit_code"] != 0:
                    summary["mode"] = "reference_migration_apply_failed"
                    summary["blockers"].append("liquidity_reference_migration_apply_failed")
                    return _finalize_trading_rehearsal(summary)
                repair_rescan = _run_python_step(
                    name="historical_repair_scan_post_reference_migration",
                    script="scripts/run_historical_compatibility_repair.py",
                    env={**env, "HISTORICAL_REPAIR_OUTPUT_JSON": "1"},
                    runner=runner,
                    cwd=cwd,
                )
                summary["steps"].append(repair_rescan)
                summary["historical_repair"] = repair_rescan.get("summary")
                if repair_rescan["exit_code"] != 0:
                    summary["mode"] = "historical_repair_rescan_failed"
                    summary["blockers"].append("historical_repair_rescan_failed")
                    return _finalize_trading_rehearsal(summary)
                repair_summary = repair_rescan.get("summary") or {}
                _append_historical_repair_decisions(summary, repair_summary, allow_reference_migration=allow_reference_migration)
            else:
                summary["mode"] = "reference_migration_manual_apply_required"
                summary["blockers"].append("liquidity_reference_migration_pending_manual_apply")
                return _finalize_trading_rehearsal(summary)

    autonomous = _run_python_step(
        name="autonomous_cycle",
        script="scripts/run_autonomous_cycle.py",
        env={
            **env,
            "AUTONOMOUS_OUTPUT_JSON": "1",
            "AUTONOMOUS_REQUIRE_DB": "1",
            "AUTONOMOUS_REQUIRE_SCHEMA": "1",
            "AUTONOMOUS_APPLY_MIGRATIONS": env.get("AUTONOMOUS_APPLY_MIGRATIONS", "1"),
            "AUTONOMOUS_REQUIRE_DATA_READINESS": env.get("AUTONOMOUS_REQUIRE_DATA_READINESS", "1"),
            "AUTONOMOUS_BACKFILL_CANDLES": env.get("AUTONOMOUS_BACKFILL_CANDLES", "1"),
            "AUTONOMOUS_RUN_PIPELINE": env.get("AUTONOMOUS_RUN_PIPELINE", "1"),
            "BACKTEST_REQUIRE_QUALITY_GATE": env.get("BACKTEST_REQUIRE_QUALITY_GATE", "1"),
        },
        runner=runner,
        cwd=cwd,
    )
    summary["steps"].append(autonomous)
    summary["autonomous_cycle"] = autonomous.get("summary")
    if autonomous["exit_code"] != 0 or not (autonomous.get("summary") or {}).get("passed"):
        summary["mode"] = "autonomous_cycle_failed"
        summary["blockers"].extend(classify_autonomous_cycle_blockers(autonomous.get("summary") or {}))
        return _finalize_trading_rehearsal(summary)

    if run_deploy:
        deploy_steps = run_deploy_and_health_checks(
            env=env,
            runner=runner,
            cwd=deploy_dir,
            deploy_project=deploy_project,
            deploy_container=deploy_container,
            deploy_port=deploy_port,
            health_max_attempts=health_max_attempts,
            health_sleep_seconds=health_sleep_seconds,
            sleeper=sleeper,
        )
        summary["steps"].extend(deploy_steps)
        failed_step = next((step for step in deploy_steps if step["exit_code"] != 0), None)
        if failed_step is not None:
            summary["mode"] = "deploy_or_health_failed"
            summary["blockers"].append(f"deploy_step_failed:{failed_step['name']}")
            return _finalize_trading_rehearsal(summary)
    else:
        summary["manual_decisions"].append(
            {
                "decision": "deploy_skipped",
                "reason": "TRADING_REHEARSAL_RUN_DEPLOY=0",
            }
        )

    post_deploy_readiness = _run_python_step(
        name="post_deploy_readiness",
        script="scripts/run_database_readiness.py",
        env={**env, "READINESS_OUTPUT_JSON": "1"},
        runner=runner,
        cwd=cwd,
    )
    post_deploy_reports = _run_python_step(
        name="post_deploy_reports",
        script="scripts/run_reports.py",
        env={**env, "REPORTS_OUTPUT_JSON": "1"},
        runner=runner,
        cwd=cwd,
    )
    summary["steps"].extend([post_deploy_readiness, post_deploy_reports])
    summary["post_deploy"] = {
        "readiness": post_deploy_readiness.get("summary"),
        "reports": post_deploy_reports.get("summary"),
    }
    if post_deploy_readiness["exit_code"] != 0 or not (post_deploy_readiness.get("summary") or {}).get("passed"):
        summary["mode"] = "post_deploy_readiness_failed"
        summary["blockers"].append("post_deploy_readiness_failed")
        return _finalize_trading_rehearsal(summary)
    if post_deploy_reports["exit_code"] != 0 or not (post_deploy_reports.get("summary") or {}).get("passed"):
        summary["mode"] = "post_deploy_reports_failed"
        summary["blockers"].append("post_deploy_reports_failed")
        return _finalize_trading_rehearsal(summary)

    summary["mode"] = "rehearsal_completed"
    summary["passed"] = True
    summary["promotion_decision"] = "promote_candidate"
    return _finalize_trading_rehearsal(summary)


def run_deploy_and_health_checks(
    *,
    env: dict,
    runner,
    cwd: Path,
    deploy_project: str,
    deploy_container: str,
    deploy_port: int,
    health_max_attempts: int,
    health_sleep_seconds: int,
    sleeper,
) -> list[dict]:
    compose_file = str(cwd / "docker-compose.yml")
    steps = [
        _run_step(
            name="deploy_down",
            command=["docker", "compose", "-p", deploy_project, "-f", compose_file, "down", "--remove-orphans"],
            env=env,
            runner=runner,
            cwd=cwd,
        ),
        _run_step(
            name="deploy_up",
            command=["docker", "compose", "-p", deploy_project, "-f", compose_file, "up", "-d", "--build"],
            env=env,
            runner=runner,
            cwd=cwd,
        ),
        _run_step(
            name="deploy_ps",
            command=["docker", "compose", "-p", deploy_project, "-f", compose_file, "ps"],
            env=env,
            runner=runner,
            cwd=cwd,
        ),
    ]
    for step in steps:
        if step["exit_code"] != 0:
            return steps

    health_wait = wait_for_container_health(
        env=env,
        runner=runner,
        cwd=cwd,
        container=deploy_container,
        max_attempts=health_max_attempts,
        sleep_seconds=health_sleep_seconds,
        sleeper=sleeper,
    )
    steps.append(health_wait)
    if health_wait["exit_code"] != 0:
        return steps

    shell_checks = (
        (
            "deploy_single_project_container",
            f'test "$(docker ps --filter label=com.docker.compose.project={deploy_project} --format \'{{{{.Names}}}}\' | wc -l | tr -d \' \')" = "1"',
        ),
        (
            "deploy_named_container",
            f'test "$(docker ps --filter name=^/{deploy_container}$ --format \'{{{{.Names}}}}\' | wc -l | tr -d \' \')" = "1"',
        ),
        (
            "deploy_container_healthy",
            f'test "$(docker inspect -f \'{{{{.State.Health.Status}}}}\' {deploy_container})" = "healthy"',
        ),
    )
    for name, command in shell_checks:
        step = _run_step(
            name=name,
            command=["bash", "-lc", command],
            env=env,
            runner=runner,
            cwd=cwd,
        )
        steps.append(step)
        if step["exit_code"] != 0:
            return steps

    for name, url in (
        ("http_health", f"http://127.0.0.1:{deploy_port}/health"),
        ("http_trading_candidates", f"http://127.0.0.1:{deploy_port}/trading/candidates?instrument_id=1&limit=1"),
    ):
        step = _run_step(
            name=name,
            command=["curl", "-fsS", url],
            env=env,
            runner=runner,
            cwd=cwd,
        )
        steps.append(step)
        if step["exit_code"] != 0:
            return steps
    return steps


def wait_for_container_health(*, env: dict, runner, cwd: Path, container: str, max_attempts: int, sleep_seconds: int, sleeper) -> dict:
    attempts = []
    for attempt in range(1, max_attempts + 1):
        step = _run_step(
            name="deploy_wait_for_health",
            command=["docker", "inspect", "-f", "{{.State.Health.Status}}", container],
            env=env,
            runner=runner,
            cwd=cwd,
        )
        status = ((step.get("stdout") or "").strip() or "unknown").splitlines()[-1] if step.get("stdout") else "unknown"
        attempts.append({"attempt": attempt, "status": status, "exit_code": step["exit_code"]})
        if step["exit_code"] == 0 and status == "healthy":
            step["summary"] = {"attempts": attempts, "final_status": status}
            return step
        if attempt < max_attempts and sleep_seconds > 0:
            sleeper(sleep_seconds)
    step["summary"] = {"attempts": attempts, "final_status": attempts[-1]["status"] if attempts else "unknown"}
    if step["exit_code"] == 0 and (step.get("stdout") or "").strip() != "healthy":
        step["exit_code"] = 2
        step["status"] = "failed"
    return step


def count_safe_repair_candidates(summary: dict) -> int:
    candidate_repairs = dict(summary.get("candidate_repairs") or {})
    return sum(int(candidate_repairs.get(key) or 0) for key in SAFE_REPAIR_COUNT_KEYS)


def classify_autonomous_cycle_blockers(summary: dict) -> list[str]:
    blockers = []
    mode = summary.get("mode")
    if mode:
        blockers.append(f"autonomous_mode:{mode}")
    for reason in summary.get("next_actions") or []:
        blockers.append(f"next_action:{reason}")
    for step in summary.get("steps") or []:
        if step.get("name") == "pipeline":
            pipeline_summary = step.get("summary") or {}
            quality_gate = pipeline_summary.get("backtest_quality_gate") or {}
            for reason in quality_gate.get("reasons") or []:
                blockers.append(f"pipeline_quality_gate:{reason}")
    return blockers or ["autonomous_cycle_failed_without_diagnostics"]


def _append_historical_repair_decisions(summary: dict, repair_summary: dict, *, allow_reference_migration: bool) -> None:
    candidate_repairs = dict(repair_summary.get("candidate_repairs") or {})
    keep_history_groups = int(candidate_repairs.get("liquidity_keep_history_groups") or 0)
    reference_migration_groups = int(candidate_repairs.get("liquidity_reference_migration_groups") or 0)
    if keep_history_groups:
        _append_manual_decision(
            summary,
            {
                "decision": "preserve_keep_version_history",
                "group_count": keep_history_groups,
            },
        )
    if reference_migration_groups:
        _append_manual_decision(
            summary,
            {
                "decision": "allow_reference_migration" if allow_reference_migration else "defer_reference_migration",
                "group_count": reference_migration_groups,
            },
        )


def _append_manual_decision(summary: dict, decision: dict) -> None:
    if decision not in summary["manual_decisions"]:
        summary["manual_decisions"].append(decision)


def _run_python_step(*, name: str, script: str, env: dict, runner, cwd: Path, args: Optional[list[str]] = None) -> dict:
    return _run_step(
        name=name,
        command=[sys.executable, script, *(args or [])],
        env=env,
        runner=runner,
        cwd=cwd,
    )


def _run_step(*, name: str, command: list[str], env: dict, runner, cwd: Path) -> dict:
    result = runner(command, cwd=cwd, capture_output=True, text=True, check=False, env=env)
    return {
        "name": name,
        "command": command,
        "status": "passed" if result.returncode == 0 else "failed",
        "exit_code": result.returncode,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "summary": _parse_json(result.stdout),
    }


def _parse_json(value: str):
    if not value:
        return None
    try:
        return json.loads(value)
    except json.JSONDecodeError:
        return None


def _finalize_trading_rehearsal(summary: dict) -> dict:
    summary["passed"] = summary["passed"] and len(summary["blockers"]) == 0
    if not summary["passed"]:
        summary["promotion_decision"] = "hold"
    summary["continuation_decision"] = classify_rehearsal_continuation(summary)
    summary["finished_ts"] = datetime.now(timezone.utc).isoformat()
    return summary


def classify_rehearsal_continuation(summary: dict) -> str:
    mode = summary.get("mode")
    if mode == "rehearsal_completed":
        return "advance"
    if mode in {
        "reference_migration_deferred",
        "reference_migration_conflicts_blocked",
        "reference_migration_manual_apply_required",
        "checklist_drift_failed",
        "historical_repair_scan_failed",
        "historical_repair_apply_failed",
        "historical_repair_rescan_failed",
        "historical_repair_safe_candidates_remain",
        "historical_repair_manual_apply_required",
        "autonomous_cycle_failed",
        "deploy_or_health_failed",
        "post_deploy_readiness_failed",
        "post_deploy_reports_failed",
        "reference_migration_apply_failed",
    }:
        return "blocked"
    return "continue"


def main(env=None) -> int:
    env = build_trading_rehearsal_env(env=env)
    summary = run_trading_delivery_rehearsal(env=env)
    if _get_bool(env, "TRADING_REHEARSAL_OUTPUT_JSON", False):
        print(json.dumps(summary, ensure_ascii=False, sort_keys=True))
        return 0 if summary["passed"] else 2

    print(f"Trading rehearsal mode: {summary['mode']}")
    print(f"Trading rehearsal passed: {summary['passed']}")
    print(f"Promotion decision: {summary['promotion_decision']}")
    if summary["blockers"]:
        print(f"Blockers: {', '.join(summary['blockers'])}")
    if summary["manual_decisions"]:
        for item in summary["manual_decisions"]:
            print(f"Manual decision: {item['decision']} group_count={item.get('group_count', 0)}")
    for step in summary["steps"]:
        rendered = " ".join(shlex.quote(part) for part in step["command"])
        print(f"Step {step['name']}: {step['status']} exit_code={step['exit_code']} -> {rendered}")
    return 0 if summary["passed"] else 2


def _parse_dotenv_value(value: str) -> str:
    if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
        return value[1:-1]
    if " #" in value:
        value = value.split(" #", 1)[0].rstrip()
    return 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 _get_int(env, key: str, default: int) -> int:
    value = env.get(key)
    if value is None or value == "":
        return default
    return int(value)


def time_sleep(seconds: int) -> None:
    import time

    time.sleep(seconds)


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