scripts/analyze_44factor_variant_failures.py

#!/usr/bin/env python3
"""Analyze failed 44-factor gate/weight searches to choose the next factor work."""

from __future__ import annotations

import argparse
import html
import json
import sys
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

from scripts.optimize_factor_weights import compute_event_metrics, metric_deltas  # noqa: E402
from scripts.search_multi_factor_weight_config import MultiFactorConfig, apply_multi_factor_config  # noqa: E402


def _root_path(value: str | Path) -> Path:
    path = Path(value)
    return path if path.is_absolute() else ROOT / path


def _round(value: float) -> float:
    if pd.isna(value) or np.isinf(value):
        return 0.0
    return round(float(value), 6)


def _confirmation_events(events: pd.DataFrame, report: dict[str, Any]) -> pd.DataFrame:
    slices = (report.get("date_split") or {}).get("slices") or {}
    confirmation = slices.get("confirmation") or {}
    if not confirmation.get("start") or not confirmation.get("end"):
        return events.copy()
    dates = pd.to_datetime(events["date"], errors="coerce")
    start = pd.to_datetime(confirmation["start"])
    end = pd.to_datetime(confirmation["end"])
    return events.loc[(dates >= start) & (dates <= end)].reset_index(drop=True)


def _slice_events(events: pd.DataFrame, report: dict[str, Any]) -> dict[str, pd.DataFrame]:
    slices = (report.get("date_split") or {}).get("slices") or {}
    parsed = pd.to_datetime(events["date"], errors="coerce")
    output: dict[str, pd.DataFrame] = {}
    for name in ("discovery", "calibration", "confirmation"):
        info = slices.get(name) or {}
        if not info.get("start") or not info.get("end"):
            continue
        start = pd.to_datetime(info["start"])
        end = pd.to_datetime(info["end"])
        output[name] = events.loc[(parsed >= start) & (parsed <= end)].reset_index(drop=True)
    if not output:
        output["confirmation"] = events.copy()
    return output


def _variant_summary(report: dict[str, Any]) -> list[dict[str, Any]]:
    rows = report.get("comparison_matrix") or []
    grouped: dict[str, list[dict[str, Any]]] = {}
    for row in rows:
        grouped.setdefault(str(row.get("variant_id")), []).append(row)
    summary = []
    for variant, items in sorted(grouped.items()):
        ranked = sorted(
            items,
            key=lambda row: (
                row.get("research_passed", False),
                row["deltas"]["buy_precision_delta_pp"],
                row["deltas"]["direction_accuracy_delta_pp"],
                row["deltas"]["sell_precision_delta_pp"],
            ),
            reverse=True,
        )
        best = ranked[0]
        summary.append(
            {
                "variant_id": variant,
                "tested": len(items),
                "research_passed": sum(bool(row.get("research_passed")) for row in items),
                "best_name": best["name"],
                "best_calibration_deltas": best["deltas"],
                "best_signal_ratio": best["signal_ratio"],
                "best_buy_signal_ratio": best["buy_signal_ratio"],
            }
        )
    return summary


def _factor_alignment(events: pd.DataFrame) -> list[dict[str, Any]]:
    y = pd.to_numeric(events["y_true"], errors="coerce").fillna(0).to_numpy(dtype=float)
    buy_base_rate = float((y == 1).mean() * 100.0) if len(y) else 0.0
    sell_base_rate = float((y == -1).mean() * 100.0) if len(y) else 0.0
    rows = []
    for column in sorted(col for col in events.columns if col.startswith("factor__")):
        values = pd.to_numeric(events[column], errors="coerce").fillna(0.0)
        arr = values.to_numpy(dtype=float)
        nonzero = arr != 0.0
        coverage = float(nonzero.mean()) if len(arr) else 0.0
        corr = pd.Series(arr).corr(pd.Series(y), method="spearman") if len(arr) > 2 else 0.0
        pos = values > values.quantile(0.9)
        neg = values < values.quantile(0.1)
        buy_lift = float((y[pos.to_numpy()] == 1).mean() * 100.0 - buy_base_rate) if pos.any() else 0.0
        sell_lift = float((y[neg.to_numpy()] == -1).mean() * 100.0 - sell_base_rate) if neg.any() else 0.0
        rows.append(
            {
                "factor": column.removeprefix("factor__"),
                "nonzero_coverage": _round(coverage),
                "spearman_to_label": _round(corr),
                "buy_lift_top_decile_pp": _round(buy_lift),
                "sell_lift_bottom_decile_pp": _round(sell_lift),
                "risk_flag": _factor_risk_flag(corr, buy_lift, sell_lift, coverage),
            }
        )
    return sorted(rows, key=lambda row: (row["risk_flag"] != "ok", abs(row["spearman_to_label"])), reverse=True)


def _factor_slice_stats(events: pd.DataFrame, factor: str) -> dict[str, Any]:
    column = f"factor__{factor}"
    if events.empty or column not in events:
        return {
            "event_count": 0,
            "coverage": 0.0,
            "spearman_to_label": 0.0,
            "positive_buy_lift_pp": 0.0,
            "negative_sell_lift_pp": 0.0,
            "positive_count": 0,
            "negative_count": 0,
        }
    y = pd.to_numeric(events["y_true"], errors="coerce").fillna(0).to_numpy(dtype=float)
    values = pd.to_numeric(events[column], errors="coerce").fillna(0.0)
    arr = values.to_numpy(dtype=float)
    nonzero = arr != 0.0
    positive = arr > 0.0
    negative = arr < 0.0
    buy_base_rate = float((y == 1).mean() * 100.0) if len(y) else 0.0
    sell_base_rate = float((y == -1).mean() * 100.0) if len(y) else 0.0
    corr = pd.Series(arr).corr(pd.Series(y), method="spearman") if len(arr) > 2 else 0.0
    positive_buy_lift = float((y[positive] == 1).mean() * 100.0 - buy_base_rate) if positive.any() else 0.0
    negative_sell_lift = float((y[negative] == -1).mean() * 100.0 - sell_base_rate) if negative.any() else 0.0
    return {
        "event_count": int(len(events)),
        "coverage": _round(float(nonzero.mean()) if len(nonzero) else 0.0),
        "spearman_to_label": _round(corr),
        "positive_buy_lift_pp": _round(positive_buy_lift),
        "negative_sell_lift_pp": _round(negative_sell_lift),
        "positive_count": int(positive.sum()),
        "negative_count": int(negative.sum()),
    }


def _direction_bucket(stats: dict[str, Any]) -> str:
    if float(stats.get("coverage", 0.0)) < 0.01:
        return "low_coverage"
    corr = float(stats.get("spearman_to_label", 0.0))
    buy_lift = float(stats.get("positive_buy_lift_pp", 0.0))
    sell_lift = float(stats.get("negative_sell_lift_pp", 0.0))
    if corr >= 0.01 and buy_lift >= 0.5:
        return "buy_aligned"
    if corr <= -0.01 and sell_lift >= 0.5:
        return "sell_aligned"
    if corr <= -0.01 and buy_lift <= -0.5:
        return "possible_inverted_buy"
    if abs(corr) < 0.005 and abs(buy_lift) < 0.5 and abs(sell_lift) < 0.5:
        return "weak"
    return "mixed"


def _stability_flag(slice_stats: dict[str, dict[str, Any]]) -> str:
    calibration = _direction_bucket(slice_stats.get("calibration", {}))
    confirmation = _direction_bucket(slice_stats.get("confirmation", {}))
    discovery = _direction_bucket(slice_stats.get("discovery", {}))
    if confirmation == "low_coverage":
        return "low_coverage_confirmation"
    if calibration in {"buy_aligned", "sell_aligned"} and confirmation == calibration:
        return "stable_" + confirmation
    if calibration in {"buy_aligned", "sell_aligned"} and confirmation in {"weak", "mixed"}:
        return "calibration_only"
    if calibration in {"buy_aligned", "sell_aligned"} and confirmation in {"buy_aligned", "sell_aligned"} and confirmation != calibration:
        return "direction_flip"
    if confirmation == "possible_inverted_buy":
        return "confirmation_inverted"
    if discovery == calibration == confirmation == "weak":
        return "consistently_weak"
    return "unstable"


def _factor_slice_stability(events: pd.DataFrame, report: dict[str, Any]) -> list[dict[str, Any]]:
    slices = _slice_events(events, report)
    factors = sorted(column.removeprefix("factor__") for column in events.columns if column.startswith("factor__"))
    rows = []
    for factor in factors:
        slice_stats = {name: _factor_slice_stats(frame, factor) for name, frame in slices.items()}
        buckets = {name: _direction_bucket(stats) for name, stats in slice_stats.items()}
        rows.append(
            {
                "factor": factor,
                "stability_flag": _stability_flag(slice_stats),
                "slice_buckets": buckets,
                "slice_stats": slice_stats,
            }
        )
    return sorted(rows, key=lambda row: row["stability_flag"])


def _factor_risk_flag(corr: float, buy_lift: float, sell_lift: float, coverage: float) -> str:
    if coverage < 0.01:
        return "low_coverage"
    if corr < -0.02 and buy_lift < 0:
        return "possible_inverted_buy"
    if buy_lift < -1.0 and sell_lift < 0:
        return "noisy_against_both_sides"
    if abs(corr) < 0.005 and abs(buy_lift) < 0.5 and abs(sell_lift) < 0.5:
        return "weak"
    return "ok"


def _gate_cohorts(events: pd.DataFrame, report: dict[str, Any]) -> dict[str, Any]:
    seed_config = report.get("locked_finalists", [{}])[0].get("config") or None
    baseline = report.get("baseline", {}).get("metrics") or {}
    rows = {}
    if seed_config:
        pred = apply_multi_factor_config(events, MultiFactorConfig(**seed_config))
        rows["displayed_candidate_without_gate"] = compute_event_metrics(events, pred)
        rows["displayed_candidate_delta_vs_report_baseline"] = metric_deltas(baseline, rows["displayed_candidate_without_gate"])
    y = events["y_true"].to_numpy(dtype=int)
    cohorts = {
        "guard_available": pd.to_numeric(events.get("factor__intraday_60k_volume_low_guard", 0), errors="coerce").fillna(0.0).to_numpy() > 0,
        "break_active": pd.to_numeric(events.get("factor__intraday_60k_volume_low_break", 0), errors="coerce").fillna(0.0).to_numpy() < 0,
        "recent_break_age_le_5": (
            (pd.to_numeric(events.get("factor__intraday_60k_volume_low_break", 0), errors="coerce").fillna(0.0).to_numpy() < 0)
            & (pd.to_numeric(events.get("intraday_60k_volume_low_age_sessions", np.nan), errors="coerce").fillna(999).to_numpy() <= 5)
        ),
    }
    for name, mask in cohorts.items():
        if not mask.any():
            rows[name] = {"count": 0}
            continue
        rows[name] = {
            "count": int(mask.sum()),
            "buy_label_rate": _round(float((y[mask] == 1).mean() * 100.0)),
            "sell_label_rate": _round(float((y[mask] == -1).mean() * 100.0)),
            "hold_label_rate": _round(float((y[mask] == 0).mean() * 100.0)),
        }
    return rows


def analyze(args: argparse.Namespace) -> dict[str, Any]:
    report_path = _root_path(args.report)
    events_path = _root_path(args.events_cache)
    report = json.loads(report_path.read_text(encoding="utf-8"))
    events = pd.read_pickle(events_path)
    events["date"] = pd.to_datetime(events["date"], errors="coerce").dt.date.astype(str)
    confirmation = _confirmation_events(events, report)
    alignment = _factor_alignment(confirmation)
    slice_stability = _factor_slice_stability(events, report)
    risks = {}
    for row in alignment:
        risks.setdefault(row["risk_flag"], []).append(row["factor"])
    interventions = _intervention_scenarios(confirmation, report, risks)
    return {
        "experiment": "44factor_variant_failure_audit",
        "source_report": str(report_path),
        "events_cache": str(events_path),
        "source_research_candidates": report.get("research_screen", {}).get("passing_count"),
        "source_strict_candidates": report.get("strict_promotion_gate", {}).get("passing_count"),
        "event_counts": {
            "all": int(len(events)),
            "confirmation": int(len(confirmation)),
            "stocks": int(confirmation["stock"].astype(str).nunique()) if not confirmation.empty else 0,
        },
        "top_confirmation_result": (report.get("confirmation_results") or [{}])[0],
        "variant_summary": _variant_summary(report),
        "factor_alignment": alignment,
        "factor_slice_stability": slice_stability,
        "risk_groups": {name: sorted(values) for name, values in sorted(risks.items())},
        "gate_cohorts": _gate_cohorts(confirmation, report),
        "intervention_scenarios": interventions,
        "next_recommendations": _recommendations(alignment, report, interventions, slice_stability),
    }


def _intervention_scenarios(events: pd.DataFrame, report: dict[str, Any], risks: dict[str, list[str]]) -> list[dict[str, Any]]:
    baseline_metrics = report.get("baseline", {}).get("metrics") or {}
    comparison = report.get("comparison_matrix") or []
    candidates = [row for row in comparison if row.get("variant_id") == "baseline_ungated"]
    if not candidates:
        return []
    best = max(
        candidates,
        key=lambda row: (
            row["deltas"]["buy_precision_delta_pp"],
            row["deltas"]["direction_accuracy_delta_pp"],
            row["deltas"]["sell_precision_delta_pp"],
        ),
    )
    config = MultiFactorConfig(**best["config"])
    scenarios: list[dict[str, Any]] = []

    def add(name: str, frame: pd.DataFrame) -> None:
        pred = apply_multi_factor_config(frame, config)
        metrics = compute_event_metrics(frame, pred)
        scenarios.append(
            {
                "name": name,
                "base_config": best["name"],
                "metrics": metrics,
                "deltas": metric_deltas(baseline_metrics, metrics),
            }
        )

    add("best_calibration_baseline_ungated", events)
    inverted = risks.get("possible_inverted_buy", [])
    low_or_noisy = risks.get("low_coverage", []) + risks.get("noisy_against_both_sides", [])
    flipped = events.copy()
    for factor in inverted:
        column = f"factor__{factor}"
        if column in flipped:
            flipped[column] = -pd.to_numeric(flipped[column], errors="coerce").fillna(0.0)
    add("flip_possible_inverted_factor_values", flipped)
    zeroed = events.copy()
    for factor in low_or_noisy:
        column = f"factor__{factor}"
        if column in zeroed:
            zeroed[column] = 0.0
    add("zero_low_coverage_and_noisy_factors", zeroed)
    combined = flipped.copy()
    for factor in low_or_noisy:
        column = f"factor__{factor}"
        if column in combined:
            combined[column] = 0.0
    add("flip_inverted_zero_low_noisy", combined)
    return scenarios


def _recommendations(
    alignment: list[dict[str, Any]],
    report: dict[str, Any],
    interventions: list[dict[str, Any]],
    slice_stability: list[dict[str, Any]],
) -> list[str]:
    recommendations = []
    if not report.get("research_screen", {}).get("passing_count"):
        recommendations.append("Do not run 1000+ confirmation yet; refreshed 217-stock research screen has zero passing candidates.")
    inverted = [row["factor"] for row in alignment if row["risk_flag"] == "possible_inverted_buy"]
    weak = [row["factor"] for row in alignment if row["risk_flag"] == "weak"]
    low = [row["factor"] for row in alignment if row["risk_flag"] == "low_coverage"]
    if inverted:
        recommendations.append("Review factor sign/definition for possible inverted BUY behavior: " + ", ".join(inverted[:12]))
    if weak:
        recommendations.append("Down-rank or split weak low-signal factors before another weight search: " + ", ".join(weak[:12]))
    if low:
        recommendations.append("Keep low-coverage factors as shadow/data-only until explicit data coverage improves: " + ", ".join(low[:12]))
    calibration_only = [row["factor"] for row in slice_stability if row["stability_flag"] == "calibration_only"]
    direction_flip = [row["factor"] for row in slice_stability if row["stability_flag"] == "direction_flip"]
    stable_buy = [row["factor"] for row in slice_stability if row["stability_flag"] == "stable_buy_aligned"]
    stable_sell = [row["factor"] for row in slice_stability if row["stability_flag"] == "stable_sell_aligned"]
    if calibration_only:
        recommendations.append("Treat calibration-only factors as overfit-prone in the next search: " + ", ".join(calibration_only[:12]))
    if direction_flip:
        recommendations.append("Do not optimize direction-flip factors with a single positive weight until definitions are split or gated: " + ", ".join(direction_flip[:12]))
    confirmation_inverted = [row["factor"] for row in slice_stability if row["stability_flag"] == "confirmation_inverted"]
    if confirmation_inverted:
        recommendations.append(
            "Confirmation-inverted factors need a regime-aware definition check before weight tuning: "
            + ", ".join(confirmation_inverted[:12])
        )
    if stable_buy or stable_sell:
        recommendations.append(
            "Use stable slice-aligned factors as the next seed cluster; BUY: "
            + ", ".join(stable_buy[:8])
            + "; SELL: "
            + ", ".join(stable_sell[:8])
        )
    best_intervention = max(
        interventions,
        key=lambda row: row["deltas"].get("buy_precision_delta_pp", -999.0),
        default=None,
    )
    if best_intervention and best_intervention["deltas"].get("buy_precision_delta_pp", 0.0) > 1.0:
        recommendations.append(
            "Prioritize a sign/definition fix experiment; "
            f"{best_intervention['name']} reached BUY {best_intervention['deltas']['buy_precision_delta_pp']}pp "
            f"and SELL {best_intervention['deltas']['sell_precision_delta_pp']}pp on confirmation."
        )
    recommendations.append("Next optimizer should test factor removal/split diagnostics, not only non-zero positive weights.")
    return recommendations


def render_html(payload: dict[str, Any], output: Path) -> None:
    factor_rows = "".join(
        "<tr>"
        f"<td>{html.escape(row['factor'])}</td><td>{row['risk_flag']}</td>"
        f"<td>{row['nonzero_coverage']}</td><td>{row['spearman_to_label']}</td>"
        f"<td>{row['buy_lift_top_decile_pp']}</td><td>{row['sell_lift_bottom_decile_pp']}</td>"
        "</tr>"
        for row in payload["factor_alignment"]
    )
    variant_rows = "".join(
        "<tr>"
        f"<td>{html.escape(row['variant_id'])}</td><td>{row['tested']}</td><td>{row['research_passed']}</td>"
        f"<td>{row['best_calibration_deltas'].get('buy_precision_delta_pp')}</td>"
        f"<td>{row['best_calibration_deltas'].get('sell_precision_delta_pp')}</td>"
        f"<td>{row['best_calibration_deltas'].get('direction_accuracy_delta_pp')}</td>"
        "</tr>"
        for row in payload["variant_summary"]
    )
    stability_rows = "".join(
        "<tr>"
        f"<td>{html.escape(row['factor'])}</td><td>{html.escape(row['stability_flag'])}</td>"
        f"<td>{html.escape(row['slice_buckets'].get('discovery', ''))}</td>"
        f"<td>{html.escape(row['slice_buckets'].get('calibration', ''))}</td>"
        f"<td>{html.escape(row['slice_buckets'].get('confirmation', ''))}</td>"
        f"<td>{row['slice_stats'].get('confirmation', {}).get('coverage', 0)}</td>"
        f"<td>{row['slice_stats'].get('confirmation', {}).get('spearman_to_label', 0)}</td>"
        f"<td>{row['slice_stats'].get('confirmation', {}).get('positive_buy_lift_pp', 0)}</td>"
        f"<td>{row['slice_stats'].get('confirmation', {}).get('negative_sell_lift_pp', 0)}</td>"
        "</tr>"
        for row in payload["factor_slice_stability"]
    )
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(
        f"""<!doctype html><html lang="zh-Hant"><head><meta charset="utf-8"><title>44-Factor Failure Audit</title>
<style>body{{font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",sans-serif;margin:24px;color:#17202a}}table{{border-collapse:collapse;width:100%;font-size:13px;margin:12px 0}}th,td{{border-bottom:1px solid #e5e7eb;padding:8px;text-align:left}}th{{background:#f8fafc}}code,pre{{white-space:pre-wrap;word-break:break-word}}</style></head>
<body><h1>44-Factor Failure Audit</h1>
<h2>Summary</h2><pre>{html.escape(json.dumps({k: payload[k] for k in ['source_research_candidates','source_strict_candidates','event_counts','risk_groups','next_recommendations']}, ensure_ascii=False, indent=2))}</pre>
<h2>Variant Summary</h2><table><tr><th>Variant</th><th>Tested</th><th>Research passed</th><th>Best BUY pp</th><th>Best SELL pp</th><th>Best Direction pp</th></tr>{variant_rows}</table>
<h2>Gate Cohorts</h2><pre>{html.escape(json.dumps(payload['gate_cohorts'], ensure_ascii=False, indent=2))}</pre>
<h2>Intervention Scenarios</h2><pre>{html.escape(json.dumps(payload['intervention_scenarios'], ensure_ascii=False, indent=2))}</pre>
<h2>Factor Slice Stability</h2><table><tr><th>Factor</th><th>Stability</th><th>Discovery</th><th>Calibration</th><th>Confirmation</th><th>Confirmation coverage</th><th>Confirmation Spearman</th><th>Confirmation BUY lift</th><th>Confirmation SELL lift</th></tr>{stability_rows}</table>
<h2>Factor Alignment</h2><table><tr><th>Factor</th><th>Risk</th><th>Coverage</th><th>Spearman</th><th>BUY lift top decile</th><th>SELL lift bottom decile</th></tr>{factor_rows}</table>
</body></html>""",
        encoding="utf-8",
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--report", default="docs/validation_runs/factor44_gate_variant_3y_top220_refreshed.json")
    parser.add_argument("--events-cache", default="docs/validation_runs/factor44_gate_variant_3y_top220_refreshed_events.pkl")
    parser.add_argument("--output-prefix", default="docs/validation_runs/factor44_gate_variant_3y_top220_refreshed_failure_audit")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    payload = analyze(args)
    prefix = _root_path(args.output_prefix)
    json_path = prefix.with_suffix(".json")
    html_path = prefix.with_suffix(".html")
    json_path.parent.mkdir(parents=True, exist_ok=True)
    json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    render_html(payload, html_path)
    print("=== 44-Factor Failure Audit ===")
    print(f"Research candidates: {payload['source_research_candidates']}")
    print(f"Strict candidates: {payload['source_strict_candidates']}")
    print(f"Confirmation events: {payload['event_counts']['confirmation']}")
    print(f"JSON: {json_path}")
    print(f"HTML: {html_path}")
    return 0


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