""" One-time bootstrap: ingest Form 4 + SC 13D/G for the last N quarters. Usage (inside the container): python scripts/bootstrap_form4_2y.py Quarters bootstrapped = settings.SEC_FORM4_BOOTSTRAP_QUARTERS (default 8 = 2 years). Idempotent: already-indexed accessions are skipped via upsert ON CONFLICT DO NOTHING. """ import asyncio import fcntl import logging import os import sys import tempfile from datetime import datetime, timezone # Ensure project root is on sys.path when invoked directly sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger("bootstrap") _LOCK_FILE = "/tmp/bootstrap_form4.lock" # Use Archives endpoint for probe — browse-edgar can be 200 while Archives is still 429 _SEC_PROBE_URL = "https://www.sec.gov/Archives/edgar/full-index/2026/QTR1/company.idx" def _quarters_to_process(n: int): """Return list of (year, quarter) tuples for the last n quarters.""" now = datetime.now(timezone.utc) current_q = (now.month - 1) // 3 + 1 results = [] year, q = now.year, current_q for _ in range(n): results.append((year, q)) q -= 1 if q == 0: q = 4 year -= 1 return results async def _check_sec_available() -> bool: """Return True if SEC EDGAR is responding (not rate-limiting us).""" import aiohttp headers = {"User-Agent": "Stock Oracle bootstrap@stockoracle.internal"} try: async with aiohttp.ClientSession() as session: async with session.get( _SEC_PROBE_URL, headers=headers, timeout=aiohttp.ClientTimeout(total=10), allow_redirects=True, ) as resp: if resp.status == 429: logger.warning(f"SEC EDGAR is rate-limiting us (429). Try again in 30+ minutes.") return False return resp.status < 500 except Exception as e: logger.warning(f"SEC EDGAR probe failed: {e}") return False async def bootstrap() -> None: from app.core.config import settings from app.core.database import AsyncSessionLocal from app.services.sec_full_index_service import SECFullIndexService from app.services.insider_transaction_service import InsiderTransactionService from app.services.activist_ownership_service import ActivistOwnershipService # Pre-flight: abort if SEC is rate-limiting us to avoid wasted retries if not await _check_sec_available(): logger.error("Aborting bootstrap — SEC EDGAR is rate-limiting this IP. Re-run in 30-60 minutes.") return n_quarters = settings.SEC_FORM4_BOOTSTRAP_QUARTERS quarters = _quarters_to_process(n_quarters) logger.info(f"Bootstrapping {n_quarters} quarters: {quarters}") index_svc = SECFullIndexService() txn_svc = InsiderTransactionService() activist_svc = ActivistOwnershipService() for year, quarter in quarters: logger.info(f"Processing {year}/Q{quarter} ...") # Form 4 via form345.zip zip_path = os.path.join(tempfile.gettempdir(), f"form345_{year}_Q{quarter}.zip") if not os.path.exists(zip_path): logger.info(f" Downloading form345.zip for {year}/Q{quarter} ...") try: await index_svc._http.fetch_form345_zip(year, quarter, zip_path) except Exception as e: logger.warning(f" form345.zip download failed: {e}; falling back to company.idx") zip_path = None form4_entries = [] if zip_path and os.path.exists(zip_path): from app.services.sec_full_index_service import parse_form345_zip, FORM4_TYPES form4_entries = parse_form345_zip(zip_path, form_types=FORM4_TYPES) logger.info(f" Form 4 entries from zip: {len(form4_entries)}") else: form4_entries = await index_svc.fetch_quarterly_form4_entries(year, quarter) logger.info(f" Form 4 entries from company.idx: {len(form4_entries)}") async with AsyncSessionLocal() as db: inserted = await txn_svc.index_form4_from_index_entries(db, form4_entries) logger.info(f" Form 4 inserted: {inserted}") # 13D/G via company.idx activist_entries = await index_svc.fetch_quarterly_activist_entries(year, quarter) logger.info(f" 13D/G entries from company.idx: {len(activist_entries)}") async with AsyncSessionLocal() as db: inserted = await activist_svc.ingest_from_index_entries(db, activist_entries) logger.info(f" 13D/G index-only inserted: {inserted}") logger.info("Bootstrap complete. Run the activist enrich job to populate ownership_pct.") if __name__ == "__main__": # Prevent multiple concurrent runs via file lock lock_fh = open(_LOCK_FILE, "w") try: fcntl.flock(lock_fh, fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: logger.error("Bootstrap is already running (lock file held). Exiting.") sys.exit(1) try: asyncio.run(bootstrap()) finally: fcntl.flock(lock_fh, fcntl.LOCK_UN) lock_fh.close() try: os.unlink(_LOCK_FILE) except OSError: pass