#!/usr/bin/env python3 """How late is Form 4 data? — reproducible filing-delay study. Measures the gap between when an insider transaction happened (`transaction_date`, as filed) and when the Form 4 reporting it reached EDGAR (`filed_date`), across every Form 4 filed in a fixed window. This is a data-infrastructure measurement, not a compliance audit and not investment advice: the number that matters for research is "when could anyone have known," and that is the filed date, not the transaction date. Backtests keyed on transaction_date look ahead by the exact distribution this script prints. Reproducible: a fixed --since/--until window returns the same population on any tier, any day you run it (the API's `until=` bound makes the window immutable). Runs on the free tier: a one-month window is ~80 requests (free tier: 10/min, so expect ~10 minutes; paid tiers are faster). Usage: export FILINGPULSE_API_KEY=fp_... # free key: filingpulse.io/signup.html python form4_filing_delay_study.py --since 2026-07-01 --until 2026-07-31 \ --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 form4_filing_delay_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 datetime import date, timedelta BASE_URL = os.environ.get("FILINGPULSE_BASE_URL", "https://api.filingpulse.io") API_KEY = os.environ.get("FILINGPULSE_API_KEY") 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() # SEC deadline context: Section 16 requires a Form 4 "before the end of the # second business day" after the transaction (17 CFR 240.16a-3(g)). Business # days exclude weekends and federal holidays. EDGAR observes federal holidays, # so the table below is the federal (not NYSE) calendar, observed dates. FEDERAL_HOLIDAYS = { # 2025 "2025-01-01", "2025-01-20", "2025-02-17", "2025-05-26", "2025-06-19", "2025-07-04", "2025-09-01", "2025-10-13", "2025-11-11", "2025-11-27", "2025-12-25", # 2026 (observed: Jul 4 falls on a Saturday -> observed Fri Jul 3) "2026-01-01", "2026-01-19", "2026-02-16", "2026-05-25", "2026-06-19", "2026-07-03", "2026-09-07", "2026-10-12", "2026-11-11", "2026-11-26", "2026-12-25", } CAL_BUCKETS = [(0, "same day"), (1, "1 day"), (2, "2 days"), (3, "3 days"), (5, "4-5 days"), (10, "6-10 days"), (30, "11-30 days"), (90, "31-90 days"), (None, "over 90 days")] BD_BUCKETS = [(0, "0"), (1, "1"), (2, "2 (deadline)"), (3, "3"), (4, "4"), (5, "5"), (10, "6-10"), (None, "over 10")] DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") def parse_iso(s): """Strict YYYY-MM-DD -> date, else None (dates are strings as filed and are occasionally malformed — that is data, and this script counts it).""" if not isinstance(s, str) or not DATE_RE.match(s): return None try: return date.fromisoformat(s) except ValueError: return None def is_business_day(d: date) -> bool: return d.weekday() < 5 and d.isoformat() not in FEDERAL_HOLIDAYS def business_days_between(txn: date, filed: date) -> int: """Business days in (txn, filed] — 'filed N business days after the transaction.' 0 means same day (or filed on a later non-business day with no business day in between).""" if filed <= txn: return 0 n, d = 0, txn while d < filed: d += timedelta(days=1) if is_business_day(d): n += 1 return n 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.""" 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 _ 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 body = e.read().decode("utf-8", "replace")[:200] sys.exit(f"API error {e.code} on {url}: {body}") sys.exit("giving up after repeated 429s") def fetch_window(since: date, until: date): """Walk the window one filed-date at a time (since=D&until=D), paging inside each day. Daily volume stays far below the 10,000 offset cap, so every filing in the window is reachable — and the per-day totals double as the coverage table.""" filings, daily = {}, [] d = since while d <= until: day = d.isoformat() got, total, offset = 0, None, 0 while total is None or got < total: page = api_get("/v1/insider-trades", {"since": day, "until": day, "limit": 200, "offset": offset}) total = page["total"] for obj in page["data"]: filings[obj["accession"]] = obj got += len(page["data"]) offset += 200 if not page["data"]: break time.sleep(0.05) daily.append({"date": day, "filed": total or 0}) print(f" {day}: {total or 0} filings", file=sys.stderr) d += timedelta(days=1) return filings, daily def edgar_index_counts(since: date, until: date, contact: str): """Distinct Form 4 (+4/A) accessions per day from EDGAR's own daily master index — the ground truth the API's per-day counts are checked against. Two 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 be earlier — re-disseminated corrections keep their original filed date (we found one filed 06-15 sitting in the 07-23 index). Membership in the index is NOT "filed that day"; only rows whose Date Filed matches the index day are counted, since that is the population a filed_date window returns.""" counts = {} d = since while d <= until: 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"form4-delay-study ({contact})"}) try: with urllib.request.urlopen(req, timeout=60, context=_SSL_CTX) as resp: text = resp.read().decode("latin-1") accessions, redisseminated = set(), set() for line in text.splitlines(): parts = line.split("|") if len(parts) == 5 and parts[2] in ("4", "4/A"): acc = parts[4].rsplit("/", 1)[-1].removesuffix(".txt") if parts[3] == day_compact: accessions.add(acc) else: redisseminated.add(acc) counts[d.isoformat()] = len(accessions) if redisseminated: print(f" note: {d.isoformat()} index carries " f"{len(redisseminated)} re-disseminated filing(s) with " f"an earlier filed date (not counted)", file=sys.stderr) except urllib.error.HTTPError as e: if e.code in (403, 404): counts[d.isoformat()] = None # no index published that day else: raise print(f" EDGAR index {d.isoformat()}: {counts[d.isoformat()]}", file=sys.stderr) time.sleep(1.1) # SEC fair access: stay well under their rate cap d += timedelta(days=1) return counts def bucketize(value, buckets): for upper, label in buckets: if upper is None or value <= upper: return label return buckets[-1][1] def percentile(sorted_vals, p): """Nearest-rank percentile — deterministic, no interpolation.""" if not sorted_vals: return None k = max(1, -(-len(sorted_vals) * p // 100)) # ceil without floats return sorted_vals[int(k) - 1] def analyze(filings: dict, daily: list, window: dict) -> dict: form4a = sum(1 for f in filings.values() if f.get("form_type") == "4/A") originals = [f for f in filings.values() if f.get("form_type") == "4"] holdings_only = undated = filed_before_txn = 0 cal_delays, bd_delays = [], [] cal_counts = {label: 0 for _, label in CAL_BUCKETS} bd_counts = {label: 0 for _, label in BD_BUCKETS} by_code = {} for f in originals: filed = parse_iso(f.get("filed_date")) txns = f.get("transactions") or [] if f.get("holdings_only") or not txns: holdings_only += 1 continue txn_dates = [parse_iso(t.get("transaction_date")) for t in txns] valid = [t for t in txn_dates if t is not None] if not valid or filed is None: undated += 1 continue earliest = min(valid) if filed < earliest: filed_before_txn += 1 # anomaly: filed date precedes transaction continue cal = (filed - earliest).days bd = business_days_between(earliest, filed) cal_delays.append(cal) bd_delays.append(bd) cal_counts[bucketize(cal, CAL_BUCKETS)] += 1 bd_counts[bucketize(bd, BD_BUCKETS)] += 1 # leg-level, by transaction code (each leg against its own date) for t, td in zip(txns, txn_dates): if td is None or filed < td: continue code = (t.get("transaction_code") or "?").upper() row = by_code.setdefault(code, {"legs": 0, "delays": [], "within_2bd": 0}) row["legs"] += 1 row["delays"].append((filed - td).days) if business_days_between(td, filed) <= 2: row["within_2bd"] += 1 cal_delays.sort() bd_delays.sort() on_time = sum(1 for b in bd_delays if b <= 2) analyzed = len(cal_delays) code_table = {} for code, row in sorted(by_code.items()): if row["legs"] < 25: # tiny cells are noise, fold into "other" other = code_table.setdefault( "other", {"legs": 0, "delays": [], "within_2bd": 0}) other["legs"] += row["legs"] other["delays"] += row["delays"] other["within_2bd"] += row["within_2bd"] else: code_table[code] = row for code, row in code_table.items(): row["delays"].sort() row["median_calendar_days"] = percentile(row["delays"], 50) row["share_within_2bd"] = ( f"{row['within_2bd'] / row['legs']:.4f}" if row["legs"] else None) del row["delays"] return { "study": "form4_filing_delay", "version": 1, "window": window, "filings": { "total_in_window": len(filings), "form_4": len(originals), "form_4a_amendments_excluded": form4a, "analyzed": analyzed, "holdings_only_excluded": holdings_only, "undated_excluded": undated, "filed_before_transaction_anomalies": filed_before_txn, }, "calendar_delay_days": { "buckets": [[label, cal_counts[label]] for _, label in CAL_BUCKETS], "percentiles": {f"p{p}": percentile(cal_delays, p) for p in (50, 75, 90, 95, 99)}, "max": cal_delays[-1] if cal_delays else None, }, "business_day_delay": { "buckets": [[label, bd_counts[label]] for _, label in BD_BUCKETS], "within_2_business_days": on_time, "share_within_2bd": f"{on_time / analyzed:.4f}" if analyzed else None, }, "by_transaction_code": {c: {k: v for k, v in r.items()} for c, r in sorted(code_table.items())}, "daily": daily, "generated_by": "form4_filing_delay_study.py v1", } def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--since", default="2026-07-01") ap.add_argument("--until", default="2026-07-31") ap.add_argument("--out", default="form4_delay_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)") since, until = parse_iso(args.since), parse_iso(args.until) if not since or not until or until < since: sys.exit("--since/--until must be YYYY-MM-DD with since <= until") years = {y for y in range(since.year, until.year + 1)} covered = {d[:4] for d in FEDERAL_HOLIDAYS} if not all(str(y) in covered for y in years): print("WARNING: window outside the built-in federal-holiday table — " "business-day figures will treat holidays as business days", file=sys.stderr) print(f"fetching Form 4 filings {since} .. {until}", file=sys.stderr) filings, daily = fetch_window(since, until) 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) index_counts = edgar_index_counts(since, until, contact) for row in daily: row["edgar_index"] = index_counts.get(row["date"]) row["match"] = (None if row["edgar_index"] is None else row["filed"] == row["edgar_index"]) results = analyze(filings, daily, {"since": since.isoformat(), "until": until.isoformat()}) with open(args.out, "w", encoding="utf-8") as fh: json.dump(results, fh, indent=2) fh.write("\n") f = results["filings"] c = results["calendar_delay_days"] b = results["business_day_delay"] print(f"\n{f['total_in_window']} filings in window " f"({f['form_4']} Form 4, {f['form_4a_amendments_excluded']} 4/A " f"excluded); {f['analyzed']} analyzed") print(f"calendar delay: p50={c['percentiles']['p50']}d " f"p90={c['percentiles']['p90']}d p99={c['percentiles']['p99']}d " f"max={c['max']}d") if b["share_within_2bd"] is not None: print(f"filed within 2 business days: {b['within_2_business_days']} " f"({float(b['share_within_2bd']):.1%})") if args.verify_coverage: checked = [r for r in daily if r.get("match") is not None] matched = sum(1 for r in checked if r["match"]) print(f"coverage vs EDGAR daily index: {matched}/{len(checked)} " f"days match exactly") print(f"full results -> {args.out}") if __name__ == "__main__": main()