#!/usr/bin/env python3 """When insider filings get corrected — reproducible Form 4 amendment study. A Form 4 is not immutable. Insiders re-file corrections as Form 4/A, and each 4/A carries `amendment_date` — the date of the original submission it amends (as filed). This script measures, over a fixed window: 1. How much Form 4 traffic is corrections (4/A share of all filings). 2. Correction lag: when a 4/A arrives, how old is the submission it corrects? That lag is exactly how long the superseded version was the only version anyone had. 3. A cohort view: of the originals filed in one month, how many had been amended by a fixed later date — a lower bound on the eventual rate, because corrections keep arriving after any observation horizon. This is a data-infrastructure measurement, not a compliance audit and not investment advice. The question it answers is point-in-time correctness: any dataset snapshotted at time T still holds the pre-correction version of every filing whose 4/A had not yet arrived at T. Reproducible: all three date bounds are fixed arguments, and the API's `until=` filter makes a fixed window return the same population on any tier, any day you run it. Runs on the free tier: the default ~8-week fetch range is ~160 requests (free tier: 10/min, so expect ~20 minutes; paid tiers are faster). Usage: export FILINGPULSE_API_KEY=fp_... # free key: filingpulse.io/signup.html python form4_amendments_study.py --since 2026-07-01 --until 2026-07-31 \ --observed-through 2026-08-24 --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_amendments_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() # Correction lag: days between the original submission date a 4/A names # (`amendment_date`, as filed) and the day the 4/A itself reached EDGAR. LAG_BUCKETS = [(0, "same day"), (1, "1 day"), (3, "2-3 days"), (7, "4-7 days"), (14, "8-14 days"), (30, "15-30 days"), (90, "31-90 days"), (365, "91-365 days"), (None, "over a year")] 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 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 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_range(start: date, end: date): """Walk the range 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 range is reachable — and the per-day totals double as the coverage table.""" filings, daily = {}, [] d = start while d <= end: 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(start: date, end: 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. 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 = start while d <= 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"form4-amendments-study ({contact})"}) text = None for attempt in range(4): 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 raise except (OSError, TimeoutError) as e: if attempt == 3: raise wait = 5 * (attempt + 1) print(f" network error ({e}) — retrying in {wait}s", file=sys.stderr) time.sleep(wait) if text is None: counts[d.isoformat()] = None else: accessions = set() for line in text.splitlines(): parts = line.split("|") if len(parts) == 5 and parts[2] in ("4", "4/A"): if parts[3] == day_compact: acc = parts[4].rsplit("/", 1)[-1].removesuffix(".txt") accessions.add(acc) counts[d.isoformat()] = len(accessions) 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 owner_key(filing: dict): """Stable identity for the insiders on a filing: sorted owner CIKs (name as fallback when a CIK is missing, '?' when both are).""" owners = filing.get("reporting_owners") or [] return tuple(sorted((o.get("cik") or o.get("name") or "?") for o in owners)) def analyze(filings: dict, daily: list, window: dict) -> dict: since = date.fromisoformat(window["since"]) until = date.fromisoformat(window["until"]) observed = date.fromisoformat(window["observed_through"]) all_form4 = [f for f in filings.values() if f.get("form_type") == "4"] all_4a = [f for f in filings.values() if f.get("form_type") == "4/A"] # -- correction lag over every 4/A in the fetch range --------------------- undated = anomalies = 0 lags = [] # (lag_days, amendment) for the analyzed set lag_counts = {label: 0 for _, label in LAG_BUCKETS} oldest_original = None for f in all_4a: filed = parse_iso(f.get("filed_date")) original = parse_iso(f.get("amendment_date")) if filed is None or original is None: undated += 1 continue if original > filed: anomalies += 1 # amendment claims to predate its own filing continue lag = (filed - original).days lags.append((lag, f)) lag_counts[bucketize(lag, LAG_BUCKETS)] += 1 if oldest_original is None or original < oldest_original: oldest_original = original lag_values = sorted(l for l, _ in lags) # -- the cohort: originals filed in [since, until] ------------------------ in_cohort_window = [f for f in filings.values() if (d := parse_iso(f.get("filed_date"))) is not None and since <= d <= until] cohort_originals = [f for f in in_cohort_window if f.get("form_type") == "4"] cohort_4a = [f for f in in_cohort_window if f.get("form_type") == "4/A"] # amendments (any filed date in range) whose named original submission # date falls in the cohort window cohort_amendments = [(lag, f) for lag, f in lags if since <= parse_iso(f["amendment_date"]) <= until] # distinct original submissions, estimated: a 4/A names the original's # date but not its accession, so (issuer CIK, owners, original date) is # the best available identity — stated as an estimate, not a join. distinct_originals = { ((f.get("issuer") or {}).get("cik"), owner_key(f), f["amendment_date"]) for _, f in cohort_amendments} cohort_lags = sorted(l for l, _ in cohort_amendments) n_orig = len(cohort_originals) n_est = len(distinct_originals) return { "study": "form4_amendments", "version": 1, "window": window, "fetched": { "days": len(daily), "total": len(filings), "form_4": len(all_form4), "form_4a": len(all_4a), }, "cohort_window_filings": { "total": len(in_cohort_window), "form_4": len(cohort_originals), "form_4a": len(cohort_4a), }, "amendments_all": { "total": len(all_4a), "undated_excluded": undated, "original_after_filed_anomalies": anomalies, "analyzed": len(lag_values), "lag_buckets": [[label, lag_counts[label]] for _, label in LAG_BUCKETS], "percentiles": {f"p{p}": percentile(lag_values, p) for p in (50, 75, 90, 95, 99)}, "max": lag_values[-1] if lag_values else None, "oldest_original_submission": oldest_original.isoformat() if oldest_original else None, }, "cohort": { "originals": n_orig, "amendments_observed": len(cohort_amendments), "estimated_originals_amended": n_est, "rate_lower_bound": f"{n_est / n_orig:.4f}" if n_orig else None, "min_observation_days": (observed - until).days, "within_days": {str(k): sum(1 for l in cohort_lags if l <= k) for k in (2, 7, 30)}, "median_lag_days": percentile(cohort_lags, 50), }, "daily": daily, "generated_by": "form4_amendments_study.py v1", } def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--since", default="2026-07-01", help="cohort window start (originals filed from this date)") ap.add_argument("--until", default="2026-07-31", help="cohort window end (inclusive)") ap.add_argument("--observed-through", default="2026-08-24", help="fixed observation horizon: amendments filed through " "this date (inclusive) are counted") ap.add_argument("--out", default="form4_amendments_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 = parse_iso(args.since) until = parse_iso(args.until) observed = parse_iso(args.observed_through) if not since or not until or until < since: sys.exit("--since/--until must be YYYY-MM-DD with since <= until") if not observed or observed < until: sys.exit("--observed-through must be YYYY-MM-DD, on or after --until") print(f"fetching Form 4 + 4/A filings {since} .. {observed}", file=sys.stderr) filings, daily = fetch_range(since, observed) 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, observed, 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(), "observed_through": observed.isoformat()}) with open(args.out, "w", encoding="utf-8") as fh: json.dump(results, fh, indent=2) fh.write("\n") fe = results["fetched"] am = results["amendments_all"] co = results["cohort"] print(f"\n{fe['total']} filings fetched over {fe['days']} days " f"({fe['form_4']} Form 4, {fe['form_4a']} Form 4/A)") if am["analyzed"]: print(f"correction lag: p50={am['percentiles']['p50']}d " f"p90={am['percentiles']['p90']}d p99={am['percentiles']['p99']}d " f"max={am['max']}d (over {am['analyzed']} amendments)") if co["rate_lower_bound"] is not None: print(f"cohort {results['window']['since']}..{results['window']['until']}: " f"{co['estimated_originals_amended']} of {co['originals']} " f"originals amended within the horizon " f"({float(co['rate_lower_bound']):.2%}, a lower bound)") 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()