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

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

import pandas as pd


@dataclass(frozen=True)
class WindowScript:
    path: str
    scope: str
    category: str
    expectation: str


WINDOW_SCRIPTS = [
    WindowScript("dragon_backtest.py", "workbook window", "evaluation", "dual_bound"),
    WindowScript("dragon_cost_stress_test.py", "fixed release window", "evaluation", "dual_bound"),
    WindowScript("dragon_deep_oversold_confirmation_experiments.py", "workbook window", "experiment", "dual_bound"),
    WindowScript("dragon_deep_oversold_experiments.py", "workbook window", "experiment", "dual_bound"),
    WindowScript("dragon_deep_oversold_selective_veto_experiments.py", "workbook window", "experiment", "dual_bound"),
    WindowScript("dragon_equity_curve_review.py", "fixed release window", "evaluation", "dual_bound"),
    WindowScript("dragon_glued_alpha_candidate.py", "workbook window", "evaluation", "dual_bound"),
    WindowScript("dragon_glued_refined_branch_review.py", "hybrid bounded window", "evaluation", "dual_bound"),
    WindowScript("dragon_glued_refined_removed_trade_attribution.py", "fixed release window", "attribution", "dual_bound"),
    WindowScript("dragon_glued_refined_sensitivity.py", "fixed release window", "evaluation", "dual_bound"),
    WindowScript("dragon_glued_refine_experiments.py", "workbook window", "experiment", "dual_bound"),
    WindowScript("dragon_predictive_break_experiments.py", "workbook window", "experiment", "dual_bound"),
    WindowScript("dragon_rc1_release.py", "fixed release window", "release", "dual_bound"),
    WindowScript("dragon_refined_alpha_attribution.py", "fixed release window", "attribution", "dual_bound"),
    WindowScript("dragon_refined_execution_validation.py", "fixed release window", "evaluation", "dual_bound"),
    WindowScript("dragon_rule_ablation.py", "workbook window", "evaluation", "dual_bound"),
    WindowScript("dragon_short_holding_audit.py", "workbook window", "audit", "dual_bound"),
    WindowScript("dragon_short_holding_experiments.py", "workbook window", "experiment", "dual_bound"),
    WindowScript("dragon_strategy_overview.py", "fixed release window", "overview", "dual_bound"),
    WindowScript("dragon_threshold_perturbation.py", "workbook window", "evaluation", "dual_bound"),
    WindowScript("dragon_daily_signal_pipeline.py", "live / forward window", "live", "live_exception"),
]


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


def _fmt_pct(value: object) -> str:
    if value is None or pd.isna(value):
        return "NA"
    return f"{float(value):.2%}"


def _fmt_num(value: object, digits: int = 2) -> str:
    if value is None or pd.isna(value):
        return "NA"
    return f"{float(value):.{digits}f}"


def _window_filter_status(text: str, expectation: str) -> tuple[str, str]:
    has_buy = 'trades["buy_date"]' in text
    has_sell = 'trades["sell_date"]' in text
    if expectation == "live_exception":
        return "EXEMPT", "Live / forward pipeline intentionally keeps open-ended trade coverage."
    if has_buy and has_sell:
        return "PASS", "Trade filter constrains both buy_date and sell_date."
    if has_buy and not has_sell:
        return "FAIL", "Trade filter constrains buy_date only."
    return "FAIL", "No recognizable trade-window filter found."


def build_window_consistency_report(base_dir: Path) -> pd.DataFrame:
    rows: list[dict[str, object]] = []
    for item in WINDOW_SCRIPTS:
        text = (base_dir / item.path).read_text(encoding="utf-8")
        status, note = _window_filter_status(text, item.expectation)
        rows.append(
            {
                "script": item.path,
                "scope": item.scope,
                "category": item.category,
                "expectation": item.expectation,
                "status": status,
                "note": note,
            }
        )
    df = pd.DataFrame(rows)
    passed = int((df["status"] == "PASS").sum())
    exempt = int((df["status"] == "EXEMPT").sum())
    failed = int((df["status"] == "FAIL").sum())
    lines = [
        "# Dragon Review - Window Consistency",
        "",
        "## Scope",
        "- Goal: verify that in-sample / workbook-window trade statistics do not keep window-external exits.",
        "- Rule: for bounded research windows, trade filters must constrain both `buy_date` and `sell_date`.",
        "",
        "## Summary",
        f"- PASS: `{passed}`",
        f"- EXEMPT: `{exempt}`",
        f"- FAIL: `{failed}`",
        "",
        "## Result Table",
    ]
    for _, row in df.iterrows():
        lines.append(
            f"- `{row['script']}` | {row['category']} | {row['scope']} | `{row['status']}` | {row['note']}"
        )
    if failed == 0:
        lines.extend(
            [
                "",
                "## Judgment",
                "- The bounded research/evaluation pack is now on a consistent dual-bound trade-window rule.",
                "- `dragon_daily_signal_pipeline.py` remains intentionally exempt because it serves the live / forward chain rather than workbook-window evaluation.",
            ]
        )
    (base_dir / "dragon_review_window_consistency.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    return df


def _branch_source_frames(base_dir: Path) -> dict[str, pd.DataFrame]:
    frames: dict[str, pd.DataFrame] = {}
    overview = _load_csv(base_dir / "dragon_strategy_overview.csv")
    if not overview.empty:
        frames["dragon_strategy_overview.csv"] = overview
    glued = _load_csv(base_dir / "dragon_glued_refined_branch_summary.csv")
    if not glued.empty:
        frames["dragon_glued_refined_branch_summary.csv"] = glued
    alpha = _load_csv(base_dir / "dragon_alpha_first_branch_summary.csv")
    if not alpha.empty:
        frames["dragon_alpha_first_branch_summary.csv"] = alpha
    return frames


def _iter_metric_values(base_dir: Path) -> Iterable[dict[str, object]]:
    metrics = [
        "trades",
        "win_rate",
        "avg_return",
        "median_return",
        "profit_factor",
        "compounded_return",
        "cagr",
        "real_buy_overlap",
        "real_sell_overlap",
    ]
    for source, df in _branch_source_frames(base_dir).items():
        for _, row in df.iterrows():
            branch = str(row["branch"])
            for metric in metrics:
                if metric in row.index:
                    yield {
                        "source": source,
                        "branch": branch,
                        "metric": metric,
                        "value": row[metric],
                    }
    rc1_path = base_dir / "dragon_rc1_config_snapshot.json"
    if rc1_path.exists():
        payload = json.loads(rc1_path.read_text(encoding="utf-8"))
        for metric, source_metric in [
            ("trades", "trade_count"),
            ("win_rate", "win_rate"),
            ("avg_return", "avg_return"),
            ("median_return", "median_return"),
            ("profit_factor", "profit_factor"),
            ("compounded_return", "compounded_return"),
            ("cagr", "cagr"),
        ]:
            yield {
                "source": "dragon_rc1_config_snapshot.json",
                "branch": str(payload["branch_name"]),
                "metric": metric,
                "value": payload[source_metric],
            }


def build_branch_metric_consistency(base_dir: Path) -> pd.DataFrame:
    raw = pd.DataFrame(list(_iter_metric_values(base_dir)))
    if raw.empty:
        raw.to_csv(base_dir / "dragon_review_branch_metric_consistency.csv", index=False, encoding="utf-8-sig")
        (base_dir / "dragon_review_branch_metric_consistency.md").write_text(
            "# Dragon Review - Branch Metric Consistency\n\n- No source data found.\n",
            encoding="utf-8",
        )
        return raw

    raw["value"] = pd.to_numeric(raw["value"], errors="coerce")
    pivot = raw.pivot_table(index=["branch", "metric"], columns="source", values="value", aggfunc="last").reset_index()
    source_cols = [col for col in pivot.columns if col not in {"branch", "metric"}]
    records: list[dict[str, object]] = []
    for _, row in pivot.iterrows():
        values = [row[col] for col in source_cols if pd.notna(row[col])]
        max_abs_diff = float(max(values) - min(values)) if values else float("nan")
        if len(values) <= 1:
            status = "single_source"
        else:
            tol = 1e-12 if row["metric"] in {"trades", "real_buy_overlap", "real_sell_overlap"} else 1e-9
            status = "match" if max_abs_diff <= tol else "mismatch"
        rec = {
            "branch": row["branch"],
            "metric": row["metric"],
            "source_count": int(len(values)),
            "min_value": float(min(values)) if values else float("nan"),
            "max_value": float(max(values)) if values else float("nan"),
            "max_abs_diff": max_abs_diff,
            "status": status,
        }
        for col in source_cols:
            rec[col] = row[col]
        records.append(rec)
    result = pd.DataFrame(records).sort_values(["branch", "metric"]).reset_index(drop=True)
    result.to_csv(base_dir / "dragon_review_branch_metric_consistency.csv", index=False, encoding="utf-8-sig")

    mismatches = result[result["status"] == "mismatch"].copy()
    lines = [
        "# Dragon Review - Branch Metric Consistency",
        "",
        "## Scope",
        "- Sources compared:",
        "- `dragon_strategy_overview.csv`",
        "- `dragon_glued_refined_branch_summary.csv`",
        "- `dragon_alpha_first_branch_summary.csv`",
        "- `dragon_rc1_config_snapshot.json`",
        "",
        f"- Compared rows: `{len(result)}`",
        f"- Mismatches: `{len(mismatches)}`",
        "",
    ]
    if mismatches.empty:
        lines.append("- All compared branch metrics are consistent across current outputs.")
    else:
        lines.extend(["## Mismatches"])
        for _, row in mismatches.iterrows():
            parts = []
            for col in source_cols:
                if pd.notna(row[col]):
                    parts.append(f"{col}={row[col]}")
            lines.append(
                f"- `{row['branch']}` / `{row['metric']}` | spread `{row['max_abs_diff']}` | " + "; ".join(parts)
            )
    lines.extend(
        [
            "",
            "## Judgment",
            "- `match` means the same branch/metric agrees across all currently available source files.",
            "- `mismatch` means at least one report family is using a different metric definition or evaluation window.",
            "- `single_source` means only one current artifact exposes that metric, so cross-check confidence is lower.",
        ]
    )
    (base_dir / "dragon_review_branch_metric_consistency.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    return result


def _calc_removed_trade_over_removal_count(base_dir: Path) -> float:
    df = _load_csv(base_dir / "dragon_glued_refined_removed_trade_attribution.csv")
    if df.empty or "recommendation" not in df.columns:
        return float("nan")
    return float((df["recommendation"].astype(str) == "OVER_REMOVAL").sum())


def _calc_local_sensitivity_robust_case_count(base_dir: Path) -> float:
    df = _load_csv(base_dir / "dragon_glued_refined_sensitivity.csv")
    if df.empty or "label" not in df.columns:
        return float("nan")
    candidate = df[df["label"] == "refined_candidate_baseline"].copy()
    if candidate.empty:
        return float("nan")
    candidate_row = candidate.iloc[0]
    neighborhood = df[~df["label"].isin(["current_alpha_control", "refined_candidate_baseline"])].copy()
    robust = neighborhood[
        (neighborhood["avg_return"] >= float(candidate_row["avg_return"]) - 0.0015)
        & (neighborhood["profit_factor"] >= float(candidate_row["profit_factor"]) - 0.20)
        & (neighborhood["real_buy_overlap"] >= int(candidate_row["real_buy_overlap"]) - 1)
        & (neighborhood["real_sell_overlap"] >= int(candidate_row["real_sell_overlap"]) - 1)
    ]
    return float(len(robust))


def build_execution_monitor_review(base_dir: Path) -> None:
    monitor = _load_csv(base_dir / "dragon_daily_monitor_snapshot.csv")
    weekly = _load_csv(base_dir / "dragon_forward_weekly_summary.csv")
    html_text = (base_dir / "dragon_forward_weekly_review.html").read_text(encoding="utf-8")
    daily_text = (base_dir / "dragon_daily_signal_pipeline.py").read_text(encoding="utf-8")
    exec_text = (base_dir / "dragon_refined_execution_validation.py").read_text(encoding="utf-8")

    actual_removed = _calc_removed_trade_over_removal_count(base_dir)
    actual_robust = _calc_local_sensitivity_robust_case_count(base_dir)
    monitor_map = {
        str(row["metric"]): float(row["actual_value"])
        for _, row in monitor.iterrows()
        if pd.notna(row["actual_value"])
    }
    removed_match = pd.notna(actual_removed) and abs(monitor_map.get("removed_trade_over_removal_count", float("nan")) - actual_removed) < 1e-12
    robust_match = pd.notna(actual_robust) and abs(monitor_map.get("local_sensitivity_robust_case_count", float("nan")) - actual_robust) < 1e-12
    daily_uses_nan = 'float("nan") if buy_next is None' in daily_text and 'float("nan") if sell_next is None' in daily_text
    exec_uses_nan = 'float("nan") if buy_next is None' in exec_text and 'float("nan") if sell_next is None' in exec_text
    weekly_has_system_monitor = not weekly.empty and "system_monitor" in set(weekly["branch"].astype(str))
    weekly_html_has_system_monitor = "system_monitor" in html_text

    lines = [
        "# Dragon Review - Execution And Monitor Consistency",
        "",
        "## Governance Metrics",
        f"- `removed_trade_over_removal_count`: monitor `{monitor_map.get('removed_trade_over_removal_count')}` vs source-derived `{actual_removed}` -> `{'match' if removed_match else 'mismatch'}`",
        f"- `local_sensitivity_robust_case_count`: monitor `{monitor_map.get('local_sensitivity_robust_case_count')}` vs source-derived `{actual_robust}` -> `{'match' if robust_match else 'mismatch'}`",
        "",
        "## Execution Fallback Rule",
        f"- `dragon_daily_signal_pipeline.py` uses NaN on missing next-bar execution prices: `{daily_uses_nan}`",
        f"- `dragon_refined_execution_validation.py` uses NaN on missing next-bar execution prices: `{exec_uses_nan}`",
        "",
        "## Weekly Monitor Separation",
        f"- `dragon_forward_weekly_summary.csv` includes `system_monitor` row: `{weekly_has_system_monitor}`",
        f"- `dragon_forward_weekly_review.html` includes `system_monitor` text: `{weekly_html_has_system_monitor}`",
        "",
        "## Judgment",
        "- The monitor chain is trustworthy only if governance metrics are derived from current source artifacts rather than hard-coded constants.",
        "- The execution-aware chain is trustworthy only if missing next-bar prices do not silently fall back to same-bar close.",
        "- Weekly summary is cleaner now because branch rows and system-level monitor counts are separated.",
    ]
    (base_dir / "dragon_review_execution_monitor.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


def build_reporting_integrity_review(base_dir: Path) -> None:
    latest_bar = json.loads((base_dir / "dragon_forward_observation_state.json").read_text(encoding="utf-8")).get("latest_bar_date", "latest")
    files_to_check = [
        base_dir / "dragon_reports_index.html",
        base_dir / "dragon_daily_signal_report.html",
        base_dir / "dragon_forward_weekly_review.html",
        base_dir / "dragon_historical_trade_details.html",
        base_dir / "dragon_indicator_strategy_guide_cn.html",
        base_dir / "html_reports" / "index.html",
        base_dir / "html_reports" / f"dragon_daily_signal_report_{latest_bar}.html",
        base_dir / "html_reports" / f"dragon_forward_weekly_review_{latest_bar}.html",
        base_dir / "html_reports" / f"dragon_historical_trade_details_{latest_bar}.html",
    ]
    existence_rows = [{"path": str(path.relative_to(base_dir)), "exists": path.exists()} for path in files_to_check]

    index_text = (base_dir / "dragon_reports_index.html").read_text(encoding="utf-8")
    archive_index_text = (base_dir / "html_reports" / "index.html").read_text(encoding="utf-8")
    detail_text = (base_dir / "dragon_historical_trade_details.html").read_text(encoding="utf-8")
    daily_text = (base_dir / "dragon_daily_signal_report.html").read_text(encoding="utf-8")

    checks = [
        ("root index links to root daily report", 'href="dragon_daily_signal_report.html"' in index_text),
        (
            "root index links to archived daily report",
            f'href="html_reports/dragon_daily_signal_report_{latest_bar}.html"' in index_text,
        ),
        (
            "archive index links locally inside html_reports",
            f'href="dragon_daily_signal_report_{latest_bar}.html"' in archive_index_text,
        ),
        ("detail page contains snapshot summary strip", "snapshot-summary" in detail_text),
        ("detail page contains event summary labels", "总事件" in detail_text and "前一条：" in detail_text),
        ("detail page contains query filters", "branch-filter" in detail_text and "keyword-filter" in detail_text),
        ("daily report links to historical detail page", 'href="dragon_historical_trade_details.html"' in daily_text),
    ]

    lines = [
        "# Dragon Review - Reporting Integrity",
        "",
        "## File Existence",
    ]
    for row in existence_rows:
        lines.append(f"- `{row['path']}` -> `{row['exists']}`")
    lines.extend(["", "## Link / Feature Checks"])
    for label, passed in checks:
        lines.append(f"- {label}: `{passed}`")
    lines.extend(
        [
            "",
            "## Judgment",
            "- Root and archive HTML outputs are present and linked through the expected root-vs-archive relative paths.",
            "- Historical detail reporting currently includes the new indicator snapshot event-summary strip and deep-link filter controls.",
            "- Terminal mojibake remains a shell-display issue; these checks only validate file presence and embedded text markers, not browser rendering fidelity.",
        ]
    )
    (base_dir / "dragon_review_reporting_integrity.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


def build_system_final(base_dir: Path, branch_consistency: pd.DataFrame) -> None:
    mismatch_branches = branch_consistency[branch_consistency["status"] == "mismatch"].copy()
    trusted_direct = [
        "dragon_alpha_first_branch_summary.csv",
        "dragon_glued_refined_branch_summary.csv",
        "dragon_rc1_config_snapshot.json",
        "dragon_daily_monitor_snapshot.csv",
    ]
    if mismatch_branches.empty:
        trusted_direct.append("dragon_strategy_overview.csv")
    trusted_labeled = [
        "dragon_refined_execution_stress.csv",
        "dragon_cost_stress_test.csv",
        "dragon_forward_weekly_summary.csv",
        "dragon_historical_trade_details.html",
    ]
    not_recommended = []
    if not mismatch_branches.empty:
        not_recommended.append("dragon_strategy_overview.csv")

    lines = [
        "# Dragon System Review Final",
        "",
        "## Overall Judgment",
        "- The current workspace is much closer to a trustworthy research pack after the window-consistency sweep and monitor-fallback fixes.",
        "",
        "## Trust Tiers",
        "- 可信可直接使用:",
    ]
    if mismatch_branches.empty:
        lines.insert(4, "- Cross-report headline metrics are currently aligned across the main branch-summary, release, and overview artifacts.")
        lines.insert(5, "- The remaining cautions are about report interpretation and forward monitoring, not internal metric drift.")
    else:
        lines.insert(4, "- The main remaining risk is not strategy logic but metric-definition drift between a few report families.")
    lines.extend([f"- `{name}`" for name in trusted_direct])
    lines.append("- 可信但需标注口径:")
    lines.extend([f"- `{name}`" for name in trusted_labeled])
    lines.append("- 暂不建议直接引用:")
    if not_recommended:
        lines.extend([f"- `{name}`" for name in not_recommended])
    else:
        lines.append("- `none`")

    if not mismatch_branches.empty:
        lines.extend(["", "## Remaining Review Findings"])
        for _, row in mismatch_branches.iterrows():
            lines.append(
                f"- `{row['branch']}` / `{row['metric']}` remains inconsistent across report families; see `dragon_review_branch_metric_consistency.csv`."
            )

    lines.extend(
        [
            "",
            "## Practical Meaning",
            "- Use the branch-specific summary/release artifacts as the primary basis for governance decisions.",
            "- Use the consistency reports as an audit trail before external distribution of top-line metrics.",
            "- `dragon_strategy_overview.csv` is now aligned with the main branch artifacts and can be used as the compact comparison view.",
        ]
    )
    (base_dir / "dragon_system_review_final.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    build_window_consistency_report(base_dir)
    branch_consistency = build_branch_metric_consistency(base_dir)
    build_execution_monitor_review(base_dir)
    build_reporting_integrity_review(base_dir)
    build_system_final(base_dir, branch_consistency)


if __name__ == "__main__":
    main()