scripts/check_stock_predictor_harness_rules.py

#!/usr/bin/env python3
"""Hard-rule checker for Stock Predictor factor configs and survey reports.

This script is intentionally separate from historical validation.  It catches
configuration and harness-contract violations before a candidate is treated as
production-adjacent.
"""

from __future__ import annotations

import argparse
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 models.multi_factor_overlay import build_runtime_signed_factors  # noqa: E402


DEFAULT_OVERLAY_CONFIG = ROOT / "config" / "multi_factor_overlay.json"
DEFAULT_COMBINE_SPLIT_REPORT = ROOT / "docs" / "validation_runs" / "factor_combine_split_strategy_survey_20260528.html"


class CheckResult:
    def __init__(self) -> None:
        self.failures: list[str] = []
        self.warnings: list[str] = []
        self.passes: list[str] = []

    def pass_(self, message: str) -> None:
        self.passes.append(message)

    def warn(self, message: str) -> None:
        self.warnings.append(message)

    def fail(self, message: str) -> None:
        self.failures.append(message)

    @property
    def ok(self) -> bool:
        return not self.failures


def _load_json(path: Path) -> dict[str, Any]:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise SystemExit(f"missing required file: {path}") from exc
    except json.JSONDecodeError as exc:
        raise SystemExit(f"invalid JSON in {path}: {exc}") from exc


def _weights(payload: dict[str, Any]) -> dict[str, float]:
    selected = payload.get("selected_config") or payload.get("config") or {}
    raw = selected.get("weights") or {}
    return {str(name): float(value) for name, value in raw.items()}


def _failed_gate_checks(payload: dict[str, Any]) -> list[str]:
    gate = payload.get("gate") or {}
    explicit = gate.get("failed_checks")
    if isinstance(explicit, list):
        return [str(item) for item in explicit]
    checks = gate.get("checks") or {}
    return [str(name) for name, passed in checks.items() if not bool(passed)]


def _delta(payload: dict[str, Any], name: str, default: float = 0.0) -> float:
    try:
        return float(((payload.get("validation") or {}).get("deltas") or {}).get(name, default))
    except Exception:
        return default


def _max_zero_allowed(payload: dict[str, Any], default: int) -> int:
    thresholds = payload.get("selection_thresholds") or {}
    try:
        return int(thresholds.get("max_zero_factor_weights", default))
    except Exception:
        return default


def _check_runtime_support(weights: dict[str, float], result: CheckResult) -> None:
    closes = pd.Series(np.linspace(80.0, 120.0, 180))
    df = pd.DataFrame(
        {
            "open": closes - 0.4,
            "high": closes + 1.2,
            "low": closes - 1.0,
            "close": closes,
            "volume": np.linspace(10000.0, 40000.0, len(closes)),
        }
    )
    features = pd.DataFrame(index=df.index)
    try:
        factors = build_runtime_signed_factors(features, df, list(weights))
    except Exception as exc:
        result.fail(f"runtime factor support failed: {exc}")
        return
    if list(factors.columns) != list(weights):
        result.fail("runtime factor output columns do not match config factor order")
        return
    result.pass_(f"runtime supports all {len(weights)} configured factors")


def _check_big_player_factor_contract(weights: dict[str, float], result: CheckResult) -> None:
    legacy = {
        "big_player_power_1m_4m",
        "big_player_power_gt_4m",
        "big_player_buy_ratio_1m_4m",
        "big_player_sell_ratio_1m_4m",
        "big_player_buy_ratio_gt_4m",
        "big_player_sell_ratio_gt_4m",
    }
    quantity = {"big_player_buy", "big_player_sell"}
    present_legacy = sorted(legacy & set(weights))
    present_quantity = sorted(quantity & set(weights))
    if present_legacy:
        result.fail(
            "current overlay config must use explicit big-player buy/sell quantity strength, "
            f"not legacy ratio/signed fields: {', '.join(present_legacy)}"
        )
    if not present_quantity:
        result.fail("current overlay config must include big_player_buy and big_player_sell quantity factors")
    elif set(present_quantity) != quantity:
        missing = sorted(quantity - set(present_quantity))
        result.fail(f"big-player quantity factors must be configured as a complete buy/sell pair set; missing: {', '.join(missing)}")
    else:
        result.pass_("big-player factors use complete explicit buy/sell quantity strength set")


def _check_engulf_factor_contract(weights: dict[str, float], result: CheckResult) -> None:
    if "bt_double_bull_engulf_one_bear" in weights:
        result.fail("current overlay config must use split engulf factors, not bt_double_bull_engulf_one_bear")
    split = {"bt_bull_engulf", "bt_bull_engulf_bear"}
    present = sorted(split & set(weights))
    if not present:
        result.fail("current overlay config must include split bull engulf factors")
    elif set(present) != split:
        missing = sorted(split - set(present))
        result.fail(f"bull engulf split factors must be configured together; missing: {', '.join(missing)}")
    else:
        result.pass_("bull engulf factors are split into bullish and bearish-risk signals")


def check_overlay_config(path: Path, *, max_zero_default: int) -> CheckResult:
    result = CheckResult()
    payload = _load_json(path)
    weights = _weights(payload)
    if not weights:
        result.fail("overlay config has no selected_config.weights")
        return result

    out_of_range = [name for name, value in weights.items() if value < 0.0 or value > 1.0]
    if out_of_range:
        result.fail(f"weights out of [0, 1] range: {', '.join(out_of_range)}")
    else:
        result.pass_("all weights are inside [0, 1]")

    zero_weights = [name for name, value in weights.items() if abs(value) <= 1e-12]
    max_zero = _max_zero_allowed(payload, max_zero_default)
    if len(zero_weights) > max_zero:
        result.fail(f"zero-weight count {len(zero_weights)} exceeds max {max_zero}: {', '.join(zero_weights)}")
    else:
        result.pass_(f"zero-weight count {len(zero_weights)} <= max {max_zero}")

    _check_big_player_factor_contract(weights, result)
    _check_engulf_factor_contract(weights, result)

    status = str(payload.get("integration_status") or "")
    strict_promoted = bool(payload.get("strict_promoted"))
    failed = _failed_gate_checks(payload)
    sell_delta = _delta(payload, "sell_precision_delta_pp")
    buy_delta = _delta(payload, "buy_precision_delta_pp")

    if status == "strict_golden" or strict_promoted:
        if failed:
            result.fail(f"strict golden config has failed checks: {', '.join(failed)}")
        if sell_delta < 0:
            result.fail(f"strict golden config has negative SELL precision delta: {sell_delta}pp")
        if buy_delta < 0:
            result.fail(f"strict golden config has negative BUY precision delta: {buy_delta}pp")
        if not payload.get("strict_source_has_golden_config"):
            result.fail("strict golden config does not cite a source golden config")
        if result.ok:
            result.pass_("strict golden promotion gates are internally consistent")
    else:
        if failed:
            result.pass_(f"failed gates are contained in non-strict config: {', '.join(failed)}")
        if sell_delta < 0 and status != "candidate_only":
            result.fail("negative SELL precision delta must remain candidate_only")
        if sell_delta < 0 and status == "candidate_only":
            note = str(((payload.get("runtime") or {}).get("note") or "")).lower()
            if "sell" not in note and "candidate" not in note:
                result.warn("candidate_only config has negative SELL precision but runtime note does not explain it")
            result.pass_("negative SELL precision delta is not promoted as strict golden")

    _check_runtime_support(weights, result)
    return result


def check_combine_split_report(path: Path) -> CheckResult:
    result = CheckResult()
    if not path.exists():
        result.warn(f"combine/split report not found: {path}")
        return result
    text = path.read_text(encoding="utf-8")
    required = [
        "Combine / Split Probability",
        "新增 Related Factor Survey",
        "Validation Plan",
        "Source Evidence",
        "SELL precision",
        "ADX trend regime gate",
        "exhaustion_cluster_confirmed",
    ]
    missing = [item for item in required if item not in text]
    if missing:
        result.fail(f"combine/split report missing required sections/terms: {', '.join(missing)}")
    else:
        result.pass_("combine/split report includes required sections and P0 recommendations")
    return result


def merge_results(results: list[CheckResult]) -> CheckResult:
    merged = CheckResult()
    for item in results:
        merged.failures.extend(item.failures)
        merged.warnings.extend(item.warnings)
        merged.passes.extend(item.passes)
    return merged


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Check Stock Predictor factor harness hard rules.")
    parser.add_argument("--overlay-config", default=str(DEFAULT_OVERLAY_CONFIG))
    parser.add_argument("--combine-split-report", default=str(DEFAULT_COMBINE_SPLIT_REPORT))
    parser.add_argument("--max-zero-factor-weights", type=int, default=5)
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    overlay = Path(args.overlay_config)
    if not overlay.is_absolute():
        overlay = ROOT / overlay
    report = Path(args.combine_split_report)
    if not report.is_absolute():
        report = ROOT / report

    result = merge_results(
        [
            check_overlay_config(overlay, max_zero_default=args.max_zero_factor_weights),
            check_combine_split_report(report),
        ]
    )

    print("Stock Predictor harness rule check")
    for message in result.passes:
        print(f"PASS: {message}")
    for message in result.warnings:
        print(f"WARN: {message}")
    for message in result.failures:
        print(f"FAIL: {message}")
    print("RESULT:", "PASS" if result.ok else "FAIL")
    return 0 if result.ok else 1


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