From 04e2c5c3be0ba5f984bf9c1585fc9a99f5c1baea Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Fri, 9 Oct 2026 22:20:43 -0700 Subject: [PATCH] research: commit the 2026-10-07/08 studies left in the working tree, overnight mid/small (STOP) Studies run by earlier sessions and documented in CLAUDE.md, but never committed: FINRA short volume (SVR) and the research-only svr_momentum veto, SEC fails-to-deliver, 13D activists and the 13D event sleeve, 13F institutional flow, GDELT news volume, S&P 400/600 Form 4 insiders, union_shield, shield_lab; plus the mid/small insider and Tiingo fetcher changes and midsmall_rebuild. overnight_midsmall: the overnight-only momentum book on S&P 400/600 earns 0.93 net Sharpe at 3 bp a side (2019-26) with corr 0.83 to the S&P 500 book (STOP). Co-Authored-By: Claude Opus 5.5 --- research/lab/activist_strategy.py | 139 ++++++++++++++ research/lab/activists.py | 175 ++++++++++++++++++ research/lab/fails.py | 129 +++++++++++++ research/lab/fetch_13d.py | 126 +++++++++++++ research/lab/fetch_13d_targets.py | 93 ++++++++++ research/lab/fetch_13f.py | 106 +++++++++++ research/lab/fetch_fails.py | 147 +++++++++++++++ research/lab/fetch_gdelt.py | 123 +++++++++++++ research/lab/fetch_insider.py | 95 ++++++---- research/lab/fetch_shortvolume.py | 148 +++++++++++++++ research/lab/fetch_tiingo.py | 11 +- research/lab/insider_mid.py | 286 +++++++++++++++++++++++++++++ research/lab/inst_flow.py | 152 +++++++++++++++ research/lab/news_volume.py | 78 ++++++++ research/lab/overnight_midsmall.py | 178 ++++++++++++++++++ research/lab/short_volume.py | 121 ++++++++++++ research/lab/union_shield.py | 142 ++++++++++++++ research/midsmall_rebuild.py | 40 +++- research/shield_lab.py | 132 +++++++++++++ research/svr_overlay.py | 91 +++++++++ strategies/svr_momentum.py | 76 ++++++++ 21 files changed, 2543 insertions(+), 45 deletions(-) create mode 100644 research/lab/activist_strategy.py create mode 100644 research/lab/activists.py create mode 100644 research/lab/fails.py create mode 100644 research/lab/fetch_13d.py create mode 100644 research/lab/fetch_13d_targets.py create mode 100644 research/lab/fetch_13f.py create mode 100644 research/lab/fetch_fails.py create mode 100644 research/lab/fetch_gdelt.py create mode 100644 research/lab/fetch_shortvolume.py create mode 100644 research/lab/insider_mid.py create mode 100644 research/lab/inst_flow.py create mode 100644 research/lab/news_volume.py create mode 100644 research/lab/overnight_midsmall.py create mode 100644 research/lab/short_volume.py create mode 100644 research/lab/union_shield.py create mode 100644 research/shield_lab.py create mode 100644 research/svr_overlay.py create mode 100644 strategies/svr_momentum.py diff --git a/research/lab/activist_strategy.py b/research/lab/activist_strategy.py new file mode 100644 index 0000000..bb1d039 --- /dev/null +++ b/research/lab/activist_strategy.py @@ -0,0 +1,139 @@ +"""Activist event sleeve: hold fresh SC 13D targets, lab-engine evaluation. + +Event study first (research/lab/activists.py): +1.2% CAR1 excess in S&P +400/600 (t 4.2), +2.2% CAR21 in the S&P 500 (t 2.2); low-SVR events beat +high-SVR events (+3.6% vs +0.8% CAR21). + +Strategy: on each session, equal-weight all panel members whose SC 13D was +filed in the last H sessions (most recent K at most; daily rebalance to the +target basket, 20bp costs are built into the mid/small panels). PIT: the +weights row for day D includes filings with `filed <= D` and is executed at +the next open by the engine, i.e. after the filing is public. + + PYTHONPATH=. uv run python -m research.lab.activist_strategy [mid|small|midsmall|us] +""" +from __future__ import annotations + +import json +import os +import pickle +import sys + +import numpy as np +import pandas as pd + +from research.lab.activists import cik_ticker_map +from research.lab.engine import simulate +from research.lab.factor_scan import stats + +TARGETS = "data/research/edgar_13d_targets.csv" +SVR_PATH = "data/research/short_volume.pkl" + + +def load_events() -> pd.DataFrame: + tg = pd.read_csv(TARGETS) + tg = tg[tg["target_cik"].fillna(0) > 0].copy() + cik2t = cik_ticker_map() + tg["ticker"] = tg["target_cik"].map(lambda c: cik2t.get(int(c))) + tg = tg[tg["ticker"].notna()] + # fallback by exact name + if os.path.exists("data/research/company_tickers.json"): + raw = json.load(open("data/research/company_tickers.json")) + by_name = {str(v["title"]).strip().upper(): str(v["ticker"]).upper().replace(".", "-") + for v in raw.values()} + miss = tg["ticker"].isna() + tg.loc[miss, "ticker"] = tg.loc[miss, "target_name"].str.strip().str.upper().map(by_name) + tg = tg.sort_values("filed").drop_duplicates(subset=["ticker", "filed"], keep="first") + return tg[["ticker", "filed"]].dropna() + + +def svr21() -> pd.DataFrame | None: + if not os.path.exists(SVR_PATH): + return None + with open(SVR_PATH, "rb") as fh: + d = pickle.load(fh) + return (d["short"] / d["total"].replace(0.0, np.nan)).rolling(21, min_periods=10).mean() + + +def build_weights(p, events: pd.DataFrame, hold: int, k_max: int, + svr_filter: str | None = None, svr_lag: int = 2) -> pd.DataFrame: + dates = p.dates + cols = list(p.close.columns) + pos = {c: i for i, c in enumerate(cols)} + W = np.zeros((len(dates), len(cols))) + svr = svr21() + if svr is not None: + svr = svr.reindex(columns=cols).shift(svr_lag) + med = svr.where(p.member).median(axis=1) + + acts: list[tuple[int, int]] = [] # (start_idx, end_idx, col_idx) + for _, ev in events.iterrows(): + i = pos.get(ev["ticker"]) + if i is None: + continue + t0 = dates.searchsorted(pd.Timestamp(ev["filed"]), side="left") + if t0 >= len(dates) - 2: + continue + start, end = t0 + 1, min(t0 + 1 + hold, len(dates) - 1) + if svr_filter and svr is not None and dates[t0] in svr.index: + v = svr.at[dates[t0], ev["ticker"]] if ev["ticker"] in svr.columns else np.nan + m = med.get(dates[t0], np.nan) + if np.isfinite(v) and np.isfinite(m): + if svr_filter == "low" and v >= m: + continue + if svr_filter == "high" and v < m: + continue + acts.append((start, end, i)) + if not acts: + return pd.DataFrame(0.0, index=dates, columns=cols) + starts = np.array([a[0] for a in acts]) + ends = np.array([a[1] for a in acts]) + idx = np.array([a[2] for a in acts]) + for d in range(len(dates)): + active = np.where((starts <= d) & (d <= ends))[0] + if len(active) == 0: + continue + if k_max and len(active) > k_max: + active = active[np.argsort(-starts[active])][:k_max] # most recent filings + W[d, idx[active]] = 1.0 / len(active) + return pd.DataFrame(W, index=dates, columns=cols) + + +def run(mkt: str, events: pd.DataFrame, hold: int, k_max: int, + svr_filter: str | None = None) -> dict: + if mkt == "us": + with open("data/research/us.panel.pkl", "rb") as fh: + p = pickle.load(fh) + else: + from research.lab.panel_mid import load_us_mid + p = load_us_mid(mkt) + W = build_weights(p, events, hold, k_max, svr_filter) + r = simulate(p, target_weights=W, exec_at="open", min_trade=0.002) + blocks = {} + for a, b in (("2016-09-27", "2021-09-26"), ("2021-09-27", "2026-09-24")): + st = stats(r.equity.loc[a:b], r.turnover.loc[a:b]) + blocks[f"{a[:4]}-{b[:4]}"] = st + return {"mkt": mkt, "hold": hold, "k": k_max, "svr": svr_filter or "none", + "blocks": blocks} + + +def main() -> None: + mkts = sys.argv[1:] or ["midsmall"] + events = load_events() + print(f"events loaded: {len(events)}") + results = [] + for mkt in mkts: + for hold, k_max, svr_filter in ((21, 20, None), (21, 20, "low"), (63, 20, None), + (63, 20, "low"), (21, 10, None), (21, 10, "low")): + res = run(mkt, events, hold, k_max, svr_filter) + results.append(res) + line = f"{mkt:8s} H{hold:<3d} K{k_max:<3d} svr={res['svr']:4s}" + for blk, st in res["blocks"].items(): + line += f" | {blk} cagr {st['cagr']*100:5.1f}% sh {st['sharpe']:5.2f} mdd {st['mdd']*100:5.1f}%" + print(line) + json.dump(results, open("data/research/results/activist_sleeve.json", "w"), indent=1) + print("saved data/research/results/activist_sleeve.json") + + +if __name__ == "__main__": + main() diff --git a/research/lab/activists.py b/research/lab/activists.py new file mode 100644 index 0000000..ad7fc3d --- /dev/null +++ b/research/lab/activists.py @@ -0,0 +1,175 @@ +"""SC 13D activist-event study on the PIT US panels. + +Event: a new SC 13D filing (originals only; /A amendments excluded; 13G +passive stakes tracked separately) for a company whose CIK maps to a ticker +in the panel universe. EDGAR disseminates same-day, so the conservative +trade date is the next session; CARs run from close(t0) where t0 = first +session at/after the filing date. + +Reports market-adjusted forward returns (event minus same-panel member mean, +same window) over 1/5/21/63 sessions, split 13D vs 13G, with yearly means, +plus conditioning on 63d momentum quintile and SVR (short-volume ratio) +tercile to test the squeeze reading. + + PYTHONPATH=. uv run python -m research.lab.activists [us|midsmall|all] +""" +from __future__ import annotations + +import json +import os +import pickle +import sys + +import numpy as np +import pandas as pd +import requests + +from research.lab.fetch_us import _proxy_env + +EVENTS_CSV = "data/research/edgar_13d.csv" +TICKERS_CACHE = "data/research/company_tickers.json" +UA = {"User-Agent": "quant-research/0.1 gahow@example.com", "Accept-Encoding": "gzip"} +HORIZONS = (1, 5, 21, 63) + + +def cik_ticker_map() -> dict[int, str]: + if not os.path.exists(TICKERS_CACHE): + r = requests.get("https://www.sec.gov/files/company_tickers.json", headers=UA, timeout=60) + r.raise_for_status() + json.dump(r.json(), open(TICKERS_CACHE, "w")) + raw = json.load(open(TICKERS_CACHE)) + out = {} + for v in raw.values(): + t = str(v["ticker"]).upper().replace(".", "-") + out.setdefault(int(v["cik_str"]), t) + return out + + +def load_svr() -> pd.DataFrame | None: + path = "data/research/short_volume.pkl" + if not os.path.exists(path): + return None + with open(path, "rb") as fh: + d = pickle.load(fh) + return d["short"] / d["total"].replace(0.0, np.nan) + + +def run(name: str, close: pd.DataFrame, member: pd.DataFrame, + events: pd.DataFrame, svr: pd.DataFrame | None) -> dict: + close = close.ffill() + member = member.reindex_like(close).fillna(False) + idx = close.index + rets = close.pct_change(fill_method=None) + # cross-sectional member mean daily return (market proxy) + mkt = rets.where(member).mean(axis=1) + cum = (1 + rets.fillna(0.0)).cumprod() + cum_mkt = (1 + mkt.fillna(0.0)).cumprod() + + mom = close.pct_change(63) + svr21 = svr.rolling(21, min_periods=10).mean() if svr is not None else None + + rows = [] + for _, ev in events.iterrows(): + t = ev["ticker"] + if t not in close.columns: + continue + pos = idx.searchsorted(pd.Timestamp(ev["filed"]), side="left") + if pos >= len(idx) - max(HORIZONS) - 1: + continue + t0 = idx[pos] + if not bool(member.at[t0, t]): + continue + svr_val = np.nan + if svr21 is not None and t in svr21.columns and t0 in svr21.index: + v = svr21.at[t0, t] + if np.isfinite(v): + svr_val = float(v) + mom_val = mom.at[t0, t] if t in mom.columns else np.nan + entry = {"form": ev["form"], "ticker": t, "date": t0, "cik": ev["cik"], + "mom": float(mom_val) if np.isfinite(mom_val) else np.nan, + "svr": svr_val} + for h in HORIZONS: + r = float(cum.iloc[pos + h][t] / cum.iloc[pos][t] - 1.0) + rm = float(cum_mkt.iloc[pos + h] / cum_mkt.iloc[pos] - 1.0) + entry[f"car{h}"] = r - rm + rows.append(entry) + df = pd.DataFrame(rows) + print(f"\n=== {name}: {len(df)} events mapped to panel " + f"({df['date'].min().date() if len(df) else '-'}..{df['date'].max().date() if len(df) else '-'}) ===") + out = {"universe": name, "n_events": len(df)} + if len(df) == 0: + return out + for form in ("SC 13D", "SC 13G"): + sub = df[df["form"] == form] + if len(sub) < 30: + continue + line = [f"{form:7s} n={len(sub):5d}"] + for h in HORIZONS: + x = sub[f"car{h}"].dropna() + t = x.mean() / x.std() * np.sqrt(len(x)) if len(x) > 1 else np.nan + line.append(f"CAR{h}: {x.mean()*100:5.2f}% (t {t:4.1f}, hit {np.mean(x>0)*100:4.0f}%)") + print(" " + " ".join(line)) + out[form] = {f"car{h}": {"mean": float(sub[f'car{h}'].mean()), + "t": float(sub[f'car{h}'].mean() / sub[f'car{h}'].std() * np.sqrt(len(sub))), + "hit": float((sub[f'car{h}'] > 0).mean())} for h in HORIZONS} + yr = sub.groupby(sub["date"].dt.year)["car21"].mean() + print(" yearly CAR21: " + " ".join(f"{y}:{v*100:+.0f}%" for y, v in yr.items())) + # conditioning on momentum / svr (13D only) + d13 = df[df["form"] == "SC 13D"].copy() + if len(d13) >= 100: + d13["mom_q"] = d13["mom"].rank(pct=True) + hi, lo = d13[d13["mom_q"] > 0.66], d13[d13["mom_q"] < 0.34] + if len(hi) > 20 and len(lo) > 20: + print(f" 13D x momentum: hi-mom CAR21 {hi['car21'].mean()*100:5.2f}% (n={len(hi)}) " + f"| lo-mom CAR21 {lo['car21'].mean()*100:5.2f}% (n={len(lo)})") + out["13D_mom_hi_car21"] = float(hi["car21"].mean()) + out["13D_mom_lo_car21"] = float(lo["car21"].mean()) + if d13["svr"].notna().sum() > 100: + d13["svr_t"] = d13["svr"].rank(pct=True) + hi, lo = d13[d13["svr_t"] > 0.66], d13[d13["svr_t"] < 0.34] + if len(hi) > 20 and len(lo) > 20: + print(f" 13D x SVR: hi-fails-short CAR21 {hi['car21'].mean()*100:5.2f}% (n={len(hi)}) " + f"| lo CAR21 {lo['car21'].mean()*100:5.2f}% (n={len(lo)})") + out["13D_svr_hi_car21"] = float(hi["car21"].mean()) + out["13D_svr_lo_car21"] = float(lo["car21"].mean()) + return out + + +def main() -> None: + _proxy_env() + which = sys.argv[1] if len(sys.argv) > 1 else "all" + targets_csv = os.environ.get("TR9_TARGETS_CSV", "data/research/edgar_13d_targets.csv") + if not os.path.exists(targets_csv): + raise SystemExit("run research.lab.fetch_13d_targets first") + tg = pd.read_csv(targets_csv) + tg = tg[(tg["target_cik"].fillna(0) > 0)] + cik2t = cik_ticker_map() + tg["ticker"] = tg["target_cik"].map(lambda c: cik2t.get(int(c))) + # fallback: exact name match against the SEC ticker map + if os.path.exists(TICKERS_CACHE): + raw = json.load(open(TICKERS_CACHE)) + by_name = {str(v["title"]).strip().upper(): str(v["ticker"]).upper().replace(".", "-") + for v in raw.values()} + miss = tg["ticker"].isna() + tg.loc[miss, "ticker"] = tg.loc[miss, "target_name"].str.strip().str.upper().map(by_name) + ev = tg[tg["ticker"].notna()].copy() + ev["form"] = "SC 13D" + # one event per target per day (multiple filers on the same stake day) + ev = ev.sort_values("filed").drop_duplicates(subset=["ticker", "filed"], keep="first") + print(f"13D originals mapped to tickers: {len(ev)}") + svr = load_svr() + results = [] + if which in ("us", "all"): + with open("data/research/us.panel.pkl", "rb") as fh: + p = pickle.load(fh) + results.append(run("S&P 500", p.close, p.member, ev, svr)) + if which in ("midsmall", "all"): + with open("data/research/us_midsmall.panel.pkl", "rb") as fh: + p = pickle.load(fh) + results.append(run("S&P 400+600", p.close, p.member, ev, svr)) + json.dump(results, open("data/research/results/activists.json", "w"), indent=1, default=str) + print("saved data/research/results/activists.json") + + +if __name__ == "__main__": + main() diff --git a/research/lab/fails.py b/research/lab/fails.py new file mode 100644 index 0000000..717ff6a --- /dev/null +++ b/research/lab/fails.py @@ -0,0 +1,129 @@ +"""Fails-to-deliver factor scan on PIT US panels. + +FTD value / 20-day dollar volume = fails intensity: the share of a stock's +daily liquidity that is stuck unsettled. High intensity marks short-sale +constraint / squeeze fuel and, on the classic evidence, predicts negative +returns in small caps. + +PIT: SEC publishes the CNMS fails files roughly 3 weeks after settlement; +the scan shifts the FTD panel forward by 21 calendar days before +reindexing to trading days (ffill, staleness-capped). + + PYTHONPATH=. uv run python -m research.lab.fails [us|midsmall|all] +""" +from __future__ import annotations + +import pickle +import sys + +import numpy as np +import pandas as pd + +FAILS_PATH = "data/research/fails.pkl" +PUBLISH_LAG_DAYS = 21 + + +def load_ftd(close: pd.DataFrame, volume: pd.DataFrame | None) -> dict[str, pd.DataFrame]: + with open(FAILS_PATH, "rb") as fh: + d = pickle.load(fh) + shares, value = d["shares"], d["value"] + cols = [c for c in close.columns if c in shares.columns] + f = {} + v = value[cols] + f["ftd_value"] = v + f["ftd_shares"] = shares[cols] + if volume is not None and len(volume): + vol = volume.reindex(columns=cols).ffill() + px = close.reindex(columns=cols).ffill() + adv = (px * vol).rolling(20, min_periods=10).mean() + f["ftd_int"] = v / adv.replace(0.0, np.nan) + sh = shares[cols] + f["ftd_sh"] = sh / vol.rolling(20, min_periods=10).mean().replace(0.0, np.nan) + return f + + +def scan(name: str, close: pd.DataFrame, member: pd.DataFrame, + volume: pd.DataFrame | None) -> None: + raw = load_ftd(close, volume) + member = member.reindex_like(close).fillna(False) + print(f"\n=== {name}: {close.shape[1]} symbols, " + f"{close.index.min().date()}..{close.index.max().date()} ===") + fwd1 = close.shift(-1) / close - 1.0 + fwd21 = close.shift(-21) / close - 1.0 + months = fwd1.index.to_period("M") + rows = [] + lagged: dict[str, pd.DataFrame] = {} + for fname, fraw in raw.items(): + f = fraw.copy() + f.index = f.index + pd.Timedelta(days=PUBLISH_LAG_DAYS) + f = (f.reindex(index=close.index.union(f.index), columns=close.columns) + .ffill(limit=45).reindex(close.index)) + f = f.where(member) + lagged[fname] = f + ics1, ics21 = [], [] + for per in sorted(set(months)): + idx = months == per + xdf = f.loc[idx] + if xdf.notna().sum(axis=1).max() < 50: + continue + for ydf, out in ((fwd1.loc[idx], ics1), (fwd21.loc[idx], ics21)): + mask = xdf.notna().to_numpy() & np.isfinite(ydf.to_numpy()) + if mask.sum() < 100: + continue + xr = xdf.rank(axis=1).to_numpy()[mask] + yr = ydf.where(mask).rank(axis=1).to_numpy()[mask] + xm = xr - xr.mean() + ym = yr - yr.mean() + denom = np.sqrt((xm ** 2).sum() * (ym ** 2).sum()) + if denom > 0: + out.append(float((xm * ym).sum() / denom)) + ic1, ic21 = np.array(ics1), np.array(ics21) + rows.append({"factor": fname, + "IC1": ic1.mean(), "t1": ic1.mean() / ic1.std() * np.sqrt(len(ic1)) if len(ic1) > 1 else np.nan, + "IC21": ic21.mean(), "t21": ic21.mean() / ic21.std() * np.sqrt(len(ic21)) if len(ic21) > 1 else np.nan, + "months": len(ic21)}) + print(pd.DataFrame(rows).round(4).to_string(index=False)) + + for fname in [k for k in ("ftd_int", "ftd_sh") if k in lagged]: + f = lagged[fname] + pairs = [] + dates = f.index + for i in range(0, len(dates) - 21, 10): + t0 = dates[i] + x = f.loc[t0].dropna() + if len(x) < 50: + continue + q = pd.qcut(x.rank(method="first"), 5, labels=False) + lo, hi = q[q == 0].index, q[q == 4].index + r = close.iloc[i + 21] / close.iloc[i] - 1.0 + pairs.append((t0, float(r[hi].mean() - r[lo].mean()))) + if not pairs: + continue + s = np.array([p[1] for p in pairs]) + ys = pd.Series(s, index=pd.DatetimeIndex([p[0] for p in pairs])) + yearly = (1 + ys).groupby(ys.index.year).prod() - 1 + print(f"{fname}: Q5-Q1 21d spread mean {s.mean()*100:.2f}% " + f"t={s.mean()/s.std()*np.sqrt(len(s)):.2f} n={len(s)} yearly: " + + " ".join(f"{y}:{v*100:+.0f}%" for y, v in yearly.items())) + + +def main() -> None: + which = sys.argv[1] if len(sys.argv) > 1 else "all" + if which in ("us", "all"): + with open("data/research/us.panel.pkl", "rb") as fh: + p = pickle.load(fh) + vol = None + try: + vol = pd.read_csv("data/us_volume.csv", index_col=0, parse_dates=True) + except Exception: # noqa: BLE001 + pass + scan("S&P 500 ever-members", p.close, p.member, vol) + if which in ("midsmall", "all"): + with open("data/research/us_midsmall.panel.pkl", "rb") as fh: + p = pickle.load(fh) + vol = getattr(p, "volume", None) + scan("S&P 400+600 (midsmall)", p.close, p.member, vol) + + +if __name__ == "__main__": + main() diff --git a/research/lab/fetch_13d.py b/research/lab/fetch_13d.py new file mode 100644 index 0000000..3efe563 --- /dev/null +++ b/research/lab/fetch_13d.py @@ -0,0 +1,126 @@ +"""Fetch SC 13D / 13G filings (activist and 5% passive stakes) from EDGAR. + +Quarterly full-index files list every filing: + + https://www.sec.gov/Archives/edgar/full-index/YYYY/QTRn/form.idx + +Each is ~40-60 MB fixed-width text. Rows for SC 13D, SC 13D/A, SC 13G and +SC 13G/A are kept (filed date, CIK, company, archive path). Amendments are +flagged via the /A suffix so an event study can use originals only. + +Writes data/research/edgar_13d.csv (append-safe per quarter; a quarter +already present in the CSV is refetched only with --refresh). + + PYTHONPATH=. uv run python -m research.lab.fetch_13d [--quarters N] +""" +from __future__ import annotations + +import io +import os +import re +import sys +import time + +import pandas as pd +import requests + +from research.lab.fetch_us import _proxy_env + +OUT = os.path.join("data", "research", "edgar_13d.csv") +URL = "https://www.sec.gov/Archives/edgar/full-index/{y}/QTR{q}/form.idx" +UA = {"User-Agent": "quant-research/0.1 gahow@example.com", "Accept-Encoding": "gzip"} +ROW = re.compile( + r"^(?P
.{1,16}?)\s{2,}(?P.+?)\s+(?P\d{4,10})\s+" + r"(?P\d{4}-\d{2}-\d{2})\s+(?Pedgar/\S+)\s*$") +KEEP = ("SC 13D", "SC 13G") + + +def quarters() -> list[tuple[int, int]]: + out = [] + for y in range(2016, 2027): + for q in range(1, 5): + if y == 2026 and q > 3: + continue + out.append((y, q)) + return out + + +def fetch_quarter(y: int, q: int) -> list[dict] | None: + url = URL.format(y=y, q=q) + for attempt in range(6): + try: + r = requests.get(url, headers=UA, timeout=300) + r.raise_for_status() + # SEC throttles sustained bulk downloads with small non-index pages + if len(r.content) < 5_000_000 or "Master Index of EDGAR" not in r.text[:2000]: + print(f" {y}Q{q}: throttled or unexpected content " + f"({len(r.content)}B), backing off") + time.sleep(20 * (attempt + 1)) + continue + rows = [] + text = io.StringIO(r.text) + for line in text: + if not (line.startswith("SC 13D") or line.startswith("SC 13G")): + continue + m = ROW.match(line.rstrip("\n")) + if not m: + continue + d = m.groupdict() + if not d["form"].startswith(KEEP): + continue + rows.append({"quarter": f"{y}Q{q}", "form": d["form"].strip(), + "company": d["company"].strip(), "cik": int(d["cik"]), + "filed": d["date"], "path": d["file"]}) + return rows + except Exception as exc: # noqa: BLE001 + if attempt == 5: + print(f" FAILED {y}Q{q}: {exc}") + else: + time.sleep(5 * (attempt + 1)) + return None + + +def main() -> None: + _proxy_env() + limit = None + if "--quarters" in sys.argv: + limit = int(sys.argv[sys.argv.index("--quarters") + 1]) + done: set[str] = set() + prev = None + if os.path.exists(OUT): + prev = pd.read_csv(OUT) + done = set(prev["quarter"].unique()) + if "--refresh-since" in sys.argv: + since = sys.argv[sys.argv.index("--refresh-since") + 1].upper() + def ge(quarter: str) -> bool: + y, q = int(quarter[:4]), int(quarter[-1]) + sy, sq = int(since[:4]), int(since[-1]) + return (y, q) >= (sy, sq) + if prev is not None: + prev = prev[~prev["quarter"].apply(ge)] + done = {q for q in done if not ge(q)} + todo = [(y, q) for (y, q) in quarters() if f"{y}Q{q}" not in done] + if limit: + todo = todo[:limit] + print(f"quarters: {len(quarters())}, stored: {len(done)}, fetching: {len(todo)}") + frames = [prev] if prev is not None else [] + for y, q in todo: + rows = fetch_quarter(y, q) + if rows is None: + continue + df = pd.DataFrame(rows) + if len(df) == 0: + print(f" {y}Q{q}: empty, skipped") + continue + frames.append(df) + print(f" {y}Q{q}: {len(df)} SC13 rows") + pd.concat(frames).drop_duplicates(subset=["form", "cik", "filed", "path"]).to_csv(OUT, index=False) + time.sleep(4) + if frames: + allrows = pd.concat(frames).drop_duplicates(subset=["form", "cik", "filed", "path"]) + allrows.to_csv(OUT, index=False) + print(f"stored {len(allrows)} rows -> {OUT}") + + +if __name__ == "__main__": + main() diff --git a/research/lab/fetch_13d_targets.py b/research/lab/fetch_13d_targets.py new file mode 100644 index 0000000..c58cb2b --- /dev/null +++ b/research/lab/fetch_13d_targets.py @@ -0,0 +1,93 @@ +"""Extract the SUBJECT company from SC 13D filing headers. + +The EDGAR full index lists the FILER (the activist) in the CIK/company +columns; the target company appears only inside the document. For Schedule +13D filings the SGML header's first COMPANY block is the subject company +(verified on samples: FTE Networks, 30DC, Helius Medical, Genco). This +fetcher reads data/research/edgar_13d.csv, keeps SC 13D originals, and for +each one fetches the .txt header, extracting the first company name + CIK. + +Writes data/research/edgar_13d_targets.csv (resume-safe: paths already +extracted are skipped). + + PYTHONPATH=. uv run python -m research.lab.fetch_13d_targets [--limit N] +""" +from __future__ import annotations + +import os +import re +import sys +import time +from concurrent.futures import ThreadPoolExecutor, as_completed + +import pandas as pd +import requests + +from research.lab.fetch_us import _proxy_env + +IN = "data/research/edgar_13d.csv" +OUT = "data/research/edgar_13d_targets.csv" +UA = {"User-Agent": "quant-research/0.1 gahow@example.com", "Accept-Encoding": "gzip"} +NAME_RE = re.compile(r"COMPANY CONFORMED NAME:\s*(.+)") +CIK_RE = re.compile(r"CENTRAL INDEX KEY:\s*(\d+)") +WORKERS = 5 + + +def fetch_target(path: str) -> tuple[str, str, int] | None: + url = "https://www.sec.gov/Archives/" + path + for attempt in range(3): + try: + r = requests.get(url, headers=UA, timeout=60) + if r.status_code == 404: + return path, "", 0 + r.raise_for_status() + head = r.text[:20000] + name = NAME_RE.search(head) + cik = CIK_RE.search(head) + return path, (name.group(1).strip() if name else ""), int(cik.group(1)) if cik else 0 + except Exception: # noqa: BLE001 + time.sleep(1.0 * (attempt + 1)) + print(f" FAILED {path}") + return None + + +def main() -> None: + _proxy_env() + limit = None + if "--limit" in sys.argv: + limit = int(sys.argv[sys.argv.index("--limit") + 1]) + ev = pd.read_csv(IN) + ev = ev[ev["form"] == "SC 13D"].copy() + done: set[str] = set() + frames = [] + if os.path.exists(OUT): + prev = pd.read_csv(OUT) + frames.append(prev) + done = set(prev["path"]) + todo = ev[~ev["path"].isin(done)] + if limit: + todo = todo.head(limit) + print(f"13D originals: {len(ev)}, done: {len(done)}, fetching: {len(todo)}") + rows = [] + with ThreadPoolExecutor(max_workers=WORKERS) as ex: + futs = {ex.submit(fetch_target, p): p for p in todo["path"]} + n = 0 + for fut in as_completed(futs): + res = fut.result() + n += 1 + if res is not None: + path, name, cik = res + rows.append({"path": path, "target_name": name, "target_cik": cik}) + if n % 500 == 0: + print(f" {n}/{len(todo)}") + df = pd.DataFrame(rows) + if len(df): + meta = todo.set_index("path")[["filed", "cik", "company"]] + df = df.join(meta, on="path") + allrows = pd.concat(frames + [df]) if frames or len(df) else pd.DataFrame() + allrows.to_csv(OUT, index=False) + print(f"stored {len(allrows)} rows -> {OUT}") + + +if __name__ == "__main__": + main() diff --git a/research/lab/fetch_13f.py b/research/lab/fetch_13f.py new file mode 100644 index 0000000..111deed --- /dev/null +++ b/research/lab/fetch_13f.py @@ -0,0 +1,106 @@ +"""Fetch SEC DERA Form 13F structured data sets and aggregate per quarter. + +One ZIP per quarter (2016q1..2023q4 on this host), each holding TSVs of every +13F filing parsed by the SEC (COVERPAGE + INFOTABLE). We aggregate the info +table by CUSIP: total shares, total value, number of filers, and a +representative issuer name. + + https://www.sec.gov/files/structureddata/data/form-13f-data-sets/YYYYqQ_form13f.zip + +Writes data/research/dera13f/agg_YYYYqQ.csv (resume-safe: zips are cached and +quarters with an aggregate CSV are skipped) and prints a summary line per +quarter. + + PYTHONPATH=. uv run python -m research.lab.fetch_13f +""" +from __future__ import annotations + +import io +import os +import sys +import zipfile +from concurrent.futures import ThreadPoolExecutor, as_completed + +import pandas as pd +import requests + +from research.lab.fetch_us import _proxy_env + +DIR = os.path.join("data", "research", "dera13f") +BASE = "https://www.sec.gov/files/structureddata/data/form-13f-data-sets/" +UA = {"User-Agent": "quant-research/0.1 gahow@example.com"} +QUARTERS = [f"{y}q{q}" for y in range(2016, 2024) for q in (1, 2, 3, 4)] + + +def download(q: str) -> str | None: + os.makedirs(DIR, exist_ok=True) + path = os.path.join(DIR, f"{q}_form13f.zip") + if os.path.exists(path) and os.path.getsize(path) > 1_000_000: + return path + url = f"{BASE}{q}_form13f.zip" + for attempt in range(4): + try: + r = requests.get(url, headers=UA, timeout=600) + r.raise_for_status() + with open(path, "wb") as fh: + fh.write(r.content) + return path + except Exception as exc: # noqa: BLE001 + print(f" download {q} attempt {attempt}: {exc}") + return None + + +def aggregate(q: str, path: str) -> str | None: + out = os.path.join(DIR, f"agg_{q}.csv") + if os.path.exists(out): + return out + try: + zf = zipfile.ZipFile(path) + names = zf.namelist() + info = next((n for n in names if "INFOTABLE" in n.upper() and n.lower().endswith(".tsv")), None) + if info is None: + print(f" {q}: no INFOTABLE in {names}") + return None + with zf.open(info) as fh: + df = pd.read_csv(io.TextIOWrapper(fh, encoding="latin-1"), sep="\t", dtype=str, + usecols=lambda c: c.upper() in + ("NAMEOFISSUER", "CUSIP", "VALUE", "SSHPRNAMT", "SSHPRNAMTTYPE"), + low_memory=False) + df.columns = [c.upper() for c in df.columns] + for col in ("VALUE", "SSHPRNAMT"): + df[col] = pd.to_numeric(df[col], errors="coerce") + df = df[df.get("SSHPRNAMTTYPE", "SH").fillna("SH").str.upper().isin(["", "SH"])] + g = df.groupby("CUSIP", dropna=True) + agg = pd.DataFrame({ + "issuer": g["NAMEOFISSUER"].first(), + "shares": g["SSHPRNAMT"].sum(), + "value": g["VALUE"].sum(), + "n_filers": g.size(), + }).reset_index() + agg.to_csv(out, index=False) + print(f" {q}: {len(agg)} cusips, {len(df)} rows", flush=True) + return out + except Exception as exc: # noqa: BLE001 + print(f" {q}: FAILED {exc}") + return None + + +def main() -> None: + _proxy_env() + only = sys.argv[1:] or QUARTERS + with ThreadPoolExecutor(max_workers=3) as ex: + futs = {ex.submit(download, q): q for q in only} + paths = {} + for fut in as_completed(futs): + q = futs[fut] + p = fut.result() + if p: + paths[q] = p + print(f" downloaded {q}", flush=True) + for q in only: + if q in paths: + aggregate(q, paths[q]) + + +if __name__ == "__main__": + main() diff --git a/research/lab/fetch_fails.py b/research/lab/fetch_fails.py new file mode 100644 index 0000000..9256d3d --- /dev/null +++ b/research/lab/fetch_fails.py @@ -0,0 +1,147 @@ +"""Fetch SEC fails-to-deliver (FTD) data for the research universes. + +Two public ZIP files per month (a = 1st half, b = 2nd half): + + https://www.sec.gov/files/data/fails-deliver-data/cnsfailsYYYYMM[a|b].zip + +Each archive holds a pipe-delimited txt with one row per symbol and +settlement date: SETTLEMENT DATE|CUSIP|SYMBOL|QUANTITY (FAILS)|DESCRIPTION| +PRICE|. The archive covers 2018-01 onward on the current URL scheme (older +months 404; SEC also lags ~3 weeks for the newest half-month). + +Writes data/research/fails.pkl: {"shares": DataFrame, "value": DataFrame} +(float32, settlement-date x symbol), filtered to the symbols of data/us.csv +plus the US mid/small panels. Resume-safe; UA follows the repo convention. + + PYTHONPATH=. uv run python -m research.lab.fetch_fails +""" +from __future__ import annotations + +import io +import json +import os +import pickle +import sys +import time +import zipfile +from concurrent.futures import ThreadPoolExecutor, as_completed + +import pandas as pd +import requests + +from research.lab.fetch_us import _proxy_env +from research.lab.fetch_shortvolume import SYMBOLS_CACHE, universe_symbols + +OUT = os.path.join("data", "research", "fails.pkl") +URL = "https://www.sec.gov/files/data/fails-deliver-data/cnsfails{ym}{half}.zip" +UA = {"User-Agent": "quant-research/0.1 gahow@example.com", "Accept-Encoding": "gzip"} +WORKERS = 4 + + +def candidates() -> list[str]: + out = [] + for y in range(2018, 2027): + for m in range(1, 13): + for half in ("a", "b"): + out.append(f"{y}{m:02d}{half}") + return out + + +def fetch_one(ym: str, symbols: set[str]) -> tuple[str, dict, dict] | None: + url = URL.format(ym=ym[:6], half=ym[6]) + for attempt in range(4): + try: + r = requests.get(url, headers=UA, timeout=60) + if r.status_code == 404: + return ym, {}, {} + r.raise_for_status() + zf = zipfile.ZipFile(io.BytesIO(r.content)) + shares: dict = {} + value: dict = {} + for name in zf.namelist(): + with zf.open(name) as fh: + text = io.TextIOWrapper(fh, encoding="latin-1") + header = True + for line in text: + if header: + header = False + continue + parts = line.rstrip("\n").split("|") + if len(parts) < 6: + continue + date, sym = parts[0], parts[2] + if sym not in symbols: + continue + try: + qty = float(parts[3]) + px = float(parts[5]) if parts[5] else 0.0 + except ValueError: + continue + key = (date, sym) + shares[key] = shares.get(key, 0.0) + qty + value[key] = value.get(key, 0.0) + qty * px + return ym, shares, value + except Exception as exc: # noqa: BLE001 + if attempt == 3: + print(f" FAILED {ym}: {exc}") + else: + time.sleep(1.0 * (attempt + 1)) + return None + + +def main() -> None: + _proxy_env() + limit = None + if "--files-limit" in sys.argv: + limit = int(sys.argv[sys.argv.index("--files-limit") + 1]) + symbols = set(universe_symbols()) + print(f"universe symbols: {len(symbols)}") + + prev_rows = None + have: set[str] = set() + if os.path.exists(OUT): + with open(OUT, "rb") as fh: + d = pickle.load(fh) + prev_rows = d + have = {str(i) for i in d["shares"].index} + todo = [ym for ym in candidates() if ym not in have] + if limit: + todo = todo[:limit] + print(f"candidates: {len(candidates())}, stored: {len(have)}, fetching: {len(todo)}") + + rows_shares: dict = {} + rows_value: dict = {} + with ThreadPoolExecutor(max_workers=WORKERS) as ex: + futs = {ex.submit(fetch_one, ym, symbols): ym for ym in todo} + done = 0 + for fut in as_completed(futs): + res = fut.result() + done += 1 + if res is not None: + ym, shares, value = res + for (date, sym), q in shares.items(): + rows_shares.setdefault(date, {})[sym] = q + rows_value.setdefault(date, {})[sym] = value[(date, sym)] + if done % 24 == 0: + print(f" {done}/{len(todo)} files") + + def frame(rows: dict) -> pd.DataFrame: + df = pd.DataFrame.from_dict(rows, orient="index") + df.index = pd.to_datetime(df.index) + return df.sort_index() + + new_shares = frame(rows_shares) + new_value = frame(rows_value) + parts_s = [f for f in (prev_rows["shares"] if prev_rows else None, new_shares) if f is not None and len(f)] + parts_v = [f for f in (prev_rows["value"] if prev_rows else None, new_value) if f is not None and len(f)] + df_s = pd.concat(parts_s).sort_index() + df_s = df_s[~df_s.index.duplicated(keep="last")].astype("float32") + df_v = pd.concat(parts_v).sort_index() + df_v = df_v[~df_v.index.duplicated(keep="last")].astype("float32") + with open(OUT, "wb") as fh: + pickle.dump({"shares": df_s, "value": df_v}, fh) + print(f"stored shares {df_s.shape} value {df_v.shape} -> {OUT}") + + +if __name__ == "__main__": + main() diff --git a/research/lab/fetch_gdelt.py b/research/lab/fetch_gdelt.py new file mode 100644 index 0000000..2ede83b --- /dev/null +++ b/research/lab/fetch_gdelt.py @@ -0,0 +1,123 @@ +"""Fetch GDELT news-volume timelines for the S&P 500 panel. + +One GDELT DOC 2.0 timelinevol query per company returns the full daily +series since 2017 (news volume intensity = share of all monitored news). +Company names come from the SEC ticker map (company_tickers.json); legal +suffixes are trimmed for the phrase query. The public API is burst-limited +(429s), so requests are paced with adaptive backoff. + +Cache: data/research/gdelt/.json (resume-safe). Output panel: +data/research/gdelt_volume.pkl {"volume": DataFrame} on daily dates. + + PYTHONPATH=. uv run python -m research.lab.fetch_gdelt [--limit N] +""" +from __future__ import annotations + +import json +import os +import pickle +import re +import sys +import time + +import pandas as pd +import requests + +from research.lab.fetch_us import _proxy_env + +OUT_DIR = os.path.join("data", "research", "gdelt") +PANEL = os.path.join("data", "research", "gdelt_volume.pkl") +TICKERS_CACHE = "data/research/company_tickers.json" +UA = {"User-Agent": "quant-research/0.1 gahow@example.com"} +BASE = ("https://api.gdeltproject.org/api/v2/doc/doc?query={q}%20sourcelang:english" + "&mode=timelinevol&format=json&startdatetime=20170101000000&enddatetime=20261007000000") +SUFFIX = re.compile(r",?\s*(inc|incorporated|corp|corporation|co|company|ltd|limited|plc|" + r"holdings?|group|the|nv|sa|ag|llc|lp|trust|reit)\.?$", re.I) + + +def clean_name(title: str) -> str: + name = title.strip().rstrip(".") + prev = None + while prev != name: + prev = name + name = SUFFIX.sub("", name).strip().rstrip(",") + return name + + +def query(name: str) -> list | None: + q = requests.utils.quote(f'"{name}"') + url = BASE.format(q=q) + delay = 5.0 + for attempt in range(5): + try: + r = requests.get(url, headers=UA, timeout=90) + if r.status_code == 429: + time.sleep(delay) + delay = min(delay * 2, 60) + continue + r.raise_for_status() + d = r.json() + series = d.get("timeline", [{}])[0].get("data", []) + return series + except Exception as exc: # noqa: BLE001 + if attempt == 4: + print(f" FAILED {name}: {exc}") + else: + time.sleep(delay) + delay = min(delay * 2, 60) + return None + + +def main() -> None: + _proxy_env() + os.makedirs(OUT_DIR, exist_ok=True) + limit = None + if "--limit" in sys.argv: + limit = int(sys.argv[sys.argv.index("--limit") + 1]) + close = pd.read_csv("data/us.csv", index_col=0, nrows=0) + tickers = list(close.columns) + raw = json.load(open(TICKERS_CACHE)) + title_by_ticker = {} + for v in raw.values(): + t = str(v["ticker"]).upper().replace(".", "-") + title_by_ticker.setdefault(t, str(v["title"])) + todo = [t for t in tickers if t in title_by_ticker and + not os.path.exists(os.path.join(OUT_DIR, f"{t}.json"))] + if limit: + todo = todo[:limit] + print(f"tickers: {len(tickers)}, cached: {len(tickers)-len(todo)}, fetching: {len(todo)}") + for i, t in enumerate(todo): + series = query(clean_name(title_by_ticker[t])) + if series is None: + continue + with open(os.path.join(OUT_DIR, f"{t}.json"), "w") as fh: + json.dump(series, fh) + if (i + 1) % 25 == 0: + print(f" {i+1}/{len(todo)} (last {t}, {len(series)} points)", flush=True) + time.sleep(12.0) + + # assemble panel + frames = {} + for t in tickers: + path = os.path.join(OUT_DIR, f"{t}.json") + if not os.path.exists(path): + continue + try: + series = json.load(open(path)) + idx = pd.to_datetime([p["date"][:8] for p in series], format="%Y%m%d") + vals = [p["value"] for p in series] + frames[t] = pd.Series(vals, index=idx) + except Exception: # noqa: BLE001 + continue + if frames: + df = pd.DataFrame(frames).sort_index() + df = df[~df.index.duplicated(keep="last")] + with open(PANEL, "wb") as fh: + pickle.dump({"volume": df.astype("float32")}, fh) + print(f"panel {df.shape} -> {PANEL}") + else: + print("no cached series yet") + + +if __name__ == "__main__": + main() diff --git a/research/lab/fetch_insider.py b/research/lab/fetch_insider.py index 50016f1..6644bc9 100644 --- a/research/lab/fetch_insider.py +++ b/research/lab/fetch_insider.py @@ -1,9 +1,12 @@ """ Fetch SEC "Insider Transactions Data Sets" (structured Form 3/4/5 data, quarterly zips from 2006Q1) and keep the open-market trades of officers and -directors of S&P 500 ever-members in the research price panel. +directors of every requested panel universe's ever-members. The zips are +parsed once; the issuer->ticker mapping and price checks run per universe +(us = S&P 500 research panel, mid = S&P 400, small = S&P 600). - uv run python -m research.lab.fetch_insider + uv run python -m research.lab.fetch_insider # us (insider_trades.pkl) + PYTHONPATH=. uv run python -m research.lab.fetch_insider --universe mid small Source: https://www.sec.gov/data-research/sec-markets-data/insider-transactions-data-sets The zip links are read from that page (the 2026Q2 file moved to a new path). @@ -30,7 +33,7 @@ filing's own ISSUERTRADINGSYMBOL, limited to the dates that symbol was used and to dates before the ticker's current owner first filed, and kept only if trade prices match the ticker's unadjusted closes (median error < 25%). -Writes data/research/insider_trades.pkl: +Writes data/research/insider_trades{,_{mid,small}}.pkl: {"trades": DataFrame, one row per kept transaction: ticker, cik, insider_cik, role, title, ten_pct, filed, trans_date, code, shares, price, value, price_ok, plan_10b51, plan_fn, ownership, @@ -52,6 +55,7 @@ Rule 10b5-1 (available in every year). from __future__ import annotations +import argparse import csv import os import pickle @@ -219,13 +223,14 @@ def norm_symbols(sym) -> list[str]: return [re.sub(r"\s+", "", x).replace(".", "-") for x in parts if x.strip()] -def raw_closes(tickers: list[str]) -> pd.DataFrame | None: +def raw_closes(tickers: list[str], fund: str = FUND_RAW) -> pd.DataFrame | None: """Unadjusted closes (split-adjusted close x later split ratios), from the - fundamentals fetch; used to check that an issuer's trade prices belong to - the ticker, and to value trades whose reported price is off.""" - if not os.path.exists(FUND_RAW): + fundamentals fetch of that universe; used to check that an issuer's trade + prices belong to the ticker, and to value trades whose reported price is + off.""" + if not os.path.exists(fund): return None - with open(FUND_RAW, "rb") as fh: + with open(fund, "rb") as fh: raw = pickle.load(fh) close_sa = raw["close_sa"].reindex(columns=tickers) ratio = raw["splits"].reindex(index=close_sa.index, columns=tickers).fillna(0.0).replace(0.0, 1.0) @@ -317,29 +322,28 @@ def map_tickers(trades: pd.DataFrame, filings: pd.DataFrame, panel_tickers: list return trades, tm.reset_index(drop=True) -def main() -> None: - for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"): - os.environ.pop(k, None) # direct connection for cik_map as well - sess = _session() - paths = download(sess) - parts, fil, t0 = [], [], time.time() - for path in paths: - tr, f = parse_quarter(path) - parts.append(tr) - fil.append(f) - print(f" {os.path.basename(path)}: {len(tr):,} P/S lines by officers/directors " - f"[{time.time() - t0:.0f}s]", flush=True) - trades = pd.concat(parts, ignore_index=True) - filings = pd.concat(fil, ignore_index=True).drop_duplicates("ACCESSION_NUMBER") - trades = trades.drop_duplicates(["accession", "trans_date", "code", "shares", "price", "ownership"]) - trades["cik"] = trades["cik"].astype("int64") - filings["cik"] = filings["cik"].astype("int64") - print(f"{len(trades):,} kept lines, {len(filings):,} Form 4 filings", flush=True) +UNIVERSE_OUT = {"us": OUT, + "mid": os.path.join("data", "research", "insider_trades_mid.pkl"), + "small": os.path.join("data", "research", "insider_trades_small.pkl")} - with open(os.path.join("data", "research", "us_ohlcv.pkl"), "rb") as fh: - close = pickle.load(fh)["close"] + +def universe_inputs(u: str) -> tuple[pd.DataFrame, pd.DataFrame | None]: + """(panel closes, unadjusted closes) of one universe's research panel.""" + if u == "us": + with open(os.path.join("data", "research", "us_ohlcv.pkl"), "rb") as fh: + close = pickle.load(fh)["close"] + fund = FUND_RAW + else: + from research.lab.panel_mid import load_us_mid + close = load_us_mid(u).close + fund = os.path.join("data", "research", f"us_{u}_fund_raw.pkl") + return close, raw_closes(list(close.columns), fund) + + +def finish(trades: pd.DataFrame, filings: pd.DataFrame, close: pd.DataFrame, + raw: pd.DataFrame | None, out: str, label: str) -> None: + """Map the parsed trades to one panel's tickers, value them, and save.""" tickers = list(close.columns) - raw = raw_closes(tickers) trades, tm = map_tickers(trades, filings, tickers, raw, close) # value at the reported price unless it is > 2x off the issuer's usual @@ -359,13 +363,40 @@ def main() -> None: "shares", "price", "value", "price_ok", "plan_10b51", "plan_fn", "ownership", "accession"] trades = trades[cols].sort_values(["filed", "ticker"]).reset_index(drop=True) unmapped = sorted(set(tickers) - set(tm["ticker"])) - print(f"{len(trades):,} trades in {trades['ticker'].nunique()} panel tickers " + print(f"[{label}] {len(trades):,} trades in {trades['ticker'].nunique()} panel tickers " f"({(trades['code'] == 'P').sum():,} P / {(trades['code'] == 'S').sum():,} S); " f"map: {(tm['source'] == 'sec_map').sum()} by CIK, {(tm['source'] == 'symbol').sum()} by symbol; " f"unmapped {unmapped}", flush=True) - with open(OUT, "wb") as fh: + with open(out, "wb") as fh: pickle.dump({"trades": trades, "tickers": tm, "unmapped": unmapped}, fh, protocol=4) - print(f"Saved {OUT}") + print(f"Saved {out}", flush=True) + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--universe", nargs="+", default=["us"], choices=["us", "mid", "small"]) + args = ap.parse_args() + for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"): + os.environ.pop(k, None) # direct connection for cik_map as well + sess = _session() + paths = download(sess) + parts, fil, t0 = [], [], time.time() + for path in paths: + tr, f = parse_quarter(path) + parts.append(tr) + fil.append(f) + print(f" {os.path.basename(path)}: {len(tr):,} P/S lines by officers/directors " + f"[{time.time() - t0:.0f}s]", flush=True) + trades = pd.concat(parts, ignore_index=True) + filings = pd.concat(fil, ignore_index=True).drop_duplicates("ACCESSION_NUMBER") + trades = trades.drop_duplicates(["accession", "trans_date", "code", "shares", "price", "ownership"]) + trades["cik"] = trades["cik"].astype("int64") + filings["cik"] = filings["cik"].astype("int64") + print(f"{len(trades):,} kept lines, {len(filings):,} Form 4 filings", flush=True) + + for u in args.universe: + close, raw = universe_inputs(u) + finish(trades.copy(), filings, close, raw, UNIVERSE_OUT[u], u) if __name__ == "__main__": diff --git a/research/lab/fetch_shortvolume.py b/research/lab/fetch_shortvolume.py new file mode 100644 index 0000000..6988047 --- /dev/null +++ b/research/lab/fetch_shortvolume.py @@ -0,0 +1,148 @@ +"""Fetch FINRA REG SHO daily short-sale volume for the research universes. + +One public file per trading day covers every NMS symbol (no auth): + + https://cdn.finra.org/equity/regsho/daily/CNMSshvolYYYYMMDD.txt + Date|Symbol|ShortVolume|ShortExemptVolume|TotalVolume|Market + +Writes data/research/short_volume.pkl: {"short": DataFrame, "total": DataFrame} +(float32, date x symbol), filtered to the symbols of data/us.csv plus the US +mid/small panels (us_mid / us_midsmall). Resume-safe: re-running fetches only +days that are missing from the store. Network goes through QUANT_MARKET_PROXY +when set (same convention as the other fetchers). + + PYTHONPATH=. uv run python -m research.lab.fetch_shortvolume [--days-limit N] +""" +from __future__ import annotations + +import json +import os +import pickle +import sys +from concurrent.futures import ThreadPoolExecutor, as_completed + +import pandas as pd +import requests + +from research.lab.fetch_us import _proxy_env + +OUT = os.path.join("data", "research", "short_volume.pkl") +SYMBOLS_CACHE = os.path.join("data", "research", "shortvolume_symbols.json") +URL = "https://cdn.finra.org/equity/regsho/daily/CNMSshvol{day}.txt" +HEADERS = {"User-Agent": "Mozilla/5.0 (research; contact local)"} +WORKERS = 6 + + +def universe_symbols() -> list[str]: + if os.path.exists(SYMBOLS_CACHE): + return json.load(open(SYMBOLS_CACHE)) + syms = set() + us = pd.read_csv("data/us.csv", index_col=0, nrows=0) + syms.update(us.columns) + for path in ("data/research/us_mid.panel.pkl", "data/research/us_midsmall.panel.pkl"): + if os.path.exists(path): + with open(path, "rb") as fh: + panel = pickle.load(fh) + cols = getattr(panel, "close").columns if hasattr(panel, "close") else None + if cols is None and isinstance(panel, dict): + cols = panel["close"].columns + syms.update(str(c).replace(".", "-") for c in cols) + syms.discard("") + out = sorted(syms) + json.dump(out, open(SYMBOLS_CACHE, "w")) + return out + + +def trading_days() -> list[str]: + close = pd.read_csv("data/us.csv", index_col=0, usecols=[0], parse_dates=True) + days = close.index + if os.path.exists("data/etfs.csv"): + etfs = pd.read_csv("data/etfs.csv", index_col=0, usecols=[0], parse_dates=True) + days = days.union(etfs.index) + return [d.strftime("%Y%m%d") for d in days] + + +def fetch_day(day: str, symbols: set[str]) -> tuple[str, dict, dict] | None: + url = URL.format(day=day) + for attempt in range(4): + try: + r = requests.get(url, headers=HEADERS, timeout=40) + if r.status_code in (403, 404): + # FINRA's public archive keeps only ~8 years; older days are gone. + return day, {}, {} + r.raise_for_status() + short, total = {}, {} + for line in r.text.splitlines()[1:]: + parts = line.split("|") + if len(parts) < 5: + continue + sym = parts[1] + if sym not in symbols: + continue + try: + s, t = float(parts[2]), float(parts[4]) + except ValueError: + continue + short[sym] = short.get(sym, 0.0) + s + total[sym] = total.get(sym, 0.0) + t + return day, short, total + except Exception as exc: # noqa: BLE001 + if attempt == 3: + print(f" FAILED {day}: {exc}") + else: + import time + time.sleep(1.5 * (attempt + 1)) + return None + + +def main() -> None: + _proxy_env() + limit = None + if "--days-limit" in sys.argv: + limit = int(sys.argv[sys.argv.index("--days-limit") + 1]) + symbols = set(universe_symbols()) + print(f"universe symbols: {len(symbols)}") + days = trading_days() + + prev_short = prev_total = None + have: set[str] = set() + if os.path.exists(OUT): + with open(OUT, "rb") as fh: + d = pickle.load(fh) + prev_short, prev_total = d["short"], d["total"] + have = {pd.Timestamp(i).strftime("%Y%m%d") for i in prev_short.index} + todo = [d for d in days if d not in have] + if limit: + todo = todo[:limit] + print(f"days: {len(days)}, already stored: {len(have)}, fetching: {len(todo)}") + + short_rows: dict[str, dict] = {} + total_rows: dict[str, dict] = {} + with ThreadPoolExecutor(max_workers=WORKERS) as ex: + futs = {ex.submit(fetch_day, d, symbols): d for d in todo} + done = 0 + for fut in as_completed(futs): + res = fut.result() + done += 1 + if res is not None: + day, short, total = res + short_rows[day] = short + total_rows[day] = total + if done % 200 == 0: + print(f" {done}/{len(todo)} fetched") + + new_short = pd.DataFrame.from_dict(short_rows, orient="index") + new_total = pd.DataFrame.from_dict(total_rows, orient="index") + for df in (new_short, new_total): + df.index = pd.to_datetime(df.index, format="%Y%m%d") + frames_short = [f for f in (prev_short, new_short) if f is not None and len(f)] + frames_total = [f for f in (prev_total, new_total) if f is not None and len(f)] + df_short = pd.concat(frames_short).sort_index() + df_total = pd.concat(frames_total).sort_index() + with open(OUT, "wb") as fh: + pickle.dump({"short": df_short.astype("float32"), "total": df_total.astype("float32")}, fh) + print(f"stored {df_short.shape} short / {df_total.shape} total -> {OUT}") + + +if __name__ == "__main__": + main() diff --git a/research/lab/fetch_tiingo.py b/research/lab/fetch_tiingo.py index ea7da95..4b38765 100644 --- a/research/lab/fetch_tiingo.py +++ b/research/lab/fetch_tiingo.py @@ -5,8 +5,10 @@ Used for: (1) completing delisted/missing names in the research panels, from TIINGO_API_KEY (environment or .env) and is never logged or written. Writes data/research/tiingo/.csv (resume-safe; a cached symbol never -counts against the monthly quota again). Columns: as-traded open/high/low/ -close/volume (Tiingo's close is NOT split-adjusted), split/dividend-adjusted +counts against the monthly quota again) and .miss for symbols Tiingo +does not serve (also skipped on later runs, so dead names do not re-burn the +500-symbol monthly quota). Columns: as-traded open/high/low/close/volume +(Tiingo's close is NOT split-adjusted), split/dividend-adjusted adjOpen..adjVolume, plus splitFactor and divCash. PYTHONPATH=. uv run python -m research.lab.fetch_tiingo TICKER [TICKER ...] @@ -91,7 +93,8 @@ def main() -> None: os.makedirs(DIR, exist_ok=True) key = api_key() - todo = [t for t in tickers if not os.path.exists(os.path.join(DIR, f"{t.upper()}.csv"))] + todo = [t for t in tickers if not os.path.exists(os.path.join(DIR, f"{t.upper()}.csv")) + and not os.path.exists(os.path.join(DIR, f"{t.upper()}.miss"))] print(f"{len(tickers)} tickers, {len(todo)} to fetch", flush=True) fetched = failed = 0 last = 0.0 @@ -107,6 +110,8 @@ def main() -> None: continue if df is None: failed += 1 + with open(os.path.join(DIR, f"{t.upper()}.miss"), "w") as fh: + fh.write("404\n") print(f"[{i + 1}/{len(todo)}] {t:8s} not found", flush=True) continue df.to_csv(os.path.join(DIR, f"{t.upper()}.csv")) diff --git a/research/lab/insider_mid.py b/research/lab/insider_mid.py new file mode 100644 index 0000000..1d50764 --- /dev/null +++ b/research/lab/insider_mid.py @@ -0,0 +1,286 @@ +""" +Form 4 officer/director open-market purchase factors on the PIT S&P 400 and +S&P 600 panels (fetch_insider.py --universe mid|small, SEC quarterly data +sets 2006 ->; a trade is usable the session after its filing). The S&P 500 +scan (research/lab/insider.py) found nothing; the literature places the +insider effect in the neglected small caps, so this is the mid/small test. + + PYTHONPATH=. uv run python -m research.lab.insider_mid --universe mid + PYTHONPATH=. uv run python -m research.lab.insider_mid --universe small + +Factors (same definitions as insider.py): insider_buyers_90d (net distinct +buyers, 90d), insider_net_value_180d (net $ / mcap, 180d), insider_ +opportunistic (CMP 2012 classification, net distinct, 90d), insider_cluster_ +buy (>= 2 distinct buyers, 30d). Sleeves use the midcap_sleeve construction +(top 30, equal weight, every 21 sessions, buffer 1, next-open fills, 20 bp a +side; 40 bp reported for S&P 600), averaged over the 21 rebalance phases. +The holdout hurdle is the EW PIT-member basket (which beat IJR by +2.8%/yr, +t 3.2, in the S&P 600 replication), not just the ETF. + +Result (2026-10-09): all four factors REJECT in both universes. S&P 400 +holdout 2017-26 (alpha vs the EW member basket): buyers -1.8%/yr (t -0.79), +net_value -1.9% (-0.79), opportunistic -0.1% (-0.06), cluster +0.3% (0.20; +halves +3.1/-2.8, the same 2017-21/2022-26 flip as the quality sleeves). +S&P 600: buyers -4.2%/yr (t -1.48 in both halves negative — insider buyers +cluster in broken names), net_value +0.7% (0.22, halves +5.2/-3.4), +opportunistic -1.6% (-0.86), cluster -1.6% (-0.89); at 40 bp all worse. The +pooled 400+600 event study finds a real but tiny drift after any purchase +(+5d excess +0.24%, t 2.8 in 2017-26; +63d +0.95%, t 1.8) that does not +survive the 20 bp round trip at portfolio scale, and the opportunistic (CMP) +split shows nothing (n ~1,000). Form 4 purchases do not extend the book into +mid/small caps — the same answer as the S&P 500 scan. +""" + +from __future__ import annotations + +import argparse +import dataclasses +import json +import os +import pickle +import warnings + +import numpy as np +import pandas as pd + +from research.lab import insider as ins +from research.lab.alpha import alpha_stats +from research.lab.engine import Costs +from research.lab.factors import overnight_mom, rec_126 +from research.lab.panel_mid import load_us_mid +from research.lab.us_final import monthly_blend +from research.lab.us_sweep import _blend, run + +warnings.filterwarnings("ignore") +OUT = os.path.join("data", "research", "results", "insider_midsmall") +BOOK = os.path.join("data", "research", "results", "universe_midcap", "returns.pkl") +TRADES = {"mid": os.path.join("data", "research", "insider_trades_mid.pkl"), + "small": os.path.join("data", "research", "insider_trades_small.pkl")} +ETF = {"mid": "IJH", "small": "IJR"} +HOLDOUT = ("2017-01-03", "2026-09-24") +HALVES = [("2017-01-03", "2021-12-31"), ("2022-01-03", "2026-09-24")] +REFERENCE = ("2010-01-04", "2016-12-30") +HORIZONS = (5, 21, 63) +EV_PERIODS = {"2010-2016": ("2010-01-01", "2016-12-30"), + "2017-2026": ("2017-01-03", "2026-09-24"), + "2017-2021": ("2017-01-03", "2021-12-31"), + "2022-2026": ("2022-01-03", "2026-09-24")} +KW = dict(top_n=30, buffer=1.0, every=21) +PREREG = { + "question": "Do officer/director Form 4 open-market purchase factors earn alpha over the EW PIT-member " + "basket among S&P 400 / S&P 600 members?", + "why": "the S&P 500 scan (insider.py) showed nothing; the insider-purchase effect is classically strongest " + "in neglected small caps, which a small account can trade", + "arms": ["c0_momentum (control)", "insider_buyers_90d", "insider_net_value_180d", + "insider_opportunistic", "insider_cluster_buy"], + "construction": "top 30, equal weight, every 21 sessions, buffer 1, next-open fills, 20 bp a side " + "(40 bp reported for S&P 600), 21 phases averaged (same as midcap_sleeve.py)", + "holdout": "2017-01-03..2026-09-24; halves 2017-2021 / 2022-2026", + "accept_if_all": [ + "A1: active return vs the EW PIT-member basket, holdout, Newey-West t >= 2.0, or t >= 1.5 with " + "active > 0 in both halves", + "A2: Sharpe(2/3 book + 1/3 sleeve) > Sharpe(2/3 book + 1/3 ETF), monthly reset, holdout " + "(book = universe_midcap base arm)", + ], + "on_accept": "forward-only validation book; no production change", + "multiple_testing": "4 factors x 2 universes = 8 sleeves; a lone t just above 2 is weak evidence; report all", + "reference_only": f"{REFERENCE[0]}..{REFERENCE[1]}: price coverage 55-78%, survivor-biased", + "status": "holdout partly seen in earlier mid/small studies; Tiingo-extended panel build 2026-10-08; " + "not independent OOS", +} + + +def build_factors(p, uni: str, factors: dict | None = None) -> tuple[dict, pd.DataFrame, pd.DataFrame]: + """(factor panels, exploded trades, ticker map) for one universe.""" + with open(TRADES[uni], "rb") as fh: + raw = pickle.load(fh) + tr, tm = raw["trades"], raw["tickers"] + if factors is not None: + tr = tr.assign(cmp=factors) + else: + tr = tr.assign(cmp=ins.classify_cmp(tr)) + ex = ins._explode(tr, tm) + col = pd.Series(np.arange(len(p.close.columns)), index=p.close.columns) + ex = ex[ex["ticker"].isin(col.index)].copy() + ex["col"] = col.reindex(ex["ticker"]).values + dates, tickers = p.dates, list(p.close.columns) + cov = ins.covered(tm, dates, tickers) + n = len(tickers) + buy, sell = ex[ex["code"] == "P"], ex[ex["code"] == "S"] + + def frame(x): + return pd.DataFrame(x, index=dates, columns=tickers).where(cov).astype("float32") + + from research.lab import fundamentals as fu + F = {} + F["insider_buyers_90d"] = frame(ins.window(buy, dates, n, 90) - ins.window(sell, dates, n, 90)) + signed = np.where(ex["code"] == "P", 1.0, -1.0) * ex["value"].fillna(0.0).values + net = ins.window(ex, dates, n, 180, weights=signed) + mcap = fu.load_mid(p, which=uni)["mcap"].reindex(index=dates, columns=tickers) + F["insider_net_value_180d"] = frame(net / mcap.values) + ob, osl = buy[buy["cmp"] == "opportunistic"], sell[sell["cmp"] == "opportunistic"] + opp = ins.window(ob, dates, n, 90) - ins.window(osl, dates, n, 90) + opp[dates < pd.Timestamp(ins.OPP_START)] = np.nan + F["insider_opportunistic"] = frame(opp) + F["insider_cluster_buy"] = frame((ins.window(buy, dates, n, 30) >= 2).astype(float)) + return F, ex, tm + + +def event_study(p, events: dict[str, pd.DataFrame]) -> dict: + """Market-adjusted CARs with month-clustered t (mirrors insider.event_study + with mid/small periods).""" + X = {h: ins.excess_fwd(p, h) for h in HORIZONS} + ym = np.asarray(p.dates.year * 100 + p.dates.month) + out = {} + for kind, e in events.items(): + out[kind] = {} + for per, (a, b) in EV_PERIODS.items(): + ra = p.dates.searchsorted(pd.Timestamp(a)) + rb = p.dates.searchsorted(pd.Timestamp(b), side="right") + ee = e[(e["row"] >= ra) & (e["row"] < rb)] + out[kind][per] = {} + for h in HORIZONS: + x = X[h][ee["row"].values, ee["col"].values] + m, t, n = ins.clustered_t(x, ym[ee["row"].values]) + out[kind][per][f"+{h}"] = {"mean": m, "t": t, "n": n} + return out + + +def score(f: pd.DataFrame, eligible: pd.DataFrame) -> pd.DataFrame: + return _blend(f.where(eligible).rank(axis=1, pct=True)).where(eligible) + + +def sharpe(r: pd.Series) -> float: + return float(r.mean() / r.std() * np.sqrt(252)) + + +def describe(r: pd.Series, ew: pd.Series, etf: pd.Series, book: pd.Series) -> dict: + """Holdout stats; ``ew`` is the EW PIT-member basket (the hurdle), ``etf`` + the universe ETF (reported).""" + h = r.loc[HOLDOUT[0]:HOLDOUT[1]] + e = (1 + h).cumprod() + a_ew, a_etf = alpha_stats(h, ew), alpha_stats(h, etf) + halves = [alpha_stats(r.loc[x:y], ew)["alpha"] for x, y in HALVES] + R = pd.DataFrame({"book": book, "s": r, "etf": etf}).loc[HOLDOUT[0]:HOLDOUT[1]].dropna() + with_s = monthly_blend(R, {"book": 2 / 3, "s": 1 / 3}) + with_e = monthly_blend(R, {"book": 2 / 3, "etf": 1 / 3}) + ref = r.loc[REFERENCE[0]:REFERENCE[1]] + return {"cagr": float(e.iloc[-1] ** (252 / len(h)) - 1), "sharpe": sharpe(h), + "mdd": float((e / e.cummax() - 1).min()), + "alpha_ew": float(a_ew["alpha"]), "alpha_ew_t": float(a_ew["t"]), + "alpha_etf": float(a_etf["alpha"]), "alpha_etf_t": float(a_etf["t"]), + "alpha_halves_vs_ew": [float(x) for x in halves], + "corr_book": float(R["s"].corr(R["book"])), + "book_sharpe_with_sleeve": sharpe(with_s), "book_sharpe_with_etf": sharpe(with_e), + "reference_sharpe": sharpe(ref), + "years": {int(k): float(v) for k, v in ((1 + h).groupby(h.index.year).prod() - 1).items()}} + + +def verdict(d: dict) -> dict: + a1 = d["alpha_ew_t"] >= 2.0 or (d["alpha_ew_t"] >= 1.5 and min(d["alpha_halves_vs_ew"]) > 0) + a2 = d["book_sharpe_with_sleeve"] > d["book_sharpe_with_etf"] + checks = {"A1": a1, "A2": a2} + return {"checks": {k: bool(v) for k, v in checks.items()}, + "verdict": "ACCEPT" if all(checks.values()) else "REJECT"} + + +def run_universe(uni: str) -> dict: + out_dir = os.path.join(OUT, uni) + os.makedirs(out_dir, exist_ok=True) + pre = os.path.join(out_dir, "prereg.json") + if not os.path.exists(pre): + with open(pre, "w") as fh: + json.dump({**PREREG, "universe": uni, "benchmark_hurdle": "EW members", + "etf": ETF[uni]}, fh, indent=2) + res_path = os.path.join(out_dir, "results.json") + done = json.load(open(res_path)) if os.path.exists(res_path) else {} + if done: + raise SystemExit(f"{res_path} exists; this study does not overwrite itself") + + p = load_us_mid(uni) + bench = p.extra["etf"][ETF[uni]].pct_change(fill_method=None) + book = pd.read_pickle(BOOK)["base"] + F, ex, tm = build_factors(p, uni) + mem = p.member + S = {"c0_momentum": _blend(overnight_mom(p).where(mem).rank(axis=1, pct=True), + rec_126(p).where(mem).rank(axis=1, pct=True)).where(mem)} + for name in ("insider_buyers_90d", "insider_net_value_180d", "insider_opportunistic", + "insider_cluster_buy"): + S[name] = score(F[name], mem) + + span = [(REFERENCE[0], HOLDOUT[1])] + costs = {"": p.costs} if uni == "mid" else {"": p.costs, "@40bp": Costs(buy=0.004, sell=0.004)} + rets = {} + for tag, c in costs.items(): + pc = dataclasses.replace(p, costs=c) + for name, sc in S.items(): + rets[name + tag] = run(pc, sc, span, **KW)["_rets"] + print(f"ran {uni} {name}{tag}", flush=True) + ew = p.close.pct_change(fill_method=None).where(p.member.shift(1, fill_value=False)).mean(axis=1) + rets["bench_ew_members"] = ew + rets[f"bench_{ETF[uni].lower()}"] = bench + + for name, r in rets.items(): + d = describe(r.reindex(p.dates).fillna(0.0), ew, bench, book) + if name.startswith("insider_") and not name.endswith("@40bp"): + d.update(verdict(d)) + done[name] = d + if not name.startswith("bench"): + print(f"{name:26s} cagr {d['cagr']*100:5.1f}% sh {d['sharpe']:.2f} mdd {d['mdd']*100:6.1f}% " + f"a_EW {d['alpha_ew']*100:+5.1f}% t {d['alpha_ew_t']:+.2f} halves " + f"{[round(x*100, 1) for x in d['alpha_halves_vs_ew']]} a_ETF t {d['alpha_etf_t']:+.2f} " + f"book+sleeve {d['book_sharpe_with_sleeve']:.2f} vs book+ETF {d['book_sharpe_with_etf']:.2f} " + f"ref {d['reference_sharpe']:.2f} {d.get('verdict', '')} {d.get('checks', '')}", flush=True) + + # ---- pooled event study on the midsmall panel + with open(TRADES["mid"], "rb") as fh: + raw_mid = pickle.load(fh) + with open(TRADES["small"], "rb") as fh: + raw_sm = pickle.load(fh) + pm = load_us_mid("midsmall") + tr = pd.concat([raw_mid["trades"], raw_sm["trades"]], ignore_index=True) + tr = tr.drop_duplicates(["accession", "code", "ticker", "trans_date", "shares", "price", "ownership"]) + tr = tr.assign(cmp=ins.classify_cmp(tr)) + tm2 = pd.concat([raw_mid["tickers"], raw_sm["tickers"]], ignore_index=True).drop_duplicates(["ticker", "cik"]) + exm = ins._explode(tr, tm2) + colm = pd.Series(np.arange(len(pm.close.columns)), index=pm.close.columns) + exm = exm[exm["ticker"].isin(colm.index)].copy() + exm["col"] = colm.reindex(exm["ticker"]).values + group = ins.share_class_group(tm2) + B, Sl = exm["code"] == "P", exm["code"] == "S" + ev = { + "all purchases": ins.trade_events(pm, exm[B], group), + "opportunistic purchases": ins.trade_events(pm, exm[B & (exm["cmp"] == "opportunistic")], group), + "routine purchases": ins.trade_events(pm, exm[B & (exm["cmp"] == "routine")], group), + "all sales": ins.trade_events(pm, exm[Sl], group), + } + es = event_study(pm, ev) + done["event_study_midsmall"] = es + print("\n=== pooled S&P 400+600 event study: excess vs EW members, % (t clustered by month) [n]") + for kind, dd in es.items(): + for h in HORIZONS: + cells = " ".join(f"{per}: {dd[per][f'+{h}']['mean']*100:+5.2f} ({dd[per][f'+{h}']['t']:+4.1f})" + f"[{dd[per][f'+{h}']['n']:5d}]" for per in EV_PERIODS) + print(f"{(kind if h == 5 else ''):26s} +{h:<3d} {cells}", flush=True) + + with open(res_path, "w") as fh: + json.dump(done, fh, indent=2, default=float) + pd.DataFrame({k: v for k, v in rets.items()}).to_pickle(os.path.join(out_dir, "returns.pkl")) + from research.lab.trials import append_trials + append_trials(os.path.join("data", "research", "results", "trials_us.jsonl"), + [{"name": f"insider_mid[{uni}]:{k}", **{x: y for x, y in v.items() if x != "years"}} + for k, v in done.items() if k in rets], + {"study": "insider_midsmall", "role": "preregistered; holdout partly seen; not independent OOS"}) + return done + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--universe", nargs="+", default=["mid", "small"], choices=["mid", "small"]) + args = ap.parse_args() + for uni in args.universe: + run_universe(uni) + + +if __name__ == "__main__": + main() diff --git a/research/lab/inst_flow.py b/research/lab/inst_flow.py new file mode 100644 index 0000000..5694286 --- /dev/null +++ b/research/lab/inst_flow.py @@ -0,0 +1,152 @@ +"""Institutional-flow factor scan from the DERA 13F aggregates. + +Per quarter, total institutional shares per CUSIP are aggregated by +fetch_13f.py. Here they map to tickers (issuer name -> SEC title map) and +become the quarterly factor: + + inst_flow = shares(q) / shares(q-1) - 1 (net institutional accumulation) + filer_chg = n_filers(q) - n_filers(q-1) (breadth of accumulation) + +PIT: the 13F deadline is 45 days after quarter end; the factor is applied +conservatively from quarter_end + 60 calendar days. + + PYTHONPATH=. uv run python -m research.lab.inst_flow [us|midsmall] +""" +from __future__ import annotations + +import glob +import json +import os +import pickle +import re +import sys + +import numpy as np +import pandas as pd + +DIR = os.path.join("data", "research", "dera13f") +TICKERS_CACHE = "data/research/company_tickers.json" +PUBLISH_LAG_DAYS = 60 + +_SUFFIX = re.compile(r"[,\s]*(inc|incorporated|corp|corporation|co|company|ltd|limited|plc|" + r"holdings?|group|the|nv|sa|ag|llc|lp|trust|reit|new|cl|a|b)[.\s]*$", re.I) + + +def norm_name(name: str) -> str: + if not isinstance(name, str): + return "" + n = name.upper().strip().rstrip(".") + n = re.sub(r"[^A-Z0-9 ]+", " ", n) + n = re.sub(r"\s+", " ", n) + prev = None + while prev != n: + prev = n + n = _SUFFIX.sub("", n).strip() + return n + + +def load_quarter(path: str) -> pd.DataFrame: + d = pd.read_csv(path, dtype={"CUSIP": str}) + d = d[d["CUSIP"].notna() & (d["CUSIP"] != "000000000")] + d["issuer"] = d["issuer"].astype(str) + d["norm"] = d["issuer"].map(norm_name) + return d + + +def main() -> None: + which = sys.argv[1] if len(sys.argv) > 1 else "us" + files = sorted(glob.glob(os.path.join(DIR, "agg_*.csv"))) + if not files: + raise SystemExit("no agg files; run fetch_13f first") + print(f"quarters: {len(files)}") + raw = json.load(open(TICKERS_CACHE)) + name2t = {} + for v in raw.values(): + name2t.setdefault(norm_name(str(v["title"])), str(v["ticker"]).upper().replace(".", "-")) + # combine quarters: cusip -> norm name from latest, and shares series + shares = {} + filers = {} + name_by_cusip = {} + for path in files: + q = os.path.basename(path)[4:-4] # YYYYqQ + d = load_quarter(path) + d = d[d["shares"].fillna(0) > 0] + shares[q] = d.set_index("CUSIP")["shares"] + filers[q] = d.set_index("CUSIP")["n_filers"] + for c, n in d.set_index("CUSIP")["norm"].items(): + name_by_cusip[c] = n + qs = [os.path.basename(p)[4:-4] for p in files] + matched, ticker_by_cusip = 0, {} + for c, n in name_by_cusip.items(): + t = name2t.get(n) + if t: + ticker_by_cusip[c] = t + matched += 1 + print(f"cusip->ticker matched: {matched}/{len(name_by_cusip)}") + + # factor frame on quarter end dates (apply from +PUBLISH_LAG_DAYS) + rows = {} + prev = None + for q in qs: + if prev is None: + prev = q + continue + s1 = shares[prev].reindex(shares[q].index) + flow = shares[q] / s1.replace(0.0, np.nan) - 1.0 + d_fil = filers[q] - filers[prev].reindex(filers[q].index) + # quarter-end date + y, qq = int(q[:4]), int(q[-1]) + qend = pd.Timestamp(y, qq * 3, 1) + pd.offsets.MonthEnd(0) + avail = qend + pd.Timedelta(days=PUBLISH_LAG_DAYS) + t = pd.Series({ticker_by_cusip[c]: f for c, f in flow.items() if c in ticker_by_cusip and np.isfinite(f)}) + tc = pd.Series({ticker_by_cusip[c]: f for c, f in d_fil.items() if c in ticker_by_cusip and np.isfinite(f)}) + if len(t): + rows[avail] = (t, tc) + prev = q + print(f"quarters with factors: {len(rows)}") + + if which == "us": + with open("data/research/us.panel.pkl", "rb") as fh: + p = pickle.load(fh) + else: + from research.lab.panel_mid import load_us_mid + p = load_us_mid(which) + close, member = p.close, p.member.reindex_like(p.close).fillna(False) + fwd21 = close.shift(-21) / close - 1.0 + # broadcast each quarterly factor until the next availability date + fac = pd.DataFrame(np.nan, index=close.index, columns=close.columns) + facf = fac.copy() + dates_avail = sorted(rows) + for i, dt in enumerate(dates_avail): + t, tc = rows[dt] + end = dates_avail[i + 1] if i + 1 < len(dates_avail) else close.index[-1] + sel = (close.index >= dt) & (close.index < end) + for name, ser in (("flow", t), ("filers", tc)): + frame = fac if name == "flow" else facf + frame.loc[sel, frame.columns.isin(ser.index)] = ser.reindex( + frame.columns[frame.columns.isin(ser.index)]).values + months = fwd21.index.to_period("M") + for name, frame in (("inst_flow", fac), ("filer_chg", facf)): + f = frame.where(member) + ics = [] + for per in sorted(set(months)): + idx = months == per + xdf = f.loc[idx] + if xdf.notna().sum(axis=1).max() < 50: + continue + ydf = fwd21.loc[idx] + mask = xdf.notna().to_numpy() & np.isfinite(ydf.reindex(columns=xdf.columns).to_numpy()) + if mask.sum() < 100: + continue + xr = xdf.rank(axis=1).to_numpy()[mask] + yr = ydf.reindex(columns=xdf.columns).where(mask).rank(axis=1).to_numpy()[mask] + xm, ym = xr - xr.mean(), yr - yr.mean() + denom = np.sqrt((xm ** 2).sum() * (ym ** 2).sum()) + if denom > 0: + ics.append(float((xm * ym).sum() / denom)) + ic = np.array(ics) + print(f"{name:10s} IC21 {ic.mean():+.4f} t={ic.mean()/ic.std()*np.sqrt(len(ic)):+.2f} months={len(ic)}") + + +if __name__ == "__main__": + main() diff --git a/research/lab/news_volume.py b/research/lab/news_volume.py new file mode 100644 index 0000000..1aedd39 --- /dev/null +++ b/research/lab/news_volume.py @@ -0,0 +1,78 @@ +"""GDELT news-volume factor scan on the S&P 500 PIT panel. + +Factors from the daily news-intensity series (share of monitored news): + news_spike = volume / 60-day rolling median + news_chg5 = 5-day mean / 21-day mean - 1 + news_z = z-score of the 5-day mean vs its 252-day history + +PIT: GDELT's day-D series is final at D's end; signals are lagged 1 session +(usable at the close of D+1 at the earliest, matching the engine convention). + + PYTHONPATH=. uv run python -m research.lab.news_volume +""" +from __future__ import annotations + +import json +import os +import pickle + +import numpy as np +import pandas as pd + +GDELT_PATH = "data/research/gdelt_volume.pkl" + + +def factors(vol: pd.DataFrame) -> dict[str, pd.DataFrame]: + med60 = vol.rolling(60, min_periods=20).median() + f = {} + f["news_spike"] = vol / med60.replace(0.0, np.nan) + m5 = vol.rolling(5, min_periods=3).mean() + m21 = vol.rolling(21, min_periods=10).mean() + f["news_chg5"] = m5 / m21.replace(0.0, np.nan) - 1.0 + f["news_z"] = (m5 - m5.rolling(252, min_periods=60).mean()) / m5.rolling( + 252, min_periods=60).std().replace(0.0, np.nan) + return f + + +def main() -> None: + with open("data/research/us.panel.pkl", "rb") as fh: + p = pickle.load(fh) + with open(GDELT_PATH, "rb") as fh: + vol = pickle.load(fh)["volume"].reindex(columns=p.close.columns) + print(f"gdelt panel: {vol.shape}, tickers with data: {int(vol.notna().any().sum())}") + close = p.close + member = p.member.reindex_like(close).fillna(False) + fwd1 = close.shift(-1) / close - 1.0 + fwd21 = close.shift(-21) / close - 1.0 + months = fwd1.index.to_period("M") + rows = [] + for fname, fraw in factors(vol).items(): + f = fraw.shift(1).reindex(index=close.index).where(member) + ics1, ics21 = [], [] + for per in sorted(set(months)): + idx = months == per + xdf = f.loc[idx] + if xdf.notna().sum(axis=1).max() < 50: + continue + for ydf, out in ((fwd1.loc[idx], ics1), (fwd21.loc[idx], ics21)): + mask = xdf.notna().to_numpy() & np.isfinite(ydf.reindex(columns=xdf.columns).to_numpy()) + if mask.sum() < 100: + continue + xr = xdf.rank(axis=1).to_numpy()[mask] + yr = ydf.reindex(columns=xdf.columns).where(mask).rank(axis=1).to_numpy()[mask] + xm, ym = xr - xr.mean(), yr - yr.mean() + denom = np.sqrt((xm ** 2).sum() * (ym ** 2).sum()) + if denom > 0: + out.append(float((xm * ym).sum() / denom)) + ic1, ic21 = np.array(ics1), np.array(ics21) + rows.append({"factor": fname, + "IC1": ic1.mean(), "t1": ic1.mean() / ic1.std() * np.sqrt(len(ic1)) if len(ic1) > 1 else np.nan, + "IC21": ic21.mean(), "t21": ic21.mean() / ic21.std() * np.sqrt(len(ic21)) if len(ic21) > 1 else np.nan, + "months": len(ic21)}) + df = pd.DataFrame(rows) + print(df.round(4).to_string(index=False)) + df.to_csv("data/research/results/news_volume_scan.csv", index=False) + + +if __name__ == "__main__": + main() diff --git a/research/lab/overnight_midsmall.py b/research/lab/overnight_midsmall.py new file mode 100644 index 0000000..89fc06f --- /dev/null +++ b/research/lab/overnight_midsmall.py @@ -0,0 +1,178 @@ +""" +The overnight-only momentum book on S&P 400/600 names: the same rule as +research/lab/overnight_only.py (top 10 by the live sleeve's score, bought at +the close, sold at the next open, flat in the day), on a second universe. Is +it a second, partly independent stream next to the S&P 500 one? + + PYTHONPATH=. uv run python -m research.lab.overnight_midsmall + +Rules (untuned copy of overnight_only's sleeve_score candidate): + universe PIT S&P 400 or S&P 600 members on t-2 (research/lab/panel_mid.py, + midsmall), close > $5 and 60-day median dollar volume >= $20m + (auction capacity), both read at t-2 + score overnight momentum (12-1) + distance from the 126-day low, + percentile blend among the eligible names, read at the close of t-2 + book top 10, equal weight, held close t-1 -> open t, rebuilt nightly + costs both auctions every night: c per side, base c = 3 bp (mid/small + auctions), stress c = 6 bp; 1 and 2 bp reported + cash earns nothing; Sharpe = mean / sd of the book's own returns + +Windows: membership coverage in the panel is >= 85% only from 2019 (Yahoo and +Tiingo miss delisted mid caps before), so the primary window is 2019-01-02 .. +2026-09-24 (halves 2019-2022 / 2023-2026). 2010-2018 is reported, survivor-biased. + +Combination (gate S4): the S&P 500 overnight sleeve (overnight_only daily.pkl, +1 bp) and this book at 3 bp, half the capital each, rebalanced daily. + +Fidelity (S5): Yahoo/Tiingo opens against the Alpaca 09:30 bar +(overnight_only.fidelity), 2016+. + +Every period of these panels was examined in earlier mid/small studies (all +REJECT/FAIL for all-day holding); the overnight split was not. The S&P 500 +overnight result was seen before this rule. The rule is written to +prereg.json before the run; the script refuses to overwrite results.json. + +Result (2026-10-09): STOP. S2 and S5 pass; S1, S3 and S4 fail. + + sleeve_score, primary 2019-01..2026-09 + gross 15.3 bp/day + net Sharpe by cost per side: 1 bp 1.32, 2 bp 1.12, 3 bp 0.93, 6 bp 0.33 + at 3 bp: 22.5%/yr, MaxDD -33%, halves 0.78 / 1.09 + 2010-18 reference (survivor-biased): 0.65 + overnight_12_1 (reported): 0.76 at 3 bp + Fidelity: 1,254 names, corr 0.975 on all name-days, 0.996 on selected; + mean diff -0.2 bp. + + - The gross overnight edge is about the S&P 500's (15.3 vs 13.5 bp/day), + but mid/small auctions cost more. + - It is not a second stream: daily corr with the S&P 500 overnight sleeve + is 0.83, and a 50/50 combination lowers the Sharpe (1.17 -> 1.09). +Interpretation: overnight momentum is one common factor across size +buckets, not a per-universe edge. More names do not diversify it; cheaper +auctions (the S&P 500) are where to hold it. +""" + +from __future__ import annotations + +import json +import os + +import numpy as np +import pandas as pd + +from research.lab.factor_scan import RESULTS +from research.lab.factors import overnight_mom, rec_126 +from research.lab.overnight_only import describe, fidelity, legs, nightly, top_mask +from research.lab.panel_mid import load_us_mid +from research.lab.us_sweep import _blend + +OUT = os.path.join(RESULTS, "overnight_midsmall") +LARGE = os.path.join(RESULTS, "overnight_only", "daily.pkl") +PRIMARY = ("2019-01-02", "2026-09-24") +HALVES = [("2019-01-02", "2022-12-30"), ("2023-01-03", "2026-09-24")] +REFERENCE = ("2010-01-04", "2018-12-31") +TOP_N = 10 +COST_BASE, COST_STRESS = 0.0003, 0.0006 +MIN_PRICE, MIN_DV = 5.0, 20e6 + +PREREG = { + "question": "Does the overnight-only momentum book (S&P 500 rule, untuned) earn a net Sharpe >= 1.0 on " + "S&P 400/600 names at mid/small auction costs, and add to the S&P 500 overnight sleeve?", + "rules": "PIT S&P 400/600 at t-2, close > $5, 60d median $ volume >= $20m; sleeve score at t-2; top 10 EW; " + "close t-1 -> open t; 2 x c per night, c = 3 bp base / 6 bp stress", + "primary": PRIMARY, "halves": HALVES, "reference_survivor_biased": REFERENCE, + "gates": {"S1": "primary net Sharpe (3 bp) >= 1.0", + "S2": "net Sharpe (3 bp) >= 0.7 in both primary halves", + "S3": "primary stress Sharpe (6 bp) >= 0.5", + "S4": "primary: daily corr with the S&P 500 overnight sleeve <= 0.6 and the 50/50 combination's " + "Sharpe >= the S&P 500 sleeve's + 0.1", + "S5": "open fidelity vs Alpaca (2016+): corr >= 0.9 on all name-days, |mean diff| <= 3 bp on " + "selected name-days"}, + "decision": {"CONTINUE": "S1-S5 pass -> add a mid/small overnight leg to the paper test", + "DATA": "S5 fails -> rebuild from Alpaca opens first", + "STOP": "otherwise"}, + "reported": "overnight_12_1 score, costs 1/2 bp, 2010-2018 reference, yearly returns, auction $ capacity", + "data_status": "panels examined in earlier mid/small studies; overnight split new; S&P 500 overnight seen", +} + + +def main() -> None: + os.makedirs(OUT, exist_ok=True) + res_path = os.path.join(OUT, "results.json") + if os.path.exists(res_path): + raise SystemExit(f"{res_path} exists; this study does not overwrite results") + with open(os.path.join(OUT, "prereg.json"), "w") as fh: + json.dump(PREREG, fh, indent=2) + + p = load_us_mid("midsmall") + dv = (p.close * p.volume).rolling(60, min_periods=40).median() + elig = p.member & (p.close > MIN_PRICE) & (dv >= MIN_DV) + on, _ = legs(p) + scores = { + "sleeve_score": _blend(*[f.where(elig).rank(axis=1, pct=True) + for f in (overnight_mom(p), rec_126(p))]).where(elig), + "overnight_12_1": overnight_mom(p).where(elig), + } + large = pd.read_pickle(LARGE)["sleeve_score"] + res, sels, daily = {"candidates": {}}, {}, {} + for name, sc in scores.items(): + sel = top_mask(sc.shift(2), TOP_N) + sels[name] = sel + gross = nightly(sel, on) + d = {} + for c in (0.0001, 0.0002, COST_BASE, COST_STRESS): + net = (gross - 2 * c).loc[REFERENCE[0]:PRIMARY[1]] + d[f"{c*1e4:.0f}bp"] = {"primary": describe(net.loc[PRIMARY[0]:PRIMARY[1]]), + "halves": [describe(net.loc[a:b]) for a, b in HALVES], + "reference": describe(net.loc[REFERENCE[0]:REFERENCE[1]])} + base = (gross - 2 * COST_BASE).loc[PRIMARY[0]:PRIMARY[1]] + daily[name] = gross - 2 * COST_BASE + L = large.reindex(base.index) + combo = 0.5 * base.fillna(0.0) + 0.5 * L.fillna(0.0) + d["corr_with_large"] = float(base.corr(L)) + d["large_primary"] = describe(L.dropna()) + d["combo_primary"] = describe(combo) + d["gross_primary"] = describe(gross.loc[PRIMARY[0]:PRIMARY[1]]) + d["yearly_3bp"] = {int(y): float(v) for y, v in + ((1 + daily[name].loc[REFERENCE[0]:PRIMARY[1]]).groupby( + daily[name].loc[REFERENCE[0]:PRIMARY[1]].index.year).prod() - 1).items()} + held_dv = dv.shift(2).where(sel).loc[PRIMARY[0]:PRIMARY[1]] + d["selected_median_dollar_volume"] = float(np.nanmedian(held_dv.values)) + d["eligible_per_day"] = float(elig.loc[PRIMARY[0]:PRIMARY[1]].sum(axis=1).mean()) + res["candidates"][name] = d + b = d["3bp"] + print(f"{name:15s} gross {d['gross_primary']['mean_bp_per_day']:5.1f}bp/d | 3bp: cagr " + f"{b['primary']['cagr']*100:6.1f}% sh {b['primary']['sharpe']:5.2f} mdd {b['primary']['mdd']*100:6.1f}% " + f"halves {b['halves'][0]['sharpe']:5.2f} {b['halves'][1]['sharpe']:5.2f} | ref 2010-18 sh " + f"{b['reference']['sharpe']:5.2f} | 1bp {d['1bp']['primary']['sharpe']:5.2f} 2bp " + f"{d['2bp']['primary']['sharpe']:5.2f} 6bp {d['6bp']['primary']['sharpe']:5.2f} | corr large " + f"{d['corr_with_large']:.2f} combo sh {d['combo_primary']['sharpe']:.2f} (large " + f"{d['large_primary']['sharpe']:.2f}) | eligible {d['eligible_per_day']:.0f} med $vol " + f"{d['selected_median_dollar_volume']/1e6:.0f}m", flush=True) + print(" yearly net 3bp %:", {y: round(v * 100, 1) for y, v in d["yearly_3bp"].items()}) + + fid = fidelity(p, sels) + res["fidelity"] = fid + print("fidelity:", {k: (round(v, 3) if isinstance(v, float) else v) for k, v in fid.items()}) + s5 = (fid.get("symbols", 0) > 0 and fid["corr_all"] >= 0.9 + and all(abs(fid[f"{c}_mean_diff_bp"]) <= 3 for c in scores if fid.get(f"{c}_name_days", 0) > 0)) + d = res["candidates"]["sleeve_score"] + checks = {"S1": d["3bp"]["primary"]["sharpe"] >= 1.0, + "S2": all(h["sharpe"] >= 0.7 for h in d["3bp"]["halves"]), + "S3": d["6bp"]["primary"]["sharpe"] >= 0.5, + "S4": d["corr_with_large"] <= 0.6 and d["combo_primary"]["sharpe"] >= d["large_primary"]["sharpe"] + 0.1, + "S5": s5} + decision = "DATA" if not s5 else ("CONTINUE" if all(checks.values()) else "STOP") + res["verdict"] = {"checks": checks, "decision": decision} + print(f"checks {checks} -> {decision}") + with open(res_path, "w") as fh: + json.dump(res, fh, indent=2) + pd.DataFrame(daily).to_pickle(os.path.join(OUT, "daily.pkl")) + from research.lab.trials import append_trials + append_trials(os.path.join(RESULTS, "trials_us.jsonl"), + [{"name": f"overnight_midsmall:{n}", **res["candidates"][n]["3bp"]["primary"]} for n in scores], + {"study": "overnight_midsmall", "role": "preregistered quick check"}) + + +if __name__ == "__main__": + main() diff --git a/research/lab/short_volume.py b/research/lab/short_volume.py new file mode 100644 index 0000000..259ce73 --- /dev/null +++ b/research/lab/short_volume.py @@ -0,0 +1,121 @@ +"""Short-volume (FINRA REG SHO daily) factor scan on PIT US panels. + +SVR = ShortVolume / TotalVolume per symbol-day. The REG SHO file for day D is +published after D's close; the scan therefore shifts all signals by 2 sessions +(conservative: usable at the close of D+2 at the earliest) before correlating +with forward returns. + +Universes: research/lab panels us (S&P 500 ever-members) and midsmall +(S&P 400 + 600) — the mid/small cut is where a sub-$1M book has a capacity +edge over large funds. Output: monthly Spearman IC (1d/21d forward), quintile +spread means, and per-year ICs for the best factor. + + PYTHONPATH=. uv run python -m research.lab.short_volume [us|midsmall|all] +""" +from __future__ import annotations + +import pickle +import sys + +import numpy as np +import pandas as pd + +SHORT_PATH = "data/research/short_volume.pkl" + + +def load_svr(panel_close: pd.DataFrame) -> pd.DataFrame: + with open(SHORT_PATH, "rb") as fh: + d = pickle.load(fh) + short, total = d["short"], d["total"] + cols = [c for c in panel_close.columns if c in short.columns] + svr = (short[cols] / total[cols].replace(0.0, np.nan)).astype("float32") + svr = svr.reindex(panel_close.index.union(svr.index)).sort_index() + return svr + + +def factors(svr: pd.DataFrame) -> dict[str, pd.DataFrame]: + f = {} + f["svr5"] = svr.rolling(5, min_periods=3).mean() + f["svr21"] = svr.rolling(21, min_periods=10).mean() + m = svr.rolling(252, min_periods=60).mean() + s = svr.rolling(252, min_periods=60).std() + f["svr_z"] = (f["svr21"] - m) / s.replace(0.0, np.nan) + f["svr_chg"] = f["svr5"] / f["svr21"].replace(0.0, np.nan) - 1.0 + return f + + +def scan(name: str, close: pd.DataFrame, member: pd.DataFrame, svr: pd.DataFrame) -> None: + fac = {k: v.reindex(close.index) for k, v in factors(svr).items()} + close = close.reindex(columns=svr.columns).ffill() + member = member.reindex_like(close).fillna(False).astype(bool) + fwd1 = close.shift(-1) / close - 1.0 + fwd21 = close.shift(-21) / close - 1.0 + lag = 2 + print(f"\n=== {name}: {close.shape[1]} symbols, " + f"{close.index.min().date()}..{close.index.max().date()} ===") + rows = [] + months = fwd1.index.to_period("M") + for fname, fraw in fac.items(): + f = fraw.shift(lag).where(member) + ics1, ics21 = [], [] + for per in sorted(set(months)): + idx = months == per + xdf = f.loc[idx] + if xdf.notna().sum(axis=1).max() < 50: + continue + for ydf, out in ((fwd1.loc[idx], ics1), (fwd21.loc[idx], ics21)): + mask = xdf.notna().to_numpy() & np.isfinite(ydf.to_numpy()) + if mask.sum() < 100: + continue + xr = xdf.rank(axis=1).to_numpy()[mask] + yr = ydf.where(mask).rank(axis=1).to_numpy()[mask] + xm = xr - xr.mean() + ym = yr - yr.mean() + denom = np.sqrt((xm ** 2).sum() * (ym ** 2).sum()) + if denom > 0: + out.append(float((xm * ym).sum() / denom)) + ic1 = np.array(ics1) + ic21 = np.array(ics21) + rows.append({ + "factor": fname, + "IC1_mean": ic1.mean(), "IC1_t": ic1.mean() / ic1.std() * np.sqrt(len(ic1)) if len(ic1) > 1 else np.nan, + "IC21_mean": ic21.mean(), "IC21_t": ic21.mean() / ic21.std() * np.sqrt(len(ic21)) if len(ic21) > 1 else np.nan, + "months": len(ic21)}) + print(pd.DataFrame(rows).round(4).to_string(index=False)) + + # quintile spread, 21-day holding, gross + for fname in ("svr21", "svr_chg", "svr_z"): + f = fac[fname].shift(lag).where(member) + pairs = [] + dates = f.index + for i in range(0, len(dates) - 21, 21): + t0 = dates[i] + x = f.loc[t0].dropna() + if len(x) < 50: + continue + q = pd.qcut(x.rank(method="first"), 5, labels=False) + lo, hi = q[q == 0].index, q[q == 4].index + r = close.iloc[i + 21] / close.iloc[i] - 1.0 + pairs.append((t0, float(r[hi].mean() - r[lo].mean()))) + s = np.array([p[1] for p in pairs]) + ys = pd.Series(s, index=pd.DatetimeIndex([p[0] for p in pairs])) + yearly = (1 + ys).groupby(ys.index.year).prod() - 1 + print(f"{fname}: Q5-Q1 21d spread mean {s.mean()*100:.2f}% " + f"t={s.mean()/s.std()*np.sqrt(len(s)):.2f} n={len(s)} " + f"yearly: " + " ".join(f"{y}:{v*100:+.0f}%" for y, v in yearly.items())) + + +def main() -> None: + which = sys.argv[1] if len(sys.argv) > 1 else "all" + if which in ("us", "all"): + with open("data/research/us.panel.pkl", "rb") as fh: + p = pickle.load(fh) + scan("S&P 500 ever-members", p.close, p.member, load_svr(p.close)) + if which in ("midsmall", "all"): + with open("data/research/us_midsmall.panel.pkl", "rb") as fh: + p = pickle.load(fh) + scan("S&P 400+600 (midsmall)", p.close, p.member, load_svr(p.close)) + + +if __name__ == "__main__": + main() diff --git a/research/lab/union_shield.py b/research/lab/union_shield.py new file mode 100644 index 0000000..6fc474f --- /dev/null +++ b/research/lab/union_shield.py @@ -0,0 +1,142 @@ +""" +Left-tail control on the S&P 500 ∪ S&P 400 union book: does a daily +market-regime shield change the universe_midcap verdict? + +Background (research/lab/universe_midcap.py): ranking S&P 400 names inside +overnight_recovery_top10 cost 4.7pp of CAGR and 0.12 of Sharpe (2019-26), +and its excess losses concentrated in down years (2022: -26pp). This applies +the same overlay to both arms and asks whether the union becomes acceptable +once the crash tail is cut. + + PYTHONPATH=. uv run python -m research.lab.union_shield + +Prereg (written before the run): exposure_t = 1{SPY close through t-1 above +its 150-day average through t-1}, else cash (0 return); applied to the +tranche-averaged daily book returns saved by universe_midcap.py; +20 bp charged per unit of exposure change. One mechanism, no variants. + +Result (2026-10-09): FAIL. The shield hurts every block of the base book +(whipsaw: it sits out 24-38% of 2020/2021/2025 bull legs) — base +29.4%/0.98 -> base+shield 16.1%/0.70 — and union+shield (15.5%/0.67) lands +below both base+shield (P1) and the plain unshielded base (P2). MaxDDs are +unchanged to the decimal (-34.3% base, -38.8% union+shield), so the filter +buys no tail either. The union ordering (union >= adv20 > minus_top3) is +unchanged under the shield. Conclusion: left-tail control does not salvage +the S&P 400 extension; a SPY>MA150 regime filter is simply bad for this +momentum book. +""" + +from __future__ import annotations + +import json +import os +import warnings + +import numpy as np +import pandas as pd + +from research.lab.alpha import alpha_stats +from research.lab.panel_mid import load_us_mid + +warnings.filterwarnings("ignore") +RESULTS = os.path.join("data", "research", "results", "universe_midcap") +OUT = os.path.join("data", "research", "results", "union_shield") +PRIMARY = ("2019-01-02", "2026-09-24") +BLOCKS = [("2019-01-02", "2020-12-31"), ("2021-01-04", "2022-12-30"), + ("2023-01-03", "2024-12-31"), ("2025-01-02", "2026-09-24")] +REFERENCE = ("2007-01-03", "2018-12-31") +COST = 0.002 +PREREG = { + "question": "With a SPY>MA150 regime shield (else cash) applied to both arms, does adding PIT " + "S&P 400 members to overnight_recovery_top10 still lose?", + "arms": ["base", "base+shield", "union", "union+shield", "union_adv20+shield (reported)"], + "mechanics": "exposure_t = 1{SPY close through t-1 > its 150d average through t-1}, else 0 (cash); " + "r'_t = e_t r_t - 0.002 |e_t - e_{t-1}|; inputs = universe_midcap/returns.pkl " + "(tranche-averaged, 20 bp a side already inside)", + "primary_window": PRIMARY, + "pass_if_all": [ + "P1: Sharpe(union+shield) > Sharpe(base+shield) (universe A/B under the same overlay)", + "P2: Sharpe(union+shield) >= Sharpe(base) (vs the live book, unshielded)", + "P3: MaxDD(union+shield) >= MaxDD(base) - 5pp", + ], + "on_pass": "the union deserves a shield-equipped forward check; no production change from this alone", + "status": "the window was already examined; not independent OOS; one mechanism, no variants", +} + + +def window_stats(r: pd.Series, a: str, b: str) -> dict: + x = r.loc[a:b] + e = (1 + x).cumprod() + yrs = len(x) / 252 + return {"cagr": float(e.iloc[-1] ** (1 / yrs) - 1), + "sharpe": float(x.mean() / x.std() * np.sqrt(252)), + "mdd": float((e / e.cummax() - 1).min())} + + +def main() -> None: + os.makedirs(OUT, exist_ok=True) + res_path = os.path.join(OUT, "results.json") + if os.path.exists(res_path): + raise SystemExit(f"{res_path} exists; this study does not overwrite itself") + if not os.path.exists(os.path.join(OUT, "prereg.json")): + with open(os.path.join(OUT, "prereg.json"), "w") as fh: + json.dump(PREREG, fh, indent=2) + + R = pd.read_pickle(os.path.join(RESULTS, "returns.pkl")) + p = load_us_mid("mid") + spy = p.extra["etf"]["SPY"] + ma = spy.rolling(150, min_periods=120).mean() + expo = (spy.shift(1) > ma.shift(1)).astype(float).reindex(R.index).ffill().fillna(0.0) + d = expo.diff().abs().fillna(0.0) + + S = {} + for arm in R.columns: + S[arm] = R[arm] + S[f"{arm}+shield"] = expo * R[arm] - COST * d + + res = {} + for arm, r in S.items(): + res[arm] = {w: window_stats(r, *ab) for w, ab in + [("primary", PRIMARY), ("reference", REFERENCE)]} + res[arm]["blocks"] = [round(window_stats(r, *ab)["sharpe"], 3) for ab in BLOCKS] + res[arm]["years"] = {int(k): round(float(v), 4) for k, v in + ((1 + r.loc[PRIMARY[0]:PRIMARY[1]]).groupby( + r.loc[PRIMARY[0]:PRIMARY[1]].index.year).prod() - 1).items()} + + a = alpha_stats(S["union+shield"].loc[PRIMARY[0]:PRIMARY[1]], + S["base"].loc[PRIMARY[0]:PRIMARY[1]]) + res["union+shield"]["alpha_vs_base"] = {k: float(a[k]) for k in ("alpha", "t")} + res["exposure_share_by_year"] = {int(y): round(float(v), 3) for y, v in + expo.loc[PRIMARY[0]:PRIMARY[1]].groupby( + expo.loc[PRIMARY[0]:PRIMARY[1]].index.year).mean().items()} + + base_, base_sh = res["base"]["primary"], res["base+shield"]["primary"] + union_sh = res["union+shield"]["primary"] + checks = {"P1_union_shield_gt_base_shield": union_sh["sharpe"] > base_sh["sharpe"], + "P2_union_shield_ge_base": union_sh["sharpe"] >= base_["sharpe"], + "P3_mdd_not_worse_5pp": union_sh["mdd"] >= base_["mdd"] - 0.05} + verdict = "PASS" if all(checks.values()) else "FAIL" + summary = {"prereg": PREREG, "results": res, "checks": {k: bool(v) for k, v in checks.items()}, + "verdict": verdict} + with open(res_path, "w") as fh: + json.dump(summary, fh, indent=2, default=float) + pd.DataFrame(S).to_pickle(os.path.join(OUT, "returns.pkl")) + + for arm in ("base", "base+shield", "union", "union+shield", "union_adv20+shield", "union_minus_top3+shield"): + s = res[arm] + print(f"{arm:22s} primary: cagr {s['primary']['cagr']*100:5.1f}% sh {s['primary']['sharpe']:.2f} " + f"mdd {s['primary']['mdd']*100:5.1f}% blocks {s['blocks']} ref {s['reference']['sharpe']:.2f}") + print("exposure share by year:", res["exposure_share_by_year"]) + print("union+shield alpha vs base:", res["union+shield"]["alpha_vs_base"]) + print("checks:", summary["checks"], "→", verdict) + + from research.lab.trials import append_trials + append_trials(os.path.join("data", "research", "results", "trials_us.jsonl"), + [{"name": f"union_shield:{k}", **{f"{w}_{x}": y for w in ("primary", "reference") + for x, y in v.items() if isinstance(y, (int, float))}} + for k, v in res.items() if isinstance(v, dict) and "primary" in v], + {"study": "union_shield", "role": "preregistered overlay on a previously seen window"}) + + +if __name__ == "__main__": + main() diff --git a/research/midsmall_rebuild.py b/research/midsmall_rebuild.py index 58ac3be..ce1c59c 100644 --- a/research/midsmall_rebuild.py +++ b/research/midsmall_rebuild.py @@ -1,24 +1,41 @@ """Convert mid/small Tiingo caches, rebuild the PIT panels, report the -survivorship change (pre-Tiingo backup vs rebuilt member counts per year). +survivorship change (pre-Tiingo backup and the previous build vs rebuilt +member counts per year). Converts every cached ticker on the mid/small plan +(plus the batch list), so it is re-runnable after each monthly Tiingo batch. PYTHONPATH=. uv run python -m research.midsmall_rebuild """ from __future__ import annotations +import json import os import pickle -import shutil import pandas as pd +def convert_list() -> list[str]: + plan = json.load(open("data/research/tiingo_plan.json")) + txt = [l.strip() for l in open("data/research/tiingo_midsmall.txt") if l.strip()] + want = set(plan["mid_small"]) | set(txt) + cached = [f[:-4] for f in os.listdir("data/research/tiingo") if f.endswith(".csv")] + return sorted(set(cached) & want) + + def main() -> None: - stage = [l.strip() for l in open("data/research/tiingo_midsmall.txt")][:230] - cached = [n for n in stage if os.path.exists(f"data/research/tiingo/{n}.csv")] - print(f"mid/small tickers: {len(stage)} listed, {len(cached)} cached", flush=True) + conv = convert_list() + print(f"mid/small cached tickers to convert: {len(conv)}", flush=True) from research.lab.tiingo_merge import write_yahoo_mid - write_yahoo_mid(cached) + write_yahoo_mid(conv) + + prev_counts = None + pkl = "data/research/us_midsmall.panel.pkl" + if os.path.exists(pkl): + with open(pkl, "rb") as fh: + prev = pickle.load(fh) + prev_counts = prev.member.sum(axis=1).resample("YE").last() + del prev for f in ("us_mid_bars.pkl", "us_mid.panel.pkl", "us_small.panel.pkl", "us_midsmall.panel.pkl"): p = os.path.join("data", "research", f) @@ -28,17 +45,20 @@ def main() -> None: from research.lab.panel_mid import load_us_mid new = load_us_mid("midsmall") + load_us_mid("mid") + load_us_mid("small") with open("data/research/us_midsmall.panel.pre_tiingo.pkl", "rb") as fh: old = pickle.load(fh) n, o = new.member.sum(axis=1), old.member.sum(axis=1) - comp = pd.DataFrame({"post": n.resample("YE").last(), "pre": o.resample("YE").last()}).dropna() - comp["delta"] = comp["post"] - comp["pre"] + comp = pd.DataFrame({"post": n.resample("YE").last(), "pre_tiingo": o.resample("YE").last(), + "prev_build": prev_counts}).dropna(subset=["pre_tiingo"]) + comp["delta_prev"] = comp["post"] - comp["prev_build"] comp.index = comp.index.year - print("\nPIT members per year (pre -> post):") + print("\nPIT members per year (pre_tiingo -> previous build -> post):") print(comp.to_string()) from research.lab.fetch_us_mid import load_bars - restored = [t for t in cached if t in new.close.columns + restored = [t for t in conv if t in new.close.columns and new.member[t].sum() > 0 and load_bars(t) is not None and t not in old.close.columns] print(f"\nnew PIT-covered names not in old panel: {len(restored)}") diff --git a/research/shield_lab.py b/research/shield_lab.py new file mode 100644 index 0000000..c278ddc --- /dev/null +++ b/research/shield_lab.py @@ -0,0 +1,132 @@ +"""Shielded stock sleeves: V9 risk machinery on top of stock selection. + +The frontier battle so far lives in the leveraged-ETF trend family because +the cash-funded ledger forbids margin: only 3x instruments can reach the +CAGR bar. This lab quantifies the alternative the selection camp asks for — +take the BEST stock-selection sleeves (overnight_recovery top-10, low_risk +top-20) and wrap them in the V9/V10 machinery: + + regime shield : SPY > MA150 (through t-1) -> hold sleeve; else hold the + momentum leader of GLD/DBC + vol target : optional, scale = clip(target/vol60, min, 1), shift(1) + trail stop : optional basket-level trailing stop -> risk-off asset + +PIT: all overlays use data through t-1 and one terminal shift, same +convention as TrendRiderV3. + + PYTHONPATH=. uv run python -m research.shield_lab +""" +from __future__ import annotations + +import numpy as np +import pandas as pd + +import trader +from strategies.base import BarStrategy +from strategies.robust_momentum import RobustMomentumStrategy + +import research.tr9_lab as lab + + +class ShieldedMomentum(RobustMomentumStrategy): + """Stock sleeve + SPY-regime shield (and optional vol target / trail).""" + + def __init__(self, signal: str, ma_long: int = 150, risk_off: tuple[str, ...] = ("GLD", "DBC"), + mom_lookback: int = 63, vol_target: float | None = None, + trail_pct: float | None = None, trail_window: int = 30, **kw) -> None: + super().__init__(signal, **kw) + self.ma_long = ma_long + self.risk_off = risk_off + self.mom_lookback = mom_lookback + self.vol_target = vol_target + self.trail_pct = trail_pct + self.trail_window = trail_window + + def generate_weights(self, bars: dict[str, pd.DataFrame]) -> pd.DataFrame: + w_s = super().generate_weights(bars) # already .shift(1)'d, stocks only + close = bars["close"] + # keep the overlay instruments out of the base sleeve's selection + extras_present = [c for c in ("SPY", *self.risk_off) if c in w_s.columns] + if extras_present and w_s[extras_present].to_numpy().any(): + w_s = w_s.copy() + w_s[extras_present] = 0.0 + w_s = w_s.div(w_s.sum(axis=1).replace(0.0, np.nan), axis=0).fillna(0.0) + if "SPY" not in close.columns: + return w_s + spy = close["SPY"] + risk_on = (spy > spy.rolling(self.ma_long).mean()).shift(1).fillna(False) + + # momentum leader among risk-off assets (through t-1) + ro = [c for c in self.risk_off if c in close.columns] + if ro: + mom = close[ro].pct_change(self.mom_lookback).shift(1) + leader = mom.idxmax(axis=1).where(mom.max(axis=1).notna(), ro[0]) + else: + leader = pd.Series("", index=close.index) + + out = pd.DataFrame(0.0, index=close.index, + columns=sorted(set(w_s.columns) | set(ro))) + out[w_s.columns] = w_s + off_rows = ~risk_on + for c in w_s.columns: + out.loc[off_rows, c] = 0.0 + for i in np.where(off_rows)[0]: + out.iat[i, out.columns.get_loc(leader.iloc[i])] = 1.0 + + if self.trail_pct: + # basket-level trailing stop on the sleeve NAV; ultra-conservative + # lag: the trigger at output row i uses NAV through row i-2 + base_ret = (w_s * close[w_s.columns].pct_change(fill_method=None)).sum(axis=1) + nav = (1 + base_ret.fillna(0.0)).cumprod() + trig = ((nav / nav.rolling(self.trail_window, min_periods=5).max() - 1.0) + <= -self.trail_pct).shift(2).fillna(False) + state, park_rows = False, [] + for i in range(len(trig)): + if not risk_on.iloc[i]: + state = False + elif trig.iloc[i]: + state = True + park_rows.append(state) + park = pd.Series(park_rows, index=close.index) & risk_on + for i in np.where(park)[0]: + for c in w_s.columns: + out.iat[i, out.columns.get_loc(c)] = 0.0 + out.iat[i, out.columns.get_loc(leader.iloc[i])] = 1.0 + + if self.vol_target: + rets = (out.shift(0) * close.pct_change(fill_method=None).reindex_like(out)).sum(axis=1) + realized = rets.rolling(60, min_periods=21).std() * np.sqrt(252) + scale = (self.vol_target / realized).clip(0.6, 1.0).shift(1).fillna(1.0) + out = out.mul(scale, axis=0) + return out + + +def register() -> list[str]: + reg = trader.STRATEGY_REGISTRY + names = [] + variants = { + "sh_orr_shield": lambda: ShieldedMomentum("overnight_recovery", top_n=10, rebal_freq=10, + buffer=2.0), + "sh_orr_shield_vt": lambda: ShieldedMomentum("overnight_recovery", top_n=10, rebal_freq=10, + buffer=2.0, vol_target=0.28), + "sh_orr_full": lambda: ShieldedMomentum("overnight_recovery", top_n=10, rebal_freq=10, + buffer=2.0, vol_target=0.28, trail_pct=0.20), + "sh_lowrisk_shield_vt": lambda: ShieldedMomentum("low_risk", top_n=20, rebal_freq=21, + buffer=2.0, vol_target=0.22), + } + for n, f in variants.items(): + reg[n] = f + trader.MIXED_STRATEGY_EXTRA_TICKERS[n] = ["SPY", "GLD", "DBC"] + names.append(n) + return names + + +def main() -> None: + names = register() + curves = lab.run(["overnight_recovery_top10", "lowrisk_top20", "us_three_sleeve", + "trend_rider_v10"] + names, end="2026-10-07") + lab.report(curves, "shielded sleeves vs frontier") + + +if __name__ == "__main__": + main() diff --git a/research/svr_overlay.py b/research/svr_overlay.py new file mode 100644 index 0000000..bc9c202 --- /dev/null +++ b/research/svr_overlay.py @@ -0,0 +1,91 @@ +"""Short-volume veto overlay test on the momentum sleeve (official engine). + +Hypothesis from the first scan (research/lab/short_volume.py): names in the +top quintile of 21-day mean REG SHO short-volume ratio (SVR = ShortVolume / +TotalVolume) underperform over the next month (S&P 500: IC21 -0.020, t -3.1; +Q5-Q1 spread -0.40%/mo). Test: veto those names at selection time (score -> +NaN before top-N selection), on the flagship overnight_recovery_top10 sleeve. + +PIT: the SVR panel is shifted by 2 sessions inside the veto (REG SHO files +are published after the close), matching the conservative convention of the +scan. + + PYTHONPATH=. uv run python -m research.svr_overlay +""" +from __future__ import annotations + +import pickle + +import numpy as np +import pandas as pd + +import trader +from strategies.robust_momentum import RobustMomentumStrategy, select_with_buffer + +import research.tr9_lab as lab + +SVR_PATH = "data/research/short_volume.pkl" +_SVR_CACHE: dict = {} + + +def svr21_panel() -> pd.DataFrame: + if "svr" not in _SVR_CACHE: + with open(SVR_PATH, "rb") as fh: + d = pickle.load(fh) + svr = (d["short"] / d["total"].replace(0.0, np.nan)) + _SVR_CACHE["svr"] = svr.rolling(21, min_periods=10).mean() + return _SVR_CACHE["svr"] + + +class RobustMomentumSvr(RobustMomentumStrategy): + """RobustMomentum + veto of top-quantile short-volume names at selection.""" + + def __init__(self, signal: str, svr_quant: float = 0.8, svr_lag: int = 2, **kw) -> None: + super().__init__(signal, **kw) + self.svr_quant = svr_quant + self.svr_lag = svr_lag + + def generate_weights(self, bars: dict[str, pd.DataFrame]) -> pd.DataFrame: + close = bars["close"] + panel = {f: bars[f].reindex(index=close.index, columns=close.columns) for f in self.fields} + if "eligible" in bars: + panel["eligible"] = bars["eligible"] + score = self._score(panel) + eligible = close.notna() + if "eligible" in bars: + eligible &= bars["eligible"].reindex_like(close).fillna(False).astype(bool) + score = score.where(eligible) + + svr = svr21_panel().reindex(index=close.index, columns=close.columns).shift(self.svr_lag) + q = svr.rank(axis=1, pct=True) + veto = (q >= self.svr_quant) & svr.notna() + score = score.where(~veto) + + weights = select_with_buffer(score, self.top_n, self.rebal_freq, self.buffer, self.warmup, + prices=close) + return weights.shift(1).fillna(0.0) + + +def register() -> list[str]: + reg = trader.STRATEGY_REGISTRY + names = [] + reg["overnight_recovery_svr80"] = lambda **kw: RobustMomentumSvr( + "overnight_recovery", top_n=10, rebal_freq=10, buffer=2.0, svr_quant=0.8) + names.append("overnight_recovery_svr80") + reg["overnight_recovery_svr90"] = lambda **kw: RobustMomentumSvr( + "overnight_recovery", top_n=10, rebal_freq=10, buffer=2.0, svr_quant=0.9) + names.append("overnight_recovery_svr90") + reg["overnight_recovery_svr70"] = lambda **kw: RobustMomentumSvr( + "overnight_recovery", top_n=10, rebal_freq=10, buffer=2.0, svr_quant=0.7) + names.append("overnight_recovery_svr70") + return names + + +def main() -> None: + names = register() + curves = lab.run(["overnight_recovery_top10", "lowrisk_top20"] + names) + lab.report(curves, "SVR veto overlay") + + +if __name__ == "__main__": + main() diff --git a/strategies/svr_momentum.py b/strategies/svr_momentum.py new file mode 100644 index 0000000..99216ea --- /dev/null +++ b/strategies/svr_momentum.py @@ -0,0 +1,76 @@ +"""RobustMomentum + short-volume veto (FINRA REG SHO). RESEARCH ONLY — not registered. + +Status: a 70%-quantile veto on overnight_recovery_top10 looked good on the +2026-10-07 window (Sharpe 0.87 -> 0.89, MaxDD -47.4% -> -39.4%), but the +quantile neighborhood killed it: veto at 0.60/0.75/0.80 all score BELOW the +baseline (Sharpe 0.81/0.82/0.84 vs 0.87). A lone spike, not a factor — the +strategy was registered as `overnight_recovery_svr70` and then reverted the +same day. Kept as an artifact; the veto idea is not robust on this sleeve. + +Context: the raw cross-sectional reading stands (21-day SVR is a negative +predictor on the S&P 500 panel, IC -0.020 with t -3.1; within small-cap +momentum the high-SVR half underperforms -0.61%/mo), but turning that into a +top-10-portfolio exclusion rule is too concentrated to be stable. + +Data: data/research/short_volume.pkl (research/lab/fetch_shortvolume.py; +missing file = no veto). REG SHO files publish after the close, so the veto +inside is lagged 2 sessions. +""" +from __future__ import annotations + +import os +import pickle + +import numpy as np +import pandas as pd + +from strategies.robust_momentum import RobustMomentumStrategy, select_with_buffer + +SVR_PATH = os.path.join("data", "research", "short_volume.pkl") +_SVR_CACHE: dict = {} + + +def svr21_panel() -> pd.DataFrame | None: + if "svr" not in _SVR_CACHE: + if not os.path.exists(SVR_PATH): + _SVR_CACHE["svr"] = None + else: + with open(SVR_PATH, "rb") as fh: + d = pickle.load(fh) + svr = d["short"] / d["total"].replace(0.0, np.nan) + _SVR_CACHE["svr"] = svr.rolling(21, min_periods=10).mean() + return _SVR_CACHE["svr"] + + +class RobustMomentumSvr(RobustMomentumStrategy): + """RobustMomentum with a top-quantile short-volume veto at selection.""" + + def __init__(self, signal: str, svr_quant: float = 0.7, svr_lag: int = 2, **kw) -> None: + super().__init__(signal, **kw) + self.svr_quant = svr_quant + self.svr_lag = svr_lag + + def generate_weights(self, bars: dict[str, pd.DataFrame]) -> pd.DataFrame: + close = bars["close"] + panel = {f: bars[f].reindex(index=close.index, columns=close.columns) for f in self.fields} + if "eligible" in bars: + panel["eligible"] = bars["eligible"] + score = self._score(panel) + eligible = close.notna() + if "eligible" in bars: + eligible &= bars["eligible"].reindex_like(close).fillna(False).astype(bool) + score = score.where(eligible) + + svr = svr21_panel() + if svr is not None: + svr = svr.reindex(index=close.index, columns=close.columns).shift(self.svr_lag) + q = svr.rank(axis=1, pct=True) + veto = (q >= self.svr_quant) & svr.notna() + score = score.where(~veto) + + weights = select_with_buffer(score, self.top_n, self.rebal_freq, self.buffer, self.warmup, + prices=close) + return weights.shift(1).fillna(0.0) + + +__all__ = ["RobustMomentumSvr"]