ilia
31c656d66f
Some checks failed
CI / security-scan (pull_request) Failing after 28s
CI / lint-and-test (pull_request) Failing after 31s
CI / dependency-scan (pull_request) Successful in 22s
CI / docker-build-test (pull_request) Failing after 34s
CI / workflow-summary (pull_request) Successful in 4s
Switch HouseWatcherClient to public STOCK Act JSON (legacy housestockwatcher.com is down), add httpx/scikit-learn to pyproject, wire SMTP settings through Settings, fix get_session() as a context manager, and use savepoints in TradeLoader so one bad row does not roll back the batch.
93 lines
3.2 KiB
Python
Executable File
93 lines
3.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Fetch recent congressional trades from House Stock Watcher and ingest into DB.
|
|
Usage: python scripts/fetch_congressional_trades.py [--days N] [--limit N]
|
|
"""
|
|
|
|
import argparse
|
|
import logging
|
|
|
|
from pote.db import SessionLocal
|
|
from pote.ingestion.house_watcher import HouseWatcherClient
|
|
from pote.ingestion.trade_loader import TradeLoader
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def main():
|
|
"""Fetch and ingest congressional trades."""
|
|
parser = argparse.ArgumentParser(description="Fetch congressional trades (free API)")
|
|
parser.add_argument(
|
|
"--days", type=int, default=30, help="Number of days to look back (default: 30)"
|
|
)
|
|
parser.add_argument(
|
|
"--limit", type=int, default=None, help="Maximum number of transactions to fetch"
|
|
)
|
|
parser.add_argument("--all", action="store_true", help="Fetch all transactions (ignore --days)")
|
|
|
|
args = parser.parse_args()
|
|
|
|
logger.info("=== Fetching Congressional Trades (public STOCK Act data) ===")
|
|
logger.info("Source: public JSON feeds (see HouseWatcherClient.data_urls)")
|
|
|
|
try:
|
|
with HouseWatcherClient() as client:
|
|
if args.all:
|
|
logger.info(f"Fetching all transactions (limit={args.limit})")
|
|
transactions = client.fetch_all_transactions(limit=args.limit)
|
|
else:
|
|
logger.info(f"Fetching transactions from last {args.days} days")
|
|
transactions = client.fetch_recent_transactions(days=args.days)
|
|
|
|
if args.limit:
|
|
transactions = transactions[: args.limit]
|
|
|
|
if not transactions:
|
|
logger.warning("No transactions fetched!")
|
|
return
|
|
|
|
logger.info(f"Fetched {len(transactions)} transactions")
|
|
|
|
# Show sample
|
|
logger.info("\nSample transaction:")
|
|
sample = transactions[0]
|
|
for key, val in sample.items():
|
|
logger.info(f" {key}: {val}")
|
|
|
|
# Ingest into database
|
|
logger.info("\n=== Ingesting into database ===")
|
|
with SessionLocal() as session:
|
|
loader = TradeLoader(session)
|
|
counts = loader.ingest_transactions(transactions)
|
|
|
|
logger.info("\n=== Summary ===")
|
|
logger.info(f"✓ Officials created/updated: {counts['officials']}")
|
|
logger.info(f"✓ Securities created/updated: {counts['securities']}")
|
|
logger.info(f"✓ Trades ingested: {counts['trades']}")
|
|
|
|
# Query some stats
|
|
with SessionLocal() as session:
|
|
from sqlalchemy import func, select
|
|
|
|
from pote.db.models import Official, Trade
|
|
|
|
total_trades = session.scalar(select(func.count(Trade.id)))
|
|
total_officials = session.scalar(select(func.count(Official.id)))
|
|
|
|
logger.info("\nDatabase totals:")
|
|
logger.info(f" Total officials: {total_officials}")
|
|
logger.info(f" Total trades: {total_trades}")
|
|
|
|
logger.info("\n✅ Done!")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch/ingest trades: {e}", exc_info=True)
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit(main())
|