"""All tables and figures in the paper. Reads panel.pkl from load.py, writes results.json."""
import json, pickle, re
import numpy as np, pandas as pd, statsmodels.api as sm
from load import *

P = pickle.load(open("panel.pkl", "rb"))
c, close, opn, vol, ret, ar, sig, dv, pc, spy, nc, npr, sessions = [P[k] for k in "c close opn vol ret ar sig dv pc spy nc npr sessions".split()]
T = len(sessions)
tix = {t: i for i, t in enumerate(sessions)}
start_i = int(np.searchsorted(sessions, SAMPLE_START))
common = set(c.loc[c.common, "ticker"]) & set(close.columns)
common.discard("SPY")
syms = sorted(common)
print("common syms", len(syms), "sessions", T - start_i)

closeA = close[syms].values; openA = opn[syms].values; volA = vol[syms].values
retA = ret[syms].values; arA = ar[syms].values; sigA = sig[syms].values; dvA = dv[syms].values; pcA = pc[syms].values
spyr = spy.values; spyc = close["SPY"].values; spyo = opn["SPY"].values
six = {s: i for i, s in enumerate(syms)}
# A corporate action the feed missed looks like a huge move on ordinary volume: spinoffs, ADR ratio changes,
# acquisition closings. Blank those returns everywhere. Rule fixed after reading the largest events by hand.
_vm0 = pd.DataFrame(volA).rolling(20, min_periods=10).median().shift(1).values
_art = (np.abs(arA) >= 0.30) & (volA / _vm0 < 2.0)
_manual = [("CTVA", "2026-10-01"), ("RNA", "2026-02-27"), ("AZN", "2026-02-02")]
for _s, _dt in _manual:
    if _s in six and _dt in tix: _art[tix[_dt], six[_s]] = True
ARTIFACTS = int(np.nansum(_art))
retA = np.where(_art, np.nan, retA); arA = np.where(_art, np.nan, arA)
# Forward windows compound the cleaned daily returns, so a split the feed missed cannot enter them.
_alive = pd.DataFrame(closeA).ffill(limit=3).notna().values
closeF = np.cumprod(1 + np.nan_to_num(retA), axis=0).astype(float)
closeF[~_alive] = np.nan

def bhar(i, t, h0, h1):
    """Buy-and-hold return from close t+h0 to close t+h1, net of SPY. NaN when off the panel."""
    a, b = t + h0, t + h1
    ok = (a >= 0) & (b < T)
    out = np.full(len(i), np.nan)
    ii, aa, bb = i[ok], a[ok], b[ok]
    out[ok] = closeF[bb, ii] / closeF[aa, ii] - spyc[bb] / spyc[aa]
    return out

# ---------- news flags on the stock-session grid
anyitem = np.zeros(arA.shape, bool)
wire = np.zeros(arA.shape, int)
_ck = pd.read_csv("data/cik.tsv", sep="\t", header=None, names=["id", "ticker", "cik"], dtype={"cik": str})
_ck["ticker"] = _ck.ticker.str.strip().str.upper()
_ck = _ck[_ck.cik.str.strip("0") != ""]
_grp = _ck.groupby("cik").ticker.apply(lambda v: sorted(set(v)))
_multi = _grp[_grp.map(len) > 1]
ALIAS = {t: [o for o in g if o in common] for g in _multi for t in g}
print("firms with more than one ticker", len(_multi))
def _expand(df):
    extra = []
    hit = df[df.ticker.isin(ALIAS)]
    for t_, g_ in hit.groupby("ticker"):
        for o in ALIAS[t_]:
            if o != t_: extra.append(g_.assign(ticker=o))
    return pd.concat([df] + extra, ignore_index=True) if extra else df
ncs = _expand(nc); ncs = ncs[ncs.ticker.isin(common)]
np.add.at(anyitem, (ncs.si.values, ncs.ticker.map(six).values), True)
e = _expand(npr[~npr.law & (npr.si >= 0)]); e = e[e.ticker.isin(common)].copy()
e["i"] = e.ticker.map(six)
np.add.at(wire, (e.si.values, e.i.values), 1)

# primary catalyst per stock-session: named beats other, pre-market beats after-hours beats in-session
order = {0: 0, 2: 1, 1: 2}
e["named"] = (e.catalyst == "other").astype(int)
e["to"] = e.timing.map(order)
prim = e.sort_values(["si", "i", "named", "to", "et"]).drop_duplicates(["si", "i"])

rows = []
tt, ii = np.where(~np.isnan(arA))
d = pd.DataFrame({"t": tt, "i": ii})
d = d[(d.t >= start_i)]
d["ar"] = arA[d.t, d.i]; d["sig"] = sigA[d.t, d.i]; d["dv"] = dvA[d.t, d.i]; d["pc"] = pcA[d.t, d.i]
d["wire"] = wire[d.t, d.i]; d["any"] = anyitem[d.t, d.i]
d = d[(d.pc >= MIN_PRICE) & (d.dv >= MIN_DV) & d.sig.notna()].reset_index(drop=True)
d["z"] = d.ar / d.sig
d["news"] = d.wire > 0
d["quiet"] = ~d["any"]
nxt = np.zeros(arA.shape, bool); nxt[:-1] = anyitem[1:]
d["quiet2"] = d.quiet & ~nxt[d.t, d.i]
d["shock"] = (d.ar.abs() >= SHOCK_ABS) & (d.z.abs() >= SHOCK_Z)
d["up"] = d.ar > 0
ti, ia = d.t.values, d.i.values
for name, h0, h1 in [("r1", 0, 1), ("r5", 0, 5), ("r10", 0, 10), ("r20", 0, 20), ("pre5", -6, -1), ("pre20", -21, -1), ("pre1", -2, -1)]:
    d[name] = bhar(ia, ti, h0, h1)
d["gap"] = openA[ti, ia] / pcA[ti, ia] - 1 - (spyo[ti] / spyc[ti - 1] - 1)
d["oc"] = closeA[ti, ia] / openA[ti, ia] - 1 - (spyc[ti] / spyo[ti] - 1)
vm = pd.DataFrame(volA).rolling(20, min_periods=10).median().shift(1).values
d["vr"] = volA[ti, ia] / vm[ti, ia]
d["date"] = sessions[ti]
d["sym"] = np.array(syms)[ia]
p = prim.set_index(["si", "i"])
key = pd.MultiIndex.from_arrays([d.t, d.i])
d["catalyst"] = p["catalyst"].reindex(key).values
d["timing"] = p["timing"].reindex(key).values
d["nid"] = p["newsitem_id"].reindex(key).values
d["title"] = p["title"].reindex(key).values
d["liq"] = pd.qcut(d.dv, 3, labels=["low", "mid", "high"])
# Same-day controls: the non-shock average for the same liquidity and volatility tercile on the same date.
d["vt"] = d.groupby("t").sig.transform(lambda v: pd.qcut(v.rank(method="first"), 3, labels=False))
d["cell"] = d.t.values * 100 + d.liq.cat.codes.values * 10 + d.vt.values.astype(int)
_ctrl = ~d.shock
def _bench(v):
    v = pd.Series(np.asarray(v, float), index=d.index)
    lo_, hi_ = v.quantile([.005, .995])
    return v.clip(lo_, hi_)[_ctrl].groupby(d.cell[_ctrl]).mean()
for _h in ("r1", "r5", "r10", "r20", "pre5"):
    d[_h + "_spy"] = d[_h]
    d[_h] = d[_h] - d.cell.map(_bench(d[_h]))
print(len(d), "stock sessions", d.news.sum(), "news", d.quiet.sum(), "quiet", d.shock.sum(), "shocks")

def mean_se(y, t, i, wins=True):
    y = np.asarray(y, float); ok = ~np.isnan(y)
    y, t, i = y[ok], np.asarray(t)[ok], np.asarray(i)[ok]
    n = len(y)
    if n < 50: return dict(n=int(n), mean=None, med=None, se=None, t=None)
    if wins:
        lo, hi = np.percentile(y, [0.5, 99.5]); yw = np.clip(y, lo, hi)
    else: yw = y
    res = sm.OLS(yw, np.ones(n)).fit(cov_type="cluster", cov_kwds={"groups": np.column_stack([t, i])})
    return dict(n=int(n), mean=float(res.params[0]), med=float(np.median(y)), se=float(res.bse[0]), t=float(res.tvalues[0]), raw=float(y.mean()))

def ms(sub, col): return mean_se(sub[col].values, sub.t.values, sub.i.values)

R = {}
R["sample"] = dict(start=sessions[start_i], end=sessions[-1], sessions=int(T - start_i), symbols=int(d.i.nunique()), stock_sessions=int(len(d)),
    raw_items=int(((nc.si >= start_i)).sum()), wire_items_npr=int((npr.si >= start_i).sum()), law_removed=int(npr.law.sum()),
    news_sessions=int(d.news.sum()), quiet_sessions=int(d.quiet.sum()), shock_sessions=int(d.shock.sum()), news_share=float(d.news.mean()), quiet_share=float(d.quiet.mean()), med_dv=float(d.dv.median()),
    any_item_not_wire=int(((~d.quiet) & (~d.news)).sum()))

# ---------- A. arrival clock
w = nc[(nc.si >= start_i) & (nc.etd <= SAMPLE_END)].copy()
w["kind"] = np.where(w.parser == 6, "sec", "wire")
w["bucket"] = (w.mins // 30) * 30
dow = w.et.dt.dayofweek
clock = w[dow < 5].groupby(["kind", "bucket"]).size().unstack(0).fillna(0).astype(int)
R["clock"] = [dict(min=int(k), wire=int(v.get("wire", 0)), sec=int(v.get("sec", 0))) for k, v in clock.iterrows()]
for kd in ["wire", "sec"]:
    x = w[(w.kind == kd) & (dow < 5)]
    R[f"clock_{kd}"] = dict(n=int(len(x)), pre=float((x.mins < 570).mean()), reg=float(((x.mins >= 570) & (x.mins < 960)).mean()), post=float((x.mins >= 960).mean()))
wd = w.groupby(["kind", dow]).size().unstack(0).fillna(0).astype(int)
R["weekday"] = [dict(d=int(k), wire=int(v.get("wire", 0)), sec=int(v.get("sec", 0))) for k, v in wd.iterrows()]
top = w[w.kind == "wire"].groupby(["etd", "mins"]).size()
wm = w[(w.kind == "wire") & (dow < 5)].mins
R["wire_top_minutes"] = [dict(min=int(k), n=int(v)) for k, v in wm.value_counts().head(8).items()]
R["wire_on_half_hour"] = float((wm % 30 == 0).mean())
h = pd.read_csv("data/hist.tsv", sep="\t")
R["by_year"] = [dict(y=int(k), wire=int(v.get("wire", 0)), sec=int(v.get("sec", 0))) for k, v in h.assign(kind=np.where(h.parser == 6, "sec", "wire")).groupby(["y", "kind"]).n.sum().unstack(1).fillna(0).astype(int).iterrows()]

# ---------- B. what news days look like
cats = ["earnings", "clinical", "mna", "guidance", "offering", "analyst", "contract", "listing", "legal", "leadership", "capital", "other"]
q = d[d.quiet]; qabs = float(q.ar.abs().mean()); qmed = float(q.ar.abs().median())
R["baseline"] = dict(quiet_abs=qabs, quiet_abs_med=qmed, news_abs=float(d[d.news].ar.abs().mean()), news_abs_med=float(d[d.news].ar.abs().median()), quiet_gt5=float((q.ar.abs() >= .05).mean()), news_gt5=float((d[d.news].ar.abs() >= .05).mean()), quiet_vr=float(q.vr.median()), news_vr=float(d[d.news].vr.median()))
tab = []
for k in cats:
    s = d[d.catalyst == k]
    m = ms(s, "ar"); za = s.z.abs()
    tab.append(dict(cat=k, n=m["n"], mean=m["mean"], se=m["se"], t=m["t"], med=m["med"], abs_mean=float(s.ar.abs().mean()), abs_med=float(s.ar.abs().median()), gt5=float((s.ar.abs() >= .05).mean()), gt10=float((s.ar.abs() >= .10).mean()), z_abs_med=float(za.median()), up=float((s.ar > 0).mean()), vr=float(s.vr.median())))
R["catalysts"] = tab

# variance decomposition: idiosyncratic variance on each day type
dd = d.copy(); lo, hi = dd.ar.quantile([.001, .999]); dd["a2"] = dd.ar.clip(lo, hi) ** 2
tot = dd.a2.sum()
def share(mask): return dict(days=float(mask.mean()), var=float(dd.a2[mask].sum() / tot))
R["variance"] = dict(news=share(dd.news), other_item=share(~dd.news & ~dd.quiet), quiet=share(dd.quiet))
for k in ("low", "mid", "high"):
    sub = dd[dd.liq == k]; tt_ = sub.a2.sum()
    R["variance"][k] = dict(news_days=float(sub.news.mean()), news_var=float(sub.a2[sub.news].sum() / tt_), quiet_days=float(sub.quiet.mean()), quiet_var=float(sub.a2[sub.quiet].sum() / tt_))
R["shock_mix"] = dict(news=float(d[d.shock].news.mean()), quiet=float(d[d.shock].quiet.mean()), news_share_all=float(d.news.mean()))

# ---------- C. shocks: news vs no news (Chan 2003, Savor 2012)
hs = ["r1", "r5", "r10", "r20"]
def shock_table(mask_shock, q="quiet", hs=hs):
    out = {}
    for grp, m in (("news", d.news), ("quiet", d[q])):
        for sgn, u in (("up", d.up), ("down", ~d.up)):
            s = d[mask_shock & m & u]
            out[f"{grp}_{sgn}"] = dict(n=int(len(s)), ar0=float(s.ar.mean()) if len(s) else None, med0=float(s.ar.median()) if len(s) else None, **{h: ms(s, h) for h in hs})
    return out
R["shock"] = shock_table(d.shock)
_spy = shock_table(d.shock, hs=["r1_spy", "r5_spy", "r10_spy", "r20_spy"])
R["shock_spy"] = {k: {**v, "r1": v["r1_spy"], "r5": v["r5_spy"], "r10": v["r10_spy"], "r20": v["r20_spy"]} for k, v in _spy.items()}
R["shock_quiet2"] = shock_table(d.shock, "quiet2")
R["shock_10"] = shock_table((d.ar.abs() >= .10) & (d.z.abs() >= SHOCK_Z))
for lv in ("low", "mid", "high"):
    R[f"shock_{lv}"] = shock_table(d.shock & (d.liq == lv))
R["shock_px5"] = shock_table(d.shock & (d.pc >= 5))
half = sessions[start_i + (T - start_i) // 2]
R["half"] = half
R["shock_h1"] = shock_table(d.shock & (d.date < half)); R["shock_h2"] = shock_table(d.shock & (d.date >= half))

# post-shock only, rebased at the close of the event day
_b = {k: _bench(bhar(ia, ti, 0, k)) for k in range(1, 21)}
def post(mask):
    s = d[mask]; ii, tt_ = s.i.values, s.t.values; out = []
    for k in range(0, 21):
        if k == 0: out.append(dict(k=0, m=0.0, se=0.0)); continue
        v = bhar(ii, tt_, 0, k) - s.cell.map(_b[k]).values; ok = ~np.isnan(v)
        y = v[ok]
        r = mean_se(y, tt_[ok], ii[ok])
        out.append(dict(k=k, m=r["mean"], se=r["se"]))
    return out
R["post"] = {f"{g}_{s}": post(d.shock & (d.news if g == "news" else d.quiet) & (d.up if s == "up" else ~d.up)) for g in ("news", "quiet") for s in ("up", "down")}

for _lv in ("low", "mid", "high"):
    R[f"post_{_lv}"] = {f"{g}_{s_}": post(d.shock & (d.liq == _lv) & (d.news if g == "news" else d.quiet) & (d.up if s_ == "up" else ~d.up)) for g in ("news", "quiet") for s_ in ("up", "down")}

# by catalyst after shock
R["shock_cat"] = []
for k in cats:
    for sgn, u in (("up", d.up), ("down", ~d.up)):
        s = d[d.shock & (d.catalyst == k) & u]
        R["shock_cat"].append(dict(cat=k, side=sgn, n=int(len(s)), ar0=float(s.ar.mean()) if len(s) else None, r5=ms(s, "r5"), r20=ms(s, "r20")))

# ---------- D. calendar-time portfolios, which avoid overlap problems entirely
def calendar(mask, hold=5, cost_bp=0):
    s = d[mask & d.news]
    long_ = np.zeros((T, len(syms))); short = np.zeros((T, len(syms)))
    for up, arr in ((True, long_), (False, short)):
        x = s[s.up == up]
        for tt_, ii_ in zip(x.t.values, x.i.values):
            arr[tt_ + 1: min(T, tt_ + 1 + hold), ii_] = 1
    A = np.nan_to_num(arA.copy())
    res = {}
    series = []
    for t_ in range(start_i + 1, T):
        L = long_[t_] > 0; S = short[t_] > 0
        l = A[t_, L].mean() if L.sum() >= 5 else np.nan; sh = A[t_, S].mean() if S.sum() >= 5 else np.nan
        series.append((sessions[t_], l, sh))
    x = pd.DataFrame(series, columns=["date", "l", "s"]); x["ls"] = x.l - x.s
    for col in ("l", "s", "ls"):
        v = x[col].dropna().values
        r = sm.OLS(v, np.ones(len(v))).fit(cov_type="HAC", cov_kwds={"maxlags": hold})
        res[col] = dict(n=int(len(v)), daily=float(r.params[0]), t=float(r.tvalues[0]), ann=float(r.params[0] * 252), sharpe=float(v.mean() / v.std() * np.sqrt(252)))
    # turnover: new names entering per day as a share of the book
    book = (long_ > 0).sum(1)[start_i:]; enter = np.zeros(T)
    res["avg_names_long"] = float(book.mean()); res["avg_names_short"] = float((short > 0).sum(1)[start_i:].mean())
    res["cost_breakeven_bp"] = float(res["ls"]["daily"] * hold * 1e4 / 2)
    return res, x
R["calendar"] = {}; curves = {}
for lv, mask in (("all", d.shock), ("high", d.shock & (d.liq == "high")), ("low", d.shock & (d.liq == "low"))):
    R["calendar"][lv], x = calendar(mask); curves[lv] = x
x = curves["all"].fillna(0)
R["calendar_curve"] = [dict(d=a, l=float(b), s=float(c_), ls=float(e_)) for a, b, c_, e_ in zip(x.date, x.l.cumsum(), x.s.cumsum(), x.ls.cumsum())][::3]

# ---------- E. bursts: how many headlines at once, and what is new
n = d[d.news].copy()
n["b"] = pd.cut(n.wire, [0, 1, 2, 4, 1000], labels=["1", "2", "3-4", "5+"])
R["bursts"] = []
for k, s in n.groupby("b", observed=True):
    R["bursts"].append(dict(b=str(k), n=int(len(s)), abs_mean=float(s.ar.abs().mean()), abs_med=float(s.ar.abs().median()), gt5=float((s.ar.abs() >= .05).mean()), vr=float(s.vr.median()), abs_liq_high=float(s[s.liq == "high"].ar.abs().median())))
# abnormal news count vs own trailing rate, point in time
W = pd.DataFrame(wire[:, :]).rolling(60, min_periods=20).mean().shift(1).values
n["base"] = W[n.t.values, n.i.values]
n["rel"] = n.wire / n.base.clip(lower=1 / 60)
n["relb"] = pd.cut(n.rel, [0, 1, 2, 5, 1e9], labels=["At or under own average", "1-2x", "2-5x", "Over 5x"])
R["burst_rel"] = []
for k, s in n.groupby("relb", observed=True):
    R["burst_rel"].append(dict(b=str(k), n=int(len(s)), abs_med=float(s.ar.abs().median()), gt5=float((s.ar.abs() >= .05).mean()), vr=float(s.vr.median())))
# multi-source burst among non-SEC items in a tight window; signed follow through
sig_ = n[n.shock].copy(); sig_["s"] = np.where(sig_.up, 1, -1)
for h_ in ("r1", "r5", "r20"): sig_[h_ + "s"] = sig_[h_] * sig_.s
R["burst_shock"] = []
for k, s in sig_.groupby(pd.cut(sig_.wire, [0, 1, 2, 4, 1000], labels=["1", "2", "3-4", "5+"]), observed=True):
    R["burst_shock"].append(dict(b=str(k), n=int(len(s)), r1=ms(s, "r1s"), r5=ms(s, "r5s"), r20=ms(s, "r20s")))

# staleness: Tetlock's measure, overlap with the prior ten headlines on the same stock
STOP = set("a an and at by for from in inc corp co ltd llc of on or the to with new its as is are announces announced reports reported company group holdings".split())
tok = lambda s: frozenset(w for w in re.findall(r"[a-z0-9]+", s.lower()) if w not in STOP and len(w) > 1)
e2 = e[e.si >= 0].sort_values(["i", "et"])
stale = np.full(len(e2), np.nan); prev = {}
for j, (i_, tl) in enumerate(zip(e2.i.values, e2.title.fillna("").values)):
    t_ = tok(tl); hist_ = prev.setdefault(i_, [])
    if len(hist_) >= 3 and t_:
        u = frozenset().union(*hist_)
        stale[j] = len(t_ & u) / len(t_)
    hist_.append(t_)
    if len(hist_) > 10: hist_.pop(0)
e2["stale"] = stale
st = e2.groupby(["si", "i"]).stale.min().rename("stale")
n = n.join(st, on=["t", "i"])
n["sb"] = pd.cut(n.stale, [-.01, .25, .5, .75, 1.0], labels=["Mostly new words", "Some overlap", "Heavy overlap", "Near-repeat"])
R["stale"] = []
n["abs"] = n.ar.abs()
for k, s in n[n.stale.notna()].groupby("sb", observed=True):
    s2 = s.copy(); s2["sg"] = np.where(s2.up, 1, -1) * s2.r5; s2["sg1"] = np.where(s2.up, 1, -1) * s2.r1
    R["stale"].append(dict(b=str(k), n=int(len(s)), abs_mean=float(s["abs"].mean()), abs_med=float(s["abs"].median()), gt5=float((s["abs"] >= .05).mean()), vr=float(s.vr.median()), r1s=ms(s2, "sg1"), r5s=ms(s2, "sg")))
# Tetlock-style regression: does the stale-day return reverse more?
x = n[n.stale.notna() & n.r5.notna()].copy()
lo, hi = x.ar.quantile([.005, .995]); x["a"] = x.ar.clip(lo, hi); lo, hi = x.r5.quantile([.005, .995]); x["y"] = x.r5.clip(lo, hi)
X = sm.add_constant(np.column_stack([x.a, x.stale, x.a * x.stale]))
rs = sm.OLS(x.y.values, X).fit(cov_type="cluster", cov_kwds={"groups": np.column_stack([x.t, x.i])})
R["stale_reg"] = dict(n=int(len(x)), b_ar=float(rs.params[1]), t_ar=float(rs.tvalues[1]), b_stale=float(rs.params[2]), t_stale=float(rs.tvalues[2]), b_int=float(rs.params[3]), t_int=float(rs.tvalues[3]))

# ---------- F. momentum interaction
s = d[d.shock & d.pre20.notna()].copy()
qs = d[d.pre20.notna()].groupby("t").pre20.quantile([1 / 3, 2 / 3]).unstack()
s = s.join(qs, on="t")
s["mom"] = np.where(s.pre20 <= s[1 / 3], "loser", np.where(s.pre20 >= s[2 / 3], "winner", "middle"))
R["momentum"] = []
for mom in ("loser", "middle", "winner"):
    for g, m in (("news", s.news), ("quiet", s.quiet)):
        for sgn, u in (("up", s.up), ("down", ~s.up)):
            x_ = s[(s.mom == mom) & m & u]
            R["momentum"].append(dict(mom=mom, grp=g, side=sgn, n=int(len(x_)), r5=ms(x_, "r5"), r20=ms(x_, "r20")))
# plain 20-day momentum across the panel in this window, for context
all_ = d[d.pre20.notna() & d.r20.notna()].copy()
qq = all_.groupby("t").pre20.transform(lambda v: pd.qcut(v.rank(method="first"), 5, labels=False))
all_["q"] = qq
sp = all_[all_.t % 20 == 0]
R["momentum_dates"] = int(sp.t.nunique())
R["momentum_plain"] = [dict(q=int(k), r20=float(v.r20.clip(*v.r20.quantile([.005, .995])).mean()), n=int(len(v))) for k, v in sp.groupby("q")]
# news-rich versus news-poor winners and losers: 20-day formation news count (Chan's split)
Wc = pd.DataFrame(wire).rolling(20, min_periods=10).sum().shift(1).values
all_["w20"] = Wc[all_.t.values, all_.i.values]
sp = all_[all_.t % 20 == 0].copy()
sp["rich"] = sp.w20 >= 2
R["momentum_news"] = []
for k, v in sp.groupby(["q", "rich"]):
    R["momentum_news"].append(dict(q=int(k[0]), rich=bool(k[1]), n=int(len(v)), r20=ms(v, "r20")["mean"], se=ms(v, "r20")["se"]))

# ---------- G. timing and the open
R["timing"] = []
for lab, tm in (("Before the open", 0), ("During the session", 1), ("After the close", 2)):
    s = d[d.news & (d.timing == tm)]
    denom = s.ar.abs().replace(0, np.nan)
    same = (np.sign(s.gap) == np.sign(s.ar)).mean()
    R["timing"].append(dict(lab=lab, n=int(len(s)), abs_med=float(s.ar.abs().median()), gap_abs_med=float(s.gap.abs().median()), oc_abs_med=float(s.oc.abs().median()), gap_share=float((s.gap.abs() / (s.gap.abs() + s.oc.abs())).median()), same=float(same), gt5=float((s.ar.abs() >= .05).mean())))
# Gap continuation by size, news vs quiet: how much of the day was left at the bell?
gs = d[(d.gap.abs() >= .05)].copy(); gs["sg"] = np.where(gs.gap > 0, 1, -1); gs["cont"] = gs.oc * gs.sg
gs["gb"] = pd.cut(gs.gap.abs(), [.05, .1, .2, 10], labels=["5-10%", "10-20%", "Over 20%"])
R["gap"] = []
for (g_, sgn), m in {("news", "up"): gs.news & (gs.sg > 0), ("news", "down"): gs.news & (gs.sg < 0), ("quiet", "up"): gs.quiet & (gs.sg > 0), ("quiet", "down"): gs.quiet & (gs.sg < 0)}.items():
    for b_ in ("5-10%", "10-20%", "Over 20%"):
        x_ = gs[m & (gs.gb == b_)]
        R["gap"].append(dict(grp=g_, side=sgn, gap=b_, n=int(len(x_)), cont=ms(x_, "cont"), up=float((x_.cont > 0).mean()) if len(x_) else None))
R["gap_high"] = []
for (g_, sgn), m in {("news", "up"): gs.news & (gs.sg > 0), ("news", "down"): gs.news & (gs.sg < 0), ("quiet", "up"): gs.quiet & (gs.sg > 0), ("quiet", "down"): gs.quiet & (gs.sg < 0)}.items():
    x_ = gs[m & (gs.liq == "high")]
    R["gap_high"].append(dict(grp=g_, side=sgn, n=int(len(x_)), cont=ms(x_, "cont"), up=float((x_.cont > 0).mean())))
# run-up before news: any sign of the move arriving early?
s = d[d.shock & d.news]
R["runup"] = {sgn: {cat: ms(s[(s.catalyst == cat) & (s.up == (sgn == "up"))], "pre5") for cat in ("earnings", "clinical", "mna", "offering", "analyst")} for sgn in ("up", "down")}
R["runup_all"] = {g: {sgn: ms(d[d.shock & (d.news if g == "news" else d.quiet) & (d.up == (sgn == "up"))], "pre5") for sgn in ("up", "down")} for g in ("news", "quiet")}

# ---------- G2. analyst direction, insider filings, and two tests of the design itself
tl = prim.set_index(["si", "i"]).title.reindex(pd.MultiIndex.from_arrays([d.t, d.i])).fillna("").values
_tl = pd.Series(tl, index=d.index)
R["analyst"] = []
for lab, pat in (("Upgrade", r"\bupgraded by\b"), ("Downgrade", r"\bdowngraded by\b"), ("Initiation", r"\binitiated coverage\b")):
    s_ = d[_tl.str.contains(pat, case=False)]
    R["analyst"].append(dict(lab=lab, n=int(len(s_)), ar0=ms(s_, "ar"), r5=ms(s_, "r5"), r20=ms(s_, "r20"), abs_med=float(s_.ar.abs().median())))
f4 = pd.read_csv("data/f4.tsv", sep="\t", header=None, names=["nid", "title"], quoting=3, engine="python", on_bad_lines="skip")
f4 = f4.merge(ncs[["nid", "si", "ticker", "mins", "et"]].drop_duplicates("nid"), on="nid")
f4["i"] = f4.ticker.map(six)
f4["buy"] = f4.title.str.contains(r"\bbought\b", case=False)
f4["plan"] = f4.title.str.contains("pre-agreed|10b5", case=False)
f4["val"] = pd.to_numeric(f4.title.str.extract(r"\$([0-9,]+)")[0].str.replace(",", ""), errors="coerce")
f4["exec"] = f4.title.str.contains(r"\b(?:CEO|CFO|Chief Executive|Chief Financial|President)\b", case=False)
f4["own10"] = f4.title.str.contains(r"10%|Owner", case=False)
base = d.set_index(["t", "i"])
def ins(mask, lab):
    x_ = f4[mask].drop_duplicates(["si", "i"])
    j = x_[["si", "i"]].join(base[["ar", "r1", "r5", "r20", "pre20"]], on=["si", "i"]).dropna(subset=["ar"]).rename(columns={"si": "t"})
    return dict(lab=lab, items=int(mask.sum()), n=int(len(j)), ar0=ms(j, "ar"), r1=ms(j, "r1"), r5=ms(j, "r5"), r20=ms(j, "r20"), pre20_med=float(j.pre20.median()) if len(j) else None)
R["insider"] = [ins(f4.buy, "All open-market buys"), ins(f4.buy & (f4.val >= 100000), "Buys of $100K or more"), ins(f4.buy & f4["exec"], "Buys by a CEO, CFO or president"), ins(f4.buy & f4.own10, "Buys by a 10% owner"), ins(~f4.buy & ~f4.plan, "Sales outside a trading plan"), ins(~f4.buy & f4.plan, "Sales under a trading plan")]
_sec = pd.Series(d.sym.map(c.drop_duplicates("ticker").set_index("ticker").sector).values, index=d.index).fillna("na")
_arS = d.ar - d.groupby([d.t, _sec]).ar.transform("median")
_shS = (_arS.abs() >= SHOCK_ABS) & ((_arS / d.sig).abs() >= SHOCK_Z)
R["sector_robust"] = dict(n=int(_shS.sum()), news=float(d.news[_shS].mean()), quiet=float(d.quiet[_shS].mean()))
_old = d[d.t < T - 21]
R["attrition"] = dict(all=float(_old.r20.isna().mean()), news_up=float(_old[_old.shock & _old.news & _old.up].r20.isna().mean()), news_down=float(_old[_old.shock & _old.news & ~_old.up].r20.isna().mean()), quiet_up=float(_old[_old.shock & _old.quiet & _old.up].r20.isna().mean()), quiet_down=float(_old[_old.shock & _old.quiet & ~_old.up].r20.isna().mean()))
R["universe"] = dict(listed_rows=int(len(c)), common=int(c.common.sum()), with_bars=int(len(syms)), used=int(d.i.nunique()), ended_early=int((pd.DataFrame(closeA).iloc[-5:].notna().sum(0) == 0).sum()))

# How often is "nothing in the feed" really a results day? Periodic filings from SEC XBRL, by CIK.
_fl = pd.read_csv("data/filed.tsv", sep="\t", header=None, names=["cik", "filed", "form"], dtype={"cik": str})
_t2c = _ck.drop_duplicates("ticker").set_index("ticker").cik
_si = np.searchsorted(sessions, _fl.filed.values); _fl["si"] = _si
_near = set()
_c2s = _ck.groupby("cik").ticker.apply(list)
for _c, _s_i in zip(_fl.cik, _fl.si):
    for _t in _c2s.get(_c, []):
        if _t in six:
            for _o in (-1, 0, 1): _near.add((_s_i + _o, six[_t]))
_nr = np.fromiter(((a, b) in _near for a, b in zip(d.t.values, d.i.values)), bool, len(d))
d["filing_near"] = _nr
_q = d[d.shock & d.quiet]; _ql = _q[_q.liq == "high"]
R["quiet_audit"] = dict(all=float(_q.filing_near.mean()), high=float(_ql.filing_near.mean()), base=float(d[d.quiet].filing_near.mean()), n=int(len(_q)), n_high=int(len(_ql)), artifacts=ARTIFACTS, alias_firms=int(len(_multi)))
d["quiet_x"] = d.quiet & ~d.filing_near
R["shock_quietx"] = shock_table(d.shock, "quiet_x")
R["shock_quietx_high"] = shock_table(d.shock & (d.liq == "high"), "quiet_x")
R["shock_big"] = shock_table(d.shock & (d.dv >= 5e6) & (d.pc >= 5))
# Does the headline word tell you the direction?
_tt = pd.Series(np.asarray(tl, dtype=object), index=d.index)
R["words"] = []
for w_, pat in (("record", r"\brecord\b"), ("positive", r"\bpositive\b"), ("strong", r"\bstrong\b"), ("raises", r"\b(raises|raised)\b"), ("growth", r"\bgrowth\b"), ("announces", r"\bannounces\b"), ("reports", r"\breports\b"), ("update", r"\bupdate\b"), ("preliminary", r"\bpreliminary\b"), ("offering", r"\boffering\b"), ("delays", r"\b(delay|delays|delayed)\b"), ("discontinue", r"\b(discontinu\w+|terminat\w+)\b"), ("cuts", r"\b(cuts|lowers|reduces)\b")):
    m_ = _tt.str.contains(pat, case=False, regex=True) & d.news
    x_ = d[m_]
    R["words"].append(dict(w=w_, n=int(len(x_)), mean=ms(x_, "ar"), med=float(x_.ar.median()) if len(x_) else None, up=float((x_.ar > 0).mean()) if len(x_) else None, abs_med=float(x_.ar.abs().median()) if len(x_) else None, shock_down=float((x_.shock & ~x_.up).mean()) if len(x_) else None, shock_up=float((x_.shock & x_.up).mean()) if len(x_) else None))

# News against no-news inside each direction: a direct test of the difference.
def diff_test(mask, col):
    x_ = d[mask & (d.news | d.quiet) & d[col].notna()]
    lo_, hi_ = x_[col].quantile([.005, .995]); y_ = x_[col].clip(lo_, hi_).values
    X_ = sm.add_constant(x_.news.astype(float).values)
    r_ = sm.OLS(y_, X_).fit(cov_type="cluster", cov_kwds={"groups": np.column_stack([x_.t, x_.i])})
    return dict(diff=float(r_.params[1]), se=float(r_.bse[1]), t=float(r_.tvalues[1]), n=int(len(x_)))
R["diff"] = {f"{lv}_{sgn}_{h_}": diff_test(d.shock & (d.up == (sgn == "up")) & (True if lv == "all" else d.liq == lv), h_) for lv in ("all", "low", "mid", "high") for sgn in ("up", "down") for h_ in ("r5", "r20")}

# ---------- H. event candidates for the case studies, read by hand
cand = d[d.shock & d.news & (d.dv >= 5e6) & (d.pc >= 5)].sort_values("ar")
cols = "date sym catalyst timing ar z vr gap oc r1 r5 r20 wire nid title dv pc".split()
cand[cols].to_csv("cand_news.csv", index=False)
cq = d[d.shock & d.quiet2 & (d.dv >= 5e6) & (d.pc >= 5)].sort_values("ar")
cq["name"] = cq.sym.map(c.drop_duplicates("ticker").set_index("ticker").name)
cq[["date", "sym", "ar", "z", "vr", "gap", "oc", "r1", "r5", "r20", "name", "dv", "pc"]].to_csv("cand_quiet.csv", index=False)
R["cases"] = {}
for _s, _dt in (("RVMD", "2026-04-13"), ("ERAS", "2026-04-28"), ("POET", "2026-04-27"), ("ENPH", "2026-02-04"), ("MRNA", "2026-08-19"), ("FISV", "2025-10-29")):
    _r = d[(d.sym == _s) & (d.date == _dt)]
    if len(_r):
        _r = _r.iloc[0]
        R["cases"][_s] = {k: (None if pd.isna(_r[k]) else float(_r[k])) for k in ("ar", "z", "vr", "gap", "oc", "r1", "r5", "r20")}
        R["cases"][_s].update(date=_dt, wire=int(_r.wire), news=bool(_r.news), quiet=bool(_r.quiet), catalyst=None if pd.isna(_r.catalyst) else str(_r.catalyst))
    else: R["cases"][_s] = None
d[["date", "sym", "t", "i", "ar", "z", "news", "quiet", "wire", "catalyst", "up", "shock", "r1", "r5", "r20", "liq", "vr", "gap", "oc", "pre20", "pc", "dv"]].to_pickle("d.pkl")
R["catalyst_shock_share"] = {k: float((d[d.catalyst == k].shock).mean()) for k in cats}
json.dump(R, open("results.json", "w"), indent=1, default=lambda o: None)
print("wrote results")