#!/usr/bin/env python3
"""News-sentiment test with a real small LLM (FinBERT) on open news data (FNSPID).

Stream a chunk of FNSPID headlines -> FinBERT sentiment -> daily sentiment per ticker -> test if
it predicts returns. KEY: same-day corr (contemporaneous, not tradeable) vs next-day corr (the only
retail-tradeable horizon) + a net-of-cost OOS long/short trade. Honest prior (research): news real
but reaction instant + alpha dies at cost + decaying."""
import csv
import datetime
import io
import json
import math
import os
import sys
import urllib.request

import numpy as np

CHUNK = 70 * 1024 * 1024   # ~70MB stream (sorted by ticker -> early-alphabet large caps)
MAX_SCORE = 18000          # cap FinBERT inferences (GPU time)


def stream_news():
    req = urllib.request.Request("https://huggingface.co/datasets/Zihan1004/FNSPID/resolve/main/Stock_news/nasdaq_exteral_data.csv",
                                 headers={"User-Agent": "curl/8", "Range": f"bytes=0-{CHUNK}"})
    buf = urllib.request.urlopen(req, timeout=120).read().decode("utf-8", "replace")
    buf = buf[:buf.rfind("\n")]                                # drop partial last line
    rows = []
    for r in csv.reader(io.StringIO(buf)):
        if len(r) < 4 or r[1] == "Date":
            continue
        try:
            d = r[1][:10]; sym = r[3].strip(); title = r[2].strip()
            datetime.date.fromisoformat(d)
            if sym and title and len(title) > 8:
                rows.append((d, sym, title))
        except Exception:
            continue
    return rows


def yclose(sym):
    try:
        res = json.loads(urllib.request.urlopen(urllib.request.Request(
            f"https://query1.finance.yahoo.com/v8/finance/chart/{sym}?interval=1d&range=5y",
            headers={"User-Agent": "Mozilla/5.0"}), timeout=30).read())["chart"]["result"][0]
        ts = res["timestamp"]; adj = res["indicators"].get("adjclose", [{}])[0].get("adjclose") or res["indicators"]["quote"][0]["close"]
        return {datetime.datetime.utcfromtimestamp(t).strftime("%Y-%m-%d"): float(c) for t, c in zip(ts, adj) if c is not None}
    except Exception:
        return {}


def main():
    print("streaming FNSPID news chunk...")
    rows = stream_news()
    syms = {}
    for d, s, t in rows:
        syms.setdefault(s, 0); syms[s] += 1
    top = [s for s, n in sorted(syms.items(), key=lambda kv: -kv[1]) if n >= 50][:12]
    rows = [r for r in rows if r[1] in top][:MAX_SCORE]
    print(f"  {len(rows)} headlines, {len(top)} tickers: {top}")

    from transformers import pipeline
    import torch
    clf = pipeline("sentiment-analysis", model="ProsusAI/finbert", device=0 if torch.cuda.is_available() else -1, truncation=True, max_length=64)
    titles = [r[2][:200] for r in rows]
    print("  scoring with FinBERT...")
    out = clf(titles, batch_size=64)
    score = {"positive": 1.0, "negative": -1.0, "neutral": 0.0}
    sent = {}  # (sym, date) -> [scores]
    for r, o in zip(rows, out):
        sent.setdefault((r[1], r[0]), []).append(score[o["label"]] * o["score"])
    daily = {k: float(np.mean(v)) for k, v in sent.items()}

    prices = {s: yclose(s) for s in top}
    # build aligned (sentiment[t], sameday ret[t], nextday ret[t+1]) cross-sectionally
    S, R0, R1 = [], [], []
    rec = []
    for (sym, d), sc in daily.items():
        px = prices.get(sym, {})
        days = sorted(px)
        if d not in px:
            # map to next trading day
            nd = [x for x in days if x >= d]
            if not nd:
                continue
            d = nd[0]
        i = days.index(d) if d in days else -1
        if i < 1 or i + 1 >= len(days):
            continue
        r0 = px[days[i]] / px[days[i - 1]] - 1
        r1 = px[days[i + 1]] / px[days[i]] - 1
        rec.append((d, sym, sc, r0, r1))
    rec.sort()
    S = np.array([x[2] for x in rec]); R0 = np.array([x[3] for x in rec]); R1 = np.array([x[4] for x in rec])
    n = len(S)
    print(f"\n  aligned sentiment-day observations: {n}")
    print(f"  corr(sentiment, SAME-day return):  {np.corrcoef(S, R0)[0,1]:+.3f}  (contemporaneous = news real but not tradeable)")
    print(f"  corr(sentiment, NEXT-day return):  {np.corrcoef(S, R1)[0,1]:+.3f}  (the only retail-tradeable horizon)")
    # next-day long/short trade by sentiment sign, net 10bp, IS/OOS by time
    order = np.argsort([x[0] for x in rec]); S, R1 = S[order], R1[order]
    sig = np.sign(S); pnl = sig * R1 - 0.0010 * (np.abs(sig) > 0)
    sp = int(0.6 * n)
    def sh(x):
        x = x[np.isfinite(x)]; return x.mean() / (x.std() + 1e-9) * math.sqrt(252) if len(x) > 20 and x.std() > 0 else float("nan")
    print(f"  next-day sentiment trade (net 10bp): IS Sharpe {sh(pnl[:sp]):+.2f}  OOS Sharpe {sh(pnl[sp:]):+.2f}")
    print("\n  VERDICT: same-day corr >> next-day corr ~0 + OOS trade ~0/neg = news is real but its reaction is")
    print("  INSTANT (contemporaneous, HFT-captured); no tradeable next-day drift for retail. Matches the research.")


if __name__ == "__main__":
    main()