scripts/confirm_fixed_sell_pattern_gate.py

#!/usr/bin/env python3
"""Replay a fixed SELL-pattern gate and fixed weights on selected stocks.

This is a confirmation harness, not a search harness. It avoids choosing among
gates or weights on the confirmation slice.
"""

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.evaluate_sell_pattern_regime_gates import (  # noqa: E402
    _apply_gate,
    _build_gate_specs,
    _prediction_rates,
)
from scripts.run_cached_daily_factor_overlay_report import DEFAULT_FACTORS, enrich_cached_events  # noqa: E402
from scripts.search_multi_factor_weight_config import (  # noqa: E402
    MultiFactorConfig,
    _now_iso,
    apply_multi_factor_config,
)
from scripts.optimize_factor_weights import compute_event_metrics, metric_deltas  # noqa: E402


DEFAULT_WEIGHTS = {
    "san_sheng_wu_nai_breakdown": 0.01,
    "downside_follow_through": 0.005,
    "weak_rebound_after_breakdown": 0.02,
    "high_volume_downside_follow_through": 0.05,
}


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


def _split_csv(value: str) -> list[str]:
    return [item.strip() for item in value.split(",") if item.strip()]


def _parse_weights(value: str) -> dict[str, float]:
    if not value.strip():
        return dict(DEFAULT_WEIGHTS)
    out: dict[str, float] = {}
    for item in value.split(","):
        if not item.strip():
            continue
        name, raw_weight = item.split("=", 1)
        out[name.strip()] = float(raw_weight.strip())
    return out


def _json_default(value: Any) -> Any:
    if isinstance(value, np.integer):
        return int(value)
    if isinstance(value, np.floating):
        return float(value)
    if isinstance(value, np.bool_):
        return bool(value)
    if hasattr(value, "isoformat"):
        return value.isoformat()
    raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable")


def _decision(deltas: dict[str, Any], *, min_signal_ratio: float, base_metrics: dict[str, Any], candidate_metrics: dict[str, Any]) -> dict[str, Any]:
    signal_ratio = candidate_metrics["signal_count"] / base_metrics["signal_count"] if base_metrics["signal_count"] else 0.0
    checks = {
        "accuracy_delta": deltas["accuracy_delta_pp"] >= 0.0,
        "direction_accuracy_delta": deltas["direction_accuracy_delta_pp"] >= 0.0,
        "buy_precision_delta": deltas["buy_precision_delta_pp"] >= 0.0,
        "sell_precision_delta": deltas["sell_precision_delta_pp"] >= 0.0,
        "signal_ratio": signal_ratio >= min_signal_ratio,
    }
    return {"passed": all(checks.values()), "checks": checks, "signal_ratio": round(signal_ratio, 4)}


def _select_stocks(events: pd.DataFrame, *, offset: int, limit: int) -> list[str]:
    stocks = sorted(events["stock"].astype(str).unique().tolist())
    if offset:
        stocks = stocks[offset:]
    if limit and limit > 0:
        stocks = stocks[:limit]
    return stocks


def confirm(args: argparse.Namespace) -> dict[str, Any]:
    factor_names = _split_csv(args.factors) or DEFAULT_FACTORS
    weights = _parse_weights(args.weights)
    missing_weights = [factor for factor in factor_names if factor not in weights]
    if missing_weights:
        raise ValueError(f"Missing weights for factors: {missing_weights}")

    events_path = _root_path(args.events_cache)
    all_events = pd.read_pickle(events_path)
    stocks = _select_stocks(all_events, offset=args.stock_offset, limit=args.limit)
    if not stocks:
        raise ValueError("No stocks selected for confirmation.")
    events = all_events[all_events["stock"].astype(str).isin(stocks)].reset_index(drop=True)

    enriched, covered_stocks = enrich_cached_events(
        events,
        factor_names=factor_names,
        months=args.months,
        workers=args.workers,
    )

    gate_specs = {spec.name: spec for spec in _build_gate_specs(enriched)}
    if args.gate not in gate_specs:
        raise ValueError(f"Unknown gate {args.gate!r}; available: {sorted(gate_specs)}")
    gate_spec = gate_specs[args.gate]
    gated = _apply_gate(enriched, factor_names, gate_spec.mask)
    cfg = MultiFactorConfig(weights={factor: float(weights[factor]) for factor in factor_names}, hold_bias=float(args.hold_bias))

    baseline_pred = gated["base_pred"].to_numpy(dtype=int)
    candidate_pred = apply_multi_factor_config(gated, cfg)
    baseline_metrics = compute_event_metrics(gated, baseline_pred)
    candidate_metrics = compute_event_metrics(gated, candidate_pred)
    deltas = metric_deltas(baseline_metrics, candidate_metrics)
    gate_active = int(gate_spec.mask.fillna(False).astype(bool).sum())
    decision = _decision(deltas, min_signal_ratio=args.min_signal_ratio, base_metrics=baseline_metrics, candidate_metrics=candidate_metrics)

    payload = {
        "experiment": "fixed_sell_pattern_gate_confirmation",
        "generated_at": _now_iso(),
        "research_only": True,
        "production_defaults_modified": False,
        "events_cache": str(events_path),
        "stock_offset": args.stock_offset,
        "stock_limit": args.limit,
        "stocks": stocks,
        "covered_stocks": covered_stocks,
        "factor_names": factor_names,
        "fixed_gate": {"name": gate_spec.name, "description": gate_spec.description},
        "fixed_config": {"weights": cfg.weights, "hold_bias": cfg.hold_bias},
        "validation_design": {
            "mode": "fixed_gate_fixed_weights_replay",
            "selection_bias_guard": "Gate and weights are fixed before replay on the selected stock slice.",
            "months": args.months,
        },
        "gate_coverage": {
            "active_events": gate_active,
            "active_rate": round(gate_active / len(gated) * 100.0, 4) if len(gated) else 0.0,
        },
        "baseline": {"metrics": baseline_metrics, "prediction_rates": _prediction_rates(baseline_metrics)},
        "candidate": {"metrics": candidate_metrics, "prediction_rates": _prediction_rates(candidate_metrics)},
        "deltas": deltas,
        "decision": decision,
    }

    output_json = _root_path(args.output_json)
    output_json.parent.mkdir(parents=True, exist_ok=True)
    output_json.write_text(json.dumps(payload, indent=2, ensure_ascii=False, default=_json_default))
    if args.output_html:
        render_html(payload, _root_path(args.output_html))
    return payload


def _fmt(value: Any, suffix: str = "") -> str:
    if isinstance(value, float):
        return f"{value:.4f}{suffix}"
    return f"{value}{suffix}"


def render_html(payload: dict[str, Any], path: Path) -> None:
    base = payload["baseline"]["metrics"]
    cand = payload["candidate"]["metrics"]
    deltas = payload["deltas"]
    base_rates = payload["baseline"]["prediction_rates"]
    cand_rates = payload["candidate"]["prediction_rates"]
    decision = payload["decision"]
    decision_html = '<span class="pass">PASS</span>' if decision["passed"] else '<span class="fail">FAIL</span>'
    parts = [
        "<!doctype html>",
        '<html lang="zh-Hant"><head><meta charset="utf-8">',
        "<title>Fixed SELL Pattern Gate Confirmation</title>",
        "<style>",
        'body{font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",sans-serif;margin:32px;color:#1f2933;line-height:1.5}',
        "h1,h2{color:#102a43}table{border-collapse:collapse;width:100%;margin:16px 0 24px}",
        "th,td{border:1px solid #cbd2d9;padding:8px 10px;text-align:right}th:first-child,td:first-child{text-align:left}th{background:#f0f4f8}",
        ".pass{color:#0b7285;font-weight:700}.fail{color:#b42318;font-weight:700}code{background:#f0f4f8;padding:2px 4px;border-radius:4px}",
        "</style></head><body>",
        "<h1>Fixed SELL Pattern Gate Confirmation</h1>",
        f"<p><strong>Generated:</strong> {html.escape(payload['generated_at'])}</p>",
        f"<p><strong>Decision:</strong> {decision_html}</p>",
        f"<p><strong>Gate:</strong> <code>{html.escape(payload['fixed_gate']['name'])}</code> - {html.escape(payload['fixed_gate']['description'])}</p>",
        f"<p><strong>Stock slice:</strong> offset {payload['stock_offset']}, limit {payload['stock_limit']}, covered {len(payload['covered_stocks'])}</p>",
        f"<p><strong>Gate coverage:</strong> {payload['gate_coverage']['active_events']} events / {_fmt(payload['gate_coverage']['active_rate'], '%')}</p>",
        "<h2>Metrics</h2>",
        "<table><thead><tr><th>Scope</th><th>Accuracy</th><th>Direction Acc</th><th>BUY Precision</th><th>SELL Precision</th><th>Coverage</th><th>Signals</th></tr></thead><tbody>",
        f"<tr><td>Baseline</td><td>{_fmt(base['accuracy'], '%')}</td><td>{_fmt(base['direction_accuracy'], '%')}</td><td>{_fmt(base['buy_precision'], '%')}</td><td>{_fmt(base['sell_precision'], '%')}</td><td>{_fmt(base['coverage'] * 100.0, '%')}</td><td>{base['signal_count']}</td></tr>",
        f"<tr><td>Candidate</td><td>{_fmt(cand['accuracy'], '%')}</td><td>{_fmt(cand['direction_accuracy'], '%')}</td><td>{_fmt(cand['buy_precision'], '%')}</td><td>{_fmt(cand['sell_precision'], '%')}</td><td>{_fmt(cand['coverage'] * 100.0, '%')}</td><td>{cand['signal_count']}</td></tr>",
        f"<tr><td>Delta</td><td>{_fmt(deltas['accuracy_delta_pp'], 'pp')}</td><td>{_fmt(deltas['direction_accuracy_delta_pp'], 'pp')}</td><td>{_fmt(deltas['buy_precision_delta_pp'], 'pp')}</td><td>{_fmt(deltas['sell_precision_delta_pp'], 'pp')}</td><td>{_fmt(deltas['coverage_delta'] * 100.0, 'pp')}</td><td>{deltas['signal_count_delta']}</td></tr>",
        "</tbody></table>",
        "<h2>BUY / HOLD / SELL Prediction Rate</h2>",
        "<table><thead><tr><th>Scope</th><th>BUY Count</th><th>HOLD Count</th><th>SELL Count</th><th>BUY Rate</th><th>HOLD Rate</th><th>SELL Rate</th></tr></thead><tbody>",
        f"<tr><td>Baseline</td><td>{base_rates['buy_count']}</td><td>{base_rates['hold_count']}</td><td>{base_rates['sell_count']}</td><td>{_fmt(base_rates['buy_rate'], '%')}</td><td>{_fmt(base_rates['hold_rate'], '%')}</td><td>{_fmt(base_rates['sell_rate'], '%')}</td></tr>",
        f"<tr><td>Candidate</td><td>{cand_rates['buy_count']}</td><td>{cand_rates['hold_count']}</td><td>{cand_rates['sell_count']}</td><td>{_fmt(cand_rates['buy_rate'], '%')}</td><td>{_fmt(cand_rates['hold_rate'], '%')}</td><td>{_fmt(cand_rates['sell_rate'], '%')}</td></tr>",
        "</tbody></table>",
        "<h2>Fixed Weights</h2>",
        "<table><thead><tr><th>Factor</th><th>Weight</th></tr></thead><tbody>",
    ]
    for factor, weight in payload["fixed_config"]["weights"].items():
        parts.append(f"<tr><td><code>{html.escape(factor)}</code></td><td>{_fmt(weight)}</td></tr>")
    parts.extend(["</tbody></table>", "</body></html>"])
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(parts))


def main() -> int:
    parser = argparse.ArgumentParser(description="Replay a fixed SELL-pattern gate and fixed weights on a selected stock slice.")
    parser.add_argument("--events-cache", default="docs/validation_runs/factor44_split_problem_v1_buy_factor_interaction_adaptive_macro_1000_events.pkl")
    parser.add_argument("--factors", default=",".join(DEFAULT_FACTORS))
    parser.add_argument("--weights", default=",".join(f"{factor}={weight}" for factor, weight in DEFAULT_WEIGHTS.items()))
    parser.add_argument("--gate", default="high_volume_upper_shadow_sell")
    parser.add_argument("--stock-offset", type=int, default=1000)
    parser.add_argument("--limit", type=int, default=0, help="0 means all stocks after offset.")
    parser.add_argument("--months", type=int, default=36)
    parser.add_argument("--workers", type=int, default=12)
    parser.add_argument("--hold-bias", type=float, default=0.0)
    parser.add_argument("--min-signal-ratio", type=float, default=0.70)
    parser.add_argument("--output-json", default="docs/validation_runs/fixed_sell_pattern_gate_heldout_20260611.json")
    parser.add_argument("--output-html", default="docs/validation_runs/fixed_sell_pattern_gate_heldout_20260611.html")
    args = parser.parse_args()
    payload = confirm(args)
    print(f"Stocks selected: {len(payload['stocks'])}")
    print(f"Stocks covered: {len(payload['covered_stocks'])}")
    print(f"Decision: {'PASS' if payload['decision']['passed'] else 'FAIL'}")
    print(f"Baseline metrics: {payload['baseline']['metrics']}")
    print(f"Candidate metrics: {payload['candidate']['metrics']}")
    print(f"Deltas: {payload['deltas']}")
    print(f"Saved JSON -> {_root_path(args.output_json)}")
    if args.output_html:
        print(f"Saved HTML -> {_root_path(args.output_html)}")
    return 0


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