DockerSpace / tests /test_multi_factor_weight_config.py
DennisChan0909's picture
Backup current stock predictor strategies
ee37d63
Raw
History Blame Contribute Delete
32.7 kB
import pandas as pd
from scripts.search_multi_factor_weight_config import (
DEFAULT_FACTORS,
MultiFactorConfig,
SUPPORTED_FACTORS,
add_intraday_60k_factor_inputs,
add_raw_factor_inputs,
apply_multi_factor_config,
build_promotion_decision,
build_signed_factor_frame,
iter_multi_factor_grid,
search_multi_factor_configs,
_needs_intraday_60k,
_score_row,
)
def test_build_signed_factor_frame_combines_bull_and_bear_components():
features = pd.DataFrame(
{
"oldwang_triple_bull": [1.0, 0.0],
"oldwang_triple_bear": [0.0, 1.0],
"oldwang_volume_high_break": [1.0, 0.0],
"oldwang_volume_low_break": [0.0, 1.0],
"factor_intraday_60k_volume_low_guard": [1.0, 0.0],
"factor_intraday_60k_volume_low_break": [0.0, 1.0],
"big_player_buy_1m_4m": [1_000_000.0, 500_000.0],
"big_player_sell_1m_4m": [500_000.0, 2_000_000.0],
"big_player_buy_gt_4m": [4_000_000.0, 1_500_000.0],
"big_player_sell_gt_4m": [1_500_000.0, 3_000_000.0],
"big_player_power_1m_4m": [0.7, -0.4],
"big_player_power_gt_4m": [0.2, -0.8],
"factor_lower_shadow_ratio": [0.5, 0.1],
"factor_upper_shadow_ratio": [0.1, 0.6],
"factor_body_ratio": [0.5, -0.5],
"factor_high_base_score": [3.0, 0.0],
"factor_short_term_battle": [1.0, 0.0],
"factor_high_volume_upper_shadow": [0.7, 0.0],
"factor_volume_z20_prior": [2.0, -1.0],
"foreign_net_vol_ratio": [0.2, -0.2],
"trust_net_vol_ratio": [0.1, -0.1],
"dealer_net_vol_ratio": [0.05, -0.05],
"chip_score": [0.5, -0.25],
"return_20d": [0.10, -0.10],
"ma_bull_alignment": [1.0, 0.0],
"ma_bear_alignment": [0.0, 1.0],
"macd_hist_norm": [0.02, -0.02],
"macd_above_zero": [1.0, 0.0],
"rsi": [25.0, 75.0],
"bb_pct_b": [0.10, 0.90],
"foreign_trust_alignment": [1.0, -1.0],
"institutional_streak": [5.0, -5.0],
"institutional_20d_zscore": [2.0, -2.0],
"margin_5d_change_pct": [-0.2, 0.2],
"short_balance_5d_change": [-0.1, 0.1],
"short_margin_ratio": [0.1, 0.4],
"sbl_balance_5d_change": [0.0, 0.6],
"sbl_balance_ratio": [0.0, 0.2],
"margin_buy_pressure": [0.7, 0.3],
"volume_ratio": [2.0, 0.6],
"factor_volume_contraction_after_spike": [1.0, -1.0],
"amihud_zscore": [-2.0, 2.0],
"volatility_20d": [0.01, 0.08],
"atr_ratio": [0.01, 0.08],
"return_60d": [0.2, -0.2],
"return_120d": [0.1, -0.1],
"mom_minus_reversal": [0.3, -0.3],
}
)
factors = build_signed_factor_frame(
features,
[
"oldwang_trend",
"oldwang_volume",
"intraday_60k_volume_low_guard",
"intraday_60k_volume_low_break",
"big_player_buy",
"big_player_sell",
"big_player_buy_ratio_1m_4m",
"big_player_sell_ratio_1m_4m",
"big_player_buy_ratio_gt_4m",
"big_player_sell_ratio_gt_4m",
"big_player_power_1m_4m",
"big_player_power_gt_4m",
"volume_expansion",
"volume_spike",
"volume_surge",
"volume_contraction_after_spike",
"candle_shadow",
"upper_shadow",
"lower_shadow",
"bullish_upper_shadow",
"high_base",
"high_volume_upper_shadow",
"short_term_battle",
"foreign_flow",
"trust_flow",
"dealer_flow",
"chip_score",
"momentum_20d",
"ma_alignment",
"macd_momentum",
"rsi_reversal",
"bb_reversion",
"institutional_flow",
"margin_risk",
"short_pressure",
"liquidity_quality",
"volatility_risk",
"momentum_60d",
"momentum_120d",
"momentum_20_60_120",
"medium_momentum",
"reversal_pressure",
],
)
assert factors["oldwang_trend"].tolist() == [1.0, -1.0]
assert factors["oldwang_volume"].tolist() == [1.0, -1.0]
assert factors["intraday_60k_volume_low_guard"].tolist() == [1.0, 0.0]
assert factors["intraday_60k_volume_low_break"].tolist() == [-0.0, -1.0]
assert factors["big_player_buy"].tolist() == [1.0, 0.4]
assert factors["big_player_sell"].tolist() == [-0.4, -1.0]
assert [round(value, 4) for value in factors["big_player_buy_ratio_1m_4m"].tolist()] == [0.6667, 0.2]
assert [round(value, 4) for value in factors["big_player_sell_ratio_1m_4m"].tolist()] == [-0.3333, -0.8]
assert [round(value, 4) for value in factors["big_player_buy_ratio_gt_4m"].tolist()] == [0.7273, 0.3333]
assert [round(value, 4) for value in factors["big_player_sell_ratio_gt_4m"].tolist()] == [-0.2727, -0.6667]
assert factors["big_player_power_1m_4m"].tolist() == [0.7, -0.4]
assert factors["big_player_power_gt_4m"].tolist() == [0.2, -0.8]
assert factors["volume_expansion"].iloc[0] > 0
assert factors["volume_expansion"].iloc[1] < 0
assert factors["volume_spike"].tolist() == [1.0, 0.0]
assert factors["volume_surge"].iloc[0] > 0
assert factors["volume_surge"].iloc[1] < 0
assert factors["volume_contraction_after_spike"].tolist() == [1.0, -1.0]
assert [round(value, 3) for value in factors["candle_shadow"].tolist()] == [0.075, -0.565]
assert factors["upper_shadow"].tolist() == [-0.1, -0.6]
assert [round(value, 3) for value in factors["lower_shadow"].tolist()] == [0.175, 0.035]
assert factors["bullish_upper_shadow"].iloc[0] < 0
assert factors["high_base"].tolist() == [-0.75, -0.0]
assert factors["high_volume_upper_shadow"].iloc[0] < 0
assert factors["high_volume_upper_shadow"].iloc[1] == 0
assert factors["short_term_battle"].tolist() == [-1.0, -0.0]
assert factors["foreign_flow"].iloc[0] > 0
assert factors["foreign_flow"].iloc[1] < 0
assert factors["trust_flow"].iloc[0] > 0
assert factors["trust_flow"].iloc[1] < 0
assert factors["dealer_flow"].iloc[0] > 0
assert factors["dealer_flow"].iloc[1] < 0
assert factors["chip_score"].tolist() == [0.5, -0.25]
assert factors["momentum_20d"].tolist() == [0.4, -0.4]
assert factors["ma_alignment"].tolist() == [1.0, -1.0]
assert factors["macd_momentum"].iloc[0] > 0
assert factors["macd_momentum"].iloc[1] < 0
assert factors["rsi_reversal"].iloc[0] > 0
assert factors["rsi_reversal"].iloc[1] < 0
assert factors["bb_reversion"].iloc[0] > 0
assert factors["bb_reversion"].iloc[1] < 0
assert factors["institutional_flow"].iloc[0] > 0
assert factors["institutional_flow"].iloc[1] < 0
assert factors["margin_risk"].tolist() == [0.5, -0.5]
assert factors["short_pressure"].iloc[0] <= 0
assert factors["short_pressure"].iloc[1] < factors["short_pressure"].iloc[0]
assert factors["liquidity_quality"].iloc[0] > 0
assert factors["liquidity_quality"].iloc[1] < 0
assert factors["volatility_risk"].iloc[0] < 0
assert factors["volatility_risk"].iloc[1] < factors["volatility_risk"].iloc[0]
assert factors["momentum_60d"].iloc[0] > 0
assert factors["momentum_120d"].iloc[1] < 0
assert factors["momentum_20_60_120"].iloc[0] > 0
assert factors["medium_momentum"].iloc[0] > 0
assert factors["medium_momentum"].iloc[1] < 0
assert factors["reversal_pressure"].tolist() == [0.3, -0.3]
def test_add_raw_factor_inputs_derives_volume_and_shadow_inputs():
df = pd.DataFrame(
{
"open": [10.0] * 25 + [11.0],
"high": [11.0] * 25 + [13.0],
"low": [9.0] * 25 + [10.5],
"close": [10.5] * 25 + [11.2],
"volume": [1000.0] * 25 + [5000.0],
}
)
features = pd.DataFrame(index=df.index)
out = add_raw_factor_inputs(features, df)
assert out["factor_upper_shadow_ratio"].iloc[-1] > out["factor_lower_shadow_ratio"].iloc[-1]
assert out["factor_volume_z20_prior"].iloc[-1] > 0
assert "factor_high_base_score" in out.columns
assert "factor_short_term_battle" in out.columns
assert "factor_high_volume_upper_shadow" in out.columns
assert "factor_volume_contraction_after_spike" in out.columns
def test_sell_specific_breakdown_patterns_are_signed_risk_factors():
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)
enriched = add_raw_factor_inputs(pd.DataFrame(index=df.index), df)
assert enriched["factor_san_sheng_wu_nai_breakdown"].iloc[27] == 1.0
assert enriched["factor_downside_follow_through"].iloc[28] == 1.0
assert enriched["factor_weak_rebound_after_breakdown"].iloc[29] == 1.0
assert enriched["factor_high_volume_downside_follow_through"].iloc[27] > 0.0
factors = build_signed_factor_frame(
enriched,
[
"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_sell_specific_daily_patterns_do_not_require_intraday_60k_fetch():
assert not _needs_intraday_60k(
[
"san_sheng_wu_nai_breakdown",
"downside_follow_through",
"weak_rebound_after_breakdown",
"high_volume_downside_follow_through",
]
)
assert _needs_intraday_60k(["intraday_60k_volume_low_guard"])
assert _needs_intraday_60k(["intraday_60k_stale_guard_risk"])
def test_add_intraday_60k_factor_inputs_maps_latest_huge_volume_low_to_daily_rows():
daily = pd.DataFrame(
{
"date": ["2026-05-26", "2026-05-27"],
"close": [101.0, 97.0],
}
)
intraday = pd.DataFrame(
{
"date": ["2026-05-26"] * 21 + ["2026-05-27"],
"low": [99.0] * 20 + [100.0, 100.0],
"close": [101.0] * 22,
"volume": [1000.0] * 20 + [2500.0, 1000.0],
}
)
out = add_intraday_60k_factor_inputs(
pd.DataFrame(index=daily.index),
daily,
intraday_df=intraday,
)
assert out["factor_intraday_60k_volume_low"].tolist() == [100.0, 100.0]
assert out["factor_intraday_60k_volume_low_available"].tolist() == [1.0, 1.0]
assert out["factor_intraday_60k_volume_low_age_sessions"].tolist() == [0.0, 1.0]
assert out["factor_intraday_60k_volume_low_guard"].tolist() == [1.0, 0.0]
assert out["factor_intraday_60k_volume_low_break"].tolist() == [0.0, 1.0]
def test_high_base_and_short_term_battle_are_signed_risk_factors():
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]
opens = [close - 0.5 for close in closes]
highs = [close + 1.0 for close in closes]
lows = [close - 1.0 for close in closes]
opens[-1] = 119.5
highs[-1] = 125.0
lows[-1] = 118.8
closes[-1] = 120.0
df = pd.DataFrame(
{
"open": opens,
"high": highs,
"low": lows,
"close": closes,
"volume": [10000.0] * (len(closes) - 1) + [40000.0],
}
)
features = add_raw_factor_inputs(pd.DataFrame(index=df.index), df)
factors = build_signed_factor_frame(features, ["high_base", "high_volume_upper_shadow", "short_term_battle", "doji_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
assert factors["doji_battle"].iloc[-1] == factors["short_term_battle"].iloc[-1]
def test_yin_gao_pao_shadow_inputs_follow_article_rules():
df = 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],
}
)
features = add_raw_factor_inputs(pd.DataFrame(index=df.index), df)
factors = build_signed_factor_frame(features, ["lower_shadow", "candle_shadow"])
assert features["factor_yin_gao_pao_raw"].tolist() == [1.0, 0.0, 0.0]
assert features["factor_yin_gao_pao_hold"].iloc[1] == 1.0
assert features["factor_yin_gao_pao_failed"].iloc[2] == 1.0
assert factors["lower_shadow"].iloc[1] > factors["lower_shadow"].iloc[2]
def test_businesstoday_kline_candidate_factors_are_signed():
df = pd.DataFrame(
{
"open": [100.0, 91.0, 96.0, 100.0, 109.0, 101.0, 106.0, 100.0, 94.0, 100.0],
"high": [102.0, 97.0, 104.0, 111.0, 110.0, 106.0, 107.0, 108.0, 101.0, 106.0],
"low": [89.0, 90.0, 95.0, 99.0, 103.0, 99.0, 100.0, 95.0, 93.0, 99.0],
"close": [90.0, 96.0, 103.0, 110.0, 104.0, 105.0, 101.0, 107.0, 100.0, 106.0],
"volume": [10000.0] * 10,
}
)
features = add_raw_factor_inputs(pd.DataFrame(index=df.index), df)
factors = build_signed_factor_frame(
features,
[
"bt_yi_yin_ya_breakout",
"bt_yi_yang_ding_breakdown",
"bt_yang_gao_pao",
"bt_bull_engulf",
"bt_bull_engulf_bear",
"bt_bull_bear_engulf_ratio",
"bt_double_bull_engulf_one_bear",
"oldwang_guard_support",
"oldwang_guard_break_risk",
"candle_lower_support",
"candle_upper_pressure",
"yin_gao_pao_support",
"yin_gao_pao_risk",
"bt_yang_gao_pao_risk",
"bt_bull_engulf_risk",
],
)
assert factors["bt_yi_yin_ya_breakout"].iloc[2] > 0
assert factors["bt_yi_yang_ding_breakdown"].iloc[4] < 0
assert factors["bt_double_bull_engulf_one_bear"].iloc[7] > 0
assert factors["bt_bull_engulf"].iloc[7] > 0
assert factors["bt_bull_engulf_risk"].iloc[7] < 0
assert factors["bt_bull_engulf_bear"].iloc[4] < 0
assert factors["bt_bull_bear_engulf_ratio"].iloc[7] > 0
assert factors["bt_yang_gao_pao"].iloc[9] > 0
assert factors["bt_yang_gao_pao_risk"].iloc[9] < 0
def test_shadow_cluster_support_resistance_factor_breaks_and_bounces():
df = pd.DataFrame(
{
"open": [100.0, 100.0, 101.0, 100.0, 100.0, 94.0],
"high": [110.0, 110.0, 112.0, 102.0, 102.0, 96.0],
"low": [99.0, 99.0, 100.0, 90.0, 90.0, 89.0],
"close": [101.0, 101.0, 111.0, 95.0, 96.0, 89.5],
"volume": [10000.0] * 6,
}
)
features = add_raw_factor_inputs(pd.DataFrame(index=df.index), df)
factors = build_signed_factor_frame(features, ["bt_shadow_cluster_support_resistance"])
assert factors["bt_shadow_cluster_support_resistance"].iloc[2] > 0
assert factors["bt_shadow_cluster_support_resistance"].iloc[5] < 0
def test_tej_lstm_candidate_factors_capture_trend_and_chop():
closes = [100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119]
df = pd.DataFrame(
{
"open": [value - 0.4 for value in closes],
"high": [value + 1.0 for value in closes],
"low": [value - 1.0 for value in closes],
"close": closes,
"volume": [10000.0] * len(closes),
}
)
features = pd.DataFrame(
{
"rsi": [55.0, 56.0, 57.0, 58.0, 59.0, 60.0, 61.0, 62.0, 63.0, 64.0, 65.0, 66.0, 67.0, 68.0, 69.0, 70.0, 71.0, 72.0, 73.0, 74.0],
"macd_hist_norm": [0.001 * idx for idx in range(len(closes))],
"k": [60.0 + idx for idx in range(len(closes))],
"d": [55.0 + idx for idx in range(len(closes))],
},
index=df.index,
)
enriched = add_raw_factor_inputs(features, df)
factors = build_signed_factor_frame(enriched, ["tej_lstm_trend_signal", "tej_lstm_chop_filter"])
assert factors["tej_lstm_trend_signal"].iloc[-1] > 0.4
assert factors["tej_lstm_chop_filter"].iloc[-1] > 0
def test_tej_macro_risk_proxy_is_signed_from_market_context():
features = pd.DataFrame(
{
"taiex_return_5d": [0.03, -0.03],
"taiex_return_20d": [0.06, -0.06],
"taiex_ma20_ratio": [1.04, 0.96],
"usdtwd_return_5d": [-0.01, 0.02],
"sox_ret_5d": [0.04, -0.04],
"sox_ma20_ratio": [1.03, 0.97],
"tnx_change_5d": [-0.10, 0.35],
"vix_level": [16.0, 30.0],
"vix_change_5d": [-2.0, 5.0],
}
)
factors = build_signed_factor_frame(features, ["tej_macro_risk_proxy", "tej_macro_risk_contrarian"])
assert factors["tej_macro_risk_proxy"].iloc[0] > 0
assert factors["tej_macro_risk_proxy"].iloc[1] < 0
assert factors["tej_macro_risk_contrarian"].iloc[0] < 0
assert factors["tej_macro_risk_contrarian"].iloc[1] > 0
def test_regime_reversal_research_factors_are_signed_and_gated():
features = pd.DataFrame(
{
"oldwang_triple_bull": [1.0, 0.0],
"oldwang_triple_bear": [0.0, 1.0],
"oldwang_ma5_hold": [1.0, 1.0],
"oldwang_trust_ma10_guard": [1.0, 1.0],
"oldwang_foreign_ma20_guard": [1.0, 1.0],
"factor_high_base_score": [0.0, 4.0],
"factor_intraday_60k_volume_low_guard": [1.0, 1.0],
"factor_intraday_60k_volume_low_age_sessions": [2.0, 8.0],
"macd_hist_norm": [0.01, -0.01],
"macd_above_zero": [1.0, 0.0],
"rsi": [70.0, 30.0],
"factor_bt_yi_yin_ya_breakout": [1.0, 0.0],
"factor_bt_yi_yin_ya_failed": [0.0, 1.0],
}
)
factors = build_signed_factor_frame(
features,
[
"oldwang_trend_reversal_pressure",
"oldwang_guard_lowbase_support",
"oldwang_guard_highbase_risk",
"intraday_60k_fresh_low_guard",
"intraday_60k_stale_guard_risk",
"macd_reversal_pressure",
"rsi_trend_reversal",
"bt_yi_yin_ya_breakout_risk",
],
)
assert factors["oldwang_trend_reversal_pressure"].iloc[0] < 0
assert factors["oldwang_trend_reversal_pressure"].iloc[1] > 0
assert factors["oldwang_guard_lowbase_support"].iloc[0] > 0
assert factors["oldwang_guard_lowbase_support"].iloc[1] == 0
assert factors["oldwang_guard_highbase_risk"].iloc[0] == 0
assert factors["oldwang_guard_highbase_risk"].iloc[1] < 0
assert factors["intraday_60k_fresh_low_guard"].iloc[0] > 0
assert factors["intraday_60k_fresh_low_guard"].iloc[1] == 0
assert factors["intraday_60k_stale_guard_risk"].iloc[0] == 0
assert factors["intraday_60k_stale_guard_risk"].iloc[1] < 0
assert factors["macd_reversal_pressure"].iloc[0] < 0
assert factors["macd_reversal_pressure"].iloc[1] > 0
assert factors["rsi_trend_reversal"].iloc[0] < 0
assert factors["rsi_trend_reversal"].iloc[1] > 0
assert factors["bt_yi_yin_ya_breakout_risk"].iloc[0] < 0
assert factors["bt_yi_yin_ya_breakout_risk"].iloc[1] > 0
def test_apply_multi_factor_config_pushes_buy_or_sell_from_signed_values():
events = pd.DataFrame(
{
"p_buy": [0.44, 0.40],
"p_hold": [0.48, 0.48],
"p_sell": [0.08, 0.12],
"factor__a": [2.0, -2.0],
"factor__b": [0.0, 0.0],
}
)
pred = apply_multi_factor_config(
events,
MultiFactorConfig(weights={"a": 0.25, "b": 0.0}),
)
assert pred.tolist() == [1, -1]
def test_iter_multi_factor_grid_builds_each_factor_weight_combination():
configs = iter_multi_factor_grid(
["a", "b"],
grid_values=[0.0, 0.1],
hold_bias_values=[0.0, 0.02],
)
assert len(configs) == 8
assert configs[-1].weights == {"a": 0.1, "b": 0.1}
assert configs[-1].hold_bias == 0.02
def test_iter_multi_factor_grid_samples_when_candidate_space_is_too_large():
configs = iter_multi_factor_grid(
["a", "b", "c", "d"],
grid_values=[0.01, 0.02, 0.03, 0.04],
hold_bias_values=[0.0],
search_mode="auto",
max_configs=10,
random_seed=7,
)
assert len(configs) == 10
assert all(all(weight > 0 for weight in cfg.weights.values()) for cfg in configs)
def test_iter_multi_factor_grid_samples_with_zero_weight_limit():
configs = iter_multi_factor_grid(
["a", "b", "c", "d", "e"],
grid_values=[0.0, 0.25, 0.5],
hold_bias_values=[0.0],
search_mode="sampled",
max_configs=20,
random_seed=7,
max_zero_factor_weights=1,
)
constrained = [
cfg for cfg in configs
if any(abs(weight) <= 1e-12 for weight in cfg.weights.values())
and not all(abs(weight) <= 1e-12 for weight in cfg.weights.values())
]
assert constrained
assert all(
sum(1 for weight in cfg.weights.values() if abs(weight) <= 1e-12) <= 1
for cfg in constrained
)
def test_search_multi_factor_configs_finds_passing_golden_config():
rows = []
for idx in range(12):
is_late = idx >= 6
bullish = idx in {2, 3, 4, 8, 9, 10}
bearish = idx in {0, 1, 6, 7}
rows.append(
{
"stock": "2330",
"date": f"2025-04-{idx + 1:02d}",
"y_true": 1 if bullish else (-1 if bearish else 0),
"base_pred": 0,
"p_buy": 0.44,
"p_hold": 0.48,
"p_sell": 0.08,
"factor__trend": 1.0 if bullish else (-1.0 if bearish else 0.0),
"factor__volume": 0.0,
}
)
result = search_multi_factor_configs(
pd.DataFrame(rows),
factor_names=["trend", "volume"],
grid_values=[0.0, 0.05],
hold_bias_values=[0.0],
optimize_ratio=0.5,
min_accuracy_delta_pp=10.0,
min_buy_precision_delta_pp=0.0,
min_sell_precision_delta_pp=None,
min_direction_accuracy_delta_pp=0.0,
min_signal_ratio=0.5,
max_buy_count_ratio=None,
require_nonzero_weights=False,
require_all_factor_weights=False,
top_n=5,
)
assert result["golden_config"]["config"]["weights"]["trend"] == 0.05
assert result["golden_config"]["validation"]["deltas"]["accuracy_delta_pp"] > 0
assert result["passing_count"] >= 1
def test_search_multi_factor_configs_can_reject_all_zero_noop_config():
rows = []
for idx in range(8):
rows.append(
{
"stock": "2330",
"date": f"2025-04-{idx + 1:02d}",
"y_true": 0,
"base_pred": 0,
"p_buy": 0.20,
"p_hold": 0.60,
"p_sell": 0.20,
"factor__trend": 0.0,
}
)
result = search_multi_factor_configs(
pd.DataFrame(rows),
factor_names=["trend"],
grid_values=[0.0],
hold_bias_values=[0.0],
optimize_ratio=0.5,
min_accuracy_delta_pp=0.0,
min_buy_precision_delta_pp=0.0,
min_sell_precision_delta_pp=None,
min_direction_accuracy_delta_pp=0.0,
min_signal_ratio=0.5,
max_buy_count_ratio=None,
require_nonzero_weights=True,
require_all_factor_weights=False,
top_n=5,
)
assert result["golden_config"] is None
assert result["passing_count"] == 0
assert result["top_overall"][0]["gate"]["checks"]["nonzero_weights"] is False
def test_search_multi_factor_configs_can_require_every_factor_weight_nonzero():
rows = []
for idx in range(8):
rows.append(
{
"stock": "2330",
"date": f"2025-05-{idx + 1:02d}",
"y_true": 1 if idx % 2 == 0 else -1,
"base_pred": 0,
"p_buy": 0.44,
"p_hold": 0.48,
"p_sell": 0.08,
"factor__trend": 1.0 if idx % 2 == 0 else -1.0,
"factor__volume": 0.0,
}
)
result = search_multi_factor_configs(
pd.DataFrame(rows),
factor_names=["trend", "volume"],
grid_values=[0.0, 0.05],
hold_bias_values=[0.0],
optimize_ratio=0.5,
min_accuracy_delta_pp=0.0,
min_buy_precision_delta_pp=0.0,
min_sell_precision_delta_pp=None,
min_direction_accuracy_delta_pp=0.0,
min_signal_ratio=0.5,
max_buy_count_ratio=None,
require_nonzero_weights=False,
require_all_factor_weights=True,
top_n=5,
)
assert result["golden_config"]["config"]["weights"] == {"trend": 0.05, "volume": 0.05}
assert result["golden_config"]["gate"]["checks"]["all_factor_weights_nonzero"] is True
def test_search_multi_factor_configs_limits_zero_weight_count():
rows = []
for idx in range(12):
bullish = idx in {2, 3, 4, 8, 9, 10}
bearish = idx in {0, 1, 6, 7}
rows.append(
{
"stock": "2330",
"date": f"2025-05-{idx + 1:02d}",
"y_true": 1 if bullish else (-1 if bearish else 0),
"base_pred": 0,
"p_buy": 0.44,
"p_hold": 0.48,
"p_sell": 0.08,
"factor__trend": 1.0 if bullish else (-1.0 if bearish else 0.0),
"factor__volume": 0.0,
"factor__chip": 0.0,
}
)
result = search_multi_factor_configs(
pd.DataFrame(rows),
factor_names=["trend", "volume", "chip"],
grid_values=[0.0, 0.05],
hold_bias_values=[0.0],
optimize_ratio=0.5,
min_accuracy_delta_pp=0.0,
min_buy_precision_delta_pp=0.0,
min_sell_precision_delta_pp=None,
min_direction_accuracy_delta_pp=0.0,
min_signal_ratio=0.5,
max_buy_count_ratio=None,
require_nonzero_weights=True,
require_all_factor_weights=False,
max_zero_factor_weights=1,
top_n=5,
)
assert result["golden_config"]["gate"]["zero_factor_weight_count"] <= 1
assert result["golden_config"]["gate"]["checks"]["max_zero_factor_weights"] is True
def test_search_multi_factor_configs_can_gate_sell_precision_drop():
rows = []
for idx in range(12):
bullish = idx in {2, 3, 4, 8, 9, 10}
bearish = idx in {0, 1, 6, 7}
rows.append(
{
"stock": "2330",
"date": f"2025-06-{idx + 1:02d}",
"y_true": 1 if bullish else (-1 if bearish else 0),
"base_pred": -1 if bearish else 0,
"p_buy": 0.44,
"p_hold": 0.48,
"p_sell": 0.08,
"factor__trend": 1.0 if bullish else 0.0,
}
)
result = search_multi_factor_configs(
pd.DataFrame(rows),
factor_names=["trend"],
grid_values=[0.05],
hold_bias_values=[0.0],
optimize_ratio=0.5,
min_accuracy_delta_pp=0.0,
min_buy_precision_delta_pp=0.0,
min_sell_precision_delta_pp=0.0,
min_direction_accuracy_delta_pp=0.0,
min_signal_ratio=0.5,
max_buy_count_ratio=None,
require_nonzero_weights=True,
require_all_factor_weights=True,
top_n=5,
)
assert result["golden_config"] is None
assert result["top_overall"][0]["gate"]["checks"]["sell_precision_delta"] is False
def test_protected_objective_ranks_signal_quality_before_label_accuracy():
quality_preserved = _score_row(
{
"accuracy_delta_pp": 0.4,
"buy_precision_delta_pp": 0.0,
"sell_precision_delta_pp": 0.0,
"direction_accuracy_delta_pp": 0.0,
},
{"accuracy": 40.7},
1.0,
min_accuracy_delta_pp=1.0,
min_buy_precision_delta_pp=0.0,
min_sell_precision_delta_pp=0.0,
min_direction_accuracy_delta_pp=0.0,
)
accuracy_only = _score_row(
{
"accuracy_delta_pp": 2.1,
"buy_precision_delta_pp": -0.2,
"sell_precision_delta_pp": -3.0,
"direction_accuracy_delta_pp": -1.3,
},
{"accuracy": 42.3},
0.8,
min_accuracy_delta_pp=1.0,
min_buy_precision_delta_pp=0.0,
min_sell_precision_delta_pp=0.0,
min_direction_accuracy_delta_pp=0.0,
)
assert quality_preserved > accuracy_only
def test_default_factor_set_uses_nonzero_full_participation_weights():
assert "oldwang_trend" in DEFAULT_FACTORS
assert "intraday_60k_volume_low_guard" in DEFAULT_FACTORS
assert "intraday_60k_volume_low_break" in DEFAULT_FACTORS
assert "big_player_buy" in DEFAULT_FACTORS
assert "big_player_sell" in DEFAULT_FACTORS
assert "big_player_buy_ratio_1m_4m" not in DEFAULT_FACTORS
assert "big_player_sell_ratio_1m_4m" not in DEFAULT_FACTORS
assert "big_player_buy_ratio_gt_4m" not in DEFAULT_FACTORS
assert "big_player_sell_ratio_gt_4m" not in DEFAULT_FACTORS
assert "big_player_power_1m_4m" not in DEFAULT_FACTORS
assert "big_player_power_gt_4m" not in DEFAULT_FACTORS
assert "foreign_flow" in DEFAULT_FACTORS
assert "short_pressure" in DEFAULT_FACTORS
assert "momentum_120d" in DEFAULT_FACTORS
assert "bt_yi_yin_ya_breakout" in DEFAULT_FACTORS
assert "bt_yi_yang_ding_breakdown" in DEFAULT_FACTORS
assert "bt_shadow_cluster_support_resistance" in DEFAULT_FACTORS
assert "high_base" in DEFAULT_FACTORS
assert "high_volume_upper_shadow" in DEFAULT_FACTORS
assert "short_term_battle" in DEFAULT_FACTORS
assert "bt_bull_engulf" in DEFAULT_FACTORS
assert "bt_bull_engulf_bear" in DEFAULT_FACTORS
assert "bt_double_bull_engulf_one_bear" not in DEFAULT_FACTORS
assert "oldwang_guard_support" not in DEFAULT_FACTORS
assert "oldwang_guard_lowbase_support" not in DEFAULT_FACTORS
assert "macd_reversal_pressure" not in DEFAULT_FACTORS
assert "rsi_trend_reversal" not in DEFAULT_FACTORS
assert "bt_yang_gao_pao_risk" not in DEFAULT_FACTORS
for sell_pattern in [
"san_sheng_wu_nai_breakdown",
"downside_follow_through",
"weak_rebound_after_breakdown",
"high_volume_downside_follow_through",
]:
assert sell_pattern in SUPPORTED_FACTORS
assert sell_pattern not in DEFAULT_FACTORS
assert "tej_lstm_trend_signal" in DEFAULT_FACTORS
assert "tej_lstm_chop_filter" in DEFAULT_FACTORS
assert "tej_macro_risk_proxy" in DEFAULT_FACTORS
def test_promotion_decision_rejects_smoke_even_with_golden_config():
result = {
"thresholds": {"min_signal_ratio": 0.7},
"search_mode": "exhaustive",
"golden_config": {
"config": {"weights": {"trend": 0.05, "volume": 0.05}, "hold_bias": 0.0},
"gate": {"signal_ratio": 1.0},
"validation": {
"deltas": {
"accuracy_delta_pp": 2.0,
"buy_precision_delta_pp": 0.0,
"sell_precision_delta_pp": 0.0,
}
},
},
}
decision = build_promotion_decision(
result=result,
factor_names=["trend", "volume"],
covered_stocks=["2330"],
months=12,
periods=20,
min_stocks=0,
)
assert decision["passed"] is False
assert decision["checks"]["all_factor_weights_nonzero"] is True
assert decision["checks"]["not_tiny_stock_sample"] is False
assert decision["checks"]["history_at_least_3y"] is False
assert not any(name.startswith("periods_at_least") for name in decision["checks"])