From a7aefa6fe76cd287164eb220fc1397882694589c Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Mon, 27 Jul 2026 11:52:59 +0200 Subject: [PATCH] Recover SEC facts misfiled under a co-registrant CIK The fundamentals import had been failing for three days on two tracked filings the index listed but Company Facts appeared not to have. They were not lagging: SEC filed the XBRL of NEE's and DOW's 2026-07-24 combined parent/subsidiary 10-Qs under the co-registrant's CIK (Florida Power & Light, Dow Chemical), so the ticker-carrying filer's own facts file never receives that accession. This does not self-correct - an NEE filing misattributed the same way in 2014 is still misfiled. Because source_max_date advances only on a promoted run, the failure was self-perpetuating: every later run re-walked the same index day and re-hit the same two filings. - Recover from the co-registrant file. The daily index lists every co-registrant of an accession, which is the only pointer to where the facts actually landed. Rows are re-stamped to the real filer, since parse_snapshots stamps the CIK of the payload it read. - Guard the recovery with a share-count continuity check against the issuer's own history, so a subsidiary's standalone facts can never be stored as the parent's. No history, no recovery. - Bound the blocking: a filing still unresolvable after MISSING_XBRL_RETRY_DAYS promotes with a named unresolved_filing warning instead of wedging every later import. - Name the offending filings in the alert and record them in validation_json, separating not_in_companyfacts from not_in_submissions. The gate previously reported a count and discarded the accessions. Verified against live SEC data: both filings recover with the correct CIK and the guard rejects a mismatched reference. Co-Authored-By: Claude Opus 5 --- app/services/sec_fundamentals_importer.py | 311 +++++++++++++++++-- tests/unit/test_sec_fundamentals_importer.py | 149 ++++++++- 2 files changed, 436 insertions(+), 24 deletions(-) diff --git a/app/services/sec_fundamentals_importer.py b/app/services/sec_fundamentals_importer.py index 46aa2f3..d516d11 100644 --- a/app/services/sec_fundamentals_importer.py +++ b/app/services/sec_fundamentals_importer.py @@ -18,6 +18,23 @@ Guardrails (design + reviews): a tracked XBRL index accession missing from Company Facts fails the run (the two are separate SEC products that can lag) so we retry rather than record a null/partial snapshot. Non-XBRL amendments are skipped with a recorded reason. + A failure here blocks every later run (``source_max_date`` only advances on a + promoted run), so it names the offending filings in the alert and separates the + causes — ``not_in_companyfacts`` (facts lag) vs ``not_in_submissions`` (the + index row is absent from the issuer's own filing list, which no retry fixes). +- **Co-registrant recovery**, because "missing from Company Facts" is often not + missing at all: SEC files some combined parent/subsidiary filings' XBRL under + the co-registrant's CIK, so the ticker-carrying parent's own file never gets + that accession. The daily index lists every co-registrant of an accession, so + the facts are found there and re-stamped to the real filer — guarded by a + share-count continuity check so a subsidiary's standalone numbers can never be + stored as the parent's. Confirmed 2026-07-27 (NEE via FPL, DOW via Dow Chemical) + and it is not transient: an NEE filing misattributed in 2014 is still misfiled. +- **Bounded blocking.** Anything still unresolvable after ``MISSING_XBRL_RETRY_DAYS`` + stops failing the run and is promoted around, with a named ``unresolved_filing`` + warning. One filing SEC misfiled must not wedge every later import; the index + only moves forward, so a tolerated accession returns only via a reparse, and + only once SEC has re-filed it under the filer's own CIK. - ``promote`` inserts snapshots ``ON CONFLICT (accession) DO NOTHING`` (immutable), reports differing existing accessions, and applies ticker updates in the same transaction. @@ -32,8 +49,8 @@ Guardrails (design + reviews): from __future__ import annotations import logging -from collections import defaultdict -from dataclasses import dataclass, field +from collections import Counter, defaultdict +from dataclasses import dataclass, field, replace from datetime import date, datetime, timedelta, timezone from typing import Any, Callable @@ -57,6 +74,16 @@ _XBRL_FORMS = {"10-K", "10-Q", "10-K/A", "10-Q/A"} # On the one-time backfill, require this fraction of tracked issuers to yield at # least one snapshot (guards a broken fetch/parse from promoting a hollow table). MIN_BACKFILL_COVERAGE = 0.5 +# How long an index accession may stay unresolvable before the run stops failing +# on it. Genuine index↔facts lag clears within a day (a weekend stretches it to +# three); past that it is misfiled, not late, and blocking forever costs more +# than the missing filing does — see the unresolved-filing guardrail below. +MISSING_XBRL_RETRY_DAYS = 3 +# Share-count band a co-registrant-recovered row must land in, relative to the +# issuer's own last snapshot. Wide enough for buybacks/issuance, nowhere near +# wide enough to let a subsidiary shell's token float through (see _shares_continuous). +RECOVERY_SHARES_MIN = 0.5 +RECOVERY_SHARES_MAX = 2.0 _SNAPSHOT_COLS = ( "cik", "accession", "form", "filed_date", "accepted_at", "period_start", @@ -78,7 +105,11 @@ class StagedFundamentals: skipped_filings: list[dict[str, str]] = field(default_factory=list) field_issues: list[dict[str, str]] = field(default_factory=list) skipped_non_xbrl: list[dict[str, str]] = field(default_factory=list) - missing_xbrl: list[dict[str, str]] = field(default_factory=list) + # Index rows we could not resolve to Company Facts, with a per-row reason + # (not_in_companyfacts | not_in_submissions | ...) — see _missing(). + missing_xbrl: list[dict[str, Any]] = field(default_factory=list) + # Accessions parsed out of a co-registrant's Company Facts file. + recovered: list[dict[str, Any]] = field(default_factory=list) invalid_payloads: list[dict[str, str]] = field(default_factory=list) existing_accessions: set[str] = field(default_factory=set) # Tracked issuers whose registrant has NO XBRL 10-K/10-Q at all: they can @@ -117,6 +148,9 @@ class SecFundamentalsImporter: # cached by detect_revision, consumed by stage: self._resolved: ResolvedUniverse | None = None self._index_rows: list[dict[str, Any]] = [] + # accession -> the OTHER CIKs the daily index lists it under (co-registrants + # of a combined filing). Only populated for accessions a tracked issuer filed. + self._coregistrants: dict[str, list[int]] = {} self._latest_index_date: date | None = None self._backfill = False @@ -132,6 +166,7 @@ class SecFundamentalsImporter: # Reparse needs every accession restaged, not just those filed since # the last run — the facts a fixed parser now accepts were never # stored, so a reparse cannot be served from the database. + self._coregistrants = {} if last_processed is None or self.reparse: self._backfill = True self._index_rows = [] @@ -151,10 +186,12 @@ class SecFundamentalsImporter: staged = StagedFundamentals(resolved=resolved, backfill=self._backfill) cik_to_tids = resolved.cik_to_ticker_ids - filed_by_cik: dict[int, list[str]] = defaultdict(list) + # Whole index rows (not bare accessions): form + index date are what make + # an unresolvable filing diagnosable without re-walking the index by hand. + filed_by_cik: dict[int, list[dict[str, Any]]] = defaultdict(list) for r in self._index_rows: if r["cik"] in cik_to_tids: - filed_by_cik[r["cik"]].append(r["accession"]) + filed_by_cik[r["cik"]].append(r) existing = await self._ciks_with_snapshots(db, set(cik_to_tids)) if self._backfill: @@ -165,10 +202,15 @@ class SecFundamentalsImporter: backfill_ciks = {c for c in cik_to_tids if c not in existing} incremental_ciks = set(filed_by_cik) - backfill_ciks + # Continuity reference for co-registrant recovery, read once up front. + last_shares = await self._last_shares_outstanding(db, set(cik_to_tids)) + async with self._client_factory() as client: for cik in sorted(backfill_ciks | incremental_ciks): is_backfill = cik in backfill_ciks - await self._stage_issuer(client, cik, is_backfill, filed_by_cik, staged) + await self._stage_issuer( + client, cik, is_backfill, filed_by_cik, staged, last_shares + ) # Read-only discrepancy detection: an accession we reconstructed that is # already stored, differing in ANY source field (immutable → report in @@ -185,7 +227,9 @@ class SecFundamentalsImporter: staged.discrepancies.append({"accession": row.accession, "fields": fields}) return staged - async def _stage_issuer(self, client, cik, is_backfill, filed_by_cik, staged) -> None: + async def _stage_issuer( + self, client, cik, is_backfill, filed_by_cik, staged, last_shares + ) -> None: cf = await client.companyfacts(cik) bad = _companyfacts_structure_error(cf) if bad is not None: @@ -201,31 +245,61 @@ class SecFundamentalsImporter: {"cik": cik10(cik), "name": sub.get("name"), "tickers": sub.get("tickers")} ) + fiscal_year_end = sub.get("fiscal_year_end") + recovered_rows: list[SnapshotRow] = [] if is_backfill: accns = set(xbrl_meta) else: present = parser.companyfacts_accessions(cf) accns = set() - for accn in filed_by_cik.get(cik, []): + for index_row in filed_by_cik.get(cik, []): + accn = index_row["accession"] if accn in nonxbrl: staged.skipped_non_xbrl.append({"cik": cik10(cik), "accession": accn}) - elif accn in xbrl_meta and accn in present: + elif accn not in xbrl_meta: + # The daily index lists it but the issuer's own filing list does + # not (submissions lagging the index, no usable period metadata, + # or a co-registrant filing). NOT a Company-Facts lag — separate + # cause, separate fix, so it gets its own reason. + staged.missing_xbrl.append( + _missing(cik, index_row, "not_in_submissions", self.today) + ) + elif accn in present: accns.add(accn) else: - # XBRL (or unknown) filing not yet in Company Facts → the products - # have lagged; fail+retry rather than record nothing for it. - staged.missing_xbrl.append({"cik": cik10(cik), "accession": accn}) + # Filed, XBRL, but absent from this issuer's Company Facts. Try + # the co-registrant file before treating it as missing data. + row, source_cik = await self._recover_from_coregistrant( + client, cik, accn, xbrl_meta, fiscal_year_end, + last_shares.get(cik10(cik)), staged, + ) + if row is not None: + recovered_rows.append(row) + staged.recovered.append({ + "cik": cik10(cik), + "accession": accn, + "source_cik": source_cik, + "form": index_row.get("form"), + }) + else: + staged.missing_xbrl.append(_missing( + cik, index_row, + # Found, but it did not look like this issuer's own + # numbers — say so; it is not the same as absent. + "coregistrant_facts_rejected" if source_cik + else "not_in_companyfacts", + self.today, + )) # fiscalYearEnd (MMDD) is what lets the parser derive period identity from # reportDate instead of SEC's unreliable fy/fp fields. - result = parser.parse_snapshots( - cf, xbrl_meta, accns, fiscal_year_end=sub.get("fiscal_year_end") - ) + result = parser.parse_snapshots(cf, xbrl_meta, accns, fiscal_year_end=fiscal_year_end) staged.rows.extend(result.rows) + staged.rows.extend(recovered_rows) staged.skipped_filings.extend(result.skipped_filings) staged.field_issues.extend(result.field_issues) staged.issuers_fetched += 1 - if result.rows: + if result.rows or recovered_rows: staged.issuers_with_rows += 1 # SIC proposal for this issuer's tickers (read-only; applied in promote). @@ -234,14 +308,65 @@ class SecFundamentalsImporter: for tid in staged.resolved.cik_to_ticker_ids.get(cik, []): staged.sic_updates.append((tid, sic, desc)) + async def _recover_from_coregistrant( + self, client, cik: int, accn: str, xbrl_meta, fiscal_year_end, reference, staged, + ) -> tuple[SnapshotRow | None, str | None]: + """Look for ``accn``'s facts in a co-registrant's Company Facts file. + + SEC sometimes files a combined parent/subsidiary filing's XBRL under the + co-registrant's CIK rather than the filer's — the ticker-carrying parent's + own file simply never gets that accession. Verified 2026-07-27 for NEE + (facts under Florida Power & Light) and DOW (under Dow Chemical); an NEE + filing misattributed the same way in **2014** is still misattributed, so + this does not self-correct and no amount of retrying recovers it. + + Returns ``(row, source_cik)`` on success, ``(None, source_cik)`` when the + facts were found but rejected by the continuity guard, ``(None, None)`` + when no co-registrant has them. + + Incremental path only: the co-registrant map comes from the daily index, + which a backfill/reparse does not walk. A reparse therefore recovers a + filing only once SEC re-files it under the filer's own CIK. + """ + for co in self._coregistrants.get(accn, []): + try: + cf_co = await client.companyfacts(co) + except SecError: + continue # a co-registrant shell often has no facts file at all + if _companyfacts_structure_error(cf_co) is not None: + continue + result = parser.parse_snapshots( + cf_co, xbrl_meta, {accn}, fiscal_year_end=fiscal_year_end + ) + if not result.rows: + continue + row = result.rows[0] + if not _shares_continuous(row.shares_outstanding, reference): + return None, cik10(co) + # A recovered row is the one most worth flagging, so its parser caveats + # travel with it rather than being dropped on the way out. + staged.field_issues.extend(result.field_issues) + # parse_snapshots stamps the CIK of the payload it read — re-stamp to + # the issuer that actually filed, or the row lands under the shell. + return replace(row, cik=cik10(cik)), cik10(co) + return None, None + async def validate(self, db, staged: StagedFundamentals) -> ValidationResult: messages: list[str] = [] - # Consistency gate — before any write. - if staged.missing_xbrl: + # Consistency gate — before any write. Only filings still inside the retry + # window block: a failure here stops every later run too (source_max_date + # advances on promotion alone), so blocking forever on a filing SEC has + # misfiled would cost far more than the one filing it withholds. Older + # ones are carried by promote() as a warning instead. The message names + # the filings: "which ones" has to be in the alert itself, not merely + # reconstructible by re-walking the index. + blocking = _within_retry_window(staged.missing_xbrl) + if blocking: messages.append( - f"{len(staged.missing_xbrl)} tracked XBRL filing(s) not yet in " - "Company Facts (index/facts lag) — retry" + f"{len(blocking)} tracked XBRL filing(s) unresolved within the " + f"{MISSING_XBRL_RETRY_DAYS}-day retry window " + f"({_reason_counts(blocking)}) — retry: {_missing_detail(blocking)}" ) # Malformed companyfacts payloads must fail, not degrade to skipped rows. if staged.invalid_payloads: @@ -272,7 +397,11 @@ class SecFundamentalsImporter: "skipped_non_xbrl": len(staged.skipped_non_xbrl), "no_xbrl_filings": staged.no_xbrl_filings[:50], "no_xbrl_filings_count": len(staged.no_xbrl_filings), - "missing_xbrl": len(staged.missing_xbrl), + "missing_xbrl": staged.missing_xbrl[:50], + "missing_xbrl_count": len(staged.missing_xbrl), + "missing_xbrl_blocking": len(blocking), + "recovered_from_coregistrant": staged.recovered[:50], + "recovered_count": len(staged.recovered), "invalid_payloads": staged.invalid_payloads, "cik_updates": len(staged.resolved.cik_updates), # differing existing accessions (immutable — kept, reported here) @@ -332,6 +461,48 @@ class SecFundamentalsImporter: created_at=_now(), )) + # Recovered rows are real data from an unexpected place — record where they + # came from, so a wrong recovery is auditable rather than invisible. + if staged.recovered: + named = ", ".join( + f"{r['accession']} <- CIK {r['source_cik']}" for r in staged.recovered[:10] + ) + db.add(SystemEvent( + severity="warning", + source="sec_facts", + code="coregistrant_recovery", + message=( + f"{len(staged.recovered)} filing(s) were absent from the filer's " + f"own Company Facts and were parsed from a co-registrant's file " + f"instead (share count checked against the issuer's history): {named}" + )[:4000], + dedup_key=f"sec_facts:coregistrant_recovery:{run_id}", + created_at=_now(), + )) + + # Filings past the retry window: promoted WITHOUT them so one misfiled + # filing cannot wedge every later import. This is the deliberate trade — + # loud and named, because the index only moves forward and nothing will + # revisit them on its own. + tolerated = _past_retry_window(staged.missing_xbrl) + if tolerated: + db.add(SystemEvent( + severity="warning", + source="sec_facts", + code="unresolved_filing", + message=( + f"{len(tolerated)} tracked filing(s) still unresolvable after " + f"{MISSING_XBRL_RETRY_DAYS} days; promoting without them rather " + f"than blocking every later import. They are NOT retried. " + f"scripts/reparse_fundamentals.py recovers them ONLY once SEC " + f"re-files under the filer's own CIK — it walks no index, so it " + f"cannot reach facts still sitting under a co-registrant: " + f"{_missing_detail(tolerated)}" + )[:4000], + dedup_key=f"sec_facts:unresolved_filing:{run_id}", + created_at=_now(), + )) + # A tracked issuer whose registrant has no XBRL filings can never produce a # snapshot, and it is restaged on every run forever. That is a resolution # problem, not missing data, and it is silent without this. @@ -390,9 +561,25 @@ class SecFundamentalsImporter: rows: list[dict[str, Any]] = [] day = last_processed + timedelta(days=1) while day <= latest: + # Group the whole day first: a combined filing is listed once per + # co-registrant CIK, and those sibling CIKs are the only pointer to + # where SEC may have put the XBRL (see _recover_from_coregistrant). + by_accession: dict[str, list[dict[str, Any]]] = defaultdict(list) for r in await client.daily_index(day): - if r["form"] in _XBRL_FORMS and r["cik"] in tracked: - rows.append(r) + if r["form"] in _XBRL_FORMS: + by_accession[r["accession"]].append(r) + for accession, group in by_accession.items(): + filers = {r["cik"] for r in group} + tracked_filers = filers & tracked + if not tracked_filers: + continue + siblings = sorted(filers - tracked_filers) + if siblings: + self._coregistrants[accession] = siblings + for r in group: + if r["cik"] in tracked_filers: + r["index_date"] = day # not hashed (revision uses cik/accession) + rows.append(r) day += timedelta(days=1) return rows @@ -409,6 +596,30 @@ class SecFundamentalsImporter: ).scalars().all() return {int(c) for c in found} + async def _last_shares_outstanding(self, db, ciks: set[int]) -> dict[str, float]: + """Latest known shares outstanding per tracked issuer — the continuity + reference co-registrant recovery is checked against.""" + if not ciks: + return {} + rows = ( + await db.execute( + select(FundamentalSnapshot.cik, FundamentalSnapshot.shares_outstanding) + .where( + FundamentalSnapshot.cik.in_([cik10(c) for c in ciks]), + FundamentalSnapshot.shares_outstanding.is_not(None), + ) + # Last write per cik wins, so the sort must be total: an amendment + # and its original share a period_end, and an undefined tie there + # would make recovery non-deterministic across runs and dialects. + .order_by( + FundamentalSnapshot.period_end, + FundamentalSnapshot.filed_date, + FundamentalSnapshot.accession, + ) + ) + ).all() + return {cik: float(shares) for cik, shares in rows} + async def _existing_by_accession(self, db, accessions: list[str]) -> dict[str, FundamentalSnapshot]: if not accessions: return {} @@ -435,6 +646,60 @@ def _companyfacts_structure_error(cf: Any) -> str | None: return None +def _missing(cik: int, row: dict[str, Any], reason: str, today: date) -> dict[str, Any]: + """One unresolvable index row, carrying everything needed to look the filing + up by hand (EDGAR accession + the index date it was seen on) and to decide + whether it is still young enough to be worth blocking on.""" + index_date = row.get("index_date") + return { + "cik": cik10(cik), + "accession": row["accession"], + "form": row.get("form"), + "index_date": index_date, + # No index date (older cached rows) => age 0 => blocks, the safe default. + "age_days": (today - index_date).days if isinstance(index_date, date) else 0, + "reason": reason, + } + + +def _within_retry_window(missing: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [m for m in missing if m.get("age_days", 0) <= MISSING_XBRL_RETRY_DAYS] + + +def _past_retry_window(missing: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [m for m in missing if m.get("age_days", 0) > MISSING_XBRL_RETRY_DAYS] + + +def _shares_continuous(shares: float | None, reference: float | None) -> bool: + """Does a co-registrant-recovered share count look like this issuer's own? + + The failure worth preventing is storing a subsidiary's standalone facts as the + parent's. A co-registrant shell holds a token float — Florida Power & Light + against NextEra's 2.09bn shares — so any sane band separates them while still + tolerating buybacks and issuance. With no history to compare against (a newly + tracked issuer) or no share count at all, recovery is refused, not guessed. + """ + if not shares or not reference: + return False + return RECOVERY_SHARES_MIN <= shares / reference <= RECOVERY_SHARES_MAX + + +def _reason_counts(missing: list[dict[str, Any]]) -> str: + counts = Counter(m["reason"] for m in missing) + return ", ".join(f"{reason}={n}" for reason, n in sorted(counts.items())) + + +def _missing_detail(missing: list[dict[str, Any]], limit: int = 10) -> str: + detail = ", ".join( + f"{m['cik']}/{m['accession']} {m.get('form') or '?'} " + f"[{m.get('index_date') or '?'}] {m['reason']}" + for m in missing[:limit] + ) + if len(missing) > limit: + detail += f", +{len(missing) - limit} more" + return detail + + def _filing_meta(sub: dict[str, Any]) -> tuple[dict[str, FilingMeta], set[str]]: """(xbrl_meta, nonxbrl_accessions) from a submissions payload. xbrl_meta only includes 10-K/10-Q(/A) filings that are XBRL and have full period metadata.""" diff --git a/tests/unit/test_sec_fundamentals_importer.py b/tests/unit/test_sec_fundamentals_importer.py index bbd71c0..19cf5d7 100644 --- a/tests/unit/test_sec_fundamentals_importer.py +++ b/tests/unit/test_sec_fundamentals_importer.py @@ -3,6 +3,7 @@ import framework with a fake SEC client (no network).""" from __future__ import annotations +import json import os import tempfile from datetime import date, datetime, timezone @@ -14,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn from app.database import Base import app.models # noqa: F401 from app.models.fundamental_snapshot import FundamentalSnapshot +from app.models.system_event import SystemEvent from app.models.ticker import Ticker from app.services.data_import import STATUS_FAILED, STATUS_PROMOTED, run_import from app.services.sec_fundamentals_importer import SecFundamentalsImporter @@ -212,10 +214,155 @@ async def test_consistency_gate_fails_when_facts_lag_index(engine): run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine) assert run.status == STATUS_FAILED - assert "Company Facts" in (run.error_details or "") + # The gate blocks every later run until it clears, so the alert itself has to + # name the filing and say why it could not be resolved. + details = run.error_details or "" + assert "GHOST" in details and "not_in_companyfacts" in details + assert "2026-05-01" in details # index date the filing was seen on + summary = json.loads(run.validation_json or "{}") + assert summary["missing_xbrl_count"] == 1 + assert summary["missing_xbrl"][0]["accession"] == "GHOST" + assert summary["missing_xbrl"][0]["form"] == "10-Q" assert await _count(factory, FundamentalSnapshot) == 2 # nothing new written +async def test_gate_separates_missing_submissions_from_missing_facts(engine): + """An index row the issuer's own filing list does not carry is a different + failure from a Company-Facts lag, and must not be reported as one.""" + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + + incr = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, # submissions never lists ORPHAN + latest_index=date(2026, 5, 2), + daily={date(2026, 5, 1): [{"form": "10-Q", "cik": 320193, "accession": "ORPHAN"}]}, + ) + run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine) + + assert run.status == STATUS_FAILED + assert "ORPHAN" in (run.error_details or "") + assert "not_in_submissions" in (run.error_details or "") + summary = json.loads(run.validation_json or "{}") + assert summary["missing_xbrl"][0]["reason"] == "not_in_submissions" + + +def _coregistrant_client(share_fact): + """Incremental client where Q2A's facts landed in co-registrant 99999's file + instead of the filer's own — the NEE-via-FPL / DOW-via-Dow-Chemical shape.""" + cf_q2 = _rev("2025-09-28", "2026-03-28", 254940, 2026, "Q2", "Q2A") + return FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={ + 320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1]), # filer's own: no Q2A + 99999: _companyfacts([cf_q2], [share_fact], cik=99999), + }, + submissions={320193: _submissions(SUB_FILINGS + [ + _filing("Q2A", "10-Q", "2026-03-28", "2026-05-01", "2026-05-01T10:01:00.000Z")])}, + latest_index=date(2026, 5, 2), + daily={date(2026, 5, 1): [ + # one combined filing, listed by the index under both co-registrants + {"form": "10-Q", "cik": 320193, "accession": "Q2A"}, + {"form": "10-Q", "cik": 99999, "accession": "Q2A"}, + ]}, + ) + + +async def test_recovers_facts_misfiled_under_coregistrant(engine): + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + + # 14687 shares is continuous with the issuer's own history (14681 last quarter). + incr = _coregistrant_client(_shares("2026-04-17", 14687, "Q2A", 2026, "Q2")) + run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine) + + assert run.status == STATUS_PROMOTED + summary = json.loads(run.validation_json or "{}") + assert summary["recovered_count"] == 1 + assert summary["recovered_from_coregistrant"][0]["source_cik"] == "0000099999" + assert summary["missing_xbrl_count"] == 0 + + async with factory() as s: + q2 = (await s.execute( + select(FundamentalSnapshot).where(FundamentalSnapshot.accession == "Q2A") + )).scalar_one() + codes = (await s.execute(select(SystemEvent.code))).scalars().all() + # Stamped to the issuer that filed, NOT the co-registrant whose file it came from. + assert q2.cik == "0000320193" + assert q2.revenue == 254940 and q2.shares_outstanding == 14687 + assert "coregistrant_recovery" in codes + + +async def test_coregistrant_recovery_rejects_discontinuous_share_count(engine): + """A co-registrant shell's standalone facts must never be stored as the + parent's — a token float is the signature and it has to be refused.""" + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + + incr = _coregistrant_client(_shares("2026-04-17", 100, "Q2A", 2026, "Q2")) + run = await run_import(_importer(incr, today=date(2026, 5, 3)), engine=engine) + + assert run.status == STATUS_FAILED + assert "coregistrant_facts_rejected" in (run.error_details or "") + assert await _count(factory, FundamentalSnapshot) == 2 # nothing recovered + + +async def test_unresolved_filing_stops_blocking_after_retry_window(engine): + """A filing SEC has misfiled must not wedge every later import forever.""" + factory = _factory(engine) + await _seed(factory, ["AAPL"]) + backfill = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, + submissions={320193: _submissions(SUB_FILINGS)}, + latest_index=date(2026, 1, 31), + ) + await run_import(_importer(backfill), engine=engine) + + incr = FakeSecClient( + tickers={"AAPL": 320193}, + companyfacts={320193: _companyfacts([CF_K, CF_Q1], [SH_K, SH_Q1])}, # no GHOST + submissions={320193: _submissions(SUB_FILINGS + [ + _filing("GHOST", "10-Q", "2026-03-28", "2026-05-01", "2026-05-01T10:01:00.000Z")])}, + latest_index=date(2026, 5, 2), + daily={date(2026, 5, 1): [{"form": "10-Q", "cik": 320193, "accession": "GHOST"}]}, + ) + # 9 days after the index date — well past the retry window. + run = await run_import(_importer(incr, today=date(2026, 5, 10)), engine=engine) + + assert run.status == STATUS_PROMOTED # promoted around it, not blocked by it + assert run.source_max_date == date(2026, 5, 2) # and the index advances + summary = json.loads(run.validation_json or "{}") + assert summary["missing_xbrl_count"] == 1 and summary["missing_xbrl_blocking"] == 0 + + async with factory() as s: + events = (await s.execute(select(SystemEvent))).scalars().all() + unresolved = [e for e in events if e.code == "unresolved_filing"] + assert len(unresolved) == 1 and "GHOST" in unresolved[0].message + + async def test_non_xbrl_amendment_skipped_not_failed(engine): factory = _factory(engine) await _seed(factory, ["AAPL"])