#!/usr/bin/env python3 """Do insiders trade right after earnings? — reproducible timing study. Most public companies restrict when insiders may trade: a typical insider-trading policy opens a "trading window" shortly after results are disclosed and closes it as the next quarter end approaches. The policies themselves are private, but both halves of the timing are public filings: the earnings disclosure is an 8-K declaring item 2.02 (Results of Operations and Financial Condition), and the trades are Form 4 transaction legs with their own as-filed dates. This script measures, over a fixed window: For every open-market purchase (code P) and sale (code S) on a non-derivative security reported on an original Form 4, how many days passed since the issuer's most recent item-2.02 8-K? If insider trading dates were indifferent to the earnings calendar, a quarterly filer's legs would spread roughly evenly across the ~91 days between consecutive 2.02 filings. The measured distribution is the story. This is a data-infrastructure measurement, not a compliance audit and not investment advice: it counts filings by their own declared dates and codes, attributes no intent, and characterizes no trade as informed or uninformed. Known limit stated up front: Form 4 does not expose the Rule 10b5-1 plan checkbox in this dataset, so scheduled plan trades (which execute regardless of windows) cannot be split from discretionary ones here. Reproducible: all window bounds are fixed arguments, and the API's `since=`/ `until=` filters make a fixed window return the same population on any tier, any day you run it. The default windows are ~37,000 Form 4 filings plus ~39,000 8-K events ≈ ~700 requests (free tier: 10/min, so expect ~70 minutes; paid tiers are faster). Usage: export FILINGPULSE_API_KEY=fp_... # free key: filingpulse.io/signup.html python insider_trades_after_earnings_study.py \ --txn-since 2026-07-01 --txn-until 2026-09-30 \ --filed-through 2026-10-09 --events-since 2026-03-01 \ --out results.json Optional coverage cross-check (no key needed for EDGAR, but the SEC asks for a contact identity in the User-Agent; set your own): export EDGAR_CONTACT="you@example.com" python insider_trades_after_earnings_study.py ... --verify-coverage Only the Python standard library is used. """ import argparse import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request from bisect import bisect_right from datetime import date, timedelta BASE_URL = os.environ.get("FILINGPULSE_BASE_URL", "https://api.filingpulse.io") API_KEY = os.environ.get("FILINGPULSE_API_KEY") ISO_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") COMPACT_RE = re.compile(r"^\d{8}$") LOOKBACK_DAYS = 120 # a quarterly filer's gap is ~91d; beyond this, "none recent" # offset-day buckets: (label, lo, hi) inclusive OFFSET_BUCKETS = ( ("0 (same day)", 0, 0), ("1", 1, 1), ("2", 2, 2), ("3", 3, 3), ("4", 4, 4), ("5", 5, 5), ("6", 6, 6), ("7", 7, 7), ("8-10", 8, 10), ("11-14", 11, 14), ("15-21", 15, 21), ("22-30", 22, 30), ("31-45", 31, 45), ("46-60", 46, 60), ("61-90", 61, 90), ("91-120", 91, LOOKBACK_DAYS), ) def _ssl_context(): """Prefer certifi's roots when it happens to be installed — some machines carry a stale OS trust store that rejects current Let's Encrypt chains, which would make a healthy API look down. Falls back to the OS store; certifi is never required.""" try: import ssl import certifi return ssl.create_default_context(cafile=certifi.where()) except ImportError: return None _SSL_CTX = _ssl_context() def parse_iso(s): """YYYY-MM-DD as filed -> date, else None. Malformed values are data, and this script counts them instead of guessing.""" if not isinstance(s, str) or not ISO_RE.match(s): return None try: return date.fromisoformat(s) except ValueError: return None def parse_compact(s): """As-filed YYYYMMDD -> date, else None (8-K headers use this format).""" if not isinstance(s, str) or not COMPACT_RE.match(s): return None try: return date(int(s[:4]), int(s[4:6]), int(s[6:8])) except ValueError: return None def parse_cik(s): """Zero-padded CIK string as filed -> int, else None.""" if not isinstance(s, str) or not s.strip().isdigit(): return None return int(s.strip()) def api_get(path: str, params: dict) -> dict: """GET with the key header; honors 429 Retry-After so the free tier just runs slower instead of failing, and retries transient network errors — a multi-minute run will eventually meet a dropped connection, and that is not a reason to lose the whole fetch.""" qs = "&".join(f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items()) url = f"{BASE_URL}{path}?{qs}" req = urllib.request.Request(url, headers={"X-API-Key": API_KEY or ""}) for attempt in range(8): try: with urllib.request.urlopen(req, timeout=60, context=_SSL_CTX) as resp: return json.load(resp) except urllib.error.HTTPError as e: if e.code == 429: wait = int(e.headers.get("Retry-After", "10")) print(f" rate limited — waiting {wait}s", file=sys.stderr) time.sleep(max(1, wait)) continue if e.code >= 500: # transient server blip — retry, don't die wait = 10 * (attempt + 1) print(f" API {e.code} — retrying in {wait}s", file=sys.stderr) time.sleep(wait) continue body = e.read().decode("utf-8", "replace")[:200] sys.exit(f"API error {e.code} on {url}: {body}") except (OSError, TimeoutError) as e: # timeouts, resets, DNS blips wait = 5 * (attempt + 1) print(f" network error ({e}) — retrying in {wait}s", file=sys.stderr) time.sleep(wait) sys.exit("giving up after repeated failures") def fetch_days(path: str, start: date, end: date, label: str): """Walk the range one filed-date at a time (since=D&until=D), paging inside each day, deduping by accession. Fetching by single day keeps every filing reachable (daily volume stays far below the offset cap) and the per-day totals double as the coverage table.""" by_day, seen = [], set() d = start while d <= end: day = d.isoformat() rows, total, offset = [], None, 0 while total is None or len(rows) < total: page = api_get(path, {"since": day, "until": day, "limit": 200, "offset": offset}) total = page["total"] rows.extend(page["data"]) offset += 200 if not page["data"]: break time.sleep(0.05) fresh = [r for r in rows if r.get("accession") not in seen] seen.update(r.get("accession") for r in fresh) by_day.append((d, fresh)) print(f" {label} {day}: {total or 0}", file=sys.stderr) d += timedelta(days=1) return by_day def edgar_index_counts(start: date, end: date, contact: str): """Distinct accessions per filed date from EDGAR's own daily master indexes, counted for BOTH form families this study fetches: ownership Form 4 (+4/A) and 8-K (+8-K/A) — one sweep, two ground-truth tables. Three EDGAR facts the naive version of this check gets wrong: - EDGAR serves 403 (not 404) for never-published days (weekends / holidays); those come back as None. - A day's index lists everything DISSEMINATED that day, and the row's own Date Filed column can differ — re-disseminated corrections keep their original filed date. Membership in the index is NOT "filed that day"; rows are bucketed by their own Date Filed column, since that is the population a filed_date window returns. - Dissemination day and Date Filed routinely differ by one (a filing can surface only in the NEXT day's index under the prior day's date), so a single day's index is not the whole population for that date. The sweep therefore accumulates Date-Filed buckets across every index in the range, with a few buffer days on each side.""" buckets4 = {} # filed date iso -> set of Form 4/4A accessions buckets8 = {} # filed date iso -> set of 8-K/8-K/A accessions published = set() # days whose index exists buffer = timedelta(days=4) d = start - buffer sweep_end = min(end + buffer, date.today()) while d <= sweep_end: qtr = (d.month - 1) // 3 + 1 day_compact = d.strftime("%Y%m%d") url = (f"https://www.sec.gov/Archives/edgar/daily-index/{d.year}/" f"QTR{qtr}/master.{day_compact}.idx") req = urllib.request.Request( url, headers={"User-Agent": f"insider-earnings-timing-study ({contact})"}) text = None for attempt in range(6): try: with urllib.request.urlopen(req, timeout=60, context=_SSL_CTX) as resp: text = resp.read().decode("latin-1") break except urllib.error.HTTPError as e: if e.code in (403, 404): break # no index published that day if e.code == 429 or e.code >= 500: # transient throttle/blip — a 200-day sweep WILL meet # one, and losing the whole run to it is not honest # coverage checking. Back off hard and retry. if attempt == 5: sys.exit(f"EDGAR kept answering {e.code} for {url}") wait = 20 * (attempt + 1) print(f" EDGAR {e.code} — backing off {wait}s", file=sys.stderr) time.sleep(wait) continue raise except (OSError, TimeoutError) as e: if attempt == 5: raise wait = 5 * (attempt + 1) print(f" network error ({e}) — retrying in {wait}s", file=sys.stderr) time.sleep(wait) if text is not None: published.add(d.isoformat()) for line in text.splitlines(): parts = line.split("|") if len(parts) != 5: continue form = parts[2] df = parts[3] if len(df) != 8 or not df.isdigit(): continue iso = f"{df[:4]}-{df[4:6]}-{df[6:8]}" acc = parts[4].rsplit("/", 1)[-1].removesuffix(".txt") if form in ("4", "4/A"): buckets4.setdefault(iso, set()).add(acc) elif form in ("8-K", "8-K/A"): buckets8.setdefault(iso, set()).add(acc) print(f" EDGAR index {d.isoformat()}: " f"{'ok' if text is not None else 'not published'}", file=sys.stderr) time.sleep(1.1) # SEC fair access: stay well under their rate cap d += timedelta(days=1) def table(buckets, start_, end_): counts = {} d2 = start_ while d2 <= end_: iso = d2.isoformat() n = len(buckets.get(iso, ())) # an unpublished index day with nothing dated to it anywhere in # the sweep is unchecked (weekend/holiday), not zero counts[iso] = n if (iso in published or n) else None d2 += timedelta(days=1) return counts return table(buckets4, start, end), table(buckets8, start, end) def analyze(form4_days, events_days, window): txn_since = date.fromisoformat(window["txn_since"]) txn_until = date.fromisoformat(window["txn_until"]) # --- 8-K side: per-issuer sorted 2.02 filed dates ----------------------- ev_total = ev_8k = ev_8ka = ev_other = 0 ev_202 = 0 cal_202 = {} # issuer cik int -> sorted [date, ...] daily_events = [] for d, events in events_days: daily_events.append({"date": d.isoformat(), "filed": len(events)}) for ev in events: ev_total += 1 ft = ev.get("form_type") if ft == "8-K/A": ev_8ka += 1 # re-filings: counted, not analyzed continue if ft != "8-K": ev_other += 1 continue ev_8k += 1 codes = {item.get("code") for item in (ev.get("items") or [])} if "2.02" not in codes: continue cik = parse_cik((ev.get("issuer") or {}).get("cik")) if cik is None: continue fd = parse_compact(ev.get("filed_date")) or d ev_202 += 1 cal_202.setdefault(cik, []).append(fd) for cik in cal_202: cal_202[cik] = sorted(set(cal_202[cik])) # consecutive 2.02 gaps, to ground the "roughly quarterly" baseline in # the fetched data itself rather than an assumed 91 gaps = [] for ds in cal_202.values(): gaps.extend((b - a).days for a, b in zip(ds, ds[1:])) gaps.sort() def pct(values, q): if not values: return None return values[max(0, -(-len(values) * q // 100) - 1)] # --- Form 4 side: leg-by-leg offsets ------------------------------------ f_total = f_4 = f_4a = f_35 = f_other = 0 legs_total = 0 derivative_excluded = other_code_excluded = 0 undated_excluded = out_of_window_excluded = no_issuer_cik_excluded = 0 analyzed = 0 daily_form4 = [] def new_side(): return {"buckets": [0] * len(OFFSET_BUCKETS), "matched": 0, "no_recent_202": 0, "offset_sum_days": 0} sides = {"P": new_side(), "S": new_side()} issuers_analyzed, issuers_matched, issuers_no202 = set(), set(), set() filings_analyzed = set() for d, filings in form4_days: n_family = 0 for fl in filings: ft = fl.get("form_type") if ft in ("4", "4/A"): n_family += 1 f_total += 1 if ft == "4/A": f_4a += 1 # re-filings: counted, not analyzed continue if ft in ("3", "3/A", "5", "5/A"): f_35 += 1 # live-forward since 2026-09-23 — mid-window continue # coverage change; excluded by name for a # population consistent across the window if ft != "4": f_other += 1 continue f_4 += 1 cik = parse_cik((fl.get("issuer") or {}).get("cik")) for leg in (fl.get("transactions") or []): legs_total += 1 if leg.get("is_derivative"): derivative_excluded += 1 continue code = leg.get("transaction_code") if code not in ("P", "S"): other_code_excluded += 1 continue td = parse_iso(leg.get("transaction_date")) if td is None: undated_excluded += 1 continue if not (txn_since <= td <= txn_until): out_of_window_excluded += 1 continue if cik is None: no_issuer_cik_excluded += 1 continue analyzed += 1 issuers_analyzed.add(cik) filings_analyzed.add(fl.get("accession")) side = sides[code] ds = cal_202.get(cik, ()) i = bisect_right(ds, td) prior = ds[i - 1] if i else None if prior is None or (td - prior).days > LOOKBACK_DAYS: side["no_recent_202"] += 1 issuers_no202.add(cik) continue off = (td - prior).days side["matched"] += 1 side["offset_sum_days"] += off issuers_matched.add(cik) for bi, (_, lo, hi) in enumerate(OFFSET_BUCKETS): if lo <= off <= hi: side["buckets"][bi] += 1 break daily_form4.append({"date": d.isoformat(), "filed": n_family, "all_ownership_forms": len(filings)}) def side_out(s): return { "analyzed": s["matched"] + s["no_recent_202"], "matched": s["matched"], "no_recent_202": s["no_recent_202"], "buckets": [[label, s["buckets"][i]] for i, (label, _, _) in enumerate(OFFSET_BUCKETS)], "mean_offset_days": (round(s["offset_sum_days"] / s["matched"], 1) if s["matched"] else None), } both = new_side() for s in sides.values(): both["matched"] += s["matched"] both["no_recent_202"] += s["no_recent_202"] both["offset_sum_days"] += s["offset_sum_days"] for i, n in enumerate(s["buckets"]): both["buckets"][i] += n return { "study": "form4_earnings_timing", "version": 1, "window": window, "lookback_days": LOOKBACK_DAYS, "fetched": { "form4": { "days": len(daily_form4), "total_ownership_filings": f_total, "form_4": f_4, "form_4a_amendments_excluded": f_4a, "forms_3_5_excluded": f_35, "other_form_types": f_other, }, "events": { "days": len(daily_events), "total": ev_total, "form_8k": ev_8k, "form_8ka_amendments_excluded": ev_8ka, "other_form_types": ev_other, "item_202_filings": ev_202, "item_202_issuers": len(cal_202), }, }, "legs": { "total_on_form4_originals": legs_total, "derivative_excluded": derivative_excluded, "other_code_excluded": other_code_excluded, "undated_excluded": undated_excluded, "out_of_window_excluded": out_of_window_excluded, "no_issuer_cik_excluded": no_issuer_cik_excluded, "analyzed": analyzed, }, "offsets": { "all": side_out(both), "P": side_out(sides["P"]), "S": side_out(sides["S"]), }, "population": { "filings_with_analyzed_legs": len(filings_analyzed), "issuers_with_analyzed_legs": len(issuers_analyzed), "issuers_matched": len(issuers_matched), "issuers_with_no_recent_202_legs": len(issuers_no202), }, "gaps_202": { "consecutive_gaps_measured": len(gaps), "p25": pct(gaps, 25), "median": pct(gaps, 50), "p75": pct(gaps, 75), }, "daily_form4": daily_form4, "daily_events": daily_events, "generated_by": "insider_trades_after_earnings_study.py v1", } def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--txn-since", default="2026-07-01", help="transaction-date window start") ap.add_argument("--txn-until", default="2026-09-30", help="transaction-date window end (inclusive)") ap.add_argument("--filed-through", default="2026-10-09", help="fetch Form 4s with filed_date from --txn-since " "through this date (the grace period for legs filed " "after the window closes)") ap.add_argument("--events-since", default="2026-03-01", help="fetch 8-K events from this date (the lookback " "every window-start trade measures against)") ap.add_argument("--out", default="insider_earnings_timing_results.json") ap.add_argument("--verify-coverage", action="store_true", help="cross-check per-day counts against EDGAR's daily " "master index (needs EDGAR_CONTACT set)") args = ap.parse_args() if not API_KEY: sys.exit("set FILINGPULSE_API_KEY (free key: " "https://filingpulse.io/signup.html)") for v in (args.txn_since, args.txn_until, args.filed_through, args.events_since): if not ISO_RE.match(v): sys.exit("window bounds must be YYYY-MM-DD") txn_since = date.fromisoformat(args.txn_since) txn_until = date.fromisoformat(args.txn_until) filed_through = date.fromisoformat(args.filed_through) events_since = date.fromisoformat(args.events_since) if not (events_since <= txn_since <= txn_until <= filed_through): sys.exit("need --events-since <= --txn-since <= --txn-until " "<= --filed-through") print(f"fetching Form 4 filings filed {txn_since} .. {filed_through}", file=sys.stderr) form4_days = fetch_days("/v1/insider-trades", txn_since, filed_through, "form4") print(f"fetching 8-K events {events_since} .. {txn_until}", file=sys.stderr) events_days = fetch_days("/v1/events", events_since, txn_until, "8-K") window = {"txn_since": args.txn_since, "txn_until": args.txn_until, "filed_through": args.filed_through, "events_since": args.events_since, "events_until": args.txn_until} results = analyze(form4_days, events_days, window) if args.verify_coverage: contact = os.environ.get("EDGAR_CONTACT") if not contact: sys.exit("--verify-coverage needs EDGAR_CONTACT set (the SEC asks " "for a contact identity in the User-Agent)") print("verifying per-day counts against EDGAR daily indexes", file=sys.stderr) idx4, idx8 = edgar_index_counts(events_since, filed_through, contact) for row in results["daily_form4"]: row["edgar_index"] = idx4.get(row["date"]) row["match"] = (None if row["edgar_index"] is None else row["filed"] == row["edgar_index"]) for row in results["daily_events"]: row["edgar_index"] = idx8.get(row["date"]) row["match"] = (None if row["edgar_index"] is None else row["filed"] == row["edgar_index"]) with open(args.out, "w", encoding="utf-8") as fh: json.dump(results, fh, indent=2) fh.write("\n") legs = results["legs"] offs = results["offsets"]["all"] print(f"\n{legs['analyzed']} open-market legs analyzed " f"({offs['matched']} matched to a prior 2.02, " f"{offs['no_recent_202']} with none in {LOOKBACK_DAYS}d)") if offs["matched"]: within10 = sum(n for label, n in offs["buckets"] if label in ("0 (same day)", "1", "2", "3", "4", "5", "6", "7", "8-10")) print(f"within 10 days of the issuer's last earnings 8-K: " f"{within10:,} ({within10 / offs['matched']:.1%} of matched)") g = results["gaps_202"] if g["median"] is not None: print(f"median gap between consecutive 2.02 filings: {g['median']}d") if args.verify_coverage: for name in ("daily_form4", "daily_events"): checked = [r for r in results[name] if r.get("match") is not None] matched = sum(1 for r in checked if r["match"]) print(f"coverage vs EDGAR daily index ({name}): " f"{matched}/{len(checked)} days match exactly") print(f"full results -> {args.out}") if __name__ == "__main__": main()