scripts/backtest_c28.py

#!/usr/bin/env python3
"""C28: News sentiment features (news_score_1d, news_score_5d_ma).

Adds daily news sentiment from the pre-built history cache
(data/news_sentiment_history.json) as ML training features.

Must run scripts/build_news_history.py first to populate the cache.

Features added over CURRENT_FEATURES:
  news_score_1d    — prior-day sentiment score [-1, 1] (non-leaking)
  news_score_5d_ma — 5-day rolling mean of sentiment

Hypothesis: News sentiment provides leading signal beyond price/volume —
positive news precedes institutional buying, negative news precedes selling.
"""
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

import warnings; warnings.filterwarnings("ignore")
import json
import argparse
import numpy as np

from scripts.improvement_harness import (
    fetch_df, walk_forward, build_triple_barrier_labels,
    CURRENT_FEATURES, DEFAULT_STOCKS, EXTENDED_STOCKS,
    PASS_DIR_ACC, PASS_UP_PREC,
)
from models.predictor import _build_features
from data.news_history import add_news_sentiment, _load_cache

# Stock name map — needed for add_news_sentiment cache lookup
STOCK_NAMES = {
    "2330": "台積電", "0050": "元大台灣50", "2317": "鴻海",
    "2454": "聯發科", "2881": "富邦金",   "2303": "聯電",
    "2382": "廣達",   "2308": "台達電",   "2882": "國泰金",
    "2886": "兆豐金", "0056": "元大高股息", "2412": "中華電",
}

C28_FEATURES = ["news_score_1d", "news_score_5d_ma"]
NEW_FEATURES  = CURRENT_FEATURES + C28_FEATURES


def load_stock(stock_no: str, news_cache: dict):
    df = fetch_df(stock_no)
    if df is None or df.empty:
        return None, None

    df   = add_news_sentiment(df, stock_no, cache=news_cache)
    feat = _build_features(df)

    # Append news features (not in _build_features output)
    for col in C28_FEATURES:
        feat[col] = df.get(col, 0.0)

    close  = (df.set_index("date")["close"] if "date" in df.columns else df["close"]).values
    labels = build_triple_barrier_labels(close)
    return feat, labels


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--extended", action="store_true")
    args = parser.parse_args()

    stocks = EXTENDED_STOCKS if args.extended else DEFAULT_STOCKS
    tag    = "12-stock" if args.extended else "5-stock"
    suffix = "_12stock" if args.extended else ""

    print(f"\n=== C28: News sentiment features [{tag}] ===")

    news_cache = _load_cache()
    # Summarise cache coverage
    covered = {s for s in stocks if s in news_cache and len(news_cache[s]) > 4}
    print(f"  Cache: {len(news_cache)} stocks total, {len(covered)}/{len(stocks)} eval stocks covered")
    if not covered:
        print("\n  ERROR: news_sentiment_history.json is empty.")
        print("  Run:  venv/bin/python scripts/build_news_history.py --months 24")
        return 1

    print(f"  C28 features: {C28_FEATURES}\n")

    per_stock = {}
    base_results, c28_results = [], []

    for stock_no in stocks:
        print(f"  {stock_no}...", end=" ", flush=True)
        feat, labels = load_stock(stock_no, news_cache)
        if feat is None:
            print("no data")
            continue

        n_news = (feat.get("news_score_1d", 0) != 0).sum() if "news_score_1d" in feat.columns else 0
        m_base = walk_forward(feat, labels, CURRENT_FEATURES)
        m_c28  = walk_forward(feat, labels, NEW_FEATURES)

        per_stock[stock_no] = {"baseline": m_base, "c28": m_c28, "news_days": int(n_news)}
        base_results.append(m_base)
        c28_results.append(m_c28)

        b_dir = m_base.get("dir_accuracy", float("nan"))
        b_up  = m_base.get("up_precision", float("nan"))
        c_dir = m_c28.get("dir_accuracy",  float("nan"))
        c_up  = m_c28.get("up_precision",  float("nan"))
        print(f"news_days={n_news}  base dir={b_dir}% ↑prec={b_up}%  →  c28 dir={c_dir}% ↑prec={c_up}%")

    def _avg(results, key):
        vals = [m[key] for m in results if m and not np.isnan(m.get(key, float("nan")))]
        return round(float(np.mean(vals)), 1) if vals else float("nan")

    base_agg = {"dir_accuracy": _avg(base_results, "dir_accuracy"),
                "up_precision": _avg(base_results, "up_precision")}
    c28_agg  = {"dir_accuracy": _avg(c28_results,  "dir_accuracy"),
                "up_precision": _avg(c28_results,  "up_precision")}

    passed = c28_agg["dir_accuracy"] >= PASS_DIR_ACC and c28_agg["up_precision"] >= PASS_UP_PREC

    print(f"\n  Baseline avg: dir={base_agg['dir_accuracy']}%  ↑prec={base_agg['up_precision']}%")
    print(f"  C28 avg:      dir={c28_agg['dir_accuracy']}%   ↑prec={c28_agg['up_precision']}%")
    print(f"  Gate (dir≥{PASS_DIR_ACC}% AND ↑prec≥{PASS_UP_PREC}%): {'PASS ✓' if passed else 'FAIL ✗'}")

    result = {
        "experiment":   "C28",
        "description":  "News sentiment features from local LLM scoring",
        "new_features": C28_FEATURES,
        "stocks":       stocks,
        "aggregate":    {"baseline": base_agg, "c28": c28_agg},
        "passed":       passed,
        "pass_gate":    {"dir_accuracy": PASS_DIR_ACC, "up_precision": PASS_UP_PREC},
        "per_stock":    per_stock,
    }

    out = ROOT / f"docs/c28_result{suffix}.json"
    out.parent.mkdir(exist_ok=True)
    out.write_text(json.dumps(result, indent=2))
    print(f"\n  Saved: {out}")
    return 0 if passed else 1


if __name__ == "__main__":
    sys.exit(main())