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

from __future__ import annotations

import json
from pathlib import Path

import pandas as pd


def _load_csv(base_dir: Path, name: str) -> pd.DataFrame:
    return pd.read_csv(base_dir / name, encoding="utf-8-sig")


def _format_pct(value: float) -> str:
    if pd.isna(value):
        return "NA"
    if value == float("inf"):
        return "inf"
    return f"{value:.2%}"


def _format_num(value: float) -> str:
    if pd.isna(value):
        return "NA"
    if value == float("inf"):
        return "inf"
    return f"{value:.2f}"


def _wf_stats(df: pd.DataFrame, branch: str, scheme: str) -> tuple[int, int, float]:
    view = df[(df["branch"] == branch) & (df["scheme"] == scheme)].copy()
    positive = int((view["test_avg_return"] > 0).sum()) if not view.empty else 0
    total = int(len(view))
    avg_test = float(view["test_avg_return"].mean()) if not view.empty else float("nan")
    return positive, total, avg_test


def main() -> None:
    base_dir = Path(__file__).resolve().parent

    branch_summary = _load_csv(base_dir, "dragon_glued_refined_branch_summary.csv")
    walk_forward = _load_csv(base_dir, "dragon_glued_refined_branch_walk_forward.csv")
    removed = _load_csv(base_dir, "dragon_glued_refined_removed_trade_attribution.csv")

    alpha = branch_summary[branch_summary["branch"] == "alpha_first_selective_veto"].iloc[0]
    refined = branch_summary[branch_summary["branch"] == "alpha_first_glued_refined_hot_cap"].iloc[0]

    af_anchor_pos, af_anchor_total, af_anchor_avg = _wf_stats(walk_forward, "alpha_first_selective_veto", "anchored_expanding")
    ref_anchor_pos, ref_anchor_total, ref_anchor_avg = _wf_stats(walk_forward, "alpha_first_glued_refined_hot_cap", "anchored_expanding")
    af_roll_pos, af_roll_total, af_roll_avg = _wf_stats(walk_forward, "alpha_first_selective_veto", "rolling_3y")
    ref_roll_pos, ref_roll_total, ref_roll_avg = _wf_stats(walk_forward, "alpha_first_glued_refined_hot_cap", "rolling_3y")

    removed_trades = int(len(removed))
    removed_avg_return = float(removed["return_pct"].mean()) if not removed.empty else float("nan")
    removed_win_rate = float((removed["return_pct"] > 0).mean()) if not removed.empty else float("nan")
    removed_keep = int((removed["recommendation"] == "KEEP_REMOVAL").sum())
    removed_observe = int((removed["recommendation"] == "OBSERVE_REMOVAL").sum())
    removed_over = int((removed["recommendation"] == "OVER_REMOVAL").sum())

    avg_return_delta = float(refined["avg_return"] - alpha["avg_return"])
    profit_factor_delta = float(refined["profit_factor"] - alpha["profit_factor"])
    buy_overlap_delta = int(refined["real_buy_overlap"] - alpha["real_buy_overlap"])
    sell_overlap_delta = int(refined["real_sell_overlap"] - alpha["real_sell_overlap"])
    short_00_05d_delta = float(refined["short_00_05d_avg_return"] - alpha["short_00_05d_avg_return"])
    short_06_10d_delta = float(refined["short_06_10d_avg_return"] - alpha["short_06_10d_avg_return"])

    headline_quality_gate = (
        avg_return_delta >= 0.003
        and profit_factor_delta >= 0.50
        and short_00_05d_delta >= 0
        and short_06_10d_delta >= 0
    )
    stability_gate = (
        ref_anchor_pos >= af_anchor_pos
        and ref_roll_pos >= af_roll_pos
        and ref_anchor_avg >= af_anchor_avg
        and ref_roll_avg >= af_roll_avg
    )
    removal_quality_gate = removed.empty or (
        removed_over == 0
        and removed_observe <= 1
        and removed_win_rate <= 0.05
        and removed_avg_return < 0
    )
    # Automatic promotion should not occur if the candidate loses more than 8
    # additional aligned BUYs or SELLs versus the current formal alpha branch.
    # Positive deltas are improvements and should not block promotion.
    alignment_cost_gate = (buy_overlap_delta >= -8 and sell_overlap_delta >= -8)

    if headline_quality_gate and stability_gate and removal_quality_gate and alignment_cost_gate:
        final_decision = "PROMOTE_REFINED_ALPHA_BASELINE"
    elif headline_quality_gate and stability_gate and removal_quality_gate:
        final_decision = "DUAL_TRACK_GOVERNANCE"
    else:
        final_decision = "KEEP_CURRENT_ALPHA_BASELINE"

    matrix_rows = [
        {
            "branch": "alpha_first_selective_veto",
            "role": "current_formal_alpha",
            "trades": int(alpha["trades"]),
            "win_rate": float(alpha["win_rate"]),
            "avg_return": float(alpha["avg_return"]),
            "median_return": float(alpha["median_return"]),
            "profit_factor": float(alpha["profit_factor"]),
            "avg_mfe": float(alpha["avg_mfe"]),
            "avg_mae": float(alpha["avg_mae"]),
            "short_00_05d_avg_return": float(alpha["short_00_05d_avg_return"]),
            "short_06_10d_avg_return": float(alpha["short_06_10d_avg_return"]),
            "real_buy_overlap": int(alpha["real_buy_overlap"]),
            "real_sell_overlap": int(alpha["real_sell_overlap"]),
            "anchored_positive_years": af_anchor_pos,
            "anchored_total_years": af_anchor_total,
            "anchored_avg_test_return": af_anchor_avg,
            "rolling_positive_years": af_roll_pos,
            "rolling_total_years": af_roll_total,
            "rolling_avg_test_return": af_roll_avg,
            "removed_trades_vs_current_alpha": 0,
            "removed_avg_return_vs_current_alpha": float("nan"),
            "removed_win_rate_vs_current_alpha": float("nan"),
            "over_removal_count_vs_current_alpha": 0,
            "observe_removal_count_vs_current_alpha": 0,
            "keep_removal_count_vs_current_alpha": 0,
            "headline_quality_gate": None,
            "stability_gate": None,
            "removal_quality_gate": None,
            "alignment_cost_gate": None,
            "governance_decision": "CURRENT_BASELINE",
        },
        {
            "branch": "alpha_first_glued_refined_hot_cap",
            "role": "leading_candidate_alpha",
            "trades": int(refined["trades"]),
            "win_rate": float(refined["win_rate"]),
            "avg_return": float(refined["avg_return"]),
            "median_return": float(refined["median_return"]),
            "profit_factor": float(refined["profit_factor"]),
            "avg_mfe": float(refined["avg_mfe"]),
            "avg_mae": float(refined["avg_mae"]),
            "short_00_05d_avg_return": float(refined["short_00_05d_avg_return"]),
            "short_06_10d_avg_return": float(refined["short_06_10d_avg_return"]),
            "real_buy_overlap": int(refined["real_buy_overlap"]),
            "real_sell_overlap": int(refined["real_sell_overlap"]),
            "anchored_positive_years": ref_anchor_pos,
            "anchored_total_years": ref_anchor_total,
            "anchored_avg_test_return": ref_anchor_avg,
            "rolling_positive_years": ref_roll_pos,
            "rolling_total_years": ref_roll_total,
            "rolling_avg_test_return": ref_roll_avg,
            "removed_trades_vs_current_alpha": removed_trades,
            "removed_avg_return_vs_current_alpha": removed_avg_return,
            "removed_win_rate_vs_current_alpha": removed_win_rate,
            "over_removal_count_vs_current_alpha": removed_over,
            "observe_removal_count_vs_current_alpha": removed_observe,
            "keep_removal_count_vs_current_alpha": removed_keep,
            "headline_quality_gate": headline_quality_gate,
            "stability_gate": stability_gate,
            "removal_quality_gate": removal_quality_gate,
            "alignment_cost_gate": alignment_cost_gate,
            "governance_decision": final_decision,
        },
    ]
    matrix = pd.DataFrame(matrix_rows)
    matrix.to_csv(base_dir / "dragon_alpha_branch_governance_matrix.csv", index=False, encoding="utf-8-sig")

    decision_payload = {
        "current_formal_alpha": "alpha_first_selective_veto",
        "leading_candidate_alpha": "alpha_first_glued_refined_hot_cap",
        "avg_return_delta_vs_current": avg_return_delta,
        "profit_factor_delta_vs_current": profit_factor_delta,
        "buy_overlap_delta_vs_current": buy_overlap_delta,
        "sell_overlap_delta_vs_current": sell_overlap_delta,
        "headline_quality_gate": headline_quality_gate,
        "stability_gate": stability_gate,
        "removal_quality_gate": removal_quality_gate,
        "alignment_cost_gate": alignment_cost_gate,
        "final_decision": final_decision,
    }
    (base_dir / "dragon_alpha_branch_governance_decision.json").write_text(
        json.dumps(decision_payload, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )

    lines = [
        "# Dragon Alpha Branch Governance",
        "",
        "## Scope",
        "- Current formal alpha branch: `alpha_first_selective_veto`.",
        "- Leading candidate branch: `alpha_first_glued_refined_hot_cap`.",
        "- Goal: decide whether the refined branch should replace the current formal alpha branch or remain a governed candidate.",
        "",
        "## Headline Metrics",
        f"- current alpha: trades `{int(alpha['trades'])}`, avg_return `{_format_pct(float(alpha['avg_return']))}`, profit_factor `{_format_num(float(alpha['profit_factor']))}`, real BUY / SELL `{int(alpha['real_buy_overlap'])}/{int(alpha['real_sell_overlap'])}`",
        f"- refined candidate: trades `{int(refined['trades'])}`, avg_return `{_format_pct(float(refined['avg_return']))}`, profit_factor `{_format_num(float(refined['profit_factor']))}`, real BUY / SELL `{int(refined['real_buy_overlap'])}/{int(refined['real_sell_overlap'])}`",
        f"- delta: avg_return `{_format_pct(avg_return_delta)}`, profit_factor `{_format_num(profit_factor_delta)}`, BUY overlap `{buy_overlap_delta}`, SELL overlap `{sell_overlap_delta}`",
        "",
        "## Risk And Quality",
        f"- avg MFE / MAE: current `{_format_pct(float(alpha['avg_mfe']))}` / `{_format_pct(float(alpha['avg_mae']))}` vs refined `{_format_pct(float(refined['avg_mfe']))}` / `{_format_pct(float(refined['avg_mae']))}`",
        f"- short `00-05d`: current `{_format_pct(float(alpha['short_00_05d_avg_return']))}` vs refined `{_format_pct(float(refined['short_00_05d_avg_return']))}`",
        f"- short `06-10d`: current `{_format_pct(float(alpha['short_06_10d_avg_return']))}` vs refined `{_format_pct(float(refined['short_06_10d_avg_return']))}`",
        "",
        "## Walk-Forward",
        f"- anchored expanding: current `{af_anchor_pos}/{af_anchor_total}`, avg `{_format_pct(af_anchor_avg)}` vs refined `{ref_anchor_pos}/{ref_anchor_total}`, avg `{_format_pct(ref_anchor_avg)}`",
        f"- rolling 3Y: current `{af_roll_pos}/{af_roll_total}`, avg `{_format_pct(af_roll_avg)}` vs refined `{ref_roll_pos}/{ref_roll_total}`, avg `{_format_pct(ref_roll_avg)}`",
        "",
        "## Removed-Trade Attribution",
        f"- removed trades vs current alpha: `{removed_trades}`",
        f"- removed-set avg_return / win_rate: `{_format_pct(removed_avg_return)}` / `{_format_pct(removed_win_rate)}`",
        f"- recommendation mix KEEP / OBSERVE / OVER: `{removed_keep}/{removed_observe}/{removed_over}`",
        "- Interpretation: the refined branch now improves by removing only weak, losing short-holding glued trades; it does not rely on deleting profitable samples.",
        "",
        "## Upgrade Gates",
        "- `headline_quality_gate`: requires avg_return delta `>= +0.30%`, profit_factor delta `>= +0.50`, and no short-bucket deterioration.",
        f"  result: `{'PASS' if headline_quality_gate else 'FAIL'}`",
        "- `stability_gate`: requires anchored and rolling walk-forward to be no worse than the current formal alpha branch.",
        f"  result: `{'PASS' if stability_gate else 'FAIL'}`",
        "- `removal_quality_gate`: requires `OVER_REMOVAL = 0`, `OBSERVE_REMOVAL <= 1`, removed-set win_rate `<= 5%`, and removed-set avg_return `< 0`.",
        f"  result: `{'PASS' if removal_quality_gate else 'FAIL'}`",
        "- `alignment_cost_gate`: automatic promotion only if incremental overlap loss is no more than `8` additional BUYs and `8` additional SELLs versus the current formal alpha branch.",
        f"  result: `{'PASS' if alignment_cost_gate else 'FAIL'}`",
        "",
        "## Final Decision",
        f"- governance_decision: `{final_decision}`",
    ]

    if final_decision == "PROMOTE_REFINED_ALPHA_BASELINE":
        lines.extend(
            [
                "- Decision: promote `alpha_first_glued_refined_hot_cap` to the new formal alpha-first baseline.",
                "- Reason: quality, stability, removal quality, and alignment cost all pass together.",
            ]
        )
    elif final_decision == "DUAL_TRACK_GOVERNANCE":
        lines.extend(
            [
                "- Decision: keep `alpha_first_selective_veto` as the formal alpha branch and keep `alpha_first_glued_refined_hot_cap` as the governed leading candidate.",
                "- Reason: quality, stability, and removal quality all pass, but alignment loss is still large enough that promotion should be explicit rather than automatic.",
            ]
        )
    else:
        lines.extend(
            [
                "- Decision: keep `alpha_first_selective_veto` as the formal alpha branch.",
                "- Reason: the refined candidate does not pass enough upgrade gates to justify even governed promotion.",
            ]
        )

    lines.extend(["", "## Recommendation"])
    if final_decision == "PROMOTE_REFINED_ALPHA_BASELINE":
        lines.extend(
            [
                "- Promote `alpha_first_glued_refined_hot_cap` into the formal alpha branch and freeze a new governed release snapshot.",
                "- Keep `alpha_first_selective_veto` as the immediate control branch for post-promotion monitoring.",
            ]
        )
    elif final_decision == "DUAL_TRACK_GOVERNANCE":
        lines.extend(
            [
                "- Keep `alpha_first_selective_veto` as the current formal alpha branch for now.",
                "- Keep `alpha_first_glued_refined_hot_cap` as the first governed promotion candidate if the governance objective explicitly shifts toward stronger alpha.",
            ]
        )
    else:
        lines.extend(
            [
                "- Keep `alpha_first_selective_veto` as the formal alpha branch.",
                "- Re-open refined promotion only after a new candidate improves quality without failing the current governance gates.",
            ]
        )

    (base_dir / "dragon_alpha_branch_governance.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()