148 lines
5.3 KiB
Python
148 lines
5.3 KiB
Python
"""SEC filing retry queue and setup-quality gate
|
|
|
|
Revision ID: 028
|
|
Revises: 027
|
|
Create Date: 2026-08-03 00:00:00.000000
|
|
"""
|
|
from datetime import date, datetime, timezone
|
|
import json
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
revision: str = "028"
|
|
down_revision: Union[str, None] = "027"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
op.create_table(
|
|
"sec_filing_gaps",
|
|
sa.Column("id", sa.Integer(), primary_key=True),
|
|
sa.Column("cik", sa.String(length=10), nullable=False),
|
|
sa.Column("accession", sa.String(length=25), nullable=False),
|
|
sa.Column("form", sa.String(length=12), nullable=True),
|
|
sa.Column("index_date", sa.Date(), nullable=True),
|
|
sa.Column("reason", sa.String(length=64), nullable=False),
|
|
sa.Column("coregistrant_ciks_json", sa.Text(), nullable=True),
|
|
sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False),
|
|
sa.Column("last_attempted_at", sa.DateTime(timezone=True), nullable=False),
|
|
sa.Column("escalated_at", sa.DateTime(timezone=True), nullable=True),
|
|
sa.UniqueConstraint("accession", name="uq_sec_filing_gaps_accession"),
|
|
)
|
|
op.create_index("ix_sec_filing_gaps_cik", "sec_filing_gaps", ["cik"])
|
|
_backfill_retry_queue()
|
|
|
|
|
|
def downgrade() -> None:
|
|
op.drop_index("ix_sec_filing_gaps_cik", table_name="sec_filing_gaps")
|
|
op.drop_table("sec_filing_gaps")
|
|
|
|
|
|
def _as_date(value) -> date | None:
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
return date.fromisoformat(value)
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _backfill_retry_queue() -> None:
|
|
"""Materialize pre-queue promoted gaps once; runtime never scans history."""
|
|
bind = op.get_bind()
|
|
runs = sa.table(
|
|
"data_import_runs",
|
|
sa.column("source", sa.String()),
|
|
sa.column("status", sa.String()),
|
|
sa.column("validation_json", sa.Text()),
|
|
sa.column("source_max_date", sa.Date()),
|
|
sa.column("started_at", sa.DateTime(timezone=True)),
|
|
)
|
|
snapshots = sa.table(
|
|
"fundamental_snapshots",
|
|
sa.column("cik", sa.String()),
|
|
sa.column("accession", sa.String()),
|
|
sa.column("filed_date", sa.Date()),
|
|
)
|
|
gaps = sa.table(
|
|
"sec_filing_gaps",
|
|
sa.column("cik", sa.String()),
|
|
sa.column("accession", sa.String()),
|
|
sa.column("form", sa.String()),
|
|
sa.column("index_date", sa.Date()),
|
|
sa.column("reason", sa.String()),
|
|
sa.column("coregistrant_ciks_json", sa.Text()),
|
|
sa.column("first_seen_at", sa.DateTime(timezone=True)),
|
|
sa.column("last_attempted_at", sa.DateTime(timezone=True)),
|
|
sa.column("escalated_at", sa.DateTime(timezone=True)),
|
|
)
|
|
|
|
snapshot_rows = bind.execute(
|
|
sa.select(snapshots.c.cik, snapshots.c.accession, snapshots.c.filed_date)
|
|
).all()
|
|
resolved_accessions = {row.accession for row in snapshot_rows}
|
|
latest_filed_by_cik: dict[str, date] = {}
|
|
for row in snapshot_rows:
|
|
if row.filed_date is not None:
|
|
current = latest_filed_by_cik.get(row.cik)
|
|
if current is None or row.filed_date > current:
|
|
latest_filed_by_cik[row.cik] = row.filed_date
|
|
|
|
audit_rows = bind.execute(
|
|
sa.select(
|
|
runs.c.validation_json,
|
|
runs.c.source_max_date,
|
|
runs.c.started_at,
|
|
).where(
|
|
runs.c.source == "sec_facts",
|
|
runs.c.status == "promoted",
|
|
runs.c.validation_json.is_not(None),
|
|
)
|
|
).all()
|
|
now = datetime.now(timezone.utc)
|
|
candidates: dict[str, dict] = {}
|
|
for audit in audit_rows:
|
|
try:
|
|
summary = json.loads(audit.validation_json)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if not isinstance(summary, dict):
|
|
continue
|
|
for item in summary.get("missing_xbrl") or []:
|
|
accession = item.get("accession")
|
|
raw_cik = item.get("cik")
|
|
if not accession or raw_cik is None or accession in resolved_accessions:
|
|
continue
|
|
cik = str(raw_cik).zfill(10)
|
|
index_date = _as_date(item.get("index_date")) or _as_date(
|
|
audit.source_max_date
|
|
)
|
|
later_filed = latest_filed_by_cik.get(cik)
|
|
if index_date is not None and later_filed is not None and later_filed > index_date:
|
|
continue
|
|
first_seen = audit.started_at or now
|
|
existing = candidates.get(accession)
|
|
if existing is not None and existing["first_seen_at"] <= first_seen:
|
|
continue
|
|
candidates[accession] = {
|
|
"cik": cik,
|
|
"accession": accession,
|
|
"form": item.get("form"),
|
|
"index_date": index_date,
|
|
"reason": item.get("reason") or "not_in_companyfacts",
|
|
"coregistrant_ciks_json": json.dumps(item.get("coregistrants") or []),
|
|
"first_seen_at": first_seen,
|
|
"last_attempted_at": first_seen,
|
|
"escalated_at": None,
|
|
}
|
|
if candidates:
|
|
op.bulk_insert(gaps, list(candidates.values()))
|