spb/hfmarketdata

1#!/usr/bin/env python32"""One-off migration: `edgar_filings` primary key `accn` → `(cik, accn)` (co-registrants share accessions).34    # production (M3U96b) — BEFORE deploying the code that expects the composite key; stop the writers first:5    pm2 stop hfmarketdata-edgar-incremental hfmarketdata-edgar-backfill6    cd ~/hfmarketdata && HFMD_DATA_ROOT=~/firstratedata venv/bin/python scripts/migrate_edgar_filings.py7    pm2 start hfmarketdata-edgar-incremental89The API may keep serving during the copy (WAL: readers are not blocked; a write to edgar_filings would wait on10the busy timeout, hence stopping the ingestion jobs). The table is rebuilt in ONE transaction11(`edgar_filings_new` ← INSERT OR IGNORE, drop, rename, 4 indexes) — on failure nothing changes. Idempotent:12running it on a migrated database prints the state and exits 0. `--check` only reports.1314Expected duration: 1.37 M rows ≈ 212 MB → copy ~5–10 s + indexes ~10–20 s on the M3 Ultra SSD (≈ 30–60 s15worst case); the WAL grows by ~300 MB and is checkpointed at the end.1617Author: Simon-Pierre Boucher <contact@spboucher.ai>18"""19from __future__ import annotations2021import argparse22import logging23import sys24import time25from pathlib import Path2627HERE = Path(__file__).resolve().parent28sys.path.insert(0, str(HERE.parent / "hfmarketdata" / "api"))2930log = logging.getLogger("migrate_edgar_filings")313233def 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")3940    from sqlalchemy import text4142    from core.config import settings43    from core.db import engine44    from fundamentals import migrations4546    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 050    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 053    if state:54        log.info("edgar_filings already keyed on (cik, accn) — %d rows, nothing to do", n)55        return 056    log.info("edgar_filings keyed on accn alone: %d rows to copy", n)57    if args.check:58        return 259    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 1686970if __name__ == "__main__":71    sys.exit(main())72