cyb50-quant/research/dragon/v2/dragon_rollout_governance_check.py

from __future__ import annotations

from datetime import datetime
import json
from pathlib import Path

import pandas as pd

from dragon_rollout_governance import (
    FORWARD_OK,
    HOLD_AND_REVIEW,
    ROLLBACK_REVIEW_REQUIRED,
    evaluate_rollout,
    gate_rows,
)


def _load_json(path: Path) -> dict[str, object]:
    if not path.exists():
        return {}
    return json.loads(path.read_text(encoding="utf-8"))


def _load_csv(path: Path) -> pd.DataFrame:
    if not path.exists():
        return pd.DataFrame()
    return pd.read_csv(path, encoding="utf-8-sig")


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    manifest_path = base_dir / "dragon_daily_rc1_manifest.json"
    monitor_history_path = base_dir / "dragon_monitor_history.csv"
    divergence_log_path = base_dir / "dragon_branch_divergence_log.csv"
    monitor_health_path = base_dir / "dragon_monitor_health_report.md"

    manifest = _load_json(manifest_path)
    monitor_history = _load_csv(monitor_history_path)
    divergence_log = _load_csv(divergence_log_path)

    decision, gates, reasons, active_branch = evaluate_rollout(
        manifest=manifest,
        monitor_history=monitor_history,
        divergence_log=divergence_log,
        monitor_health_report_exists=monitor_health_path.exists(),
    )

    fallback_branch = "alpha_first_selective_veto"
    latest_bar_date = str(manifest.get("latest_bar_date", ""))
    request_date = str(manifest.get("as_of_request_date", ""))
    generated_at = datetime.now().isoformat(timespec="seconds")

    snapshot = pd.DataFrame(gate_rows(gates))
    snapshot.to_csv(base_dir / "dragon_rollout_governance_snapshot.csv", index=False, encoding="utf-8-sig")

    gate_status_counts = snapshot["status"].value_counts().to_dict() if not snapshot.empty else {}
    decision_payload = {
        "generated_at": generated_at,
        "request_date": request_date,
        "latest_bar_date": latest_bar_date,
        "release_version": manifest.get("release_version", "RC1"),
        "candidate_branch": manifest.get("branch", "alpha_first_glued_refined_hot_cap"),
        "fallback_branch": fallback_branch,
        "active_branch": active_branch,
        "decision": decision,
        "gate_status_counts": gate_status_counts,
        "reasons": reasons,
    }
    (base_dir / "dragon_rollout_state.json").write_text(
        json.dumps(decision_payload, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )

    lines = [
        "# Dragon Rollout Governance Report",
        "",
        f"- generated_at: `{generated_at}`",
        f"- request_date: `{request_date}`",
        f"- latest_bar_date: `{latest_bar_date}`",
        f"- candidate_branch: `{decision_payload['candidate_branch']}`",
        f"- fallback_branch: `{fallback_branch}`",
        f"- decision: `{decision}`",
        f"- active_branch: `{active_branch}`",
        "",
        "## Gate Summary",
        f"- ok: `{int(gate_status_counts.get('ok', 0))}`",
        f"- warning: `{int(gate_status_counts.get('warning', 0))}`",
        f"- hard_fail: `{int(gate_status_counts.get('hard_fail', 0))}`",
        "",
        "## Gate Details",
    ]

    if snapshot.empty:
        lines.append("- No gate rows were generated.")
    else:
        for _, row in snapshot.iterrows():
            lines.extend(
                [
                    f"### {row['gate']}",
                    f"- status: `{row['status']}`",
                    f"- value: `{row['value']}`",
                    f"- threshold: `{row['threshold']}`",
                    f"- detail: {row['detail']}",
                    f"- action: {row['action']}",
                    "",
                ]
            )

    if reasons:
        lines.extend(
            [
                "## Decision Reasons",
                *(f"- `{item}`" for item in reasons),
                "",
            ]
        )

    lines.extend(
        [
            "## Artifacts",
            "- `dragon_rollout_state.json`",
            "- `dragon_rollout_governance_snapshot.csv`",
            "- `dragon_rollout_rollback_runbook.md`",
        ]
    )
    (base_dir / "dragon_rollout_governance_report.md").write_text("\n".join(lines) + "\n", encoding="utf-8")

    runbook_lines = [
        "# Dragon Rollback Runbook",
        "",
        "## Trigger Conditions",
        "- Any gate in `dragon_rollout_governance_snapshot.csv` with `status = hard_fail`.",
        "- Decision equals `ROLLBACK_REVIEW_REQUIRED`.",
        "- Persistent warning state (`HOLD_AND_REVIEW`) for more than two bars.",
        "",
        "## Immediate Actions",
        "1. Freeze candidate branch decisions for live-facing guidance.",
        "2. Switch active branch to control branch `alpha_first_selective_veto`.",
        "3. Re-run forward chain and verify monitor status:",
        "   `py -3 dragon_forward_observation_pipeline.py`",
        "4. Inspect latest artifacts:",
        "   - `dragon_daily_monitor_snapshot.csv`",
        "   - `dragon_monitor_health_report.md`",
        "   - `dragon_branch_divergence_report.md`",
        "",
        "## Recovery Checklist",
        "1. Confirm all hard-fail gates are cleared.",
        "2. Confirm warning streaks return to threshold range.",
        "3. Confirm branch divergence drops to `none` or `mild`.",
        "4. Regenerate rollout report and ensure decision is `FORWARD_OK` before reactivating candidate branch.",
        "",
        "## Current Snapshot",
        f"- latest_bar_date: `{latest_bar_date}`",
        f"- decision: `{decision}`",
        f"- recommended_active_branch: `{active_branch}`",
        f"- fallback_branch: `{fallback_branch}`",
    ]
    (base_dir / "dragon_rollout_rollback_runbook.md").write_text("\n".join(runbook_lines) + "\n", encoding="utf-8")

    if decision == FORWARD_OK:
        return
    if decision == HOLD_AND_REVIEW:
        return
    if decision == ROLLBACK_REVIEW_REQUIRED:
        return


if __name__ == "__main__":
    main()