tests/test_44factor_gate_variant_report.py

import json
from types import SimpleNamespace

import numpy as np
import pandas as pd

from scripts.run_44factor_gate_variant_report import (
    _bootstrap_buy_delta_ci,
    _checks,
    _load_locked_finalists,
    apply_factor_transform_profile,
    build_factor_set,
    build_variants,
    render_html,
    split_events_by_date,
)
from scripts.search_multi_factor_weight_config import MultiFactorConfig


def _args(**overrides):
    values = {
        "min_buy_precision_delta_pp": 3.0,
        "research_min_sell_precision_delta_pp": 0.0,
        "strict_min_sell_precision_delta_pp": 2.0,
        "min_direction_accuracy_delta_pp": 0.0,
        "min_signal_ratio": 0.70,
        "min_buy_signal_ratio": 0.70,
    }
    values.update(overrides)
    return SimpleNamespace(**values)


def test_split_events_by_date_keeps_dates_isolated_and_purges_later_slices():
    dates = pd.date_range("2026-01-01", periods=10, freq="D").strftime("%Y-%m-%d")
    events = pd.DataFrame([{"date": date, "stock": stock} for date in dates for stock in ["1101", "2330"]])

    frames, manifest = split_events_by_date(events, ratios=(0.6, 0.2, 0.2), purge_days=1)

    discovery = set(frames["discovery"]["date"])
    calibration = set(frames["calibration"]["date"])
    confirmation = set(frames["confirmation"]["date"])
    assert not discovery & calibration
    assert not discovery & confirmation
    assert not calibration & confirmation
    assert manifest["slices"]["discovery"]["end"] == "2026-01-06"
    assert manifest["slices"]["calibration"]["start"] == "2026-01-08"
    assert manifest["slices"]["confirmation"]["start"] == "2026-01-10"


def test_checks_keep_research_and_strict_sell_floors_separate():
    config = MultiFactorConfig(weights={f"factor_{index}": 0.01 for index in range(44)})
    checks = _checks(
        args=_args(),
        config=config,
        deltas={"buy_precision_delta_pp": 3.5, "sell_precision_delta_pp": 0.5, "direction_accuracy_delta_pp": 0.1},
        signal_ratio=0.9,
        buy_signal_ratio=0.8,
        ci={"available": True, "lower_95_pp": 0.2},
        covered_stock_count=1000,
        expected_factor_count=44,
    )

    assert all(checks["research_screen"].values())
    assert checks["strict_promotion_gate"]["sell_precision_delta"] is False


def test_checks_reject_buy_signal_collapse_even_when_overall_ratio_passes():
    config = MultiFactorConfig(weights={f"factor_{index}": 0.01 for index in range(44)})
    checks = _checks(
        args=_args(),
        config=config,
        deltas={"buy_precision_delta_pp": 4.0, "sell_precision_delta_pp": 3.0, "direction_accuracy_delta_pp": 0.1},
        signal_ratio=0.95,
        buy_signal_ratio=0.2,
        ci={"available": True, "lower_95_pp": 1.0},
        covered_stock_count=1000,
        expected_factor_count=44,
    )

    assert checks["research_screen"]["signal_ratio"] is True
    assert checks["research_screen"]["buy_signal_ratio"] is False


def test_bootstrap_buy_delta_ci_is_date_blocked_and_deterministic():
    events = pd.DataFrame(
        {
            "date": ["2026-01-01"] * 2 + ["2026-01-02"] * 2 + ["2026-01-03"] * 2,
            "y_true": [1, -1, 1, -1, -1, 1],
        }
    )
    baseline = np.array([1, 1, 1, 1, 1, 1])
    candidate = np.array([1, 0, 1, 0, 0, 1])

    first = _bootstrap_buy_delta_ci(events, baseline, candidate, iterations=50, random_seed=7)
    second = _bootstrap_buy_delta_ci(events, baseline, candidate, iterations=50, random_seed=7)

    assert first == second
    assert first["available"] is True
    assert first["iterations"] == 50


def test_load_locked_finalists_replays_saved_config_without_search(tmp_path):
    config = {"weights": {f"factor_{index}": 0.01 for index in range(44)}, "hold_bias": 0.02}
    path = tmp_path / "finalists.json"
    path.write_text(json.dumps({"locked_finalists": [{"name": "winner", "variant_id": "baseline_ungated", "config": config}]}))
    variants = {item.variant_id: item for item in build_variants([-0.95], recent_break_sessions=5)}

    finalists = _load_locked_finalists(path, variants)

    assert finalists[0][0] == "winner"
    assert finalists[0][1].hold_bias == 0.02
    assert finalists[0][2].variant_id == "baseline_ungated"


def test_factor_transform_profile_is_candidate_only_and_reports_changes():
    events = pd.DataFrame(
        {
            "factor__oldwang_guard": [0.5, -0.25],
            "factor__big_player_buy": [0.4, 0.2],
            "factor__volume_surge": [0.1, -0.1],
        }
    )

    transformed, manifest = apply_factor_transform_profile(events, "flip_inverted_zero_low_noisy_v1")

    assert events["factor__oldwang_guard"].tolist() == [0.5, -0.25]
    assert transformed["factor__oldwang_guard"].tolist() == [-0.5, 0.25]
    assert transformed["factor__big_player_buy"].tolist() == [0.0, 0.0]
    assert transformed["factor__volume_surge"].tolist() == [0.1, -0.1]
    assert "oldwang_guard" in manifest["flipped"]
    assert "big_player_buy" in manifest["zeroed"]


def test_split_problem_factor_set_replaces_aggregate_factors_with_nonzero_research_components():
    seed = {
        "oldwang_guard": 0.0,
        "candle_shadow": 0.2,
        "lower_shadow": 0.01,
        "bt_yang_gao_pao": 0.03,
        "bt_bull_engulf": 0.04,
        "volume_surge": 0.01,
    }

    weights, manifest = build_factor_set(seed, "split_problem_v1")

    assert "oldwang_guard" not in weights
    assert "candle_shadow" not in weights
    assert "oldwang_guard_support" in weights
    assert "oldwang_guard_break_risk" in weights
    assert "candle_lower_support" in weights
    assert "candle_upper_pressure" in weights
    assert "bt_yang_gao_pao_risk" in weights
    assert "bt_bull_engulf_risk" in weights
    assert weights["oldwang_guard_support"] == 0.01
    assert all(value >= 0.01 for value in weights.values())
    assert manifest["factor_set"] == "split_problem_v1"


def test_regime_reversal_factor_set_replaces_inverted_candidates_without_production_defaults():
    seed = {
        "oldwang_trend": 0.02,
        "oldwang_guard": 0.0,
        "intraday_60k_volume_low_guard": 0.03,
        "macd_momentum": 0.04,
        "rsi_trend": 0.05,
        "bt_yi_yin_ya_breakout": 0.06,
        "tej_macro_risk_proxy": 0.07,
        "volume_surge": 0.01,
    }

    weights, manifest = build_factor_set(seed, "regime_reversal_v1")

    for replaced in [
        "oldwang_trend",
        "oldwang_guard",
        "intraday_60k_volume_low_guard",
        "macd_momentum",
        "rsi_trend",
        "bt_yi_yin_ya_breakout",
        "tej_macro_risk_proxy",
    ]:
        assert replaced not in weights
    assert "oldwang_trend_reversal_pressure" in weights
    assert "oldwang_guard_lowbase_support" in weights
    assert "oldwang_guard_highbase_risk" in weights
    assert "intraday_60k_fresh_low_guard" in weights
    assert "intraday_60k_stale_guard_risk" in weights
    assert "macd_reversal_pressure" in weights
    assert "rsi_trend_reversal" in weights
    assert "bt_yi_yin_ya_breakout_risk" in weights
    assert "tej_macro_risk_contrarian" in weights
    assert weights["oldwang_guard_lowbase_support"] == 0.01
    assert all(value >= 0.01 for value in weights.values())
    assert manifest["factor_set"] == "regime_reversal_v1"


def test_render_html_includes_filterable_complete_config_table(tmp_path):
    output = tmp_path / "report.html"
    payload = {
        "strict_promotion_gate": {"proposed_golden_config": None},
        "research_screen": {"passing_count": 0},
        "covered_stocks": ["1101"],
        "comparison_matrix": [
            {
                "name": "candidate_00001",
                "variant_id": "baseline_ungated",
                "deltas": {"buy_precision_delta_pp": 1.0, "sell_precision_delta_pp": 0.0, "direction_accuracy_delta_pp": 0.0},
                "signal_ratio": 1.0,
                "buy_signal_ratio": 1.0,
                "research_passed": False,
                "config": {"weights": {"factor": 0.01}, "hold_bias": 0.0},
            }
        ],
        "locked_finalists": [],
        "confirmation_results": [],
        "run_manifest": {},
        "failure_summary": {},
        "warnings": [],
    }

    render_html(payload, output)

    text = output.read_text()
    assert "All calibration combinations" in text
    assert "filterRows()" in text
    assert "candidate_00001" in text