tests/test_multi_factor_overlay_integration.py

import argparse
import json
from pathlib import Path

import numpy as np
import pandas as pd

from models.multi_factor_overlay import apply_multi_factor_overlay, build_runtime_signed_factors
from scripts.integrate_multi_factor_config import build_export_payload, select_config


def _args(**overrides):
    values = {
        "source_json": "source.json",
        "policy": "buy_focused",
        "scope": "hotspot_short_term",
        "activate": True,
        "require_strict_activation": False,
        "min_accuracy_delta_pp": 0.0,
        "min_buy_precision_delta_pp": 0.0,
        "min_signal_ratio": 0.7,
        "max_buy_count_ratio": 1.3,
        "max_sell_precision_drop_pp": None,
    }
    values.update(overrides)
    return argparse.Namespace(**values)


def _candidate(name, *, acc, buy, sell, passed=False):
    return {
        "name": name,
        "passed": passed,
        "config": {"weights": {"oldwang_trend": 0.02, "upper_shadow": 0.05}, "hold_bias": 0.0},
        "gate": {"signal_ratio": 1.0, "buy_count_ratio": 1.0, "checks": {"all_factor_weights_nonzero": True}},
        "validation": {
            "metrics": {"accuracy": 42.0, "buy_precision": 41.0, "sell_precision": 40.0},
            "deltas": {
                "accuracy_delta_pp": acc,
                "buy_precision_delta_pp": buy,
                "sell_precision_delta_pp": sell,
                "direction_accuracy_delta_pp": 0.1,
            },
        },
        "tune": {"metrics": {}, "deltas": {}},
    }


def test_select_buy_focused_config_allows_non_strict_best_candidate():
    source = {
        "generated_at": "2026-05-23T00:00:00+00:00",
        "result": {
            "golden_config": None,
            "top_overall": [
                _candidate("balanced", acc=1.0, buy=0.1, sell=-0.2),
                _candidate("buy_best", acc=0.8, buy=0.9, sell=-2.0),
            ],
        },
    }

    selected = select_config(source, _args())
    payload = build_export_payload(source, selected, _args())

    assert selected["name"] == "buy_best"
    assert payload["active"] is True
    assert payload["integration_status"] == "candidate_only"
    assert payload["selected_config"]["weights"]["upper_shadow"] == 0.05


def test_strict_activation_blocks_non_passing_candidate():
    source = {"result": {"golden_config": None, "top_overall": [_candidate("candidate", acc=1.0, buy=0.5, sell=-2.0)]}}

    selected = select_config(source, _args())
    payload = build_export_payload(source, selected, _args(require_strict_activation=True))

    assert payload["active"] is False


def test_runtime_overlay_applies_signed_factor_weights_when_active():
    close = pd.Series(np.linspace(100, 110, 30))
    df = pd.DataFrame(
        {
            "open": close - 0.2,
            "high": close + 1.0,
            "low": close - 1.0,
            "close": close,
            "volume": 10000,
        }
    )
    features = pd.DataFrame(
        {
            "oldwang_triple_bull": [0.0] * 29 + [1.0],
            "oldwang_triple_bear": 0.0,
            "factor_upper_shadow_ratio": 0.0,
            "factor_lower_shadow_ratio": 0.0,
        }
    )
    config = {
        "active": True,
        "policy": "test",
        "selected_config": {"weights": {"oldwang_trend": 0.05}, "hold_bias": 0.0},
    }

    buy, sell, meta = apply_multi_factor_overlay(
        buy_prob=0.4,
        sell_prob=0.3,
        features=features,
        df=df,
        config=config,
        enabled=True,
    )

    assert buy == 0.45
    assert sell == 0.3
    assert meta["applied"] is True
    assert meta["factor_count"] == 1


def test_runtime_overlay_side_policy_filters_unapproved_factor_sides():
    close = pd.Series(np.linspace(100, 110, 30))
    df = pd.DataFrame(
        {
            "open": close - 0.2,
            "high": close + 1.0,
            "low": close - 1.0,
            "close": close,
            "volume": 10000,
        }
    )
    features = pd.DataFrame(
        {
            "oldwang_triple_bull": [0.0] * 29 + [1.0],
            "oldwang_triple_bear": 0.0,
        }
    )
    config = {
        "active": True,
        "policy": "test",
        "selected_config": {
            "weights": {"oldwang_trend": 0.05},
            "hold_bias": 0.0,
            "side_policy": {"keep_positive": [], "keep_negative": ["oldwang_trend"]},
        },
    }

    buy, sell, meta = apply_multi_factor_overlay(
        buy_prob=0.4,
        sell_prob=0.3,
        features=features,
        df=df,
        config=config,
        enabled=True,
    )

    assert buy == 0.4
    assert sell == 0.3
    assert meta["side_policy"]["keep_negative"] == ["oldwang_trend"]


def test_runtime_lower_shadow_prefers_yin_gao_pao_shape():
    df = pd.DataFrame(
        {
            "open": [100.0, 100.0],
            "high": [102.0, 102.0],
            "low": [94.0, 92.0],
            "close": [95.0, 97.0],
            "volume": [10000.0, 10000.0],
        }
    )
    factors = build_runtime_signed_factors(
        pd.DataFrame(index=df.index),
        df,
        ["lower_shadow", "candle_shadow"],
    )

    assert factors["lower_shadow"].iloc[1] > 0.55
    assert factors["lower_shadow"].iloc[0] < 0.20
    assert factors["candle_shadow"].iloc[1] > factors["candle_shadow"].iloc[0]


def test_runtime_lower_shadow_requires_prior_low_to_hold():
    base = pd.DataFrame(
        {
            "open": [100.0, 98.0, 98.0],
            "high": [102.0, 100.0, 100.0],
            "low": [92.0, 93.0, 91.0],
            "close": [97.0, 98.0, 92.0],
            "volume": [10000.0, 10000.0, 10000.0],
        }
    )
    held_df = base.iloc[:2]
    held = build_runtime_signed_factors(pd.DataFrame(index=held_df.index), held_df, ["lower_shadow"])
    broken = build_runtime_signed_factors(pd.DataFrame(index=base.index), base, ["lower_shadow"])

    assert held["lower_shadow"].iloc[-1] > 0.30
    assert broken["lower_shadow"].iloc[-1] < held["lower_shadow"].iloc[-1]


def test_runtime_high_base_and_battle_factors_push_sell():
    closes = [50.0] * 100 + [55.0, 58.0, 61.0, 64.0, 67.0, 70.0, 72.0, 74.0, 76.0, 79.0, 82.0, 86.0, 90.0, 94.0, 98.0, 102.0, 106.0, 110.0, 114.0, 118.0, 120.0]
    df = pd.DataFrame(
        {
            "open": [close - 0.5 for close in closes],
            "high": [close + 1.0 for close in closes],
            "low": [close - 1.0 for close in closes],
            "close": closes,
            "volume": [10000.0] * (len(closes) - 1) + [40000.0],
        }
    )
    df.loc[df.index[-1], ["open", "high", "low", "close"]] = [119.5, 125.0, 118.8, 120.0]

    factors = build_runtime_signed_factors(
        pd.DataFrame(index=df.index),
        df,
        ["high_base", "high_volume_upper_shadow", "short_term_battle"],
    )

    assert factors["high_base"].iloc[-1] < 0
    assert factors["high_volume_upper_shadow"].iloc[-1] < 0
    assert factors["short_term_battle"].iloc[-1] == -1.0


def test_runtime_sell_specific_breakdown_factors_push_sell():
    rows = []
    for _ in range(25):
        rows.append({"open": 100.2, "high": 101.0, "low": 99.0, "close": 100.0, "volume": 1000.0})
    rows.extend(
        [
            {"open": 100.0, "high": 100.5, "low": 98.0, "close": 98.5, "volume": 1100.0},
            {"open": 98.7, "high": 99.0, "low": 96.0, "close": 96.5, "volume": 1200.0},
            {"open": 96.7, "high": 97.0, "low": 93.5, "close": 94.0, "volume": 3500.0},
            {"open": 94.2, "high": 95.0, "low": 92.0, "close": 93.0, "volume": 2800.0},
            {"open": 93.0, "high": 94.5, "low": 92.5, "close": 94.0, "volume": 900.0},
        ]
    )
    df = pd.DataFrame(rows)

    factors = build_runtime_signed_factors(
        pd.DataFrame(index=df.index),
        df,
        [
            "san_sheng_wu_nai_breakdown",
            "downside_follow_through",
            "weak_rebound_after_breakdown",
            "high_volume_downside_follow_through",
        ],
    )

    assert factors["san_sheng_wu_nai_breakdown"].iloc[27] < 0
    assert factors["downside_follow_through"].iloc[28] < 0
    assert factors["weak_rebound_after_breakdown"].iloc[29] < 0
    assert factors["high_volume_downside_follow_through"].iloc[27] < 0


def test_runtime_overlay_accepts_accuracy_candidate_config():
    config = json.loads(Path("config/multi_factor_overlay.json").read_text())
    closes = pd.Series(np.linspace(80.0, 120.0, 140))
    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)

    buy, sell, meta = apply_multi_factor_overlay(
        buy_prob=0.35,
        sell_prob=0.25,
        features=features,
        df=df,
        config=config,
        enabled=True,
    )

    assert meta["applied"] is True
    assert meta["selected_name"] == "fixed_side_policy_weight_opt_v1"
    assert meta["factor_count"] == 44
    assert 0.0 <= buy <= 1.0
    assert 0.0 <= sell <= 1.0