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

from __future__ import annotations

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 _classify_entry(row: pd.Series) -> str:
    a1 = float(row["buy_a1"])
    b1 = float(row["buy_b1"])
    c1 = float(row["buy_c1"])
    if b1 > 0:
        return "positive_b1_rebound"
    if c1 < 11 and a1 < -0.05 and b1 < -0.08:
        return "deep_capitulation"
    if c1 < 12 and b1 < -0.06:
        return "classic_oversold"
    if c1 >= 15 or a1 > -0.03 or b1 > -0.03:
        return "shallow_false_start"
    return "mixed_oversold"


def _quality_bucket(row: pd.Series) -> str:
    ret = float(row["return_pct"])
    mfe = float(row["mfe_pct"])
    mae = float(row["mae_pct"])
    if ret > 0.02:
        return "strong_positive"
    if ret > 0 and mfe > 0.03:
        return "weak_positive"
    if ret <= -0.03 and mfe < 0.01:
        return "immediate_failure"
    if ret <= -0.03:
        return "rebound_then_fail"
    if mae < -0.04:
        return "high_volatility_noise"
    return "flat_noise"


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    quality = _load_csv(base_dir, "dragon_trade_quality.csv")
    path_trace = _load_csv(base_dir, "dragon_trade_path_trace.csv")
    rows = quality[quality["buy_reason"].astype(str).str.startswith("deep_oversold_rebound_buy")].copy()
    rows = rows.merge(
        path_trace[
            [
                "buy_date",
                "sell_date",
                "buy_a1",
                "buy_b1",
                "buy_c1",
                "buy_aligned_with_workbook",
                "sell_aligned_with_workbook",
            ]
        ],
        on=["buy_date", "sell_date"],
        how="left",
    )
    rows["entry_subtype"] = rows.apply(_classify_entry, axis=1)
    rows["quality_bucket"] = rows.apply(_quality_bucket, axis=1)
    rows["is_winner"] = rows["return_pct"] > 0
    rows["is_fast_failure"] = (rows["holding_days"] <= 6) & (rows["return_pct"] < 0)
    rows["is_shallow_like"] = (
        (rows["buy_c1"] >= 15)
        | (rows["buy_a1"] > -0.03)
        | (rows["buy_b1"] > -0.03)
    )

    audit_cols = [
        "buy_date",
        "sell_date",
        "buy_a1",
        "buy_b1",
        "buy_c1",
        "sell_reason",
        "holding_days",
        "return_pct",
        "mfe_pct",
        "mae_pct",
        "giveback_from_peak_pct",
        "entry_subtype",
        "quality_bucket",
        "is_winner",
        "is_fast_failure",
        "is_shallow_like",
        "buy_aligned_with_workbook",
        "sell_aligned_with_workbook",
    ]
    rows[audit_cols].to_csv(base_dir / "dragon_deep_oversold_audit.csv", index=False, encoding="utf-8-sig")

    subtype_summary = (
        rows.groupby("entry_subtype")
        .agg(
            trades=("buy_date", "count"),
            win_rate=("is_winner", "mean"),
            avg_return=("return_pct", "mean"),
            avg_mfe=("mfe_pct", "mean"),
            avg_mae=("mae_pct", "mean"),
            fast_failures=("is_fast_failure", "sum"),
            shallow_like=("is_shallow_like", "sum"),
        )
        .reset_index()
        .sort_values(["avg_return", "trades"], ascending=[True, False])
    )
    subtype_summary.to_csv(base_dir / "dragon_deep_oversold_subtype_summary.csv", index=False, encoding="utf-8-sig")

    lines = [
        "# Dragon Deep Oversold Audit",
        "",
        f"- audited trades: `{len(rows)}`",
        f"- all buy dates aligned with workbook: `{bool(rows['buy_aligned_with_workbook'].all())}`",
        f"- all sell dates aligned with workbook: `{bool(rows['sell_aligned_with_workbook'].all())}`",
        f"- winners: `{int(rows['is_winner'].sum())}` / `{len(rows)}`",
        f"- fast failures (holding <= 6d and negative return): `{int(rows['is_fast_failure'].sum())}`",
        "",
        "## Entry Subtype Summary",
    ]
    for _, row in subtype_summary.iterrows():
        lines.append(
            f"- `{row['entry_subtype']}`: trades `{int(row['trades'])}`, win_rate `{float(row['win_rate']):.2%}`, "
            f"avg_return `{float(row['avg_return']):.2%}`, avg_mfe `{float(row['avg_mfe']):.2%}`, "
            f"avg_mae `{float(row['avg_mae']):.2%}`, fast_failures `{int(row['fast_failures'])}`"
        )

    lines.extend(["", "## Candidate Pressure Points"])
    for _, row in rows.sort_values(["return_pct", "holding_days"]).head(8).iterrows():
        lines.append(
            f"- `{row['buy_date']} -> {row['sell_date']}`: `{row['entry_subtype']}` / `{row['quality_bucket']}` | "
            f"buy a1 `{float(row['buy_a1']):.4f}` b1 `{float(row['buy_b1']):.4f}` c1 `{float(row['buy_c1']):.2f}` | "
            f"return `{float(row['return_pct']):.2%}`"
        )

    lines.extend(
        [
            "",
            "## Quant Judgment",
            "- This rule family cannot be bluntly removed because every current deep-oversold trade is workbook-aligned.",
            "- The weakest local pattern is not the deepest capitulation bucket; it is the shallow or positive-B1 rebound subset.",
            "- Any redesign should therefore prefer subtype gating or delayed confirmation for shallow rebounds rather than tighter global oversold thresholds.",
        ]
    )

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


if __name__ == "__main__":
    main()