fundamentals: co-déclarants (clé (cik, accn)), poller borné, dictionnaire de couverture

E1 — edgar_filings : clé primaire composite (cik, accn) + index sur accn. Un numéro
d'accession est partagé par les co-déclarants (famille Entergy 65984/66901/…) : la PK
sur accn seule faisait échouer tout _upsert_filings (IntegrityError, 18 entreprises
jamais ingérées). Migration idempotente (fundamentals/migrations.py) : inline au
démarrage si ≤ 300 k lignes, sinon avertissement + scripts/migrate_edgar_filings.py à
lancer avant le déploiement ; l'upsert passe par INSERT … ON CONFLICT DO NOTHING et
tolère l'ancien schéma pendant la transition. _known_pairs raisonne en paires (cik, accn)
sans balayer la table ; publish_filing_event produit un événement par co-déclarant.

E2 — poller incrémental : filtre des entrées Atom par TRACKED_FORMS (le flux type=8-K
renvoie des 8-K/A : Genasys ré-ingéré à chaque cycle pendant 41 h) ; TRACKED_FORMS
étendu aux amendements et rapports de transition ; garde-fou Redis edgar:seen_accn
(TTL 7 j) + edgar:failed_accn (3 tentatives) ; companyfacts_404 enregistre quand même
les dépôts et n'est plus compté comme échec (1 671 « failures » venaient de deux CIK
sans facts retentés à chaque cycle) ; failure_samples désormais cumulés (plus écrasés
par un cycle sain) ; index quotidien : jours ouvrables seulement, 403/404 = non publié
(les master.idx de fin de semaine n'existent pas) ; bulk rebuild seulement si des
lignes ont changé ; httpx en WARNING dans le script.

E3 — fund_statements.coverage (1,17 Ko × 1,22 M lignes = 1,9 Go) → dictionnaire
fund_coverage_blob(id, sha1, json) référencé par coverage_id ; lecture hybride
(JSON inline hérité OU référence) dans service._rows et bulk.build ;
_replace_statements ne réécrit que si l'empreinte des lignes a changé ;
scripts/migrate_coverage.py convertit l'existant par lots avec reprise (--vacuum
optionnel) ; auto_vacuum=INCREMENTAL sur une base jeune + incremental_vacuum dans
edgar_reconcile.

Tests : tests/test_fundamentals_ingest.py (co-déclarants, boucle du poller, retries,
échantillons, 403 index, réécriture conditionnelle, dictionnaire, migration).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Simon-Pierre Boucher committed 18 days ago (Sep 6, 2026) parent 02b779a

13 changed files +1,114 −70

modified hfmarketdata/api/bulk/build.py +4 −2

		`@@ -65,15 +65,17 @@ def build_year(year: int) -> dict:`
65	65	order_by=[t.c.filed_date.desc(), t.c.id.desc()]).label("rn")
66	66	sub = select(t, rn).where(t.c.fiscal_year == int(year)).subquery()
67	67	q = select(sub).where(sub.c.rn == 1).order_by(sub.c.ticker, sub.c.statement, sub.c.fiscal_quarter)
	68	+ from fundamentals.coverage_store import hydrate
68	69	with session() as s:
69		− df = pd.DataFrame([dict(r._mapping) for r in s.execute(q)])
	70	+ rows = [dict(r._mapping) for r in s.execute(q)]
	71	+ df = pd.DataFrame(hydrate(rows)) # coverage comes back inline (dictionary reference resolved)
70	72	if not df.empty:
71	73	df = df.drop(columns=["rn", "id"])
72	74	df["coverage"] = df["coverage"].map(lambda v: json.dumps(v) if v is not None else None)
73	75	for c in ("period_start", "period_end", "filed_date"):
74	76	df[c] = pd.to_datetime(df[c])
75	77	else:
76		− df = pd.DataFrame(columns=[c for c in t.c.keys() if c != "id"])
	78	+ df = pd.DataFrame(columns=[c for c in t.c.keys() if c not in ("id", "coverage_id")])
77	79	p = fundamentals_path(year)
78	80	tmp = p.with_suffix(".parquet.tmp")
79	81	df.to_parquet(tmp, index=False, compression="zstd")

added hfmarketdata/api/fundamentals/coverage_store.py +128 −0

		`@@ -0,0 +1,128 @@`
	1	+"""Dictionary encoding of `fund_statements.coverage`.
	2	+
	3	+The coverage document of a statement row ({account: {"reason": …} \| {"tag": …, "priority": …} \| {"computed": …}})
	4	+averages 1.1 KB and repeats massively (a company's quarters share the same null reasons): 1.22 M rows carried
	5	+1.9 GB of JSON for a few tens of thousands of distinct documents. Rows now store `coverage_id` →
	6	+`fund_coverage_blob(id, sha1, json)` and leave `coverage` NULL.
	7	+
	8	+* `intern(docs)` → ids for a batch of documents (insert-or-get by sha1 of the canonical JSON)
	9	+* `hydrate(rows)` → puts the `coverage` dict back on rows read from the table (inline legacy JSON or
	10	+ dictionary reference — both shapes coexist until `scripts/migrate_coverage.py` is done)
	11	+* `digest(doc)` → canonical sha1 (also used by the change detection of `ingest._replace_statements`)
	12	+
	13	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	14	+"""
	15	+from __future__ import annotations
	16	+
	17	+import hashlib
	18	+import json
	19	+import threading
	20	+from typing import Any
	21	+
	22	+from sqlalchemy import select
	23	+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
	24	+
	25	+from core.db import session
	26	+
	27	+from .models import FundCoverageBlob
	28	+
	29	+_lock = threading.Lock()
	30	+_id_by_sha: dict[str, int] = {} # process-wide caches (blobs are immutable)
	31	+_doc_by_id: dict[int, dict[str, Any]] = {}
	32	+CACHE_MAX = 50_000
	33	+
	34	+
	35	+def canonical(doc: dict[str, Any] \| None) -> str:
	36	+ return json.dumps(doc or {}, sort_keys=True, separators=(",", ":"), default=str)
	37	+
	38	+
	39	+def digest(doc: dict[str, Any] \| None) -> str:
	40	+ return hashlib.sha1(canonical(doc).encode()).hexdigest()
	41	+
	42	+
	43	+def _remember(sha: str, id_: int, doc: dict[str, Any] \| None) -> None:
	44	+ with _lock:
	45	+ if len(_id_by_sha) > CACHE_MAX:
	46	+ _id_by_sha.clear()
	47	+ _doc_by_id.clear()
	48	+ _id_by_sha[sha] = id_
	49	+ if doc is not None:
	50	+ _doc_by_id[id_] = doc
	51	+
	52	+
	53	+def intern(docs: list[dict[str, Any] \| None]) -> list[int]:
	54	+ """Return the blob id of every document (same order), inserting the unseen ones in one round trip."""
	55	+ shas = [digest(d) for d in docs]
	56	+ by_sha: dict[str, dict[str, Any] \| None] = {}
	57	+ for s, d in zip(shas, docs):
	58	+ by_sha.setdefault(s, d)
	59	+ with _lock:
	60	+ missing = [s for s in by_sha if s not in _id_by_sha]
	61	+ if missing:
	62	+ with session() as s:
	63	+ for i in range(0, len(missing), 500):
	64	+ chunk = missing[i:i + 500]
	65	+ s.execute(sqlite_insert(FundCoverageBlob).on_conflict_do_nothing(index_elements=["sha1"]),
	66	+ [{"sha1": sha, "json": canonical(by_sha[sha])} for sha in chunk])
	67	+ for row in s.execute(select(FundCoverageBlob.id, FundCoverageBlob.sha1).where(FundCoverageBlob.sha1.in_(chunk))):
	68	+ _remember(row.sha1, int(row.id), by_sha[row.sha1] or {})
	69	+ with _lock:
	70	+ return [_id_by_sha[sha] for sha in shas]
	71	+
	72	+
	73	+def fetch(ids: set[int]) -> dict[int, dict[str, Any]]:
	74	+ """Documents of the given blob ids (cache first, then one query for the rest)."""
	75	+ out: dict[int, dict[str, Any]] = {}
	76	+ with _lock:
	77	+ for i in ids:
	78	+ if i in _doc_by_id:
	79	+ out[i] = _doc_by_id[i]
	80	+ todo = [i for i in ids if i not in out]
	81	+ if todo:
	82	+ with session() as s:
	83	+ for i in range(0, len(todo), 500):
	84	+ for row in s.execute(select(FundCoverageBlob.id, FundCoverageBlob.sha1, FundCoverageBlob.json)
	85	+ .where(FundCoverageBlob.id.in_(todo[i:i + 500]))):
	86	+ doc = json.loads(row.json)
	87	+ out[int(row.id)] = doc
	88	+ _remember(row.sha1, int(row.id), doc)
	89	+ return out
	90	+
	91	+
	92	+def sha_of_ids(ids: set[int]) -> dict[int, str]:
	93	+ """sha1 of the given blob ids (for change detection without loading the documents)."""
	94	+ out: dict[int, str] = {}
	95	+ with _lock:
	96	+ rev = {i: s for s, i in _id_by_sha.items()}
	97	+ for i in ids:
	98	+ if i in rev:
	99	+ out[i] = rev[i]
	100	+ todo = [i for i in ids if i not in out]
	101	+ if todo:
	102	+ with session() as s:
	103	+ for i in range(0, len(todo), 500):
	104	+ for row in s.execute(select(FundCoverageBlob.id, FundCoverageBlob.sha1).where(FundCoverageBlob.id.in_(todo[i:i + 500]))):
	105	+ out[int(row.id)] = row.sha1
	106	+ _remember(row.sha1, int(row.id), None)
	107	+ return out
	108	+
	109	+
	110	+def hydrate(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
	111	+ """In place: rows read from `fund_statements` get their `coverage` dict (from the inline JSON or the
	112	+ dictionary); `coverage_id` is removed so callers see the historical row shape."""
	113	+ ids = {int(r["coverage_id"]) for r in rows if r.get("coverage") is None and r.get("coverage_id") is not None}
	114	+ docs = fetch(ids) if ids else {}
	115	+ for r in rows:
	116	+ cid = r.pop("coverage_id", None)
	117	+ if r.get("coverage") is None:
	118	+ r["coverage"] = docs.get(int(cid), {}) if cid is not None else {}
	119	+ return rows
	120	+
	121	+
	122	+def reset_cache_for_tests() -> None:
	123	+ with _lock:
	124	+ _id_by_sha.clear()
	125	+ _doc_by_id.clear()
	126	+
	127	+
	128	+__all__ = ["intern", "fetch", "hydrate", "digest", "canonical", "sha_of_ids", "reset_cache_for_tests"]

modified hfmarketdata/api/fundamentals/edgar_client.py +25 −7

		`@@ -35,8 +35,15 @@ URL_COMPANY_TICKERS = f"{WWW_BASE}/files/company_tickers.json"`
35	35	URL_COMPANY_TICKERS_EXCHANGE = f"{WWW_BASE}/files/company_tickers_exchange.json"
36	36	URL_ATOM_CURRENT = WWW_BASE + "/cgi-bin/browse-edgar?action=getcurrent&type={form}&owner=include&count=100&output=atom"
37	37
38		−TRACKED_FORMS = ("10-K", "10-Q", "8-K", "20-F", "10-K/A", "10-Q/A", "20-F/A", "40-F", "6-K")
39		−FINANCIAL_FORMS = ("10-K", "10-Q", "20-F", "40-F", "10-K/A", "10-Q/A", "20-F/A", "10-KT", "10-QT")
	38	+# Forms recorded in edgar_filings and accepted from the live feeds. Amendments (`/A`) and transition reports
	39	+# (`10-KT`, `10-QT`) are included so an accession seen in the Atom feed always lands in the table — an
	40	+# untracked form seen in the feed (e.g. an 8-K/A returned by the `type=8-K` feed) would otherwise be "new" at
	41	+# every polling cycle and re-ingest its company for ever.
	42	+TRACKED_FORMS = ("10-K", "10-Q", "8-K", "20-F", "40-F", "6-K", "10-KT", "10-QT",
	43	+ "10-K/A", "10-Q/A", "8-K/A", "20-F/A", "40-F/A", "6-K/A", "10-KT/A", "10-QT/A")
	44	+FINANCIAL_FORMS = ("10-K", "10-Q", "20-F", "40-F", "10-K/A", "10-Q/A", "20-F/A", "40-F/A", "10-KT", "10-QT",
	45	+ "10-KT/A", "10-QT/A")
	46	+MISSING_STATUS = (403, 404) # www.sec.gov/Archives (S3) answers 403 for objects that do not exist yet (weekend indexes)
40	47
41	48
42	49	def cik10(cik: int \| str) -> str:
		`@@ -105,7 +112,9 @@ class EdgarClient:`
105	112	"User-Agent": settings.sec_user_agent, "Accept-Encoding": "gzip, deflate", "Accept": "application/json,/"})
106	113
107	114	# ------------------------------------------------------------------------------------ low level
108		− def _request(self, url: str, *, accept_404: bool = False) -> httpx.Response \| None:
	115	+ def _request(self, url: str, *, accept_404: bool = False, accept_missing: bool = False) -> httpx.Response \| None:
	116	+ """GET with the token bucket and backoff. `accept_404` → None on 404; `accept_missing` → None on 403/404
	117	+ (Archives objects that are not published yet, e.g. a daily index on a weekend)."""
109	118	delay = 0.5
110	119	for attempt in range(self.max_retries + 1):
111	120	self.bucket.acquire()
		`@@ -122,6 +131,8 @@ class EdgarClient:`
122	131	continue
123	132	if r.status_code == 404 and accept_404:
124	133	return None
	134	+ if r.status_code in MISSING_STATUS and accept_missing:
	135	+ return None
125	136	if r.status_code in (429, 503, 500, 502, 504):
126	137	if attempt == self.max_retries:
127	138	self.stats.errors.append(f"{url}: HTTP {r.status_code}")
		`@@ -159,9 +170,16 @@ class EdgarClient:`
159	170	tmp.replace(cache_path)
160	171	return data
161	172
162		− def get_text(self, url: str) -> str:
163		− r = self._request(url)
164		− return r.text if r is not None else ""
	173	+ def get_text(self, url: str, *, accept_missing: bool = False) -> str \| None:
	174	+ """Body as text; None when `accept_missing` and the object does not exist (403/404)."""
	175	+ r = self._request(url, accept_missing=accept_missing)
	176	+ return r.text if r is not None else None
	177	+
	178	+ def daily_index(self, d: date) -> str \| None:
	179	+ """`master.YYYYMMDD.idx` of one day, None when not published (weekend, holiday, not yet generated).
	180	+ Indexes live under `Archives/edgar/daily-index/{year}/QTR{q}/`."""
	181	+ q = (d.month - 1) // 3 + 1
	182	+ return self.get_text(f"{WWW_BASE}/Archives/edgar/daily-index/{d.year}/QTR{q}/master.{d:%Y%m%d}.idx", accept_missing=True)
165	183
166	184	# ------------------------------------------------------------------------------------ documents
167	185	def company_tickers(self, max_age_hours: float = 24) -> dict[str, Any]:
		`@@ -228,7 +246,7 @@ class EdgarClient:`
228	246
229	247	def atom_current(self, form: str = "") -> str:
230	248	"""Fallback feed (Atom) of the latest filings on EDGAR, optionally filtered by form type."""
231		− return self.get_text(URL_ATOM_CURRENT.format(form=form))
	249	+ return self.get_text(URL_ATOM_CURRENT.format(form=form)) or ""
232	250
233	251	def close(self) -> None:
234	252	self._http.close()

modified hfmarketdata/api/fundamentals/ingest.py +233 −43

		`@@ -22,11 +22,13 @@ from pathlib import Path`
22	22	from typing import Any
23	23
24	24	import pandas as pd
25		−from sqlalchemy import delete, insert, select
	25	+from sqlalchemy import delete, insert, select, update
	26	+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
26	27
27	28	from core.config import settings
28	29	from core.db import session
29	30
	31	+from . import coverage_store as CS
30	32	from . import mapping as M
31	33	from . import utcnow
32	34	from . import normalize as N
		`@@ -39,7 +41,13 @@ from .service import ratio_inputs`
39	41
40	42	log = logging.getLogger("hfmarketdata.fundamentals.ingest")
41	43
42		−EVENT_FORMS = ("10-K", "10-Q", "8-K", "20-F", "10-K/A", "10-Q/A", "20-F/A", "40-F", "6-K")
	44	+EVENT_FORMS = TRACKED_FORMS # every tracked filing (one event per co-registrant) is published
	45	+NO_FACTS = "companyfacts_404" # permanent: no XBRL facts on EDGAR (funds, trusts, paper filers)
	46	+SEEN_KEY = "edgar:seen_accn" # Redis SET of "cik:accn" pairs already handled by the poller
	47	+SEEN_TTL_S = 7 * 86_400
	48	+FAILED_KEY = "edgar:failed_accn" # Redis HASH "cik:accn" → attempts (transient errors are retried…)
	49	+MAX_ATTEMPTS = 3 # …this many times, then the pair is marked seen and left to reconcile
	50	+FAILURE_SAMPLES_MAX = 10
43	51
44	52
45	53	def facts_dir(cik: int) -> Path:
		`@@ -145,6 +153,12 @@ class IngestResult:`
145	153	seconds: float = 0.0
146	154	error: str \| None = None
147	155	new_filings: list[str] = field(default_factory=list)
	156	+ changed: bool = False # statements rewritten (content differed from what was stored)
	157	+
	158	+ @property
	159	+ def failed(self) -> bool:
	160	+ """A real failure — `companyfacts_404` is a permanent, expected state, not an error to count."""
	161	+ return self.error is not None and self.error != NO_FACTS
148	162
149	163
150	164	def ingest_company(client: EdgarClient \| None, cik: int, *, refresh: bool = False, with_metalinks: bool = True,
		`@@ -160,7 +174,12 @@ def ingest_company(client: EdgarClient \| None, cik: int, *, refresh: bool = Fals`
160	174	cf = companyfacts if companyfacts is not None else client.companyfacts(cik, refresh=refresh)
161	175	sub = submissions if submissions is not None else client.submissions(cik, refresh=refresh)
162	176	if cf is None:
163		− res.error = "companyfacts_404"
	177	+ # no XBRL facts: still record the filings so the accessions are known (the poller must not see them
	178	+ # as new for ever) and stamp the company
	179	+ res.error = NO_FACTS
	180	+ filings = N.filings_from_submissions(sub)
	181	+ _upsert_filings(cik, filings, pd.DataFrame())
	182	+ res.filings = len(filings)
164	183	_mark_company(cik, sub, None)
165	184	return res
166	185	facts = N.facts_frame(cf)
		`@@ -179,7 +198,7 @@ def ingest_company(client: EdgarClient \| None, cik: int, *, refresh: bool = Fals`
179	198	except Exception as e: # pragma: no cover
180	199	log.warning("metalinks %s/%s: %s", cik, latest_10k.accn, e)
181	200	norm = N.normalize_company(int(cik), ticker, facts, filings, fye, metalinks)
182		− _replace_statements(cik, norm.rows)
	201	+ res.changed = _replace_statements(cik, norm.rows)
183	202	_upsert_mapping_log(cik, norm.mapping_log)
184	203	cov = compute_coverage(cik, ticker, norm.rows, len(filings), norm.stats)
185	204	_mark_company(cik, sub, fye)
		`@@ -217,39 +236,120 @@ def _known_accns(cik: int) -> set[str]:`
217	236	return set(s.scalars(select(EdgarFiling.accn).where(EdgarFiling.cik == int(cik))))
218	237
219	238
220		−def _upsert_filings(cik: int, filings: dict[str, N.Filing], facts: pd.DataFrame) -> None:
	239	+def _known_pairs(pairs: set[tuple[int, str]]) -> set[tuple[int, str]]:
	240	+ """Subset of the (cik, accn) pairs already in edgar_filings — looked up by accn in chunks, never a full scan."""
	241	+ if not pairs:
	242	+ return set()
	243	+ accns = sorted({a for _, a in pairs})
	244	+ found: set[tuple[int, str]] = set()
	245	+ with session() as s:
	246	+ for i in range(0, len(accns), 500):
	247	+ for cik, accn in s.execute(select(EdgarFiling.cik, EdgarFiling.accn).where(EdgarFiling.accn.in_(accns[i:i + 500]))):
	248	+ found.add((int(cik), accn))
	249	+ return found & pairs
	250	+
	251	+
	252	+def _upsert_filings(cik: int, filings: dict[str, N.Filing], facts: pd.DataFrame) -> int:
	253	+ """Insert the tracked filings of one company, update the XBRL/period flags of the known ones.
	254	+
	255	+ Inserts go through `INSERT … ON CONFLICT DO NOTHING`: with the composite key (cik, accn) a co-registrant's copy
	256	+ of a shared accession is a distinct row; on a database not migrated yet (legacy key on accn alone) the copy is
	257	+ skipped instead of aborting the whole transaction. Returns the number of rows inserted."""
221	258	xbrl_accns = set(facts["accn"].unique()) if not facts.empty else set()
222	259	now = utcnow()
	260	+ t = EdgarFiling.__table__
	261	+ inserted = 0
223	262	with session() as s:
224		− existing = {f.accn: f for f in s.scalars(select(EdgarFiling).where(EdgarFiling.cik == int(cik)))}
	263	+ existing = {r.accn: r for r in s.execute(select(t.c.accn, t.c.is_xbrl, t.c.period_of_report, t.c.parsed_at)
	264	+ .where(t.c.cik == int(cik)))}
	265	+ new_rows: list[dict[str, Any]] = []
225	266	for accn, f in filings.items():
226	267	if f.form not in TRACKED_FORMS and accn not in xbrl_accns:
227	268	continue
228	269	row = existing.get(accn)
229	270	if row is None:
230		− s.add(EdgarFiling(accn=accn, cik=int(cik), form=f.form, filed_date=f.filed, period_of_report=f.report_date,
231		− primary_doc=primary_doc_url(int(cik), accn, f.primary_doc), is_amendment=f.is_amendment,
232		− is_xbrl=f.is_xbrl or accn in xbrl_accns, parsed_at=now if accn in xbrl_accns else None))
233		− else:
234		− row.is_xbrl = row.is_xbrl or accn in xbrl_accns
235		− row.period_of_report = row.period_of_report or f.report_date
236		− if accn in xbrl_accns:
237		− row.parsed_at = now
	271	+ new_rows.append({"cik": int(cik), "accn": accn, "form": f.form, "filed_date": f.filed,
	272	+ "period_of_report": f.report_date, "primary_doc": primary_doc_url(int(cik), accn, f.primary_doc),
	273	+ "is_amendment": bool(f.is_amendment), "is_xbrl": bool(f.is_xbrl or accn in xbrl_accns),
	274	+ "parsed_at": now if accn in xbrl_accns else None})
	275	+ continue
	276	+ values: dict[str, Any] = {}
	277	+ if accn in xbrl_accns:
	278	+ if not row.is_xbrl:
	279	+ values["is_xbrl"] = True
	280	+ values["parsed_at"] = now
	281	+ if row.period_of_report is None and f.report_date is not None:
	282	+ values["period_of_report"] = f.report_date
	283	+ if values:
	284	+ s.execute(update(t).where(t.c.cik == int(cik), t.c.accn == accn).values(**values))
	285	+ for i in range(0, len(new_rows), 500):
	286	+ res = s.execute(sqlite_insert(t).on_conflict_do_nothing(), new_rows[i:i + 500])
	287	+ inserted += max(int(res.rowcount or 0), 0)
	288	+ return inserted
	289	+
	290	+
	291	+_STATEMENT_CONTENT_COLUMNS = tuple(c for c in fund_statements.c.keys() if c not in ("id", "coverage", "coverage_id"))
	292	+
	293	+
	294	+def _statement_fingerprint(rec: dict[str, Any], coverage_sha: str) -> str:
	295	+ """Canonical string of one statement row (dates/bools/floats normalised) + the sha1 of its coverage."""
	296	+ parts = []
	297	+ for c in _STATEMENT_CONTENT_COLUMNS:
	298	+ v = rec.get(c)
	299	+ if isinstance(v, (datetime, date)):
	300	+ v = N._d(v).isoformat() if N._d(v) else None
	301	+ elif isinstance(v, bool):
	302	+ v = int(v)
	303	+ elif isinstance(v, float):
	304	+ v = None if (math.isnan(v) or math.isinf(v)) else repr(float(v))
	305	+ elif isinstance(v, int) and not isinstance(v, bool):
	306	+ v = repr(float(v)) if c not in ("cik", "fiscal_year", "fiscal_quarter") else v
	307	+ parts.append(f"{c}={v}")
	308	+ return "\|".join(parts) + f"\|coverage={coverage_sha}"
	309	+
	310	+
	311	+def _stored_fingerprints(cik: int) -> set[str]:
	312	+ with session() as s:
	313	+ stored = [dict(r._mapping) for r in s.execute(select(fund_statements).where(fund_statements.c.cik == int(cik)))]
	314	+ ids = {int(r["coverage_id"]) for r in stored if r.get("coverage") is None and r.get("coverage_id") is not None}
	315	+ shas = CS.sha_of_ids(ids) if ids else {}
	316	+ out = set()
	317	+ for r in stored:
	318	+ if r.get("coverage") is not None:
	319	+ sha = CS.digest(r["coverage"])
	320	+ else:
	321	+ sha = shas.get(int(r["coverage_id"]), CS.digest({})) if r.get("coverage_id") is not None else CS.digest({})
	322	+ out.add(_statement_fingerprint(r, sha))
	323	+ return out
238	324
239	325
240		−def _replace_statements(cik: int, rows: list[dict[str, Any]]) -> None:
	326	+def _replace_statements(cik: int, rows: list[dict[str, Any]]) -> bool:
	327	+ """Store the standardized rows of one company; coverage documents go through the dictionary.
	328	+
	329	+ The table is rewritten (DELETE + INSERT) only when the content differs from what is stored — a poll cycle
	330	+ that re-normalises an unchanged company then costs one SELECT instead of ~350 deletes/inserts (and no
	331	+ fragmentation). Returns True when rows were written."""
241	332	cols = [c for c in fund_statements.c.keys() if c != "id"]
242		− clean = []
	333	+ clean: list[dict[str, Any]] = []
243	334	for r in rows:
244	335	rec = {k: r.get(k) for k in cols} # executemany needs identical keys on every row
245	336	for k, v in rec.items():
246	337	if isinstance(v, float) and (math.isnan(v) or math.isinf(v)):
247	338	rec[k] = None
248	339	clean.append(rec)
	340	+ shas = [CS.digest(rec.get("coverage")) for rec in clean]
	341	+ fresh = {_statement_fingerprint(rec, sha) for rec, sha in zip(clean, shas)}
	342	+ if fresh == _stored_fingerprints(cik):
	343	+ return False
	344	+ ids = CS.intern([rec.get("coverage") for rec in clean]) if clean else []
	345	+ for rec, cid in zip(clean, ids):
	346	+ rec["coverage_id"] = cid
	347	+ rec["coverage"] = None
249	348	with session() as s:
250	349	s.execute(delete(fund_statements).where(fund_statements.c.cik == int(cik)))
251	350	for i in range(0, len(clean), 500):
252	351	s.execute(insert(fund_statements), clean[i:i + 500])
	352	+ return True
253	353
254	354
255	355	def _upsert_mapping_log(cik: int, entries: list[dict[str, Any]]) -> None:
		`@@ -428,43 +528,132 @@ def publish_filing_event(cik: int, ticker: str, form: str, filed: date, period:`
428	528
429	529
430	530	# ------------------------------------------------------------------------------------ incremental poll
	531	+def _redis():
	532	+ """Sync Redis client shared with the stream broker (fakeredis under HFMD_REDIS_URL=fakeredis://); None when down."""
	533	+ try:
	534	+ from stream.broker import get_redis
	535	+ r = get_redis()
	536	+ r.ping()
	537	+ return r
	538	+ except Exception as e:
	539	+ log.warning("redis unavailable for the poller guard: %s", e)
	540	+ return None
	541	+
	542	+
	543	+def _pair_key(cik: int, accn: str) -> str:
	544	+ return f"{int(cik)}:{accn}"
	545	+
	546	+
	547	+def _redis_seen(r, pairs: set[tuple[int, str]]) -> set[tuple[int, str]]:
	548	+ if r is None or not pairs:
	549	+ return set()
	550	+ items = sorted(pairs)
	551	+ flags = r.smismember(SEEN_KEY, [_pair_key(c, a) for c, a in items])
	552	+ return {p for p, f in zip(items, flags) if f}
	553	+
	554	+
	555	+def _redis_mark_seen(r, pairs: set[tuple[int, str]]) -> None:
	556	+ if r is None or not pairs:
	557	+ return
	558	+ r.sadd(SEEN_KEY, *[_pair_key(c, a) for c, a in pairs])
	559	+ r.expire(SEEN_KEY, SEEN_TTL_S)
	560	+ r.hdel(FAILED_KEY, *[_pair_key(c, a) for c, a in pairs])
	561	+
	562	+
	563	+def _redis_mark_failed(r, pairs: set[tuple[int, str]]) -> set[tuple[int, str]]:
	564	+ """Count one failed attempt per pair; returns the pairs that reached MAX_ATTEMPTS (to be marked seen)."""
	565	+ if r is None or not pairs:
	566	+ return set()
	567	+ exhausted = set()
	568	+ for c, a in pairs:
	569	+ n = int(r.hincrby(FAILED_KEY, _pair_key(c, a), 1))
	570	+ if n >= MAX_ATTEMPTS:
	571	+ exhausted.add((c, a))
	572	+ r.expire(FAILED_KEY, SEEN_TTL_S)
	573	+ return exhausted
	574	+
	575	+
	576	+def business_days_back(today: date, n: int = 2) -> list[date]:
	577	+ """The last `n` weekdays ending today (today included when it is a weekday). EDGAR publishes no index on
	578	+ Saturdays/Sundays — asking for them only produces 403s."""
	579	+ out: list[date] = []
	580	+ d = today
	581	+ while len(out) < n:
	582	+ if d.weekday() < 5:
	583	+ out.append(d)
	584	+ d -= timedelta(days=1)
	585	+ return out
	586	+
	587	+
431	588	def poll_new_filings(client: EdgarClient, *, since: datetime \| None = None, forms: tuple[str, ...] = ("10-K", "10-Q", "8-K", "20-F"),
432	589	include_daily_index: bool = True) -> dict[str, Any]:
433		− """One polling cycle: Atom `getcurrent` per form (+ yesterday/today master index as a safety net) →
434		− filter tracked CIKs → re-ingest them (fresh companyfacts) → publish events. Returns a summary."""
	590	+ """One polling cycle: Atom `getcurrent` per form (+ the last two business-day master indexes as a safety
	591	+ net) → tracked (cik, accn) pairs of tracked forms → minus what edgar_filings and the Redis guard already know
	592	+ → re-ingest the affected companies (fresh companyfacts) → publish events. Returns a summary.
	593	+
	594	+ Termination guards (the poller re-ingested one company every cycle for 41 h before them):
	595	+ * only `TRACKED_FORMS` are considered, so every accepted accession ends up in edgar_filings;
	596	+ * pairs are keyed (cik, accn): co-registrants of one accession are all ingested, once each;
	597	+ * Redis SET `edgar:seen_accn` (7 days) remembers every pair handled — even when the ingestion could not
	598	+ record it (transient EDGAR error: retried `MAX_ATTEMPTS` times, then left to the reconcile job)."""
435	599	tracked = set(tracked_ciks())
436		− seen: dict[str, dict[str, Any]] = {}
	600	+ seen: dict[tuple[int, str], dict[str, Any]] = {}
437	601	for form in forms:
438	602	try:
439	603	for e in parse_atom(client.atom_current(form)):
440		− if e["cik"] in tracked and e["accn"]:
441		− seen[e["accn"]] = e
	604	+ if e["cik"] in tracked and e["accn"] and (e.get("form") or form) in TRACKED_FORMS:
	605	+ seen.setdefault((int(e["cik"]), e["accn"]), e)
442	606	except Exception as ex: # pragma: no cover
443	607	log.warning("atom %s: %s", form, ex)
	608	+ index_days: list[str] = []
444	609	if include_daily_index:
445		− for back in (0, 1):
446		− d = date.today() - timedelta(days=back)
	610	+ for d in business_days_back(date.today(), 2):
447	611	try:
448		− for e in parse_master_index(client.get_text(master_index_url(d))):
449		− if e["cik"] in tracked and e["form"] in TRACKED_FORMS and e["accn"] not in seen:
450		− seen[e["accn"]] = e
451		− except Exception:
	612	+ text = client.daily_index(d)
	613	+ except Exception as ex:
	614	+ log.warning("daily index %s: %s", d, ex)
452	615	continue
453		− known = _all_known_accns()
454		− new = {a: e for a, e in seen.items() if a not in known}
455		− affected = sorted({e["cik"] for e in new.values()})
456		− results = []
	616	+ if text is None:
	617	+ log.debug("daily index %s not published yet", d)
	618	+ continue
	619	+ index_days.append(d.isoformat())
	620	+ for e in parse_master_index(text):
	621	+ if e["cik"] in tracked and e["form"] in TRACKED_FORMS:
	622	+ seen.setdefault((int(e["cik"]), e["accn"]), e)
	623	+ r = _redis()
	624	+ pairs = set(seen)
	625	+ known = _known_pairs(pairs)
	626	+ guarded = _redis_seen(r, pairs - known)
	627	+ new = {p: e for p, e in seen.items() if p not in known and p not in guarded}
	628	+ affected = sorted({c for c, _ in new})
	629	+ results: list[IngestResult] = []
	630	+ handled: set[tuple[int, str]] = set()
	631	+ failed_pairs: set[tuple[int, str]] = set()
457	632	for cik in affected:
458		− results.append(ingest_company(client, cik, refresh=True, with_metalinks=False, publish=True))
	633	+ res = ingest_company(client, cik, refresh=True, with_metalinks=False, publish=True)
	634	+ results.append(res)
	635	+ mine = {p for p in new if p[0] == cik}
	636	+ if res.failed:
	637	+ failed_pairs \|= mine
	638	+ else:
	639	+ handled \|= mine
	640	+ exhausted = _redis_mark_failed(r, failed_pairs)
	641	+ if exhausted:
	642	+ log.warning("giving up on %d accession(s) after %d attempts: %s", len(exhausted), MAX_ATTEMPTS, sorted(exhausted)[:5])
	643	+ _redis_mark_seen(r, handled \| exhausted)
459	644	now = utcnow()
460	645	newest = max((datetime.fromisoformat(e["filed"]) for e in seen.values() if e.get("filed")), default=None)
461	646	lag = (now - newest).total_seconds() if newest else None
	647	+ failures = [r_.error for r_ in results if r_.failed]
462	648	_set_state("incremental", last_run_at=now, last_success_at=now, last_rss_check_at=now, lag_seconds=lag,
463	649	requests_made=client.stats.requests, last_filing_seen=newest,
464		− failures_add=sum(1 for r in results if r.error), failure_samples=[r.error for r in results if r.error][:5])
465		− _update_mapping_failure_rate()
466		− return {"seen": len(seen), "new": len(new), "affected_ciks": affected,
467		− "events": sum(len(r.new_filings) for r in results), "errors": [r.error for r in results if r.error]}
	650	+ failures_add=len(failures), failure_samples_add=[f"cik={r_.cik} {r_.error}" for r_ in results if r_.failed])
	651	+ if any(r_.changed for r_ in results):
	652	+ _update_mapping_failure_rate()
	653	+ return {"seen": len(seen), "new": len(new), "affected_ciks": affected, "index_days": index_days,
	654	+ "events": sum(len(r_.new_filings) for r_ in results), "errors": [r_.error for r_ in results if r_.error],
	655	+ "failures": failures, "rows_changed": sum(1 for r_ in results if r_.changed),
	656	+ "no_facts": [r_.cik for r_ in results if r_.error == NO_FACTS]}
468	657
469	658
470	659	def master_index_url(d: date) -> str:
		`@@ -489,13 +678,10 @@ def parse_master_index(text: str) -> list[dict[str, Any]]:`
489	678	return out
490	679
491	680
492		−def _all_known_accns() -> set[str]:
493		− with session() as s:
494		− return set(s.scalars(select(EdgarFiling.accn)))
495		−
496		−
497	681	# ------------------------------------------------------------------------------------------- state
498		−def _set_state(key: str, , failures_add: int = 0, *fields: Any) -> None:
	682	+def _set_state(key: str, , failures_add: int = 0, failure_samples_add: list[str] \| None = None, *fields: Any) -> None:
	683	+ """Update one job row. `failure_samples_add` appends (most recent last, capped at FAILURE_SAMPLES_MAX) so a
	684	+ later successful cycle never wipes the evidence of earlier failures; `failure_samples=` still replaces."""
499	685	with session() as s:
500	686	st = s.get(FundIngestState, key)
501	687	if st is None:
		`@@ -504,6 +690,9 @@ def _set_state(key: str, , failures_add: int = 0, *fields: Any) -> None:`
504	690	for k, v in fields.items():
505	691	setattr(st, k, v)
506	692	st.failures = (st.failures or 0) + failures_add
	693	+ if failure_samples_add:
	694	+ samples = list(st.failure_samples or []) + [str(x)[:2000] for x in failure_samples_add]
	695	+ st.failure_samples = samples[-FAILURE_SAMPLES_MAX:]
507	696
508	697
509	698	def _bump_state(key: str, **counters: int) -> None:
		`@@ -537,4 +726,5 @@ def state_snapshot() -> dict[str, Any]:`
537	726
538	727
539	728	__all__ = ["sync_universe", "ingest_company", "refresh_latest", "build_latest_all", "poll_new_filings",
540		− "publish_filing_event", "compute_coverage", "tracked_ciks", "parse_master_index", "master_index_url"]
	729	+ "publish_filing_event", "compute_coverage", "tracked_ciks", "parse_master_index", "master_index_url",
	730	+ "business_days_back", "IngestResult", "NO_FACTS", "SEEN_KEY", "FAILED_KEY"]

added hfmarketdata/api/fundamentals/migrations.py +151 −0

		`@@ -0,0 +1,151 @@`
	1	+"""Idempotent schema maintenance for the fundamentals tables (SQLite cannot ALTER a primary key).
	2	+
	3	+Called by `models.init_db()` at module start and by the scripts; every step is a no-op once applied:
	4	+
	5	+* `edgar_filings` key `accn` → `(cik, accn)` (co-registrants share accession numbers). The table is rebuilt
	6	+ (`edgar_filings_new` ← `INSERT OR IGNORE`, drop, rename, indexes). Small tables (≤ `INLINE_MAX_ROWS`) are
	7	+ migrated inline at startup; bigger ones (production: 1.37 M rows, ≈ 30–60 s) are left alone with a warning
	8	+ and `scripts/migrate_edgar_filings.py` must be run before the deployment — the ingestion code tolerates
	9	+ the old shape meanwhile (`INSERT … ON CONFLICT DO NOTHING`).
	10	+* `fund_statements.coverage_id` column (dictionary reference, see `coverage_store.py`).
	11	+* `PRAGMA auto_vacuum=INCREMENTAL` when the database is brand new (must be set before the first table).
	12	+
	13	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	14	+"""
	15	+from __future__ import annotations
	16	+
	17	+import logging
	18	+import time
	19	+
	20	+from sqlalchemy import text
	21	+from sqlalchemy.engine import Connection
	22	+
	23	+from core.db import engine
	24	+
	25	+log = logging.getLogger("hfmarketdata.fundamentals.migrations")
	26	+
	27	+INLINE_MAX_ROWS = 300_000
	28	+
	29	+_FILINGS_COLUMNS = ("cik", "accn", "form", "filed_date", "period_of_report", "primary_doc", "is_amendment", "is_xbrl", "parsed_at")
	30	+_FILINGS_DDL = """
	31	+CREATE TABLE {name} (
	32	+ cik INTEGER NOT NULL,
	33	+ accn VARCHAR(24) NOT NULL,
	34	+ form VARCHAR(16) NOT NULL,
	35	+ filed_date DATE NOT NULL,
	36	+ period_of_report DATE,
	37	+ primary_doc VARCHAR(512),
	38	+ is_amendment BOOLEAN NOT NULL,
	39	+ is_xbrl BOOLEAN NOT NULL,
	40	+ parsed_at DATETIME,
	41	+ PRIMARY KEY (cik, accn)
	42	+)"""
	43	+_FILINGS_INDEXES = (
	44	+ "CREATE INDEX IF NOT EXISTS ix_edgar_filings_cik_filed ON edgar_filings (cik, filed_date)",
	45	+ "CREATE INDEX IF NOT EXISTS ix_edgar_filings_accn ON edgar_filings (accn)",
	46	+ "CREATE INDEX IF NOT EXISTS ix_edgar_filings_form ON edgar_filings (form)",
	47	+ "CREATE INDEX IF NOT EXISTS ix_edgar_filings_filed_date ON edgar_filings (filed_date)",
	48	+)
	49	+
	50	+
	51	+def _table_exists(con: Connection, name: str) -> bool:
	52	+ return con.execute(text("SELECT 1 FROM sqlite_master WHERE type='table' AND name=:n"), {"n": name}).first() is not None
	53	+
	54	+
	55	+def _columns(con: Connection, table: str) -> list[dict]:
	56	+ return [dict(r._mapping) for r in con.execute(text(f"PRAGMA table_info({table})"))]
	57	+
	58	+
	59	+def filings_pk_is_composite(con: Connection) -> bool \| None:
	60	+ """True when edgar_filings is keyed on (cik, accn); False for the legacy `accn` key; None when absent."""
	61	+ if not _table_exists(con, "edgar_filings"):
	62	+ return None
	63	+ pk = sorted((c["pk"], c["name"]) for c in _columns(con, "edgar_filings") if c["pk"])
	64	+ return [n for _, n in pk] == ["cik", "accn"]
	65	+
	66	+
	67	+def filings_row_count(con: Connection) -> int:
	68	+ return int(con.execute(text("SELECT count(*) FROM edgar_filings")).scalar() or 0)
	69	+
	70	+
	71	+def migrate_filings_pk(con: Connection, *, log_progress: bool = True) -> int:
	72	+ """Rebuild edgar_filings with the composite key inside one transaction. Returns the rows copied.
	73	+
	74	+ `INSERT OR IGNORE` makes the copy safe to redo (the table is fully reconstructible by ingestion anyway)."""
	75	+ t0 = time.time()
	76	+ con.execute(text("DROP TABLE IF EXISTS edgar_filings_new"))
	77	+ con.execute(text(_FILINGS_DDL.format(name="edgar_filings_new")))
	78	+ cols = ", ".join(_FILINGS_COLUMNS)
	79	+ con.execute(text(f"INSERT OR IGNORE INTO edgar_filings_new ({cols}) SELECT {cols} FROM edgar_filings"))
	80	+ n = int(con.execute(text("SELECT count(*) FROM edgar_filings_new")).scalar() or 0)
	81	+ con.execute(text("DROP TABLE edgar_filings"))
	82	+ con.execute(text("ALTER TABLE edgar_filings_new RENAME TO edgar_filings"))
	83	+ for ddl in _FILINGS_INDEXES:
	84	+ con.execute(text(ddl))
	85	+ if log_progress:
	86	+ log.info("edgar_filings rebuilt with key (cik, accn): %d rows in %.1fs", n, time.time() - t0)
	87	+ return n
	88	+
	89	+
	90	+def ensure_filings_schema(*, inline_max_rows: int = INLINE_MAX_ROWS) -> str:
	91	+ """Startup hook: migrate small tables inline, warn for big ones. Returns 'ok' \| 'migrated' \| 'pending' \| 'absent'."""
	92	+ with engine.begin() as con:
	93	+ state = filings_pk_is_composite(con)
	94	+ if state is None:
	95	+ return "absent"
	96	+ if state:
	97	+ return "ok"
	98	+ n = filings_row_count(con)
	99	+ if n > inline_max_rows:
	100	+ log.warning("edgar_filings still keyed on accn alone (%d rows): run scripts/migrate_edgar_filings.py before "
	101	+ "deploying — co-registrant filings are skipped until then", n)
	102	+ return "pending"
	103	+ migrate_filings_pk(con)
	104	+ return "migrated"
	105	+
	106	+
	107	+def ensure_statements_schema() -> bool:
	108	+ """Add `fund_statements.coverage_id` when missing (legacy databases). Returns True when added."""
	109	+ with engine.begin() as con:
	110	+ if not _table_exists(con, "fund_statements"):
	111	+ return False
	112	+ names = {c["name"] for c in _columns(con, "fund_statements")}
	113	+ if "coverage_id" in names:
	114	+ return False
	115	+ con.execute(text("ALTER TABLE fund_statements ADD COLUMN coverage_id INTEGER"))
	116	+ log.info("fund_statements: column coverage_id added")
	117	+ return True
	118	+
	119	+
	120	+AUTO_VACUUM_MAX_PAGES = 12_800 # ≈ 50 MB at 4 KB/page: switching mode needs a VACUUM, only cheap on a young DB
	121	+
	122	+
	123	+def ensure_auto_vacuum(*, max_pages: int = AUTO_VACUUM_MAX_PAGES) -> bool:
	124	+ """Switch a young database (≤ `max_pages`) to `auto_vacuum=INCREMENTAL` (the mode only takes effect through a
	125	+ VACUUM, instantaneous on a small file). Big existing files are left alone: a one-off offline `VACUUM`
	126	+ (`scripts/migrate_coverage.py --vacuum`) is the operator's call. Returns True when the mode was switched."""
	127	+ with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as con:
	128	+ mode = int(con.execute(text("PRAGMA auto_vacuum")).scalar() or 0)
	129	+ if mode == 2:
	130	+ return False
	131	+ pages = int(con.execute(text("PRAGMA page_count")).scalar() or 0)
	132	+ if pages > max_pages:
	133	+ return False
	134	+ con.execute(text("PRAGMA auto_vacuum=INCREMENTAL"))
	135	+ con.execute(text("VACUUM"))
	136	+ log.info("sqlite auto_vacuum=INCREMENTAL enabled (%d pages)", pages)
	137	+ return True
	138	+
	139	+
	140	+def incremental_vacuum(pages: int \| None = None) -> int:
	141	+ """Release free pages when auto_vacuum=INCREMENTAL (no-op otherwise). Returns the freelist count after."""
	142	+ with engine.connect() as con:
	143	+ mode = int(con.execute(text("PRAGMA auto_vacuum")).scalar() or 0)
	144	+ if mode == 2:
	145	+ con.execute(text(f"PRAGMA incremental_vacuum{f'({int(pages)})' if pages else ''}"))
	146	+ con.commit()
	147	+ return int(con.execute(text("PRAGMA freelist_count")).scalar() or 0)
	148	+
	149	+
	150	+__all__ = ["ensure_filings_schema", "ensure_statements_schema", "ensure_auto_vacuum", "migrate_filings_pk",
	151	+ "filings_pk_is_composite", "filings_row_count", "incremental_vacuum", "INLINE_MAX_ROWS"]

modified hfmarketdata/api/fundamentals/models.py +34 −7

		@@ -4,9 +4,10 @@ Raw XBRL facts are NOT here: they live in the Parquet lake `data_root/edgar/fact
4	4	(queried with DuckDB). SQLite holds what needs indexes and point-in-time lookups:
5	5
6	6	* `edgar_companies` CIK ↔ ticker(s) (share classes, history), SIC, exchange, fiscal year end, status
7		−* `edgar_filings` one row per filing (accession number), amendments flagged
	7	+* `edgar_filings` one row per (cik, accession number) — co-registrants share accessions —, amendments flagged
8	8	* `fund_statements` wide standardized statements, versioned by filed_date (point-in-time index on
9		− (ticker, period_end, filed_date)); `coverage` JSON explains every null
	9	+ (ticker, period_end, filed_date)); `coverage_id` → `fund_coverage_blob` explains every null
	10	+* `fund_coverage_blob` dictionary of distinct coverage JSON documents (sha1-keyed)
10	11	* `fund_mapping` the prioritized tag mapping, versioned (seeded from mapping.py)
11	12	* `fund_mapping_log` unmapped / custom-extension tags seen per CIK (never guessed)
12	13	* `fund_coverage` per ticker: periods, completeness %, gaps
		`@@ -17,6 +18,7 @@ Author: Simon-Pierre Boucher <contact@spboucher.ai>`
17	18	"""
18	19	from __future__ import annotations
19	20
	21	+import logging
20	22	from datetime import date, datetime
21	23
22	24	from sqlalchemy import (JSON, Boolean, Column, Date, DateTime, Float, Index, Integer, String, Table, Text,
		`@@ -49,9 +51,14 @@ class EdgarCompany(Base):`
49	51
50	52
51	53	class EdgarFiling(Base):
	54	+ """One row per (co-)registrant and accession number.
	55	+
	56	+ An accession number is NOT unique across companies: a parent and its subsidiaries file one combined 10-Q/8-K
	57	+ (Entergy 65984 + 66901/65770/…, Southern 92122…) that appears in the submissions of every co-registrant with the
	58	+ same `accn`. The key is therefore the pair (cik, accn); `accn` alone is indexed."""
52	59	__tablename__ = "edgar_filings"
	60	+ cik: Mapped[int] = mapped_column(Integer, primary_key=True)
53	61	accn: Mapped[str] = mapped_column(String(24), primary_key=True) # 0000320193-24-000069
54		− cik: Mapped[int] = mapped_column(Integer, index=True)
55	62	form: Mapped[str] = mapped_column(String(16), index=True) # 10-K, 10-Q, 8-K, 20-F, 10-K/A…
56	63	filed_date: Mapped[date] = mapped_column(Date, index=True)
57	64	period_of_report: Mapped[date \| None] = mapped_column(Date, nullable=True)
		`@@ -59,7 +66,18 @@ class EdgarFiling(Base):`
59	66	is_amendment: Mapped[bool] = mapped_column(Boolean, default=False)
60	67	is_xbrl: Mapped[bool] = mapped_column(Boolean, default=False)
61	68	parsed_at: Mapped[datetime \| None] = mapped_column(DateTime, nullable=True)
62		− __table_args__ = (Index("ix_edgar_filings_cik_filed", "cik", "filed_date"),)
	69	+ __table_args__ = (Index("ix_edgar_filings_cik_filed", "cik", "filed_date"),
	70	+ Index("ix_edgar_filings_accn", "accn"))
	71	+
	72	+
	73	+class FundCoverageBlob(Base):
	74	+ """Dictionary of distinct `coverage` JSON documents (the null reasons repeat a lot: ~1.2 KB × 1.2 M rows
	75	+ collapsed to a few tens of thousands of distinct blobs). `fund_statements.coverage_id` references it; a
	76	+ row keeps `coverage` inline (legacy) until `scripts/migrate_coverage.py` converts it."""
	77	+ __tablename__ = "fund_coverage_blob"
	78	+ id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
	79	+ sha1: Mapped[str] = mapped_column(String(40), unique=True, nullable=False)
	80	+ json: Mapped[str] = mapped_column(Text, nullable=False)
63	81
64	82
65	83	class FundMapping(Base):
		`@@ -135,7 +153,7 @@ class FundIngestState(Base):`
135	153	# ------------------------------------------------------------------------------------------- wide tables
136	154	STATEMENT_KEY_COLUMNS = ("cik", "ticker", "statement", "fiscal_year", "fiscal_quarter", "period_start", "period_end",
137	155	"calendar_quarter", "form", "accn", "filed_date", "derived", "restated", "currency", "coverage",
138		− "mapping_version")
	156	+ "mapping_version", "coverage_id")
139	157
140	158	fund_statements = Table(
141	159	"fund_statements", Base.metadata,
		`@@ -154,8 +172,9 @@ fund_statements = Table(`
154	172	Column("derived", Boolean, default=False), # Q4 / de-cumulated quarter
155	173	Column("restated", Boolean, default=False), # differs from an earlier version of the same period
156	174	Column("currency", String(3), default="USD"),
157		− Column("coverage", JSON, default=dict), # {account: {"reason": …, "tag": …, "computed": …}}
	175	+ Column("coverage", JSON, nullable=True), # legacy inline {account: {"reason": …, …}} — NULL once interned
158	176	Column("mapping_version", String(16)),
	177	+ Column("coverage_id", Integer, nullable=True), # → fund_coverage_blob.id (see coverage_store.py)
159	178	*[Column(name, Float, nullable=True) for name in ALL_ACCOUNT_NAMES],
160	179	UniqueConstraint("cik", "statement", "fiscal_year", "fiscal_quarter", "accn", name="uq_fund_statements_version"),
161	180	Index("ix_fund_statements_pit", "ticker", "period_end", "filed_date"),
		`@@ -192,8 +211,16 @@ SCREENER_TEXT_FIELDS: tuple[str, ...] = ("ticker", "sic", "exchange", "name")`
192	211
193	212
194	213	def init_db() -> None:
195		− """Idempotent create_all + mapping seed (called at module import by routes.py and by the scripts)."""
	214	+ """Idempotent create_all + schema maintenance + mapping seed (called at module import by routes.py and by
	215	+ the scripts). Heavy migrations are never run here: see `migrations.ensure_filings_schema`."""
	216	+ from . import migrations
	217	+ try:
	218	+ migrations.ensure_auto_vacuum()
	219	+ except Exception as e: # pragma: no cover — never block the API on a housekeeping pragma
	220	+ logging.getLogger("hfmarketdata.fundamentals").warning("auto_vacuum: %s", e)
196	221	create_all()
	222	+ migrations.ensure_filings_schema()
	223	+ migrations.ensure_statements_schema()
197	224	from .mapping import MAPPING_VERSION, mapping_rows
198	225	from core.db import session
199	226	from sqlalchemy import select

modified hfmarketdata/api/fundamentals/service.py +4 −1

		`@@ -74,11 +74,14 @@ def resolve_company(ticker: str) -> EdgarCompany:`
74	74
75	75
76	76	def _rows(cik: int, statements: list[str] \| None = None) -> list[dict[str, Any]]:
	77	+ """Every stored version of the company's statements, `coverage` resolved (inline legacy JSON or dictionary)."""
77	78	q = select(fund_statements).where(fund_statements.c.cik == cik)
78	79	if statements:
79	80	q = q.where(fund_statements.c.statement.in_(statements))
80	81	with session() as s:
81		− return [dict(r._mapping) for r in s.execute(q)]
	82	+ rows = [dict(r._mapping) for r in s.execute(q)]
	83	+ from . import coverage_store
	84	+ return coverage_store.hydrate(rows)
82	85
83	86
84	87	def company_rows(cik: int, as_of: date \| None, statements: list[str] \| None = None,

modified scripts/edgar_backfill.py +1 −1

		`@@ -119,7 +119,7 @@ def main() -> int:`
119	119	except Exception as e: # pragma: no cover
120	120	r = None
121	121	log.exception("cik %s crashed: %s", cik, e)
122		− if r is not None and r.error == "companyfacts_404":
	122	+ if r is not None and r.error == ingest.NO_FACTS:
123	123	# no XBRL facts on EDGAR (funds, trusts, paper filers): permanent — not retried on resume
124	124	done[str(cik)] = {"status": "no_facts", "ticker": r.ticker, "at": utcnow().isoformat()}
125	125	elif r is None or r.error:

modified scripts/edgar_incremental.py +16 −9

		`@@ -7,11 +7,14 @@`
7	7	venv/bin/python scripts/edgar_incremental.py --once
8	8
9	9	Sources (see fundamentals.ingest.poll_new_filings): the EDGAR Atom feed `browse-edgar?action=getcurrent` for
10		−10-K / 10-Q / 8-K / 20-F (live, ~4 requests per cycle) + the daily master index of today/yesterday as a safety
11		−net. For every new accession of a tracked CIK: companyfacts + submissions are refetched (cache bypass),
12		−statements re-normalised (new versions, restatements), coverage + screener row refreshed, and a `filing`
13		−event is published on Redis (`filings` channel + `filings:stream` buffer) for the WebSocket. Lag, failures and
14		−the mapping failure rate are written to `fund_ingest_state` (GET /v1/fundamentals/_health).
	10	+10-K / 10-Q / 8-K / 20-F (live, ~4 requests per cycle) + the daily master index of the last two business days
	11	+as a safety net (weekend indexes do not exist: 403 from the SEC, skipped silently). For every new
	12	+(cik, accession) pair of a tracked form: companyfacts + submissions are refetched (cache bypass), statements
	13	+re-normalised (new versions, restatements) and rewritten only when they changed, coverage + screener row
	14	+refreshed, and a `filing` event is published on Redis (`filings` channel + `filings:stream` buffer) for the
	15	+WebSocket. A Redis SET `edgar:seen_accn` (7 days) guarantees a pair is handled once even when EDGAR misbehaves.
	16	+Lag, failures (+ samples) and the mapping failure rate are written to `fund_ingest_state`
	17	+(GET /v1/fundamentals/_health).
15	18
16	19	Author: Simon-Pierre Boucher <contact@spboucher.ai>
17	20	"""
		`@@ -46,6 +49,8 @@ def main() -> int:`
46	49	ap.add_argument("-v", "--verbose", action="store_true")
47	50	args = ap.parse_args()
48	51	logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
	52	+ if not args.verbose:
	53	+ logging.getLogger("httpx").setLevel(logging.WARNING) # one line per request otherwise (incl. expected 403/404)
49	54	signal.signal(signal.SIGTERM, _sig)
50	55	signal.signal(signal.SIGINT, _sig)
51	56
		`@@ -60,9 +65,10 @@ def main() -> int:`
60	65	t0 = time.time()
61	66	try:
62	67	summary = ingest.poll_new_filings(client, forms=forms, include_daily_index=not args.no_daily_index)
63		− log.info("cycle: seen=%s new=%s affected=%s events=%s errors=%s (%.1fs)", summary["seen"], summary["new"],
64		− summary["affected_ciks"], summary["events"], summary["errors"], time.time() - t0)
65		− if summary["affected_ciks"]:
	68	+ log.info("cycle: seen=%s new=%s affected=%s changed=%s events=%s errors=%s index=%s (%.1fs)", summary["seen"],
	69	+ summary["new"], summary["affected_ciks"], summary["rows_changed"], summary["events"], summary["errors"],
	70	+ summary["index_days"], time.time() - t0)
	71	+ if summary["rows_changed"]: # nothing new stored → the extracts are still fresh
66	72	try:
67	73	from bulk.build import available_years, build_year
68	74	for y in available_years()[-2:]: # keep the two most recent yearly extracts fresh
		`@@ -71,7 +77,8 @@ def main() -> int:`
71	77	log.warning("bulk rebuild failed: %s", e)
72	78	except Exception as e:
73	79	log.exception("cycle failed: %s", e)
74		− ingest._set_state("incremental", last_run_at=utcnow(), failures_add=1, failure_samples=[str(e)])
	80	+ ingest._set_state("incremental", last_run_at=utcnow(), failures_add=1,
	81	+ failure_samples_add=[f"cycle: {type(e).__name__}: {e}"])
75	82	if args.once:
76	83	break
77	84	for _ in range(int(max(1.0, args.interval - (time.time() - t0)))):

modified scripts/edgar_reconcile.py +5 −0

		`@@ -123,6 +123,11 @@ def main() -> int:`
123	123	"discrepancies": sum(len(c["discrepancies"]) for c in report["companies"])})
124	124	if args.json:
125	125	Path(args.json).write_text(json.dumps(report, default=str, indent=1))
	126	+ try: # housekeeping: hand free pages back to the OS when the database runs auto_vacuum=INCREMENTAL (no-op otherwise)
	127	+ from fundamentals.migrations import incremental_vacuum
	128	+ log.info("sqlite incremental_vacuum done, freelist pages left: %d", incremental_vacuum())
	129	+ except Exception as e: # pragma: no cover
	130	+ log.warning("incremental_vacuum failed: %s", e)
126	131	log.info("reconcile done: %d/%d companies with discrepancies (%s)", bad, len(sample), client.stats)
127	132	return 1 if bad else 0
128	133

added scripts/migrate_coverage.py +108 −0

		`@@ -0,0 +1,108 @@`
	1	+#!/usr/bin/env python3
	2	+"""Convert the inline `fund_statements.coverage` JSON of existing rows to `fund_coverage_blob` references.
	3	+
	4	+ cd ~/hfmarketdata && HFMD_DATA_ROOT=~/firstratedata venv/bin/python scripts/migrate_coverage.py # all rows
	5	+ venv/bin/python scripts/migrate_coverage.py --batch 5000 --max-batches 50 # a slice
	6	+ venv/bin/python scripts/migrate_coverage.py --vacuum # reclaim space (offline!)
	7	+
	8	+Runs while the API serves (WAL, short transactions of `--batch` rows; each batch commits, so it can be stopped
	9	+and resumed at any time: the work set is simply `WHERE coverage IS NOT NULL`). New rows are written through the
	10	+dictionary already (`ingest._replace_statements`), the read side understands both shapes, so this only shrinks the
	11	+past: 1.22 M rows × ~1.1 KB ≈ 1.3 GB of JSON → a few tens of thousands of blobs.
	12	+
	13	+Space is NOT returned to the OS by the conversion itself (free pages are reused by later inserts). `--vacuum`
	14	+runs `PRAGMA auto_vacuum=INCREMENTAL` + `VACUUM` afterwards: needs ~2× the file size of free disk, exclusive access
	15	+(stop the API and the jobs) and a few minutes for 3 GB — run it once, off-hours.
	16	+
	17	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	18	+"""
	19	+from __future__ import annotations
	20	+
	21	+import argparse
	22	+import json
	23	+import logging
	24	+import sys
	25	+import time
	26	+from pathlib import Path
	27	+
	28	+HERE = Path(__file__).resolve().parent
	29	+sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))
	30	+
	31	+log = logging.getLogger("migrate_coverage")
	32	+
	33	+
	34	+def convert(batch: int, max_batches: int \| None, sleep_s: float) -> tuple[int, int]:
	35	+ """Convert rows in id order; returns (rows converted, batches)."""
	36	+ from sqlalchemy import text
	37	+
	38	+ from core.db import engine
	39	+ from fundamentals import coverage_store as CS
	40	+ from fundamentals.models import init_db
	41	+
	42	+ init_db() # makes sure coverage_id / fund_coverage_blob exist
	43	+ done = batches = 0
	44	+ last_id = 0
	45	+ t0 = time.time()
	46	+ while True:
	47	+ with engine.begin() as con:
	48	+ rows = con.execute(text("SELECT id, coverage FROM fund_statements WHERE id > :last AND coverage IS NOT NULL "
	49	+ "ORDER BY id LIMIT :n"), {"last": last_id, "n": batch}).all()
	50	+ if not rows:
	51	+ break
	52	+ docs = []
	53	+ for _, cov in rows:
	54	+ if isinstance(cov, (bytes, bytearray)):
	55	+ cov = cov.decode()
	56	+ docs.append(json.loads(cov) if isinstance(cov, str) else (cov or {}))
	57	+ ids = CS.intern(docs)
	58	+ con.execute(text("UPDATE fund_statements SET coverage = NULL, coverage_id = :cid WHERE id = :id"),
	59	+ [{"cid": cid, "id": rid} for (rid, _), cid in zip(rows, ids)])
	60	+ last_id = int(rows[-1][0])
	61	+ done += len(rows)
	62	+ batches += 1
	63	+ if batches % 20 == 0:
	64	+ log.info("converted %d rows (last id %d) in %.0fs", done, last_id, time.time() - t0)
	65	+ if max_batches is not None and batches >= max_batches:
	66	+ break
	67	+ if sleep_s:
	68	+ time.sleep(sleep_s)
	69	+ return done, batches
	70	+
	71	+
	72	+def main() -> int:
	73	+ ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
	74	+ ap.add_argument("--batch", type=int, default=2000, help="rows per transaction (default 2000)")
	75	+ ap.add_argument("--max-batches", type=int, help="stop after N batches (resume later)")
	76	+ ap.add_argument("--sleep", type=float, default=0.0, help="pause between batches (seconds) to stay gentle with the API")
	77	+ ap.add_argument("--vacuum", action="store_true", help="after conversion: auto_vacuum=INCREMENTAL + VACUUM (exclusive, slow)")
	78	+ ap.add_argument("-v", "--verbose", action="store_true")
	79	+ args = ap.parse_args()
	80	+ logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
	81	+
	82	+ from sqlalchemy import text
	83	+
	84	+ from core.config import settings
	85	+ from core.db import engine
	86	+
	87	+ log.info("database: %s", settings.state_db)
	88	+ t0 = time.time()
	89	+ done, batches = convert(args.batch, args.max_batches, args.sleep)
	90	+ with engine.connect() as con:
	91	+ left = con.execute(text("SELECT count(*) FROM fund_statements WHERE coverage IS NOT NULL")).scalar()
	92	+ blobs = con.execute(text("SELECT count(*) FROM fund_coverage_blob")).scalar()
	93	+ log.info("converted %d rows in %d batches (%.0fs); rows still inline: %d; distinct blobs: %d", done, batches,
	94	+ time.time() - t0, left, blobs)
	95	+ if args.vacuum:
	96	+ t1 = time.time()
	97	+ with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as con:
	98	+ before = con.execute(text("PRAGMA page_count")).scalar()
	99	+ con.execute(text("PRAGMA auto_vacuum=INCREMENTAL"))
	100	+ con.execute(text("VACUUM"))
	101	+ after = con.execute(text("PRAGMA page_count")).scalar()
	102	+ mode = con.execute(text("PRAGMA auto_vacuum")).scalar()
	103	+ log.info("VACUUM: %s → %s pages (auto_vacuum=%s) in %.0fs", before, after, mode, time.time() - t1)
	104	+ return 0 if left == 0 or args.max_batches else 1
	105	+
	106	+
	107	+if __name__ == "__main__":
	108	+ sys.exit(main())

added scripts/migrate_edgar_filings.py +71 −0

		`@@ -0,0 +1,71 @@`
	1	+#!/usr/bin/env python3
	2	+"""One-off migration: `edgar_filings` primary key `accn` → `(cik, accn)` (co-registrants share accessions).
	3	+
	4	+ # production (M3U96b) — BEFORE deploying the code that expects the composite key; stop the writers first:
	5	+ pm2 stop hfmarketdata-edgar-incremental hfmarketdata-edgar-backfill
	6	+ cd ~/hfmarketdata && HFMD_DATA_ROOT=~/firstratedata venv/bin/python scripts/migrate_edgar_filings.py
	7	+ pm2 start hfmarketdata-edgar-incremental
	8	+
	9	+The API may keep serving during the copy (WAL: readers are not blocked; a write to edgar_filings would wait on
	10	+the busy timeout, hence stopping the ingestion jobs). The table is rebuilt in ONE transaction
	11	+(`edgar_filings_new` ← INSERT OR IGNORE, drop, rename, 4 indexes) — on failure nothing changes. Idempotent:
	12	+running it on a migrated database prints the state and exits 0. `--check` only reports.
	13	+
	14	+Expected duration: 1.37 M rows ≈ 212 MB → copy ~5–10 s + indexes ~10–20 s on the M3 Ultra SSD (≈ 30–60 s
	15	+worst case); the WAL grows by ~300 MB and is checkpointed at the end.
	16	+
	17	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	18	+"""
	19	+from __future__ import annotations
	20	+
	21	+import argparse
	22	+import logging
	23	+import sys
	24	+import time
	25	+from pathlib import Path
	26	+
	27	+HERE = Path(__file__).resolve().parent
	28	+sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))
	29	+
	30	+log = logging.getLogger("migrate_edgar_filings")
	31	+
	32	+
	33	+def main() -> int:
	34	+ ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
	35	+ ap.add_argument("--check", action="store_true", help="report the current key shape and exit")
	36	+ ap.add_argument("-v", "--verbose", action="store_true")
	37	+ args = ap.parse_args()
	38	+ logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
	39	+
	40	+ from sqlalchemy import text
	41	+
	42	+ from core.config import settings
	43	+ from core.db import engine
	44	+ from fundamentals import migrations
	45	+
	46	+ log.info("database: %s", settings.state_db)
	47	+ with engine.connect() as con:
	48	+ state = migrations.filings_pk_is_composite(con)
	49	+ n = migrations.filings_row_count(con) if state is not None else 0
	50	+ if state is None:
	51	+ log.info("edgar_filings does not exist yet — nothing to migrate (init_db will create it with the composite key)")
	52	+ return 0
	53	+ if state:
	54	+ log.info("edgar_filings already keyed on (cik, accn) — %d rows, nothing to do", n)
	55	+ return 0
	56	+ log.info("edgar_filings keyed on accn alone: %d rows to copy", n)
	57	+ if args.check:
	58	+ return 2
	59	+ t0 = time.time()
	60	+ with engine.begin() as con:
	61	+ con.execute(text("PRAGMA busy_timeout=600000"))
	62	+ copied = migrations.migrate_filings_pk(con)
	63	+ with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as con:
	64	+ con.execute(text("PRAGMA wal_checkpoint(TRUNCATE)"))
	65	+ ok = migrations.filings_pk_is_composite(con)
	66	+ log.info("done: %d rows copied (%d dropped as exact duplicates), composite key: %s, %.1fs", copied, n - copied, ok, time.time() - t0)
	67	+ return 0 if ok else 1
	68	+
	69	+
	70	+if __name__ == "__main__":
	71	+ sys.exit(main())

added tests/test_fundamentals_ingest.py +334 −0

		`@@ -0,0 +1,334 @@`
	1	+"""Ingestion guards: co-registrant filings (shared accessions), poller termination (form filter, Redis seen-set,
	2	+bounded retries), failure samples, conditional statement rewrite, coverage dictionary, schema migrations."""
	3	+from __future__ import annotations
	4	+
	5	+import types
	6	+from datetime import date
	7	+
	8	+import httpx
	9	+import pytest
	10	+from sqlalchemy import delete, select
	11	+
	12	+PARENT, SUB = 65984, 66901 # Entergy Corp + Entergy Arkansas: one 10-Q/8-K, two registrants
	13	+SHARED = "0000065984-26-000283"
	14	+OTHER = "0000065984-26-000290"
	15	+
	16	+
	17	+def _atom(entries: list[tuple[str, str, int, str]]) -> str:
	18	+ items = []
	19	+ for form, name, cik, accn in entries:
	20	+ nodash = accn.replace("-", "")
	21	+ items.append(f"""<entry>
	22	+<title>{form} - {name} ({cik:010d}) (Filer)</title>
	23	+<link rel="alternate" type="text/html" href="https://www.sec.gov/Archives/edgar/data/{cik}/{nodash}/{accn}-index.htm"/>
	24	+<updated>2026-09-04T17:27:57-04:00</updated>
	25	+<category scheme="https://www.sec.gov/" label="form type" term="{form}"/>
	26	+<id>urn:tag:sec.gov,2008:accession-number={accn}</id>
	27	+</entry>""")
	28	+ return ('<?xml version="1.0" encoding="ISO-8859-1" ?>\n<feed xmlns="http://www.w3.org/2005/Atom">\n<title>Latest Filings</title>\n'
	29	+ + "\n".join(items) + "\n</feed>")
	30	+
	31	+
	32	+class FakeClient:
	33	+ """Minimal EdgarClient stand-in for poll_new_filings: canned Atom feed, optional daily index."""
	34	+
	35	+ def __init__(self, atom: str, index: str \| None = None):
	36	+ self.atom, self.index = atom, index
	37	+ self.stats = types.SimpleNamespace(requests=0)
	38	+ self.index_calls: list[date] = []
	39	+
	40	+ def atom_current(self, form: str = "") -> str:
	41	+ self.stats.requests += 1
	42	+ return self.atom
	43	+
	44	+ def daily_index(self, d: date) -> str \| None:
	45	+ self.stats.requests += 1
	46	+ self.index_calls.append(d)
	47	+ return self.index
	48	+
	49	+
	50	+@pytest.fixture
	51	+def cofilers(fundamentals_data):
	52	+ """Two temporary tracked companies sharing accessions; everything they touch is removed afterwards."""
	53	+ from core.db import session
	54	+ from fundamentals import ingest
	55	+ from fundamentals.models import EdgarCompany, EdgarFiling, FundIngestState
	56	+ from stream.broker import get_redis
	57	+ r = get_redis()
	58	+ r.delete(ingest.SEEN_KEY, ingest.FAILED_KEY)
	59	+ with session() as s:
	60	+ before = s.get(FundIngestState, "incremental")
	61	+ snapshot = {k: getattr(before, k) for k in ("failures", "failure_samples")} if before else None
	62	+ s.add(EdgarCompany(cik=PARENT, ticker="ETR", tickers=["ETR"], name="Entergy Corp", status="active", ticker_history=[]))
	63	+ s.add(EdgarCompany(cik=SUB, ticker="EAL", tickers=["EAL"], name="Entergy Arkansas", status="active", ticker_history=[]))
	64	+ try:
	65	+ yield {"parent": PARENT, "sub": SUB, "redis": r}
	66	+ finally:
	67	+ from fundamentals.models import FundCoverage, fund_statements
	68	+ with session() as s:
	69	+ s.execute(delete(EdgarFiling).where(EdgarFiling.cik.in_([PARENT, SUB])))
	70	+ s.execute(delete(EdgarCompany).where(EdgarCompany.cik.in_([PARENT, SUB])))
	71	+ s.execute(delete(FundCoverage).where(FundCoverage.cik.in_([PARENT, SUB])))
	72	+ s.execute(delete(fund_statements).where(fund_statements.c.cik.in_([PARENT, SUB])))
	73	+ st = s.get(FundIngestState, "incremental")
	74	+ if st is not None and snapshot is not None:
	75	+ st.failures, st.failure_samples = snapshot["failures"], snapshot["failure_samples"]
	76	+ r.delete(ingest.SEEN_KEY, ingest.FAILED_KEY)
	77	+
	78	+
	79	+def _stub(monkeypatch, outcome):
	80	+ """Replace ingest_company with a recorder; `outcome(cik)` → error string or None."""
	81	+ from fundamentals import ingest
	82	+ calls: list[int] = []
	83	+
	84	+ def fake(client, cik, **kw):
	85	+ calls.append(int(cik))
	86	+ res = ingest.IngestResult(cik=int(cik), ticker=str(cik))
	87	+ res.error = outcome(int(cik))
	88	+ res.changed = res.error is None
	89	+ return res
	90	+ monkeypatch.setattr(ingest, "ingest_company", fake)
	91	+ return calls
	92	+
	93	+
	94	+# ------------------------------------------------------------------------------------- co-registrants
	95	+def test_shared_accession_is_one_row_per_registrant(cofilers):
	96	+ import pandas as pd
	97	+
	98	+ from fundamentals import ingest
	99	+ from fundamentals import normalize as N
	100	+ from fundamentals import service as S
	101	+ shared = N.Filing(accn=SHARED, form="10-Q", filed=date(2026, 7, 31), report_date=date(2026, 6, 30), primary_doc="etr-20260630.htm")
	102	+ only_parent = N.Filing(accn=OTHER, form="8-K", filed=date(2026, 8, 1), report_date=date(2026, 8, 1))
	103	+ untracked = N.Filing(accn="0000065984-26-000299", form="S-8", filed=date(2026, 8, 2))
	104	+ assert ingest._upsert_filings(PARENT, {SHARED: shared, OTHER: only_parent, "x": untracked}, pd.DataFrame()) == 2
	105	+ assert ingest._upsert_filings(SUB, {SHARED: shared}, pd.DataFrame()) == 1 # no IntegrityError any more
	106	+ assert ingest._upsert_filings(SUB, {SHARED: shared}, pd.DataFrame()) == 0 # idempotent
	107	+ pairs = {(PARENT, SHARED), (SUB, SHARED), (PARENT, OTHER), (SUB, OTHER)}
	108	+ assert ingest._known_pairs(pairs) == {(PARENT, SHARED), (SUB, SHARED), (PARENT, OTHER)}
	109	+ assert ingest._known_accns(SUB) == {SHARED}
	110	+ # both companies list the shared filing with their own EDGAR links
	111	+ for tk, cik in (("ETR", PARENT), ("EAL", SUB)):
	112	+ df, meta = S.filings(tk, form=None, date_from=None, date_to=None)
	113	+ row = df[df["accn"] == SHARED].iloc[0]
	114	+ assert meta["cik"] == cik and row["primary_doc_url"] == f"https://www.sec.gov/Archives/edgar/data/{cik}/000006598426000283/etr-20260630.htm"
	115	+ assert len(S.filings("ETR", form="10-Q,8-K", date_from=None, date_to=None)[0]) == 2
	116	+ assert len(S.filings("EAL", form=None, date_from=None, date_to=None)[0]) == 1
	117	+
	118	+
	119	+def test_poll_ingests_each_coregistrant_once(cofilers, monkeypatch):
	120	+ from fundamentals import ingest
	121	+ calls = _stub(monkeypatch, lambda cik: None)
	122	+ atom = _atom([("8-K", "Entergy Corp", PARENT, SHARED), ("8-K", "Entergy Arkansas", SUB, SHARED),
	123	+ ("S-8", "Entergy Corp", PARENT, "0000065984-26-000299"), # untracked form: ignored
	124	+ ("8-K", "Nobody Inc", 999999999, "0000999999-26-000001")]) # not a tracked CIK
	125	+ client = FakeClient(atom, index=None)
	126	+ s1 = ingest.poll_new_filings(client, forms=("8-K",))
	127	+ assert s1["seen"] == 2 and s1["new"] == 2 and s1["affected_ciks"] == [PARENT, SUB] and s1["index_days"] == []
	128	+ assert sorted(calls) == [PARENT, SUB] and s1["rows_changed"] == 2 and s1["failures"] == []
	129	+ assert all(d.weekday() < 5 for d in client.index_calls) and len(client.index_calls) == 2
	130	+ # the stub recorded nothing in edgar_filings: without the Redis guard the same pairs would be "new" for ever
	131	+ s2 = ingest.poll_new_filings(client, forms=("8-K",))
	132	+ assert s2["seen"] == 2 and s2["new"] == 0 and s2["affected_ciks"] == [] and len(calls) == 2
	133	+ assert cofilers["redis"].scard(ingest.SEEN_KEY) == 2 and cofilers["redis"].ttl(ingest.SEEN_KEY) > 6 * 86_400
	134	+
	135	+
	136	+def test_amendment_in_the_8k_feed_is_tracked(cofilers, monkeypatch):
	137	+ """The `type=8-K` Atom feed also returns 8-K/A entries — they must land in edgar_filings (Genasys loop)."""
	138	+ from fundamentals import ingest
	139	+ from fundamentals.edgar_client import TRACKED_FORMS
	140	+ assert {"8-K/A", "10-K/A", "10-Q/A", "10-KT", "10-QT", "20-F/A", "6-K/A"} <= set(TRACKED_FORMS)
	141	+ calls = _stub(monkeypatch, lambda cik: None)
	142	+ client = FakeClient(_atom([("8-K/A", "Entergy Corp", PARENT, OTHER)]))
	143	+ s1 = ingest.poll_new_filings(client, forms=("8-K",), include_daily_index=False)
	144	+ assert s1["new"] == 1 and calls == [PARENT]
	145	+
	146	+
	147	+def test_transient_failures_are_retried_then_abandoned(cofilers, monkeypatch):
	148	+ from core.db import session
	149	+ from fundamentals import ingest
	150	+ from fundamentals.models import FundIngestState
	151	+ calls = _stub(monkeypatch, lambda cik: "HTTPStatusError: 503 Service Unavailable" if cik == PARENT else None)
	152	+ client = FakeClient(_atom([("8-K", "Entergy Corp", PARENT, SHARED), ("8-K", "Entergy Arkansas", SUB, SHARED)]))
	153	+ with session() as s:
	154	+ st = s.get(FundIngestState, "incremental")
	155	+ failures0 = (st.failures or 0) if st else 0
	156	+ summaries = [ingest.poll_new_filings(client, forms=("8-K",), include_daily_index=False) for _ in range(ingest.MAX_ATTEMPTS + 1)]
	157	+ # the subsidiary succeeded once; the parent was retried MAX_ATTEMPTS times then marked seen
	158	+ assert calls.count(SUB) == 1 and calls.count(PARENT) == ingest.MAX_ATTEMPTS
	159	+ assert [s["new"] for s in summaries] == [2, 1, 1, 0]
	160	+ assert summaries[0]["failures"] == ["HTTPStatusError: 503 Service Unavailable"]
	161	+ with session() as s:
	162	+ st = s.get(FundIngestState, "incremental")
	163	+ assert st.failures == failures0 + ingest.MAX_ATTEMPTS
	164	+ # samples survive the clean 4th cycle and carry the CIK
	165	+ assert st.failure_samples and all(x.startswith(f"cik={PARENT} HTTPStatusError") for x in st.failure_samples[-ingest.MAX_ATTEMPTS:])
	166	+ assert cofilers["redis"].hlen(ingest.FAILED_KEY) == 0 # cleared once marked seen
	167	+
	168	+
	169	+def test_no_facts_is_permanent_not_a_failure(cofilers, monkeypatch):
	170	+ from core.db import session
	171	+ from fundamentals import ingest
	172	+ from fundamentals.models import FundIngestState
	173	+ calls = _stub(monkeypatch, lambda cik: ingest.NO_FACTS)
	174	+ client = FakeClient(_atom([("8-K", "Entergy Corp", PARENT, OTHER)]))
	175	+ with session() as s:
	176	+ st = s.get(FundIngestState, "incremental")
	177	+ failures0 = (st.failures or 0) if st else 0
	178	+ s1 = ingest.poll_new_filings(client, forms=("8-K",), include_daily_index=False)
	179	+ s2 = ingest.poll_new_filings(client, forms=("8-K",), include_daily_index=False)
	180	+ assert s1["new"] == 1 and s1["no_facts"] == [PARENT] and s1["failures"] == [] and s2["new"] == 0 and calls == [PARENT]
	181	+ with session() as s:
	182	+ assert (s.get(FundIngestState, "incremental").failures or 0) == failures0
	183	+
	184	+
	185	+def test_ingest_company_without_facts_still_records_filings(cofilers):
	186	+ """companyfacts 404 (funds, trusts…) must not leave the accessions unknown — the poller would loop on them."""
	187	+ from fundamentals import ingest
	188	+ sub = {"cik": PARENT, "name": "Entergy Corp", "fiscalYearEnd": "1231", "filings": {"recent": {
	189	+ "accessionNumber": [SHARED, OTHER], "form": ["10-Q", "8-K"], "filingDate": ["2026-07-31", "2026-08-01"],
	190	+ "reportDate": ["2026-06-30", "2026-08-01"], "primaryDocument": ["etr-20260630.htm", "etr-8k.htm"], "isXBRL": [1, 1]}}}
	191	+ res = ingest.ingest_company(None, PARENT, companyfacts={"cik": PARENT, "facts": {}}, submissions=sub)
	192	+ assert res.error is None or res.error == ingest.NO_FACTS
	193	+ no_facts_client = types.SimpleNamespace(companyfacts=lambda cik, refresh=False: None,
	194	+ submissions=lambda cik, refresh=False: {**sub, "cik": SUB})
	195	+ res = ingest.ingest_company(no_facts_client, SUB)
	196	+ assert res.error == ingest.NO_FACTS and not res.failed and res.filings == 2
	197	+ assert ingest._known_accns(SUB) == {SHARED, OTHER}
	198	+
	199	+
	200	+# ------------------------------------------------------------------------------------- daily index
	201	+def test_business_days_back():
	202	+ from fundamentals.ingest import business_days_back
	203	+ assert business_days_back(date(2026, 9, 6)) == [date(2026, 9, 4), date(2026, 9, 3)] # Sunday → Fri, Thu
	204	+ assert business_days_back(date(2026, 9, 7)) == [date(2026, 9, 7), date(2026, 9, 4)] # Monday → Mon, Fri
	205	+ assert business_days_back(date(2026, 9, 9), 3) == [date(2026, 9, 9), date(2026, 9, 8), date(2026, 9, 7)]
	206	+
	207	+
	208	+def test_daily_index_403_means_not_published(app, tmp_path):
	209	+ """www.sec.gov/Archives answers 403 for objects that do not exist (weekend indexes): no exception, no retry."""
	210	+ from fundamentals import edgar_client as ec
	211	+ hits: list[str] = []
	212	+
	213	+ def handler(request: httpx.Request) -> httpx.Response:
	214	+ hits.append(str(request.url))
	215	+ assert "contact@" in request.headers["user-agent"] and "gzip" in request.headers["accept-encoding"]
	216	+ if request.url.path.endswith("master.20260906.idx"):
	217	+ return httpx.Response(403, text="<Error><Code>AccessDenied</Code></Error>")
	218	+ return httpx.Response(200, text="CIK\|Company Name\|Form Type\|Date Filed\|File Name\n"
	219	+ "65984\|ENTERGY CORP\|8-K\|20260904\|edgar/data/65984/0000065984-26-000290.txt\n")
	220	+ client = ec.EdgarClient(rate=1000, raw_dir=tmp_path, http=httpx.Client(transport=httpx.MockTransport(handler), headers={
	221	+ "User-Agent": "HF Market Data (test, contact@spboucher.ai)", "Accept-Encoding": "gzip, deflate"}))
	222	+ assert client.daily_index(date(2026, 9, 6)) is None
	223	+ assert len(hits) == 1 and hits[0].endswith("/Archives/edgar/daily-index/2026/QTR3/master.20260906.idx")
	224	+ text = client.daily_index(date(2026, 9, 4))
	225	+ from fundamentals.ingest import parse_master_index
	226	+ assert parse_master_index(text) == [{"cik": 65984, "name": "ENTERGY CORP", "form": "8-K", "filed": "2026-09-04",
	227	+ "accn": "0000065984-26-000290"}]
	228	+ with pytest.raises(httpx.HTTPStatusError):
	229	+ client.get_text("https://www.sec.gov/Archives/edgar/daily-index/2026/QTR3/master.20260906.idx")
	230	+
	231	+
	232	+# ------------------------------------------------------------------------------------- statements / coverage
	233	+def test_replace_statements_only_rewrites_on_change(fundamentals_data):
	234	+ from core.db import session
	235	+ from fundamentals import ingest
	236	+ from fundamentals import service as S
	237	+ from fundamentals.models import fund_statements
	238	+ cik = 320193
	239	+ original = S._rows(cik)
	240	+ ids0 = sorted(r["id"] for r in original)
	241	+ assert original and all(isinstance(r["coverage"], dict) for r in original) and "coverage_id" not in original[0]
	242	+ assert ingest._replace_statements(cik, original) is False # identical → untouched
	243	+ with session() as s:
	244	+ assert sorted(s.scalars(select(fund_statements.c.id).where(fund_statements.c.cik == cik))) == ids0
	245	+ changed = [dict(r) for r in original]
	246	+ target = next(r for r in changed if r["statement"] == "income" and r.get("revenue") is not None)
	247	+ target["revenue"] = target["revenue"] + 1.0
	248	+ assert ingest._replace_statements(cik, changed) is True
	249	+ again = S._rows(cik)
	250	+ assert sorted(r["id"] for r in again) != ids0
	251	+ hit = next(r for r in again if r["accn"] == target["accn"] and r["statement"] == "income" and r["fiscal_quarter"] == target["fiscal_quarter"])
	252	+ assert hit["revenue"] == target["revenue"] and hit["coverage"] == target["coverage"]
	253	+ with session() as s: # written through the dictionary: inline JSON is NULL, the reference is set
	254	+ rows = s.execute(select(fund_statements.c.coverage, fund_statements.c.coverage_id).where(fund_statements.c.cik == cik)).all()
	255	+ assert rows and all(c is None and cid is not None for c, cid in rows)
	256	+ assert ingest._replace_statements(cik, original) is True # restore for the other tests
	257	+ assert S._rows(cik)[0]["coverage"] == original[0]["coverage"]
	258	+
	259	+
	260	+def test_coverage_dictionary_interns_and_hydrates(app):
	261	+ from core.db import session
	262	+ from fundamentals import coverage_store as CS
	263	+ from fundamentals.models import FundCoverageBlob
	264	+ a = {"interest_expense": {"reason": "no_mapped_tag"}, "ebitda": {"computed": "operating_income + d_and_a"}}
	265	+ b = {"ebitda": {"computed": "operating_income + d_and_a"}, "interest_expense": {"reason": "no_mapped_tag"}} # same, other order
	266	+ ids = CS.intern([a, b, {}, a])
	267	+ assert ids[0] == ids[1] == ids[3] and ids[2] != ids[0] and CS.digest(a) == CS.digest(b)
	268	+ CS.reset_cache_for_tests()
	269	+ assert CS.intern([b]) == [ids[0]] # found in the table, not re-inserted
	270	+ with session() as s:
	271	+ n = s.scalar(select(FundCoverageBlob.id).where(FundCoverageBlob.sha1 == CS.digest(a)))
	272	+ assert n == ids[0]
	273	+ rows = [{"accn": "x", "coverage": None, "coverage_id": ids[0]}, # dictionary reference
	274	+ {"accn": "y", "coverage": {"revenue": {"tag": "us-gaap:Revenues"}}, "coverage_id": None}, # legacy inline
	275	+ {"accn": "z", "coverage": None, "coverage_id": None}]
	276	+ CS.hydrate(rows)
	277	+ assert rows[0]["coverage"] == a and rows[1]["coverage"] == {"revenue": {"tag": "us-gaap:Revenues"}} and rows[2]["coverage"] == {}
	278	+ assert all("coverage_id" not in r for r in rows)
	279	+ assert CS.sha_of_ids({ids[0]}) == {ids[0]: CS.digest(a)}
	280	+
	281	+
	282	+def test_failure_samples_append_and_cap(app):
	283	+ from core.db import session
	284	+ from fundamentals import ingest
	285	+ from fundamentals.models import FundIngestState
	286	+ key = "test_samples"
	287	+ try:
	288	+ ingest._set_state(key, failures_add=2, failure_samples_add=[f"e{i}" for i in range(12)])
	289	+ ingest._set_state(key, last_run_at=None) # a clean cycle keeps the evidence
	290	+ with session() as s:
	291	+ st = s.get(FundIngestState, key)
	292	+ assert st.failures == 2 and st.failure_samples == [f"e{i}" for i in range(2, 12)]
	293	+ ingest._set_state(key, failure_samples=[]) # explicit replace still possible
	294	+ with session() as s:
	295	+ assert s.get(FundIngestState, key).failure_samples == []
	296	+ finally:
	297	+ with session() as s:
	298	+ s.execute(delete(FundIngestState).where(FundIngestState.key == key))
	299	+
	300	+
	301	+# ------------------------------------------------------------------------------------- migrations
	302	+def test_migrate_filings_pk_from_legacy_table(tmp_path):
	303	+ from sqlalchemy import create_engine, text
	304	+
	305	+ from fundamentals import migrations
	306	+ eng = create_engine(f"sqlite:///{tmp_path / 'legacy.db'}")
	307	+ with eng.begin() as con:
	308	+ con.execute(text("""CREATE TABLE edgar_filings (accn VARCHAR(24) NOT NULL, cik INTEGER NOT NULL, form VARCHAR(16) NOT NULL,
	309	+ filed_date DATE NOT NULL, period_of_report DATE, primary_doc VARCHAR(512), is_amendment BOOLEAN NOT NULL,
	310	+ is_xbrl BOOLEAN NOT NULL, parsed_at DATETIME, PRIMARY KEY (accn))"""))
	311	+ con.execute(text("CREATE INDEX ix_edgar_filings_cik ON edgar_filings (cik)"))
	312	+ con.execute(text("INSERT INTO edgar_filings VALUES ('0000065984-26-000283', 65984, '10-Q', '2026-07-31', '2026-06-30', NULL, 0, 1, NULL)"))
	313	+ con.execute(text("INSERT INTO edgar_filings VALUES ('0000320193-24-000069', 320193, '10-Q', '2024-05-03', '2024-03-30', NULL, 0, 1, NULL)"))
	314	+ assert migrations.filings_pk_is_composite(con) is False
	315	+ assert migrations.migrate_filings_pk(con) == 2
	316	+ assert migrations.filings_pk_is_composite(con) is True
	317	+ # the subsidiary's copy of the shared accession now fits
	318	+ con.execute(text("INSERT INTO edgar_filings (cik, accn, form, filed_date, is_amendment, is_xbrl) VALUES (66901, '0000065984-26-000283', '10-Q', '2026-07-31', 0, 1)"))
	319	+ assert con.execute(text("SELECT count(*) FROM edgar_filings WHERE accn='0000065984-26-000283'")).scalar() == 2
	320	+ names = {r[0] for r in con.execute(text("SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='edgar_filings'"))}
	321	+ assert {"ix_edgar_filings_accn", "ix_edgar_filings_cik_filed", "ix_edgar_filings_form", "ix_edgar_filings_filed_date"} <= names
	322	+ assert migrations.migrate_filings_pk(con) == 3 # idempotent rerun keeps every row
	323	+
	324	+
	325	+def test_startup_schema_is_idempotent(fundamentals_data):
	326	+ from core.db import engine
	327	+ from fundamentals import migrations
	328	+ assert migrations.ensure_filings_schema() == "ok"
	329	+ assert migrations.ensure_statements_schema() is False
	330	+ assert migrations.ensure_auto_vacuum() is False # already switched when the test DB was young
	331	+ with engine.connect() as con:
	332	+ from sqlalchemy import text
	333	+ assert int(con.execute(text("PRAGMA auto_vacuum")).scalar()) == 2
	334	+ migrations.incremental_vacuum()
	335