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