""" 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. uv run python -m research.lab.fetch_insider 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). Zips are cached in data/research/ext/insider/ and never downloaded twice. Every request sends the SEC User-Agent, goes out directly (no proxy) and is spaced >= 0.25 s from the previous one (SEC allows 10/s; we stay under 5/s). Kept rows (NONDERIV_TRANS joined to SUBMISSION and REPORTINGOWNER): * DOCUMENT_TYPE "4". Amendments (4/A, ~3% of Form 4s) are dropped: they mostly restate the original's trades, so keeping both double counts, and the original is what the market saw on its filing date. * TRANS_CODE P (open-market purchase, acquired) or S (open-market sale, disposed) of a non-preferred equity security, no equity swap. * The reporting owner is an officer and/or director. Pure 10% owners and "other" are dropped. A joint filing (rare) is attributed to one officer/ director owner (lowest CIK) so a trade is never counted twice. * Exact duplicate lines within a filing (same date, code, shares, price, shares owned after, ownership form) are dropped. Issuer -> panel ticker: the current SEC ticker map (company_tickers.json) by issuer CIK; share classes of one CIK (GOOG/GOOGL, ...) all map to it. Issuers missing from it (renamed, reorganized, taken private) fall back to the 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: {"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, accession "tickers": DataFrame[ticker, cik, source, first, last, n_filings, price_err] (which issuer CIK covers each panel ticker, and when), "unmapped": panel tickers with no issuer} Point in time: a trade is usable from the first trading day strictly after ``filed`` (FILING_DATE). ``trans_date`` is only used to classify traders (calendar month and year), never to decide when a trade is known. ``value`` = shares x reported price; where the price is missing or more than 2x off the issuer's usual price / unadjusted-close ratio (``price_ok`` False: mostly total proceeds typed as the price), shares x the unadjusted close. ``plan_10b51`` is the Rule 10b5-1 checkbox (Form 4 from April 2023, NaN before); ``plan_fn`` is True when a footnote or remark of the filing mentions Rule 10b5-1 (available in every year). """ from __future__ import annotations import csv import os import pickle import re import time import zipfile from urllib.parse import urljoin import numpy as np import pandas as pd import requests from research.lab.fetch_us_fundamentals import UA, cik_map PAGE = "https://www.sec.gov/data-research/sec-markets-data/insider-transactions-data-sets" ZIP_DIR = os.path.join("data", "research", "ext", "insider") OUT = os.path.join("data", "research", "insider_trades.pkl") FUND_RAW = os.path.join("data", "research", "us_fund_raw.pkl") SUB_COLS = ["ACCESSION_NUMBER", "FILING_DATE", "DOCUMENT_TYPE", "ISSUERCIK", "ISSUERTRADINGSYMBOL", "REMARKS", "AFF10B5ONE"] OWN_COLS = ["ACCESSION_NUMBER", "RPTOWNERCIK", "RPTOWNER_RELATIONSHIP", "RPTOWNER_TITLE"] TRN_COLS = ["ACCESSION_NUMBER", "SECURITY_TITLE", "TRANS_DATE", "TRANS_CODE", "EQUITY_SWAP_INVOLVED", "TRANS_SHARES", "TRANS_PRICEPERSHARE", "TRANS_ACQUIRED_DISP_CD", "SHRS_OWND_FOLWNG_TRANS", "DIRECT_INDIRECT_OWNERSHIP"] PLAN_RE = r"10b5\s*-\s*1|10b-5-1|10b5one" NOT_COMMON = r"prefer|pfd|note|debenture|warrant|bond|option" _last = [0.0] def _session() -> requests.Session: s = requests.Session() s.trust_env = False # direct connection: ignore proxy env vars s.headers.update(UA) return s def _get(sess: requests.Session, url: str, **kw) -> requests.Response: for attempt in range(5): wait = 0.25 - (time.time() - _last[0]) if wait > 0: time.sleep(wait) _last[0] = time.time() try: r = sess.get(url, timeout=300, **kw) r.raise_for_status() return r except Exception as exc: # noqa: BLE001 — retry transient SEC errors print(f" {url}: {exc}; retry {attempt + 1}", flush=True) time.sleep(5 * (attempt + 1)) raise RuntimeError(f"failed: {url}") def zip_urls(sess: requests.Session) -> dict[str, str]: html = _get(sess, PAGE).text found = re.findall(r'href="([^"]*?(\d{4}q[1-4])_form345\.zip)"', html) return {q: urljoin(PAGE, href) for href, q in sorted(found, key=lambda x: x[1])} def download(sess: requests.Session) -> list[str]: os.makedirs(ZIP_DIR, exist_ok=True) urls = zip_urls(sess) print(f"{len(urls)} quarterly zips listed ({min(urls)} .. {max(urls)})", flush=True) paths = [] for q, url in urls.items(): path = os.path.join(ZIP_DIR, f"{q}_form345.zip") if not (os.path.exists(path) and zipfile.is_zipfile(path)): r = _get(sess, url) with open(path + ".part", "wb") as fh: fh.write(r.content) os.replace(path + ".part", path) print(f" {q}: {len(r.content) / 1e6:.1f} MB", flush=True) paths.append(path) return paths # --------------------------------------------------------------------------- # parse one quarter # --------------------------------------------------------------------------- def _read(z: zipfile.ZipFile, name: str, cols: list[str]) -> pd.DataFrame: with z.open(name) as fh: return pd.read_csv(fh, sep="\t", dtype=str, quoting=csv.QUOTE_NONE, usecols=lambda c: c in cols, encoding_errors="replace") def _date(s: pd.Series) -> pd.Series: return pd.to_datetime(s, format="%d-%b-%Y", errors="coerce") def parse_quarter(path: str) -> tuple[pd.DataFrame, pd.DataFrame]: """(kept trades, every Form 4 filing's issuer CIK / symbol / date).""" z = zipfile.ZipFile(path) sub = _read(z, "SUBMISSION.tsv", SUB_COLS) sub = sub[sub["DOCUMENT_TYPE"].str.strip() == "4"].copy() sub["filed"] = _date(sub["FILING_DATE"]) sub["cik"] = pd.to_numeric(sub["ISSUERCIK"], errors="coerce") sub = sub.dropna(subset=["filed", "cik"]) filings = sub[["ACCESSION_NUMBER", "cik", "ISSUERTRADINGSYMBOL", "filed"]] own = _read(z, "REPORTINGOWNER.tsv", OWN_COLS) own = own[own["ACCESSION_NUMBER"].isin(sub["ACCESSION_NUMBER"])] rel = own["RPTOWNER_RELATIONSHIP"].fillna("").str.lower() own = own.assign(officer=rel.str.contains("officer"), director=rel.str.contains("director"), ten_pct=rel.str.contains("tenpercent"), insider_cik=pd.to_numeric(own["RPTOWNERCIK"], errors="coerce")) own = own[(own["officer"] | own["director"]) & own["insider_cik"].notna()] own = own.sort_values(["ACCESSION_NUMBER", "insider_cik"]).drop_duplicates("ACCESSION_NUMBER") own["role"] = np.where(own["officer"] & own["director"], "officer+director", np.where(own["officer"], "officer", "director")) tr = _read(z, "NONDERIV_TRANS.tsv", TRN_COLS) code = tr["TRANS_CODE"].str.strip() ad = tr["TRANS_ACQUIRED_DISP_CD"].str.strip() keep = (((code == "P") & (ad == "A")) | ((code == "S") & (ad == "D"))) keep &= tr["ACCESSION_NUMBER"].isin(own["ACCESSION_NUMBER"]) keep &= ~tr["SECURITY_TITLE"].fillna("").str.contains(NOT_COMMON, case=False, regex=True) keep &= ~tr["EQUITY_SWAP_INVOLVED"].fillna("0").str.strip().isin(["1", "true"]) tr = tr[keep].copy() tr["code"] = tr["TRANS_CODE"].str.strip() tr["trans_date"] = _date(tr["TRANS_DATE"]) for src, dst in (("TRANS_SHARES", "shares"), ("TRANS_PRICEPERSHARE", "price"), ("SHRS_OWND_FOLWNG_TRANS", "owned_after")): tr[dst] = pd.to_numeric(tr[src], errors="coerce") tr["ownership"] = tr["DIRECT_INDIRECT_OWNERSHIP"].str.strip() tr = tr[tr["shares"] > 0] tr = tr.drop_duplicates(["ACCESSION_NUMBER", "trans_date", "code", "shares", "price", "owned_after", "ownership"]) # Rule 10b5-1: the 2023 checkbox, and any footnote / remark that cites the rule fn = _read(z, "FOOTNOTES.tsv", ["ACCESSION_NUMBER", "FOOTNOTE_TXT"]) fn = fn[fn["ACCESSION_NUMBER"].isin(tr["ACCESSION_NUMBER"])] cited = set(fn.loc[fn["FOOTNOTE_TXT"].str.contains(PLAN_RE, case=False, na=False, regex=True), "ACCESSION_NUMBER"]) cited |= set(sub.loc[sub["REMARKS"].str.contains(PLAN_RE, case=False, na=False, regex=True), "ACCESSION_NUMBER"]) box = (sub["AFF10B5ONE"].str.strip().str.lower().map({"1": 1.0, "true": 1.0, "0": 0.0, "false": 0.0}) if "AFF10B5ONE" in sub else pd.Series(np.nan, index=sub.index)) sub = sub.assign(plan_10b51=box) out = (tr.merge(own[["ACCESSION_NUMBER", "insider_cik", "role", "RPTOWNER_TITLE", "ten_pct"]], on="ACCESSION_NUMBER") .merge(sub[["ACCESSION_NUMBER", "cik", "ISSUERTRADINGSYMBOL", "filed", "plan_10b51"]], on="ACCESSION_NUMBER")) out["plan_fn"] = out["ACCESSION_NUMBER"].isin(cited) out = out.rename(columns={"ACCESSION_NUMBER": "accession", "RPTOWNER_TITLE": "title", "ISSUERTRADINGSYMBOL": "symbol"}) return (out[["accession", "cik", "symbol", "insider_cik", "role", "title", "ten_pct", "filed", "trans_date", "code", "shares", "price", "plan_10b51", "plan_fn", "ownership"]], filings) # --------------------------------------------------------------------------- # issuer CIK -> panel ticker # --------------------------------------------------------------------------- def norm_symbols(sym) -> list[str]: """'(WST)' -> [WST]; 'NYSE: KRC' -> [KRC]; 'CBS, CBS.A' -> [CBS, CBS-A].""" if not isinstance(sym, str): return [] s = re.sub(r'["()\[\]]', "", sym.upper()) s = re.sub(r"\b(NYSE|NASDAQ|NYSE AMERICAN|AMEX|NMS|OTC)\s*:", "", s) parts = re.split(r"[,;/&]| AND ", s) return [re.sub(r"\s+", "", x).replace(".", "-") for x in parts if x.strip()] def raw_closes(tickers: list[str]) -> 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): return None with open(FUND_RAW, "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) after = ratio.iloc[::-1].cumprod().iloc[::-1].shift(-1).fillna(1.0) return close_sa * after def _raw_at(raw: pd.DataFrame, tickers: pd.Series, dates: pd.Series) -> np.ndarray: """raw close of each (ticker, date) pair as of that date (last close on or before).""" out = np.full(len(tickers), np.nan) ff = raw.ffill() col = {t: i for i, t in enumerate(ff.columns)} ci = tickers.map(col) ri = ff.index.searchsorted(dates.values, side="right") - 1 ok = ci.notna().values & (ri >= 0) & dates.notna().values out[ok] = ff.values[ri[ok], ci[ok].astype(int)] return out def map_tickers(trades: pd.DataFrame, filings: pd.DataFrame, panel_tickers: list[str], raw: pd.DataFrame | None, close: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]: """Assign each trade to a panel ticker; returns (trades with ticker, ticker map).""" cmap = cik_map() pset = set(panel_tickers) rev: dict[int, list[str]] = {} for t, c in cmap.items(): if t in pset: rev.setdefault(c, []).append(t) span = filings.groupby("cik")["filed"].agg(["min", "max", "count"]) rows = [] for c, ts in rev.items(): if c in span.index: for t in sorted(ts): rows.append((t, c, "sec_map", span.at[c, "min"], span.at[c, "max"], span.at[c, "count"])) # symbol fallback for issuers the current SEC map does not tie to a panel ticker fb = filings[~filings["cik"].isin(list(rev))] pairs = (fb.groupby(["cik", "ISSUERTRADINGSYMBOL"])["filed"] .agg(first="min", last="max", n="count").reset_index()) cand = [] for r in pairs.itertuples(index=False): for t in norm_symbols(r.ISSUERTRADINGSYMBOL): if t in pset: cand.append((t, int(r.cik), r.first, r.last, r.n)) cand = (pd.DataFrame(cand, columns=["ticker", "cik", "first", "last", "n"]) .groupby(["ticker", "cik"]).agg(first=("first", "min"), last=("last", "max"), n=("n", "sum")) .reset_index()) px_first = close.apply(lambda s: s.first_valid_index()) px_last = close.apply(lambda s: s.last_valid_index()) for r in cand.sort_values("n", ascending=False).itertuples(index=False): first, last = r.first, r.last owner = cmap.get(r.ticker) if owner is not None and owner in span.index: # a predecessor only, never alongside the owner last = min(last, span.at[owner, "min"] - pd.Timedelta(days=1)) if r.ticker in px_first.index and pd.notna(px_first[r.ticker]): first, last = max(first, px_first[r.ticker]), min(last, px_last[r.ticker]) taken = [(a, b) for (t, _, _, a, b, _) in rows if t == r.ticker] if last <= first or any(a <= last and first <= b for a, b in taken): continue rows.append((r.ticker, r.cik, "symbol", first, last, r.n)) tm = pd.DataFrame(rows, columns=["ticker", "cik", "source", "first", "last", "n_filings"]) # assign trades: SEC-map issuers to the filing's own share class when it is # one of theirs, fallback issuers only inside their window prim = {c: ts[0] for c, ts in ((c, sorted(ts)) for c, ts in rev.items())} t_sec = trades["cik"].map(prim) sym0 = trades["symbol"].map(lambda s: (norm_symbols(s) or [None])[0]) own_class = [s if (isinstance(s, str) and c in rev and s in rev[c]) else None for s, c in zip(sym0.values, trades["cik"].values)] trades = trades.assign(ticker=pd.Series(own_class, index=trades.index).fillna(t_sec)) for r in tm[tm["source"] == "symbol"].itertuples(index=False): m = (trades["cik"] == r.cik) & trades["filed"].between(r.first, r.last) & trades["ticker"].isna() trades.loc[m, "ticker"] = r.ticker trades = trades.dropna(subset=["ticker"]).copy() # price check: trade prices vs the ticker's unadjusted close on the trade date trades["raw_close"] = _raw_at(raw, trades["ticker"], trades["trans_date"]) if raw is not None else np.nan with np.errstate(divide="ignore", invalid="ignore"): err = np.abs(np.log(trades["price"] / trades["raw_close"])) trades["price_err"] = err.where(trades["price"] > 0) chk = trades.groupby(["ticker", "cik"])["price_err"].median().rename("price_err") tm = tm.merge(chk, left_on=["ticker", "cik"], right_index=True, how="left") bad = tm[(tm["source"] == "symbol") & (tm["price_err"] > np.log(1.25))] if len(bad): print(" symbol matches rejected on price:", bad[["ticker", "cik", "price_err"]].to_dict("records")) tm = tm.drop(bad.index) keep = pd.MultiIndex.from_frame(trades[["ticker", "cik"]]).isin( pd.MultiIndex.from_frame(bad[["ticker", "cik"]])) trades = trades[~keep] 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) with open(os.path.join("data", "research", "us_ohlcv.pkl"), "rb") as fh: close = pickle.load(fh)["close"] 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 # price / close ratio (~1 normally, ~1500 for BRK-A trades mapped to BRK-B): # such prices are unit errors (total proceeds typed as the price) and are # replaced by the ticker's unadjusted close with np.errstate(divide="ignore", invalid="ignore"): lr = pd.Series(np.log(trades["price"] / trades["raw_close"]), index=trades.index) lr = lr.where(trades["price"] > 0) usual = lr.groupby([trades["ticker"], trades["cik"]]).transform("median") trades["price_ok"] = (lr - usual).abs() < np.log(2) px = np.where(trades["price_ok"], trades["price"], np.where(trades["raw_close"] > 0, trades["raw_close"], trades["price"])) trades["value"] = trades["shares"] * np.where(px > 0, px, np.nan) trades["insider_cik"] = trades["insider_cik"].astype("int64") cols = ["ticker", "cik", "insider_cik", "role", "title", "ten_pct", "filed", "trans_date", "code", "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 " 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: pickle.dump({"trades": trades, "tickers": tm, "unmapped": unmapped}, fh, protocol=4) print(f"Saved {OUT}") if __name__ == "__main__": main()