#!/usr/bin/env python3 """Which corporate events get disclosed on Fridays? — reproducible 8-K study. Every 8-K declares what happened via the SEC's official item codes (2.02 = earnings, 5.02 = officer/director changes, 4.02 = non-reliance on prior financials, ...). This script measures, over a fixed filed-date window: 1. What actually gets filed: filings per item code, items per filing, and the item combinations that travel together. 2. The weekday pattern: how each item code's filings distribute across Monday-Friday, against the all-8-K baseline — the measurable core of the pattern popularly called the "Friday news dump". One EDGAR mechanic shapes how to read "Friday" here, so it is worth being precise: under Regulation S-T Rule 13 (17 CFR 232.13), an 8-K accepted after 5:30 p.m. Eastern is deemed filed the NEXT business day (the 10 p.m. same-day window covers Forms 3/4/5, 13D/G, 144 — not 8-Ks). So a Friday filed_date means EDGAR accepted the filing by Friday 5:30 p.m. ET, and a submission pushed out Friday evening carries Monday's date. Both edges of the week are reported. This is a data-infrastructure measurement, not a compliance audit and not investment advice: it counts disclosures by their own declared item codes and dates. It does not characterize any filing as good or bad news. Reproducible: the 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 12-month window is ~66,000 filings ≈ ~700 requests (free tier: 10/min, so expect ~70 minutes; paid tiers are faster — a single month runs in a few minutes on any tier). Usage: export FILINGPULSE_API_KEY=fp_... # free key: filingpulse.io/signup.html python eightk_weekday_study.py --since 2025-09-01 --until 2026-08-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 eightk_weekday_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 collections import Counter 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") WEEKDAYS = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") ISO_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") COMPACT_RE = re.compile(r"^\d{8}$") 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_compact(s): """As-filed YYYYMMDD -> date, else None. 8-K objects carry dates exactly as filed in EDGAR's SGML headers; malformed values are data, and this script counts them instead of guessing.""" 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 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_range(start: date, end: date): """Walk the range one filed-date at a time (since=D&until=D), paging inside each day. Fetching by single day keeps every filing reachable (daily 8-K volume stays far below the offset cap) and the per-day totals double as the coverage table. It also pins each filing's day-of-week to the window bound itself rather than to a re-parse of the object's date.""" by_day, daily = [], [] d = start while d <= end: day = d.isoformat() events, total, offset = [], None, 0 while total is None or len(events) < total: page = api_get("/v1/events", {"since": day, "until": day, "limit": 200, "offset": offset}) total = page["total"] events.extend(page["data"]) offset += 200 if not page["data"]: break time.sleep(0.05) by_day.append((d, events)) daily.append({"date": day, "filed": total or 0}) print(f" {day}: {total or 0} events", file=sys.stderr) d += timedelta(days=1) return by_day, daily def edgar_index_counts(start: date, end: date, contact: str): """Distinct 8-K (+8-K/A) accessions per filed date from EDGAR's own daily master indexes — the ground truth the API's per-day counts are checked against. 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 for 8-Ks (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. Verified first-hand: 0001493152-25-017331 carries Date Filed 2025-10-07 and appears only in the 2025-10-08 index.""" buckets = {} # filed date iso -> set of 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"eightk-weekday-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 365-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 and parts[2] in ("8-K", "8-K/A"): df = parts[3] if len(df) == 8 and df.isdigit(): iso = f"{df[:4]}-{df[4:6]}-{df[6:8]}" acc = parts[4].rsplit("/", 1)[-1].removesuffix(".txt") buckets.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) counts = {} d = start while d <= end: iso = d.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 d += timedelta(days=1) return counts def analyze(by_day: list, daily: list, window: dict) -> dict: total = form_8k = form_8ka = other_forms = 0 date_anomalies = 0 # object's as-filed date disagrees with its window day no_items = 0 # originals whose header declared no items weekday_filings = Counter() # weekday -> 8-K originals days_with_filings = Counter() # weekday -> days with >=1 event items_per_filing = Counter() # distinct-code count -> filings by_item = {} # code -> {"caption", "filings", "weekday": Counter} pair_counts = Counter() # frozen (code_a, code_b) -> filings carrying both for d, events in by_day: wd = WEEKDAYS[d.weekday()] if events: days_with_filings[wd] += 1 for ev in events: total += 1 ft = ev.get("form_type") if ft == "8-K/A": form_8ka += 1 continue # re-filings: counted, not analyzed if ft != "8-K": other_forms += 1 continue form_8k += 1 as_filed = parse_compact(ev.get("filed_date")) if as_filed is not None and as_filed != d: date_anomalies += 1 # belt and suspenders; expected 0 weekday_filings[wd] += 1 codes = [] seen = set() for item in (ev.get("items") or []): code = item.get("code") or "?" if code in seen: continue # duplicate declaration, count once seen.add(code) codes.append((code, item.get("caption") or "")) items_per_filing[min(len(codes), 5)] += 1 if not codes: no_items += 1 for code, caption in codes: rec = by_item.setdefault( code, {"caption": caption, "filings": 0, "weekday": Counter()}) if caption and not rec["caption"]: rec["caption"] = caption rec["filings"] += 1 rec["weekday"][wd] += 1 for i, (a, _) in enumerate(codes): for b, _ in codes[i + 1:]: pair_counts[tuple(sorted((a, b)))] += 1 # calendar occurrences of each weekday inside the window, so shares can # be normalized against how often that weekday even happened days_in_window = Counter() d = date.fromisoformat(window["since"]) end = date.fromisoformat(window["until"]) while d <= end: days_in_window[WEEKDAYS[d.weekday()]] += 1 d += timedelta(days=1) def wd_dict(counter): return {wd: counter.get(wd, 0) for wd in WEEKDAYS} return { "study": "eightk_weekday", "version": 1, "window": window, "fetched": { "days": len(daily), "total": total, "form_8k": form_8k, "form_8ka_amendments_excluded": form_8ka, "other_form_types": other_forms, }, "analyzed": form_8k, "asfiled_date_anomalies": date_anomalies, "no_items_filings": no_items, "items_per_filing": {str(k) if k < 5 else "5+": v for k, v in sorted(items_per_filing.items())}, "weekday": { "days_in_window": wd_dict(days_in_window), "days_with_filings": wd_dict(days_with_filings), "filings": wd_dict(weekday_filings), }, "by_item": { code: {"caption": rec["caption"], "filings": rec["filings"], "weekday": wd_dict(rec["weekday"])} for code, rec in sorted(by_item.items(), key=lambda kv: -kv[1]["filings"]) }, "top_pairs": [[a, b, n] for (a, b), n in pair_counts.most_common(10)], "daily": daily, "generated_by": "eightk_weekday_study.py v1", } def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--since", default="2025-09-01", help="window start (8-Ks with filed_date from this date)") ap.add_argument("--until", default="2026-08-31", help="window end (inclusive)") ap.add_argument("--out", default="eightk_weekday_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)") if not ISO_RE.match(args.since) or not ISO_RE.match(args.until): sys.exit("--since/--until must be YYYY-MM-DD") since = date.fromisoformat(args.since) until = date.fromisoformat(args.until) if until < since: sys.exit("--since must be on or before --until") print(f"fetching 8-K events {since} .. {until}", file=sys.stderr) by_day, daily = fetch_range(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(by_day, 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") fe = results["fetched"] wk = results["weekday"]["filings"] print(f"\n{fe['total']} events fetched over {fe['days']} days " f"({fe['form_8k']} 8-K, {fe['form_8ka_amendments_excluded']} 8-K/A)") if results["analyzed"]: base = results["analyzed"] print("weekday shares (8-K originals): " + " ".join(f"{wd}={wk[wd] / base:.1%}" for wd in ("Mon", "Tue", "Wed", "Thu", "Fri"))) top = list(results["by_item"].items())[:5] print("top items: " + ", ".join( f"{code} ({rec['filings']:,})" for code, rec in top)) 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()