spb/hfmarketdata

1"""Ingestion guards: co-registrant filings (shared accessions), poller termination (form filter, Redis seen-set,2bounded retries), failure samples, conditional statement rewrite, coverage dictionary, schema migrations."""3from __future__ import annotations45import types6from datetime import date78import httpx9import pytest10from sqlalchemy import delete, select1112PARENT, SUB = 65984, 66901            # Entergy Corp + Entergy Arkansas: one 10-Q/8-K, two registrants13SHARED = "0000065984-26-000283"14OTHER = "0000065984-26-000290"151617def _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>")303132class FakeClient:33    """Minimal EdgarClient stand-in for poll_new_filings: canned Atom feed, optional daily index."""3435    def __init__(self, atom: str, index: str | None = None):36        self.atom, self.index = atom, index37        self.stats = types.SimpleNamespace(requests=0)38        self.index_calls: list[date] = []3940    def atom_current(self, form: str = "") -> str:41        self.stats.requests += 142        return self.atom4344    def daily_index(self, d: date) -> str | None:45        self.stats.requests += 146        self.index_calls.append(d)47        return self.index484950@pytest.fixture51def cofilers(fundamentals_data):52    """Two temporary tracked companies sharing accessions; everything they touch is removed afterwards."""53    from core.db import session54    from fundamentals import ingest55    from fundamentals.models import EdgarCompany, EdgarFiling, FundIngestState56    from stream.broker import get_redis57    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 None62        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_statements68        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)777879def _stub(monkeypatch, outcome):80    """Replace ingest_company with a recorder; `outcome(cik)` → error string or None."""81    from fundamentals import ingest82    calls: list[int] = []8384    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 None89        return res90    monkeypatch.setattr(ingest, "ingest_company", fake)91    return calls929394# ------------------------------------------------------------------------------------- co-registrants95def test_shared_accession_is_one_row_per_registrant(cofilers):96    import pandas as pd9798    from fundamentals import ingest99    from fundamentals import normalize as N100    from fundamentals import service as S101    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()) == 2105    assert ingest._upsert_filings(SUB, {SHARED: shared}, pd.DataFrame()) == 1          # no IntegrityError any more106    assert ingest._upsert_filings(SUB, {SHARED: shared}, pd.DataFrame()) == 0          # idempotent107    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 links111    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]) == 2116    assert len(S.filings("EAL", form=None, date_from=None, date_to=None)[0]) == 1117118119def test_poll_ingests_each_coregistrant_once(cofilers, monkeypatch):120    from fundamentals import ingest121    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: ignored124                  ("8-K", "Nobody Inc", 999999999, "0000999999-26-000001")])      # not a tracked CIK125    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) == 2130    # the stub recorded nothing in edgar_filings: without the Redis guard the same pairs would be "new" for ever131    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) == 2133    assert cofilers["redis"].scard(ingest.SEEN_KEY) == 2 and cofilers["redis"].ttl(ingest.SEEN_KEY) > 6 * 86_400134135136def 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 ingest139    from fundamentals.edgar_client import TRACKED_FORMS140    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]145146147def test_transient_failures_are_retried_then_abandoned(cofilers, monkeypatch):148    from core.db import session149    from fundamentals import ingest150    from fundamentals.models import FundIngestState151    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 0156    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 seen158    assert calls.count(SUB) == 1 and calls.count(PARENT) == ingest.MAX_ATTEMPTS159    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_ATTEMPTS164        # samples survive the clean 4th cycle and carry the CIK165        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 seen167168169def test_no_facts_is_permanent_not_a_failure(cofilers, monkeypatch):170    from core.db import session171    from fundamentals import ingest172    from fundamentals.models import FundIngestState173    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 0178    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) == failures0183184185def 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 ingest188    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_FACTS193    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 == 2197    assert ingest._known_accns(SUB) == {SHARED, OTHER}198199200# ------------------------------------------------------------------------------------- daily index201def test_business_days_back():202    from fundamentals.ingest import business_days_back203    assert business_days_back(date(2026, 9, 6)) == [date(2026, 9, 4), date(2026, 9, 3)]      # Sunday → Fri, Thu204    assert business_days_back(date(2026, 9, 7)) == [date(2026, 9, 7), date(2026, 9, 4)]      # Monday → Mon, Fri205    assert business_days_back(date(2026, 9, 9), 3) == [date(2026, 9, 9), date(2026, 9, 8), date(2026, 9, 7)]206207208def 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 ec211    hits: list[str] = []212213    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 None223    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_index226    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")230231232# ------------------------------------------------------------------------------------- statements / coverage233def test_replace_statements_only_rewrites_on_change(fundamentals_data):234    from core.db import session235    from fundamentals import ingest236    from fundamentals import service as S237    from fundamentals.models import fund_statements238    cik = 320193239    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 → untouched243    with session() as s:244        assert sorted(s.scalars(select(fund_statements.c.id).where(fund_statements.c.cik == cik))) == ids0245    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.0248    assert ingest._replace_statements(cik, changed) is True249    again = S._rows(cik)250    assert sorted(r["id"] for r in again) != ids0251    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 set254        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 tests257    assert S._rows(cik)[0]["coverage"] == original[0]["coverage"]258259260def test_coverage_dictionary_interns_and_hydrates(app):261    from core.db import session262    from fundamentals import coverage_store as CS263    from fundamentals.models import FundCoverageBlob264    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 order266    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-inserted270    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 reference274            {"accn": "y", "coverage": {"revenue": {"tag": "us-gaap:Revenues"}}, "coverage_id": None},   # legacy inline275            {"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)}280281282def test_coverage_intern_survives_cache_eviction(app, monkeypatch):283    """Prod 2026-09-06: the 50 533rd distinct document raised KeyError because _remember() wiped the cache284    (CACHE_MAX) in the middle of intern(); ids must be resolved locally, not through the cache."""285    from fundamentals import coverage_store as CS286    CS.reset_cache_for_tests()287    monkeypatch.setattr(CS, "CACHE_MAX", 5)288    docs = [{"k%d" % i: {"reason": "r%d" % i}} for i in range(20)] + [{"k0": {"reason": "r0"}}]289    ids = CS.intern(docs)290    assert len(ids) == 21 and len(set(ids[:20])) == 20 and ids[20] == ids[0]291    assert CS.intern(docs) == ids                                          # second pass: same ids, no KeyError292293294def test_failure_samples_append_and_cap(app):295    from core.db import session296    from fundamentals import ingest297    from fundamentals.models import FundIngestState298    key = "test_samples"299    try:300        ingest._set_state(key, failures_add=2, failure_samples_add=[f"e{i}" for i in range(12)])301        ingest._set_state(key, last_run_at=None)                          # a clean cycle keeps the evidence302        with session() as s:303            st = s.get(FundIngestState, key)304            assert st.failures == 2 and st.failure_samples == [f"e{i}" for i in range(2, 12)]305        ingest._set_state(key, failure_samples=[])                        # explicit replace still possible306        with session() as s:307            assert s.get(FundIngestState, key).failure_samples == []308    finally:309        with session() as s:310            s.execute(delete(FundIngestState).where(FundIngestState.key == key))311312313# ------------------------------------------------------------------------------------- migrations314def test_migrate_filings_pk_from_legacy_table(tmp_path):315    from sqlalchemy import create_engine, text316317    from fundamentals import migrations318    eng = create_engine(f"sqlite:///{tmp_path / 'legacy.db'}")319    with eng.begin() as con:320        con.execute(text("""CREATE TABLE edgar_filings (accn VARCHAR(24) NOT NULL, cik INTEGER NOT NULL, form VARCHAR(16) NOT NULL,321                            filed_date DATE NOT NULL, period_of_report DATE, primary_doc VARCHAR(512), is_amendment BOOLEAN NOT NULL,322                            is_xbrl BOOLEAN NOT NULL, parsed_at DATETIME, PRIMARY KEY (accn))"""))323        con.execute(text("CREATE INDEX ix_edgar_filings_cik ON edgar_filings (cik)"))324        con.execute(text("INSERT INTO edgar_filings VALUES ('0000065984-26-000283', 65984, '10-Q', '2026-07-31', '2026-06-30', NULL, 0, 1, NULL)"))325        con.execute(text("INSERT INTO edgar_filings VALUES ('0000320193-24-000069', 320193, '10-Q', '2024-05-03', '2024-03-30', NULL, 0, 1, NULL)"))326        assert migrations.filings_pk_is_composite(con) is False327        assert migrations.migrate_filings_pk(con) == 2328        assert migrations.filings_pk_is_composite(con) is True329        # the subsidiary's copy of the shared accession now fits330        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)"))331        assert con.execute(text("SELECT count(*) FROM edgar_filings WHERE accn='0000065984-26-000283'")).scalar() == 2332        names = {r[0] for r in con.execute(text("SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='edgar_filings'"))}333        assert {"ix_edgar_filings_accn", "ix_edgar_filings_cik_filed", "ix_edgar_filings_form", "ix_edgar_filings_filed_date"} <= names334        assert migrations.migrate_filings_pk(con) == 3                     # idempotent rerun keeps every row335336337def test_startup_schema_is_idempotent(fundamentals_data):338    from core.db import engine339    from fundamentals import migrations340    assert migrations.ensure_filings_schema() == "ok"341    assert migrations.ensure_statements_schema() is False342    assert migrations.ensure_auto_vacuum() is False                        # already switched when the test DB was young343    with engine.connect() as con:344        from sqlalchemy import text345        assert int(con.execute(text("PRAGMA auto_vacuum")).scalar()) == 2346    migrations.incremental_vacuum()347