Spaces:
Sleeping
Sleeping
File size: 5,630 Bytes
3404590 | 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 | #!/usr/bin/env python3
"""C7 validation: BASELINE vs BASELINE + calendar features + 36-month lookback."""
import sys
import json
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
from scripts.improvement_harness import (
BASELINE_FEATURES,
build_triple_barrier_labels,
compute_metrics,
walk_forward,
RF_PARAMS,
DEFAULT_STOCKS,
)
CALENDAR_FEATURES = [
"dow_sin", "dow_cos",
"month_sin", "month_cos",
"is_options_expiry_week",
"is_earnings_season",
]
C7_FEATURES = BASELINE_FEATURES + CALENDAR_FEATURES
def fetch_df_36(stock_no: str):
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=36)
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 _mean(rows, field):
vals = [r[field] for r in rows
if isinstance(r.get(field), (int, float)) and not np.isnan(r.get(field, float("nan")))]
return round(sum(vals) / len(vals), 1) if vals else float("nan")
def main():
from models.predictor import _build_features
feature_sets = [("baseline", BASELINE_FEATURES), ("calendar", C7_FEATURES)]
per_stock = {}
agg = {"baseline": [], "calendar": []}
hdr = f"{'Stock':>6} {'Set':>12} {'Acc%':>5} {'Dir%':>5} {'↑Prec%':>7} {'Signals':>7}"
print(f"\n{hdr}\n{'-' * len(hdr)}")
for stock_no in DEFAULT_STOCKS:
print(f" computing {stock_no}...", end="\r", flush=True)
df = fetch_df_36(stock_no)
if df is None or df.empty:
print(f"{stock_no:>6} no data")
continue
feat = _build_features(df)
# Merge calendar columns from df into feat so walk_forward can use them
for col in CALENDAR_FEATURES:
if col in df.columns:
feat[col] = df[col].values
close = (df.set_index("date")["close"] if "date" in df.columns else df["close"]).values
labels = build_triple_barrier_labels(close)
per_stock[stock_no] = {}
for name, cols in feature_sets:
r = walk_forward(feat, labels, cols)
per_stock[stock_no][name] = r
if r:
agg[name].append(r)
prefix = f"{stock_no:>6}" if name == "baseline" else f"{'':>6}"
print(f"{prefix} {name:>12} {r['accuracy']:>5.1f} {r['dir_accuracy']:>5.1f} "
f"{r['up_precision']:>7.1f} {r.get('n_signals', 0):>7}")
r0 = per_stock[stock_no].get("baseline", {})
r1 = per_stock[stock_no].get("calendar", {})
if r0 and r1:
dd = r1["dir_accuracy"] - r0["dir_accuracy"]
dp = r1["up_precision"] - r0["up_precision"]
print(f"{'':>6} {'Δ':>12} {'':>5} {dd:>+5.1f} {dp:>+7.1f}")
print()
print("=== AGGREGATE ===")
agg_summary = {}
for name, rows in agg.items():
s = {k: _mean(rows, k) for k in ("accuracy", "dir_accuracy", "up_precision", "dn_precision", "n_signals")}
agg_summary[name] = s
print(f" {name:>12}: acc={s['accuracy']}% dir={s['dir_accuracy']}% "
f"↑prec={s['up_precision']}% signals={s['n_signals']}")
PASS_DIR_ACC = 43.0
PASS_UP_PREC = 54.0
cal = agg_summary.get("calendar", {})
bas = agg_summary.get("baseline", {})
cal_dir = cal.get("dir_accuracy", 0)
cal_prec = cal.get("up_precision", 0)
bas_dir = bas.get("dir_accuracy", 0)
bas_prec = bas.get("up_precision", 0)
meets_threshold = (cal_dir >= PASS_DIR_ACC) or (cal_prec >= PASS_UP_PREC)
no_regression = (
(cal_dir >= bas_dir - 1.0) and
(cal_prec >= bas_prec - 1.0)
)
passed = meets_threshold and no_regression
print(f"\n calendar dir={cal_dir}% ↑prec={cal_prec}%")
print(f" Pass (dir≥{PASS_DIR_ACC} OR ↑prec≥{PASS_UP_PREC}, no regression>1pp): {'YES' if passed else 'NO'}")
result = {
"results": per_stock,
"aggregate": agg_summary,
"passed": passed,
"pass_criterion": f"dir_accuracy >= {PASS_DIR_ACC} OR up_precision >= {PASS_UP_PREC}",
}
def _jsonify(obj):
if isinstance(obj, dict):
return {k: _jsonify(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_jsonify(v) for v in obj]
if isinstance(obj, np.integer):
return int(obj)
if isinstance(obj, np.floating):
return float(obj)
if isinstance(obj, np.bool_):
return bool(obj)
return obj
out = ROOT / "docs" / "c7_calendar_result.json"
out.parent.mkdir(exist_ok=True)
with open(out, "w") as f:
json.dump(_jsonify(result), f, indent=2)
print(f"\n Written: {out}")
if __name__ == "__main__":
main()
|