ai-exchange/scripts/check_trading_delivery_readiness.py

from dataclasses import dataclass
from pathlib import Path
import re


ROOT = Path(__file__).resolve().parents[1]
TRADING_DELIVERY_DOCS = (
    ROOT / "README.md",
    ROOT / "RUNBOOK.md",
)
TRADING_DELIVERY_SECTION_HEADING = "Trading Delivery Checklist"
TRADING_DELIVERY_COMPOSE_PREFIX = (
    "docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml"
)
TRADING_DELIVERY_COMMAND_GROUPS = (
    (
        "pre_deploy_offline",
        (
            '.venv/bin/python -m unittest discover -s tests/unit -p "test_*.py"',
            ".venv/bin/python scripts/run_offline_checks.py",
            ".venv/bin/python scripts/check_local_demo_readiness.py",
            ".venv/bin/python scripts/check_trading_delivery_readiness.py",
        ),
    ),
    (
        "database_gate",
        (
            "READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
            "HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
            ".venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs",
            "HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
            "PIPELINE_OUTPUT_JSON=1 BACKTEST_REQUIRE_QUALITY_GATE=1 .venv/bin/python scripts/run_pipeline.py",
        ),
    ),
    (
        "deploy_and_health",
        (
            f"{TRADING_DELIVERY_COMPOSE_PREFIX} down --remove-orphans",
            f"{TRADING_DELIVERY_COMPOSE_PREFIX} up -d --build",
            f"{TRADING_DELIVERY_COMPOSE_PREFIX} ps",
            'test "$(docker ps --filter label=com.docker.compose.project=ai-ict --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
            'test "$(docker ps --filter name=^/ai-ict$ --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
            'test "$(docker inspect -f \'{{.State.Health.Status}}\' ai-ict)" = "healthy"',
            "curl -fsS http://127.0.0.1:8000/health",
            'curl -fsS "http://127.0.0.1:8000/trading/candidates?instrument_id=1&limit=1"',
        ),
    ),
    (
        "post_deploy_observe",
        (
            "READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
            "REPORTS_OUTPUT_JSON=1 .venv/bin/python scripts/run_reports.py",
        ),
    ),
    (
        "rollback_boundary",
        (
            "git rev-parse HEAD",
            f"{TRADING_DELIVERY_COMPOSE_PREFIX} logs --tail=100 ai-ict",
            "git worktree add /tmp/ai-ict-rollback <last-known-good-sha>",
            "cd /tmp/ai-ict-rollback && ./scripts/deploy_server.exp",
            "rm -rf /tmp/ai-ict-rollback",
        ),
    ),
)
TRADING_DELIVERY_COMMANDS = tuple(
    command
    for _group_name, commands in TRADING_DELIVERY_COMMAND_GROUPS
    for command in commands
)
TRADING_DELIVERY_REQUIRED_SNIPPETS = (
    "must-run before every market validation or deploy",
    "only when the dry-run output shows safe candidates",
    "manual review is still required for `needs_sweep_reference_migration` and `keep_version_history`",
    "paper execution boundary",
    "Rollback trigger",
    "last-known-good SHA",
    "without relying on chat memory",
)
TRADING_DELIVERY_REQUIRED_LOCAL_PATHS = (
    ROOT / "scripts" / "check_local_demo_readiness.py",
    ROOT / "scripts" / "deploy_server.exp",
    ROOT / "scripts" / "run_database_readiness.py",
    ROOT / "scripts" / "run_historical_compatibility_repair.py",
    ROOT / "scripts" / "run_offline_checks.py",
    ROOT / "scripts" / "run_pipeline.py",
    ROOT / "scripts" / "run_reports.py",
)
TRADING_DELIVERY_COMMAND_PREFIXES = (
    ".venv/bin/python",
    "READINESS_OUTPUT_JSON=1",
    "HISTORICAL_REPAIR_OUTPUT_JSON=1",
    "PIPELINE_OUTPUT_JSON=1",
    "REPORTS_OUTPUT_JSON=1",
    "docker compose",
    "test ",
    "curl ",
    "git ",
    "cd /tmp/ai-ict-rollback &&",
    "rm -rf /tmp/ai-ict-rollback",
)


@dataclass(frozen=True)
class TradingDeliveryReadinessReport:
    checked_docs: tuple[Path, ...]
    command_lines_by_doc: tuple[tuple[Path, tuple[str, ...]], ...]
    errors: tuple[str, ...]


def check_trading_delivery_readiness(
    docs: tuple[Path, ...] = TRADING_DELIVERY_DOCS,
) -> TradingDeliveryReadinessReport:
    command_lines_by_doc = []
    errors = []

    for path in TRADING_DELIVERY_REQUIRED_LOCAL_PATHS:
        if not path.is_file():
            errors.append(f"{_display_path(path)} is missing")

    for path in docs:
        if not path.is_file():
            command_lines_by_doc.append((path, ()))
            errors.append(f"{_display_path(path)} is missing")
            continue

        text = path.read_text(encoding="utf-8")
        try:
            section = _trading_delivery_section(text)
        except ValueError as exc:
            command_lines_by_doc.append((path, ()))
            errors.append(f"{_display_path(path)} {exc}")
            continue

        command_lines = _trading_delivery_command_lines(section)
        command_lines_by_doc.append((path, command_lines))

        for snippet in TRADING_DELIVERY_REQUIRED_SNIPPETS:
            if snippet not in section:
                errors.append(f"{_display_path(path)} missing trading delivery snippet: {snippet}")
        if command_lines != TRADING_DELIVERY_COMMANDS:
            errors.append(
                f"{_display_path(path)} trading delivery commands drifted from the locked checklist shape"
            )

    return TradingDeliveryReadinessReport(
        checked_docs=tuple(docs),
        command_lines_by_doc=tuple(command_lines_by_doc),
        errors=tuple(errors),
    )


def main() -> int:
    report = check_trading_delivery_readiness()
    if report.errors:
        print("Trading delivery readiness check failed")
        for error in report.errors:
            print(f"- {error}")
        return 1

    print(
        "Trading delivery readiness check passed: "
        f"{len(report.checked_docs)} docs, "
        f"{len(TRADING_DELIVERY_COMMAND_GROUPS)} phases, "
        f"{len(TRADING_DELIVERY_COMMANDS)} locked commands."
    )
    return 0


def _trading_delivery_section(text: str) -> str:
    match = re.search(
        rf"^## {re.escape(TRADING_DELIVERY_SECTION_HEADING)}\n(?P<section>.*?)(?:\n## |\Z)",
        text,
        re.DOTALL | re.MULTILINE,
    )
    if match is None:
        raise ValueError(f"is missing the {TRADING_DELIVERY_SECTION_HEADING} section")
    return match.group("section")


def _trading_delivery_command_lines(text: str) -> tuple[str, ...]:
    return tuple(
        line.strip()
        for line in text.splitlines()
        if line.strip().startswith(TRADING_DELIVERY_COMMAND_PREFIXES)
    )


def _display_path(path: Path) -> str:
    try:
        return str(path.relative_to(ROOT))
    except ValueError:
        return str(path)


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