File size: 4,507 Bytes
e610a2f
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
ee37d63
e610a2f
 
 
 
 
ee37d63
 
 
 
 
 
 
 
 
 
 
 
 
e610a2f
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
import numpy as np
import pandas as pd

from models.predictor import (
    FEATURE_COLUMNS,
    OLDWANG_WEIGHT_OVERLAY,
    _apply_oldwang_weight_overlay,
    _build_features,
    _build_oldwang_context,
)


OLDWANG_FEATURES = [
    "oldwang_triple_bull",
    "oldwang_triple_bear",
    "oldwang_ma5_hold",
    "oldwang_trust_ma10_guard",
    "oldwang_trust_ma10_broken",
    "oldwang_foreign_ma20_guard",
    "oldwang_foreign_ma20_broken",
    "oldwang_volume_spike",
    "oldwang_volume_high_break",
    "oldwang_volume_low_guard",
    "oldwang_volume_low_break",
    "oldwang_gap_guard",
    "oldwang_gap_filled",
    "oldwang_bull_score",
    "oldwang_bear_score",
]


def _oldwang_frame(n: int = 80) -> pd.DataFrame:
    close = pd.Series(np.linspace(100, 145, n))
    volume = np.full(n, 10_000.0)
    volume[50] = 45_000.0
    df = pd.DataFrame(
        {
            "open": close.shift(1).fillna(close.iloc[0]) + 0.2,
            "high": close + 1.5,
            "low": close - 1.0,
            "close": close,
            "volume": volume,
            "ma5": close.rolling(5, min_periods=1).mean(),
            "ma10": close.rolling(10, min_periods=1).mean(),
            "ma20": close.rolling(20, min_periods=1).mean(),
            "ma60": close.rolling(60, min_periods=1).mean(),
            "rsi": 55.0,
            "bb_pct_b": 0.5,
            "k": 50.0,
            "d": 48.0,
            "volume_ratio": pd.Series(volume).rolling(20, min_periods=1).mean() / 10_000.0,
            "atr_ratio": 0.02,
            "obv_trend": 0.01,
            "volatility_20d": 0.02,
            "foreign_net": 1_000.0,
            "trust_net": 500.0,
            "dealer_net": 0.0,
            "institutional_net": 1_500.0,
        }
    )
    df["macd"] = np.linspace(-1, 1, n)
    df["macd_signal"] = np.linspace(-0.8, 0.7, n)
    df["macd_hist"] = df["macd"] - df["macd_signal"]
    return df


def test_oldwang_features_are_production_columns_and_non_null():
    df = _oldwang_frame()
    feat = _build_features(df)

    for col in OLDWANG_FEATURES:
        assert col in FEATURE_COLUMNS
        assert col in feat.columns
        assert feat[col].replace([np.inf, -np.inf], np.nan).notna().all()


def test_oldwang_bull_context_activates_on_uptrend_with_institutional_buying():
    df = _oldwang_frame()
    feat = _build_features(df)
    context = _build_oldwang_context(df, feat)

    assert feat["oldwang_triple_bull"].tail(10).max() == 1.0
    assert feat["oldwang_trust_ma10_guard"].tail(10).max() == 1.0
    assert feat["oldwang_foreign_ma20_guard"].tail(10).max() == 1.0
    assert feat["oldwang_bull_score"].tail(10).max() >= 3.0
    assert context["bull_score"] >= 3.0
    assert any("三陽開泰" in reason for reason in context["bull_reasons"])
    assert any("外資" in reason for reason in context["bull_reasons"])
    assert context["key_levels"]["ma20"] is not None


def test_oldwang_context_reports_volume_and_gap_risks():
    df = _oldwang_frame()
    feat = _build_features(df)
    feat.loc[feat.index[-1], "oldwang_volume_low_break"] = 1.0
    feat.loc[feat.index[-1], "oldwang_gap_filled"] = 1.0
    feat.loc[feat.index[-1], "oldwang_bear_score"] = 2.0

    context = _build_oldwang_context(df, feat)

    assert context["bear_score"] == 2.0
    assert any("爆大量" in reason for reason in context["risk_reasons"])
    assert any("缺口回補" in reason for reason in context["risk_reasons"])


def test_oldwang_weight_overlay_applies_validated_top40_weights():
    buy, sell, meta = _apply_oldwang_weight_overlay(
        buy_prob=0.55,
        sell_prob=0.30,
        oldwang_context={"bull_score": 3, "bear_score": 1},
        config=OLDWANG_WEIGHT_OVERLAY,
    )

    assert buy == 0.66
    assert sell == 0.30
    assert meta["applied"] is True
    assert meta["validation_stock_count"] == 40


def test_oldwang_weight_overlay_applies_validated_bear_gate():
    buy, sell, meta = _apply_oldwang_weight_overlay(
        buy_prob=0.70,
        sell_prob=0.20,
        oldwang_context={"bull_score": 1, "bear_score": 3},
        config=OLDWANG_WEIGHT_OVERLAY,
    )

    assert buy == 0.0
    assert sell == 0.20
    assert meta["applied"] is True


def test_oldwang_weight_overlay_can_be_disabled():
    buy, sell, meta = _apply_oldwang_weight_overlay(
        buy_prob=0.55,
        sell_prob=0.30,
        oldwang_context={"bull_score": 3, "bear_score": 1},
        enabled=False,
    )

    assert buy == 0.55
    assert sell == 0.30
    assert meta["applied"] is False