You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
139 lines
4.7 KiB
Python
139 lines
4.7 KiB
Python
"""
|
|
Bootstrap Form 4 insider transactions by iterating over all universe tickers.
|
|
|
|
Uses index_form4s(ticker, days=730) — the proven submissions-API path — rather
|
|
than the company.idx + per-accession Archives index.json approach (which is
|
|
aggressively rate-limited by SEC EDGAR).
|
|
|
|
Usage (inside the container):
|
|
python scripts/bootstrap_form4_by_ticker.py [--resume-from TICKER]
|
|
|
|
Options:
|
|
--resume-from TICKER Skip tickers alphabetically before TICKER (for resuming
|
|
after an interruption or rate-limit pause).
|
|
--days N Days of history to fetch per ticker (default: 730).
|
|
--delay-ms N Per-ticker delay in milliseconds (default: 300).
|
|
|
|
Idempotent: already-indexed accessions are skipped via upsert ON CONFLICT DO NOTHING.
|
|
"""
|
|
|
|
import asyncio
|
|
import fcntl
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import text
|
|
|
|
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_ticker")
|
|
|
|
_LOCK_FILE = "/tmp/bootstrap_form4_ticker.lock"
|
|
|
|
|
|
async def bootstrap(
|
|
resume_from: Optional[str] = None,
|
|
days: int = 730,
|
|
delay_ms: int = 300,
|
|
) -> None:
|
|
from app.core.database import AsyncSessionLocal
|
|
from app.services.insider_transaction_service import InsiderTransactionService
|
|
from sqlalchemy import text
|
|
|
|
txn_svc = InsiderTransactionService()
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
rows = await db.execute(
|
|
text("SELECT ticker FROM universe_ticker_registry ORDER BY ticker")
|
|
)
|
|
tickers = [r[0] for r in rows.fetchall()]
|
|
|
|
total = len(tickers)
|
|
logger.info(f"Universe: {total} tickers, fetching {days} days of Form 4 history each")
|
|
|
|
if resume_from:
|
|
resume_from = resume_from.upper()
|
|
start_idx = next((i for i, t in enumerate(tickers) if t >= resume_from), 0)
|
|
logger.info(f"Resuming from {resume_from} (index {start_idx}/{total})")
|
|
tickers = tickers[start_idx:]
|
|
|
|
inserted_total = 0
|
|
errors = 0
|
|
delay_s = delay_ms / 1000.0
|
|
_SESSION_RECYCLE = 50 # close/reopen aiohttp session every N tickers
|
|
|
|
for i, ticker in enumerate(tickers, 1):
|
|
# Recycle the aiohttp session periodically to release connection pool memory
|
|
if i % _SESSION_RECYCLE == 1 and i > 1:
|
|
await txn_svc._http.close()
|
|
txn_svc = InsiderTransactionService()
|
|
import gc
|
|
gc.collect()
|
|
logger.info(f"Session recycled at ticker {i} to free memory")
|
|
|
|
# SECHttpClient has unbounded _text_cache and _json_cache — clear before each ticker
|
|
# (disk cache still provides persistence across calls)
|
|
txn_svc._http._text_cache.clear()
|
|
txn_svc._http._json_cache.clear()
|
|
|
|
try:
|
|
async with AsyncSessionLocal() as db:
|
|
# Disable idle-in-transaction timeout: HTTP fetches for 730 days of
|
|
# filings can take several minutes, exceeding the default 90s limit.
|
|
await db.execute(text("SET SESSION idle_in_transaction_session_timeout = 0"))
|
|
await db.commit()
|
|
n = await txn_svc.index_form4s(db, ticker, days=days, force_refresh=False)
|
|
inserted_total += n
|
|
except Exception as e:
|
|
errors += 1
|
|
logger.warning(f"[{i}/{len(tickers)}] {ticker}: error — {e}")
|
|
|
|
if i % 100 == 0 or i == len(tickers):
|
|
logger.info(
|
|
f"Progress: {i}/{len(tickers)} tickers processed, "
|
|
f"{inserted_total} transactions inserted, {errors} errors"
|
|
)
|
|
|
|
if delay_s > 0:
|
|
await asyncio.sleep(delay_s)
|
|
|
|
logger.info(
|
|
f"Bootstrap complete: {len(tickers)} tickers, "
|
|
f"{inserted_total} total transactions inserted, {errors} errors"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--resume-from", default=None)
|
|
parser.add_argument("--days", type=int, default=730)
|
|
parser.add_argument("--delay-ms", type=int, default=300)
|
|
args = parser.parse_args()
|
|
|
|
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(
|
|
resume_from=args.resume_from,
|
|
days=args.days,
|
|
delay_ms=args.delay_ms,
|
|
))
|
|
finally:
|
|
fcntl.flock(lock_fh, fcntl.LOCK_UN)
|
|
lock_fh.close()
|
|
try:
|
|
os.unlink(_LOCK_FILE)
|
|
except OSError:
|
|
pass
|