News Sentiment Index

START_DAY_UTC   = "2026-03-15"
END_DAY_UTC     = "2026-04-01"   # inclusive
TOPICS          = ["oil", "crypto", "cybersecurity", "semiconductors"]
MAX_WORKERS     = 48
DAILY_LIMIT     = 100            # max articles per topic per day
MIN_TITLE_WORDS = 4
OUTPUT_DIR      = "nsi_output"   # local directory for CSV files
PROGRESS_EVERY  = 96             # print progress every N tiles (~1 day)

import io
import re
import html
import time
import zipfile
import threading
from pathlib import Path
from datetime import datetime, timedelta, timezone
from urllib.parse import urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed

import pandas as pd
import requests
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
import matplotlib.pyplot as plt
import matplotlib.ticker as mticker

Path(OUTPUT_DIR).mkdir(exist_ok=True)
#print(f"Output directory: {Path(OUTPUT_DIR).resolve()}")

TOPIC_CONFIGS = {
    "oil": {
        "anchors": [
            "oil", "crude", "brent", "wti", "opec", "opec+", "eia", "iea",
            "spr", "strategic petroleum reserve", "refinery", "refineries",
            "diesel", "gasoline", "distillate", "inventory", "inventories"
        ],
        "positive_terms": {
            "oil": 2.0, "crude": 2.2, "brent": 2.0, "wti": 2.0,
            "opec": 2.4, "opec+": 2.6, "eia": 2.4, "iea": 2.3,
            "inventory": 1.8, "inventories": 1.8, "refinery": 1.5,
            "refineries": 1.5, "spr": 2.0, "sanctions": 1.5,
            "exports": 1.2, "production": 1.6, "output": 1.4,
            "demand": 1.2, "supply": 1.2, "outage": 1.8,
            "shutdown": 1.7, "attack": 1.5, "pipeline": 1.4, "terminal": 1.3
        },
        "positive_phrases": {
            "oil prices": 1.5, "crude oil": 2.0, "eia inventories": 3.5,
            "opec+": 3.0, "strategic petroleum reserve": 3.0,
            "oil demand": 2.2, "oil supply": 2.2
        },
        "negative_terms": {
            "olive": -5.0, "essential oil": -6.0, "fish oil": -6.0,
            "cooking oil": -6.0, "beauty": -3.0, "cosmetic": -3.0
        },
        "min_score": 3.0
    },
    "crypto": {
        "anchors": [
            "crypto", "cryptocurrency", "cryptocurrencies", "bitcoin", "btc",
            "ethereum", "ether", "eth", "stablecoin", "stablecoins", "token",
            "tokens", "blockchain", "binance", "coinbase", "sec crypto",
            "bitcoin etf", "crypto etf"
        ],
        "positive_terms": {
            "crypto": 2.0, "cryptocurrency": 2.2, "cryptocurrencies": 2.2,
            "bitcoin": 2.8, "btc": 2.4, "ethereum": 2.6, "ether": 2.2,
            "eth": 2.0, "stablecoin": 2.4, "stablecoins": 2.4,
            "token": 1.4, "tokens": 1.4, "blockchain": 1.6,
            "binance": 1.8, "coinbase": 1.8, "etf": 1.2,
            "exchange": 1.1, "sec": 1.0, "regulation": 1.2,
            "hack": 1.2, "custody": 1.1
        },
        "positive_phrases": {
            "bitcoin etf": 3.2, "crypto etf": 2.8, "crypto exchange": 2.0,
            "digital asset": 2.0, "digital assets": 2.0,
            "crypto regulation": 2.6, "bitcoin treasury": 2.0
        },
        "negative_terms": {
            "cryptography": -5.0, "encryption": -2.5,
            "decrypt": -3.0, "cipher": -3.0
        },
        "min_score": 3.0
    },
    "cybersecurity": {
        "anchors": [
            "cyber", "cybersecurity", "cyberattack", "cyberattacks",
            "ransomware", "malware", "data breach", "breach", "breaches",
            "hack", "hacked", "hacking", "ddos", "exploit", "vulnerability",
            "zero-day", "zero day", "phishing", "infosec"
        ],
        "positive_terms": {
            "cyber": 2.0, "cybersecurity": 2.4, "cyberattack": 2.8,
            "cyberattacks": 2.8, "ransomware": 3.0, "malware": 2.5,
            "breach": 2.4, "breaches": 2.4, "hack": 2.0,
            "hacked": 2.0, "hacking": 2.0, "ddos": 2.2,
            "exploit": 1.8, "vulnerability": 1.8, "zero-day": 2.2,
            "phishing": 1.8, "cisa": 1.4, "nsa": 0.8,
            "microsoft": 0.6, "firewall": 1.0
        },
        "positive_phrases": {
            "data breach": 3.2, "critical vulnerability": 2.8,
            "zero day": 3.0, "cyber attack": 3.0, "cyber attacks": 3.0,
            "ransomware attack": 3.2, "security breach": 2.8
        },
        "negative_terms": {
            "cybertruck": -6.0, "video game": -2.5, "gaming": -2.0
        },
        "min_score": 3.0
    },
    "semiconductors": {
        "anchors": [
            "semiconductor", "semiconductors", "chip", "chips", "chipmaker",
            "chipmakers", "foundry", "foundries", "fab", "fabs", "wafer",
            "wafers", "tsmc", "intel", "nvidia", "amd", "asml", "hbm", "dram",
            "memory chip", "memory chips", "lithography"
        ],
        "positive_terms": {
            "semiconductor": 2.8, "semiconductors": 2.8, "chip": 1.8,
            "chips": 1.8, "chipmaker": 2.0, "chipmakers": 2.0,
            "foundry": 2.2, "foundries": 2.2, "fab": 1.8, "fabs": 1.8,
            "wafer": 1.5, "wafers": 1.5, "tsmc": 2.2, "intel": 1.5,
            "nvidia": 1.6, "amd": 1.4, "asml": 1.8, "hbm": 1.8,
            "dram": 1.8, "memory": 1.0, "lithography": 2.0,
            "export controls": 1.8, "capacity": 1.2
        },
        "positive_phrases": {
            "semiconductor industry": 3.0, "memory chip": 2.6,
            "memory chips": 2.6, "chip export": 2.2, "chip exports": 2.2,
            "chip demand": 2.2, "chip supply": 2.2
        },
        "negative_terms": {
            "potato chips": -7.0, "casino chips": -7.0,
            "tortilla chips": -7.0, "fish and chips": -7.0
        },
        "min_score": 3.0
    }
}

GDELT_V2_BASE_URL = "http://data.gdeltproject.org/gdeltv2"
GKG_USECOLS   = [0, 1, 3, 4, 7, 8, 15, 26]
GKG_COL_NAMES = ["GKGRECORDID", "DATE", "SourceCommonName", "DocumentIdentifier",
                  "Themes", "V2Themes", "V2Tone", "Extras"]
THREAD_LOCAL  = threading.local()

def _compile_topics(configs):
    out = {}
    for tid, cfg in configs.items():
        anchors = [re.escape(str(x).lower()) for x in cfg["anchors"]]
        out[tid] = {
            "anchor_re": re.compile(
                r"(?:" + "|".join(sorted(anchors, key=len, reverse=True)) + r")",
                flags=re.IGNORECASE
            ),
            "phrase_items":   [(str(p).lower(), float(w)) for p, w in cfg["positive_phrases"].items()],
            "positive_items": [(str(k).lower(), float(v)) for k, v in cfg["positive_terms"].items()],
            "negative_items": [(str(k).lower(), float(v)) for k, v in cfg["negative_terms"].items()],
            "min_score":      float(cfg["min_score"]),
        }
    return out

PREPARED = _compile_topics(TOPIC_CONFIGS)

def _utc_midnight(day_str):
    d = datetime.strptime(day_str, "%Y-%m-%d")
    return d.replace(tzinfo=timezone.utc)

def _iter_tiles(start_utc, end_exclusive_utc):
    cur = start_utc.replace(minute=(start_utc.minute // 15) * 15, second=0, microsecond=0)
    while cur < end_exclusive_utc:
        yield cur
        cur += timedelta(minutes=15)

def _session():
    s = getattr(THREAD_LOCAL, "session", None)
    if s:
        return s
    s = requests.Session()
    retry = Retry(total=3, backoff_factor=1,
                  status_forcelist=[429, 500, 502, 503, 504],
                  allowed_methods=["GET"])
    s.mount("http://", HTTPAdapter(max_retries=retry))
    s.mount("https://", HTTPAdapter(max_retries=retry))
    THREAD_LOCAL.session = s
    return s

def _norm_domain(val):
    s = str(val or "").strip().lower()
    if "://" in s:
        try:
            s = urlparse(s).netloc
        except Exception:
            pass
    if ":" in s:
        s = s.split(":", 1)[0]
    return s.lstrip("www.").strip(".")

def _extract_tag(extras, tag):
    m = re.search(fr"<{tag}>(.*?)</{tag}>", str(extras or ""), re.DOTALL)
    return m.group(1).strip() if m else ""

def _parse_ts(raw):
    digits = re.sub(r"\D", "", str(raw or ""))
    if len(digits) >= 14:
        try:
            return pd.to_datetime(digits[:14], format="%Y%m%d%H%M%S", utc=True)
        except Exception:
            pass
    return pd.NaT

def _tone_first(val):
    try:
        return float(str(val or "").split(",", 1)[0])
    except Exception:
        return float("nan")

def _score(text, topic_id):
    prep = PREPARED[topic_id]
    t = str(text or "").lower()
    if not prep["anchor_re"].search(t):
        return None
    score = sum(w for k, w in prep["positive_items"] if k in t)
    score += sum(w * 1.2 for p, w in prep["phrase_items"] if p in t)
    score += sum(w for k, w in prep["negative_items"] if k in t)
    if any(c.isdigit() for c in t):
        score += 0.3
    return float(score) if score >= prep["min_score"] else None

def _process_tile(ts_utc, start_utc, end_excl_utc, topics):
    url = f"{GDELT_V2_BASE_URL}/{ts_utc.strftime('%Y%m%d%H%M%S')}.gkg.csv.zip"
    try:
        resp = _session().get(url, timeout=20)
        if resp.status_code != 200:
            return {}
        with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
            with zf.open(zf.namelist()[0]) as f:
                raw = pd.read_csv(f, sep="\t", header=None, names=GKG_COL_NAMES,
                                  usecols=GKG_USECOLS, low_memory=False, encoding="latin-1")
    except Exception:
        return {}

    if raw is None or raw.empty:
        return {}

    df = raw.copy()
    df["gdelt_ts"] = pd.to_datetime(
        df["DATE"].fillna("").astype(str).str.split(".").str[0],
        format="%Y%m%d%H%M%S", errors="coerce", utc=True
    )
    df = df.dropna(subset=["gdelt_ts"])
    if df.empty:
        return {}

    df["Extras"] = df["Extras"].fillna("").astype(str)
    def _extract_both(extras):
        title = html.unescape(_extract_tag(extras, "PAGE_TITLE")).strip()
        pub   = _extract_tag(extras, "PAGE_PRECISEPUBTIMESTAMP")
        return title, pub
    extracted        = df["Extras"].map(_extract_both)
    df["title"]      = extracted.map(lambda x: x[0])
    df["pub_raw"]    = extracted.map(lambda x: x[1])

    df = df[df["title"].str.len() > 0].copy()
    df = df[df["title"].str.split().str.len() >= MIN_TITLE_WORDS].copy()
    if df.empty:
        return {}

    df["pub_ts"]  = df["pub_raw"].map(_parse_ts)
    df["use_ts"]  = df["pub_ts"].where(df["pub_ts"].notna(), df["gdelt_ts"])
    df["use_ts"]  = pd.to_datetime(df["use_ts"], errors="coerce", utc=True)
    df = df[df["use_ts"].notna()].copy()
    mask = (df["use_ts"] >= start_utc) & (df["use_ts"] < end_excl_utc)
    df = df[mask].copy()
    if df.empty:
        return {}

    df["day_key"] = df["use_ts"].dt.strftime("%Y-%m-%d")
    df["source"]  = df["SourceCommonName"].fillna("").astype(str).map(_norm_domain)
    df["url_norm"]   = df["DocumentIdentifier"].fillna("").astype(str).str.strip().str.lower()
    df["title_norm"] = df["title"].str.strip().str.lower()
    df["V2Tone"]  = df["V2Tone"].fillna("").astype(str)
    df["combined"] = (
        df["title_norm"] + " " +
        df["Themes"].fillna("").astype(str).str.lower() + " " +
        df["V2Themes"].fillna("").astype(str).str.lower()
    )

    result = {}
    for tid in topics:
        anchor_mask = df["combined"].str.contains(PREPARED[tid]["anchor_re"], na=False)
        sub = df[anchor_mask].copy()
        if sub.empty:
            continue
        sub["score"] = sub["combined"].map(lambda x: _score(x, tid))
        sub = sub[sub["score"].notna()].copy()
        if sub.empty:
            continue
        result[tid] = sub[[
            "day_key", "use_ts", "source",
            "url_norm", "title_norm", "title",
            "V2Tone", "score"
        ]].to_dict("records")
    return result

def _build_nsi(accumulator, tid, daily_limit):
    """Deduplicate by url+title, apply daily limit with score+time sort, compute NSI."""
    day_frames = []
    for dk, rows in accumulator[tid].items():
        df = pd.DataFrame(rows)
        # url_norm and title_norm already lowercased in _process_tile
        df["use_dt_sort"] = pd.to_datetime(df["use_ts"], errors="coerce", utc=True)
        df = df.sort_values(["score", "use_dt_sort"], ascending=[False, False])
        df = df.drop_duplicates(subset=["url_norm"],   keep="first")
        df = df.drop_duplicates(subset=["title_norm"], keep="first")
        day_frames.append(df.head(daily_limit))

    if not day_frames:
        return pd.DataFrame()

    df = pd.concat(day_frames, ignore_index=True)
    df["tone"] = df["V2Tone"].map(_tone_first)
    df = df[df["tone"].notna()].copy()

    daily = (
        df.groupby("day_key", as_index=False)
        .agg(
            article_count=("tone", "size"),
            positive_count=("tone", lambda s: int((s > 0).sum())),
            negative_count=("tone", lambda s: int((s < 0).sum())),
            mean_tone=("tone", "mean"),
            median_tone=("tone", "median"),
        )
        .sort_values("day_key")
        .reset_index(drop=True)
    )

    denom = daily["positive_count"] + daily["negative_count"]
    daily["nsi"] = 0.0
    valid = denom > 0
    daily.loc[valid, "nsi"] = (
        (daily.loc[valid, "positive_count"] - daily.loc[valid, "negative_count"]) / denom[valid]
    )

    daily["nsi_7d"]       = daily["nsi"].rolling(7, min_periods=1).mean()
    daily["mean_tone_7d"] = daily["mean_tone"].rolling(7, min_periods=1).mean()
    daily["topic_id"]     = tid
    return daily

PALETTE = {
    "bar": "#c8cdd6",
    "line": "#2563eb",
    "bg": "#ffffff",
    "grid": "#eef0f7",
    "text": "#0f172a",
    "text_light": "#64748b",
    "border": "#e2e8f0",
}

plt.rcParams.update({
    "font.family": "sans-serif",
    "font.sans-serif": ["DejaVu Sans", "Liberation Sans", "Arial"],
    "font.size": 11,
    "axes.facecolor": PALETTE["bg"],
    "figure.facecolor": PALETTE["bg"],
    "axes.edgecolor": PALETTE["border"],
    "axes.linewidth": 0.6,
    "axes.grid": True,
    "grid.color": PALETTE["grid"],
    "grid.linewidth": 0.5,
    "xtick.color": PALETTE["text_light"],
    "ytick.color": PALETTE["text_light"],
    "xtick.labelsize": 10,
    "ytick.labelsize": 10,
    "axes.spines.top": False,
    "axes.spines.right": False,
})