File size: 8,313 Bytes
ff6fd33
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
#!/usr/bin/env python3
"""C3 validation: fixed-horizon labels vs triple-barrier labels."""
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
try:
    from dotenv import load_dotenv; load_dotenv(ROOT / ".env")
except ImportError: pass

import warnings; warnings.filterwarnings("ignore")
import numpy as np
import pandas as pd
import json
from sklearn.ensemble import RandomForestClassifier
from models.predictor import FEATURE_COLUMNS, _build_features

STOCKS = ["2330", "0050", "2317", "2454", "2881"]
LABEL_HORIZON = 5
FIXED_THRESH  = 0.02
PT_SL_RATIO   = 1.0   # symmetric barriers; scale = 1 * daily_vol
VOL_LOOKBACK  = 20
MIN_TRAIN     = 240
STEP          = 21
RF = dict(n_estimators=200, max_depth=6, min_samples_leaf=10,
          class_weight="balanced", random_state=42, n_jobs=-1)


def fetch_df(stock_no):
    from services.predictor_service import _fetch_with_cache
    from indicators.technical import add_all_indicators, add_cross_asset_tw
    from data.institutional_flow import add_institutional_flow
    from data.margin_flow import add_margin_flow
    from data.fetcher import fetch_cross_asset_tw
    df = _fetch_with_cache(stock_no, months=24)
    if df is None or df.empty: return None
    df = add_all_indicators(df)
    df = add_institutional_flow(df, stock_no)
    df = add_margin_flow(df, stock_no)
    start, end = str(df["date"].min()), str(df["date"].max())
    result = fetch_cross_asset_tw(start, end)
    taiex, usdtwd = result[0], result[1]
    sox = result[2] if len(result) > 2 else None
    tnx = result[3] if len(result) > 3 else None
    df = add_cross_asset_tw(df, taiex, usdtwd, sox_close=sox, tnx_close=tnx)
    return df


def fixed_labels(close: np.ndarray) -> np.ndarray:
    """5-day forward return ±2% threshold."""
    n = len(close)
    labels = np.zeros(n, dtype=float)
    for i in range(n - LABEL_HORIZON):
        ret = (close[i + LABEL_HORIZON] - close[i]) / close[i]
        labels[i] = 1 if ret > FIXED_THRESH else (-1 if ret < -FIXED_THRESH else 0)
    labels[-LABEL_HORIZON:] = np.nan  # no forward data
    return labels


def triple_barrier_labels(close: np.ndarray) -> np.ndarray:
    """
    For each row t:
      - compute daily_vol = rolling std of log returns over VOL_LOOKBACK days
      - upper = close[t] * (1 + daily_vol * PT_SL_RATIO)
      - lower = close[t] * (1 - daily_vol * PT_SL_RATIO)
      - scan close[t+1 .. t+LABEL_HORIZON], find which barrier is hit first
      - label: 1 (upper), -1 (lower), 0 (neither = vertical barrier expired)
    """
    n = len(close)
    log_ret = np.diff(np.log(close + 1e-9))  # length n-1
    labels = np.zeros(n, dtype=float)

    for i in range(n - LABEL_HORIZON):
        # Daily vol: std of log_ret in [i-VOL_LOOKBACK, i)
        start_vol = max(0, i - VOL_LOOKBACK)
        window = log_ret[start_vol:i]
        if len(window) < 5:
            # Not enough data — fall back to fixed threshold
            ret = (close[i + LABEL_HORIZON] - close[i]) / close[i]
            labels[i] = 1 if ret > FIXED_THRESH else (-1 if ret < -FIXED_THRESH else 0)
            continue

        daily_vol = float(np.std(window))
        if daily_vol < 0.001:
            daily_vol = 0.001  # floor

        upper = close[i] * (1.0 + daily_vol * PT_SL_RATIO)
        lower = close[i] * (1.0 - daily_vol * PT_SL_RATIO)

        label = 0  # default: vertical barrier (HOLD)
        for j in range(1, LABEL_HORIZON + 1):
            c = close[i + j]
            if c >= upper:
                label = 1
                break
            if c <= lower:
                label = -1
                break

        labels[i] = label

    labels[-LABEL_HORIZON:] = np.nan
    return labels


def walk_forward(feat_df, label_arr, cols):
    avail = [c for c in cols if c in feat_df.columns]
    X_all = feat_df[avail].fillna(0).values
    n = len(feat_df)
    y_true_all, y_pred_all = [], []
    cutoff = MIN_TRAIN
    while cutoff + STEP + LABEL_HORIZON <= n:
        y_tr = label_arr[:cutoff]
        valid = ~np.isnan(y_tr)
        if valid.sum() < 2 or len(np.unique(y_tr[valid])) < 2:
            cutoff += STEP; continue
        clf = RandomForestClassifier(**RF)
        clf.fit(X_all[:cutoff][valid], y_tr[valid].astype(int))
        test_end = min(cutoff + STEP, n - LABEL_HORIZON)
        y_te = label_arr[cutoff:test_end]
        valid_te = ~np.isnan(y_te)
        if valid_te.sum() == 0:
            cutoff += STEP; continue
        y_pred = clf.predict(X_all[cutoff:test_end][valid_te])
        y_true_all.extend(y_te[valid_te].tolist())
        y_pred_all.extend(y_pred.tolist())
        cutoff += STEP
    if not y_true_all: return {}
    y_true = np.array(y_true_all)
    y_pred = np.array(y_pred_all)
    dir_mask = y_pred != 0
    acc = (y_true == y_pred).mean() * 100
    dir_acc = (y_true[dir_mask] == y_pred[dir_mask]).mean() * 100 if dir_mask.sum() > 0 else float("nan")
    up_mask = y_pred == 1
    up_prec = (y_true[up_mask] == 1).mean() * 100 if up_mask.sum() > 0 else float("nan")
    dn_mask = y_pred == -1
    dn_prec = (y_true[dn_mask] == -1).mean() * 100 if dn_mask.sum() > 0 else float("nan")
    return {"accuracy": round(acc, 1), "dir_accuracy": round(dir_acc, 1),
            "up_precision": round(up_prec, 1), "dn_precision": round(dn_prec, 1),
            "n_predictions": len(y_true), "n_features": len(avail),
            "label_up_pct": round((y_true == 1).mean() * 100, 1),
            "label_dn_pct": round((y_true == -1).mean() * 100, 1)}


def main():
    results = {}
    hdr = f"{'Stock':>6}  {'Label':>12}  {'Acc%':>5}  {'Dir%':>5}  {'↑Prec%':>7}  {'↓Prec%':>7}  {'Preds':>5}"
    print(f"\n{hdr}\n{'-'*len(hdr)}")
    agg = {"fixed": [], "triple": []}

    for stock_no in STOCKS:
        print(f"  computing {stock_no}...", end="\r", flush=True)
        df = fetch_df(stock_no)
        if df is None or df.empty:
            print(f"{stock_no:>6}  no data"); continue
        feat = _build_features(df)
        close = df.set_index("date")["close"].values if "date" in df.columns else df["close"].values

        fl = fixed_labels(close)
        tbl = triple_barrier_labels(close)

        results[stock_no] = {}
        for name, labels in [("fixed", fl), ("triple", tbl)]:
            r = walk_forward(feat, labels, FEATURE_COLUMNS)
            results[stock_no][name] = r
            if r:
                print(f"{stock_no:>6}  {name:>12}  {r['accuracy']:>5.1f}  {r['dir_accuracy']:>5.1f}  "
                      f"{r['up_precision']:>7.1f}  {r['dn_precision']:>7.1f}  {r['n_predictions']:>5}")
                agg[name].append(r)

        if results[stock_no].get("fixed") and results[stock_no].get("triple"):
            rf, rt = results[stock_no]["fixed"], results[stock_no]["triple"]
            dd = rt["dir_accuracy"] - rf["dir_accuracy"]
            dp = rt["up_precision"] - rf["up_precision"]
            print(f"{'':>6}  {'Δ':>12}  {'':>5}  {dd:>+5.1f}  {dp:>+7.1f}")
        print()

    def mf(rows, f):
        vals = [r[f] for r in rows if isinstance(r.get(f), (int, float)) and not np.isnan(r.get(f, float("nan")))]
        return round(sum(vals)/len(vals), 1) if vals else float("nan")

    print("=== AGGREGATE ===")
    agg_summary = {}
    for name in ("fixed", "triple"):
        rows = agg[name]
        s = {k: mf(rows, k) for k in ("accuracy", "dir_accuracy", "up_precision", "dn_precision")}
        agg_summary[name] = s
        print(f"  {name:>12}: acc={s['accuracy']}%  dir={s['dir_accuracy']}%  ↑prec={s['up_precision']}%  ↓prec={s['dn_precision']}%")

    tb = agg_summary["triple"]
    passed = tb["dir_accuracy"] >= 31.5 or tb["up_precision"] >= 43.2
    winner = "triple" if (tb["dir_accuracy"] > agg_summary["fixed"]["dir_accuracy"] or
                          tb["up_precision"] > agg_summary["fixed"]["up_precision"]) else "fixed"
    print(f"\n  Winner: {winner}  |  Pass: {'YES' if passed else 'NO'}")

    Path("docs").mkdir(exist_ok=True)
    with open("docs/c3_triple_barrier_result.json", "w") as f:
        json.dump({"results": results, "aggregate": agg_summary,
                   "winner": winner, "passed": passed,
                   "pass_threshold": {"dir_accuracy": 31.5, "up_precision": 43.2}}, f, indent=2)

if __name__ == "__main__":
    main()