Initial commit: POTE Phase 1 complete
- PR1: Project scaffold, DB models, price loader - PR2: Congressional trade ingestion (House Stock Watcher) - PR3: Security enrichment + deployment infrastructure - 37 passing tests, 87%+ coverage - Docker + Proxmox deployment ready - Complete documentation - Works 100% offline with fixtures
This commit is contained in:
ilia
@@ -0,0 +1,8 @@
|
||||
"""
|
||||
POTE – Public Officials Trading Explorer
|
||||
|
||||
A research-only tool for tracking and analyzing public stock trades
|
||||
by government officials. Not for investment advice.
|
||||
"""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@@ -0,0 +1,39 @@
|
||||
"""
|
||||
Configuration management using pydantic-settings.
|
||||
Loads from environment variables and .env file.
|
||||
"""
|
||||
|
||||
from pydantic import Field
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""Application settings."""
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
env_file=".env",
|
||||
env_file_encoding="utf-8",
|
||||
case_sensitive=False,
|
||||
extra="ignore",
|
||||
)
|
||||
|
||||
# Database
|
||||
database_url: str = Field(
|
||||
default="sqlite:///./pote.db",
|
||||
description="SQLAlchemy database URL",
|
||||
)
|
||||
|
||||
# API keys
|
||||
quiverquant_api_key: str = Field(default="", description="QuiverQuant API key")
|
||||
fmp_api_key: str = Field(default="", description="Financial Modeling Prep API key")
|
||||
|
||||
# Logging
|
||||
log_level: str = Field(default="INFO", description="Log level (DEBUG, INFO, WARNING, ERROR)")
|
||||
|
||||
# Application
|
||||
app_name: str = "POTE"
|
||||
app_version: str = "0.1.0"
|
||||
|
||||
|
||||
# Global settings instance
|
||||
settings = Settings()
|
||||
@@ -0,0 +1,40 @@
|
||||
"""
|
||||
Database layer: engine, session factory, and base model.
|
||||
"""
|
||||
|
||||
from collections.abc import Generator
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
|
||||
|
||||
from pote.config import settings
|
||||
|
||||
# Create engine
|
||||
engine = create_engine(
|
||||
settings.database_url,
|
||||
echo=settings.log_level == "DEBUG",
|
||||
connect_args={"check_same_thread": False} if "sqlite" in settings.database_url else {},
|
||||
)
|
||||
|
||||
# Session factory
|
||||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
|
||||
class Base(DeclarativeBase):
|
||||
"""Base class for all models."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
def get_session() -> Generator[Session, None, None]:
|
||||
"""Get a database session (use as a context manager or dependency)."""
|
||||
session = SessionLocal()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
def init_db() -> None:
|
||||
"""Create all tables. Use Alembic migrations in production."""
|
||||
Base.metadata.create_all(bind=engine)
|
||||
@@ -0,0 +1,220 @@
|
||||
"""
|
||||
SQLAlchemy ORM models for POTE.
|
||||
Matches the schema defined in docs/02_data_model.md.
|
||||
"""
|
||||
|
||||
from datetime import date, datetime, timezone
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import (
|
||||
DECIMAL,
|
||||
Date,
|
||||
DateTime,
|
||||
ForeignKey,
|
||||
Index,
|
||||
Integer,
|
||||
String,
|
||||
Text,
|
||||
UniqueConstraint,
|
||||
)
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from pote.db import Base
|
||||
|
||||
|
||||
class Official(Base):
|
||||
"""Government officials (Congress members, etc.)."""
|
||||
|
||||
__tablename__ = "officials"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
name: Mapped[str] = mapped_column(String(200), nullable=False, index=True)
|
||||
chamber: Mapped[str | None] = mapped_column(String(50)) # "House", "Senate", etc.
|
||||
party: Mapped[str | None] = mapped_column(String(50))
|
||||
state: Mapped[str | None] = mapped_column(String(2))
|
||||
bioguide_id: Mapped[str | None] = mapped_column(String(20), unique=True)
|
||||
external_ids: Mapped[str | None] = mapped_column(Text) # JSON blob for other IDs
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime, default=lambda: datetime.now(timezone.utc)
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime,
|
||||
default=lambda: datetime.now(timezone.utc),
|
||||
onupdate=lambda: datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
# Relationships
|
||||
trades: Mapped[list["Trade"]] = relationship("Trade", back_populates="official")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<Official(id={self.id}, name='{self.name}', chamber='{self.chamber}')>"
|
||||
|
||||
|
||||
class Security(Base):
|
||||
"""Securities (stocks, bonds, etc.)."""
|
||||
|
||||
__tablename__ = "securities"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
ticker: Mapped[str] = mapped_column(String(20), nullable=False, unique=True, index=True)
|
||||
name: Mapped[str | None] = mapped_column(String(200))
|
||||
exchange: Mapped[str | None] = mapped_column(String(50))
|
||||
sector: Mapped[str | None] = mapped_column(String(100))
|
||||
industry: Mapped[str | None] = mapped_column(String(100))
|
||||
asset_type: Mapped[str] = mapped_column(String(50), default="stock") # stock, bond, etc.
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime, default=lambda: datetime.now(timezone.utc)
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime,
|
||||
default=lambda: datetime.now(timezone.utc),
|
||||
onupdate=lambda: datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
# Relationships
|
||||
trades: Mapped[list["Trade"]] = relationship("Trade", back_populates="security")
|
||||
prices: Mapped[list["Price"]] = relationship("Price", back_populates="security")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<Security(id={self.id}, ticker='{self.ticker}', name='{self.name}')>"
|
||||
|
||||
|
||||
class Trade(Base):
|
||||
"""Trades disclosed by officials."""
|
||||
|
||||
__tablename__ = "trades"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
official_id: Mapped[int] = mapped_column(ForeignKey("officials.id"), nullable=False, index=True)
|
||||
security_id: Mapped[int] = mapped_column(
|
||||
ForeignKey("securities.id"), nullable=False, index=True
|
||||
)
|
||||
|
||||
# Core trade fields
|
||||
source: Mapped[str] = mapped_column(String(50), nullable=False) # "quiver", "fmp", etc.
|
||||
external_id: Mapped[str | None] = mapped_column(String(100)) # source-specific ID
|
||||
transaction_date: Mapped[date] = mapped_column(Date, nullable=False, index=True)
|
||||
filing_date: Mapped[date | None] = mapped_column(Date, index=True)
|
||||
side: Mapped[str] = mapped_column(String(20), nullable=False) # "buy", "sell", "exchange"
|
||||
|
||||
# Amount (often disclosed as a range)
|
||||
value_min: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 2))
|
||||
value_max: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 2))
|
||||
amount: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 2)) # shares/units if available
|
||||
currency: Mapped[str] = mapped_column(String(3), default="USD")
|
||||
|
||||
# Quality flags (JSON or enum list)
|
||||
quality_flags: Mapped[str | None] = mapped_column(Text) # e.g., "range_only,delayed_filing"
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime, default=lambda: datetime.now(timezone.utc)
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime,
|
||||
default=lambda: datetime.now(timezone.utc),
|
||||
onupdate=lambda: datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
# Relationships
|
||||
official: Mapped["Official"] = relationship("Official", back_populates="trades")
|
||||
security: Mapped["Security"] = relationship("Security", back_populates="trades")
|
||||
|
||||
# Constraints
|
||||
__table_args__ = (
|
||||
Index("ix_trades_official_date", "official_id", "transaction_date"),
|
||||
Index("ix_trades_security_date", "security_id", "transaction_date"),
|
||||
UniqueConstraint(
|
||||
"source", "external_id", name="uq_trades_source_external_id"
|
||||
), # dedup by source ID
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return (
|
||||
f"<Trade(id={self.id}, official_id={self.official_id}, "
|
||||
f"ticker={self.security.ticker if self.security else 'N/A'}, "
|
||||
f"side='{self.side}', date={self.transaction_date})>"
|
||||
)
|
||||
|
||||
|
||||
class Price(Base):
|
||||
"""Daily price data for securities."""
|
||||
|
||||
__tablename__ = "prices"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
security_id: Mapped[int] = mapped_column(
|
||||
ForeignKey("securities.id"), nullable=False, index=True
|
||||
)
|
||||
date: Mapped[date] = mapped_column(Date, nullable=False, index=True)
|
||||
|
||||
open: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 4))
|
||||
high: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 4))
|
||||
low: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 4))
|
||||
close: Mapped[Decimal] = mapped_column(DECIMAL(15, 4), nullable=False)
|
||||
volume: Mapped[int | None] = mapped_column(Integer)
|
||||
adjusted_close: Mapped[Decimal | None] = mapped_column(DECIMAL(15, 4))
|
||||
|
||||
source: Mapped[str] = mapped_column(String(50), default="yfinance")
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime, default=lambda: datetime.now(timezone.utc)
|
||||
)
|
||||
|
||||
# Relationships
|
||||
security: Mapped["Security"] = relationship("Security", back_populates="prices")
|
||||
|
||||
# Constraints
|
||||
__table_args__ = (UniqueConstraint("security_id", "date", name="uq_prices_security_date"),)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<Price(security_id={self.security_id}, date={self.date}, close={self.close})>"
|
||||
|
||||
|
||||
# Future analytics models (stubs for now, will implement in Phase 2)
|
||||
|
||||
|
||||
class MetricOfficial(Base):
|
||||
"""Aggregate metrics per official (Phase 2)."""
|
||||
|
||||
__tablename__ = "metrics_official"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
official_id: Mapped[int] = mapped_column(ForeignKey("officials.id"), nullable=False, index=True)
|
||||
calc_date: Mapped[date] = mapped_column(Date, nullable=False)
|
||||
calc_version: Mapped[str] = mapped_column(String(20), nullable=False)
|
||||
|
||||
# Placeholder metric fields (will expand in Phase 2)
|
||||
trade_count: Mapped[int | None] = mapped_column(Integer)
|
||||
avg_abnormal_return_1m: Mapped[Decimal | None] = mapped_column(DECIMAL(10, 6))
|
||||
cluster_label: Mapped[str | None] = mapped_column(String(50))
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime, default=lambda: datetime.now(timezone.utc)
|
||||
)
|
||||
|
||||
__table_args__ = (
|
||||
UniqueConstraint("official_id", "calc_date", "calc_version", name="uq_metrics_official"),
|
||||
)
|
||||
|
||||
|
||||
class MetricTrade(Base):
|
||||
"""Per-trade metrics (abnormal returns, etc., Phase 2)."""
|
||||
|
||||
__tablename__ = "metrics_trade"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
trade_id: Mapped[int] = mapped_column(ForeignKey("trades.id"), nullable=False, index=True)
|
||||
calc_date: Mapped[date] = mapped_column(Date, nullable=False)
|
||||
calc_version: Mapped[str] = mapped_column(String(20), nullable=False)
|
||||
|
||||
# Placeholder metric fields
|
||||
return_1m: Mapped[Decimal | None] = mapped_column(DECIMAL(10, 6))
|
||||
abnormal_return_1m: Mapped[Decimal | None] = mapped_column(DECIMAL(10, 6))
|
||||
signal_flags: Mapped[str | None] = mapped_column(Text) # JSON list
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime, default=lambda: datetime.now(timezone.utc)
|
||||
)
|
||||
|
||||
__table_args__ = (
|
||||
UniqueConstraint("trade_id", "calc_date", "calc_version", name="uq_metrics_trade"),
|
||||
)
|
||||
@@ -0,0 +1,3 @@
|
||||
"""
|
||||
Data ingestion modules for fetching external data.
|
||||
"""
|
||||
@@ -0,0 +1,187 @@
|
||||
"""
|
||||
House Stock Watcher client for fetching congressional trade data.
|
||||
Free, no API key required - scrapes from housestockwatcher.com
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import date, datetime
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class HouseWatcherClient:
|
||||
"""
|
||||
Client for House Stock Watcher API (free, community-maintained).
|
||||
|
||||
Data source: https://housestockwatcher.com/
|
||||
No authentication required.
|
||||
"""
|
||||
|
||||
BASE_URL = "https://housestockwatcher.com/api"
|
||||
|
||||
def __init__(self, timeout: float = 30.0):
|
||||
"""
|
||||
Initialize the client.
|
||||
|
||||
Args:
|
||||
timeout: Request timeout in seconds
|
||||
"""
|
||||
self.timeout = timeout
|
||||
self._client = httpx.Client(timeout=timeout)
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
self.close()
|
||||
|
||||
def close(self):
|
||||
"""Close the HTTP client."""
|
||||
self._client.close()
|
||||
|
||||
def fetch_all_transactions(self, limit: int | None = None) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Fetch all recent transactions from House Stock Watcher.
|
||||
|
||||
Args:
|
||||
limit: Maximum number of transactions to return (None = all)
|
||||
|
||||
Returns:
|
||||
List of transaction dicts with keys:
|
||||
- representative: Official's name
|
||||
- ticker: Stock ticker symbol
|
||||
- transaction_date: Date of transaction (YYYY-MM-DD)
|
||||
- disclosure_date: Date disclosed (YYYY-MM-DD)
|
||||
- transaction: Type ("Purchase", "Sale", "Exchange", etc.)
|
||||
- amount: Amount range (e.g., "$1,001 - $15,000")
|
||||
- house: Chamber ("House" or "Senate")
|
||||
- district: District (if House)
|
||||
- party: Political party
|
||||
- cap_gains_over_200_usd: Capital gains flag (bool)
|
||||
|
||||
Raises:
|
||||
httpx.HTTPError: If request fails
|
||||
"""
|
||||
url = f"{self.BASE_URL}/all_transactions"
|
||||
logger.info(f"Fetching transactions from {url}")
|
||||
|
||||
try:
|
||||
response = self._client.get(url)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
if not isinstance(data, list):
|
||||
raise ValueError(f"Expected list response, got {type(data)}")
|
||||
|
||||
logger.info(f"Fetched {len(data)} transactions from House Stock Watcher")
|
||||
|
||||
if limit:
|
||||
data = data[:limit]
|
||||
|
||||
return data
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Failed to fetch from House Stock Watcher: {e}")
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error fetching transactions: {e}")
|
||||
raise
|
||||
|
||||
def fetch_recent_transactions(self, days: int = 30) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Fetch transactions from the last N days.
|
||||
|
||||
Args:
|
||||
days: Number of days to look back
|
||||
|
||||
Returns:
|
||||
List of recent transaction dicts
|
||||
"""
|
||||
all_txns = self.fetch_all_transactions()
|
||||
|
||||
cutoff = date.today()
|
||||
# We'll filter on disclosure_date since that's when we'd see them
|
||||
recent = []
|
||||
|
||||
for txn in all_txns:
|
||||
try:
|
||||
disclosure_str = txn.get("disclosure_date", "")
|
||||
if not disclosure_str:
|
||||
continue
|
||||
|
||||
# Parse date (format: "YYYY-MM-DD" or "MM/DD/YYYY")
|
||||
if "/" in disclosure_str:
|
||||
disclosure_date = datetime.strptime(disclosure_str, "%m/%d/%Y").date()
|
||||
else:
|
||||
disclosure_date = datetime.strptime(disclosure_str, "%Y-%m-%d").date()
|
||||
|
||||
if (cutoff - disclosure_date).days <= days:
|
||||
recent.append(txn)
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
logger.warning(f"Failed to parse date '{disclosure_str}': {e}")
|
||||
continue
|
||||
|
||||
logger.info(f"Filtered to {len(recent)} transactions in last {days} days")
|
||||
return recent
|
||||
|
||||
|
||||
def parse_amount_range(amount_str: str) -> tuple[float | None, float | None]:
|
||||
"""
|
||||
Parse amount range string like "$1,001 - $15,000" to (min, max).
|
||||
|
||||
Args:
|
||||
amount_str: Amount string from API
|
||||
|
||||
Returns:
|
||||
Tuple of (min_value, max_value) or (None, None) if unparseable
|
||||
"""
|
||||
if not amount_str or amount_str == "N/A":
|
||||
return (None, None)
|
||||
|
||||
try:
|
||||
# Remove $ and commas
|
||||
clean = amount_str.replace("$", "").replace(",", "")
|
||||
|
||||
# Handle ranges like "1001 - 15000"
|
||||
if " - " in clean:
|
||||
parts = clean.split(" - ")
|
||||
min_val = float(parts[0].strip())
|
||||
max_val = float(parts[1].strip())
|
||||
return (min_val, max_val)
|
||||
|
||||
# Handle single values
|
||||
if clean.strip():
|
||||
val = float(clean.strip())
|
||||
return (val, val)
|
||||
|
||||
except (ValueError, IndexError) as e:
|
||||
logger.warning(f"Failed to parse amount '{amount_str}': {e}")
|
||||
|
||||
return (None, None)
|
||||
|
||||
|
||||
def normalize_transaction_type(txn_type: str) -> str:
|
||||
"""
|
||||
Normalize transaction type to our schema's "side" field.
|
||||
|
||||
Args:
|
||||
txn_type: Transaction type from API (e.g., "Purchase", "Sale")
|
||||
|
||||
Returns:
|
||||
Normalized side: "buy", "sell", or "exchange"
|
||||
"""
|
||||
txn_lower = txn_type.lower().strip()
|
||||
|
||||
if "purchase" in txn_lower or "buy" in txn_lower:
|
||||
return "buy"
|
||||
elif "sale" in txn_lower or "sell" in txn_lower:
|
||||
return "sell"
|
||||
elif "exchange" in txn_lower:
|
||||
return "exchange"
|
||||
else:
|
||||
# Default to the original, lowercased
|
||||
return txn_lower
|
||||
@@ -0,0 +1,209 @@
|
||||
"""
|
||||
Price data loader using yfinance.
|
||||
Fetches daily OHLCV data for securities and stores in the prices table.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from decimal import Decimal
|
||||
|
||||
import pandas as pd
|
||||
import yfinance as yf
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from pote.db.models import Price, Security
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PriceLoader:
|
||||
"""Loads price data from yfinance and stores it in the database."""
|
||||
|
||||
def __init__(self, session: Session):
|
||||
self.session = session
|
||||
|
||||
def fetch_and_store_prices(
|
||||
self,
|
||||
ticker: str,
|
||||
start_date: date | None = None,
|
||||
end_date: date | None = None,
|
||||
force_refresh: bool = False,
|
||||
) -> int:
|
||||
"""
|
||||
Fetch price data for a ticker and store in the database.
|
||||
|
||||
Args:
|
||||
ticker: Stock ticker symbol
|
||||
start_date: Start date for price history (defaults to 1 year ago)
|
||||
end_date: End date for price history (defaults to today)
|
||||
force_refresh: If True, re-fetch even if data exists
|
||||
|
||||
Returns:
|
||||
Number of price records inserted/updated
|
||||
|
||||
Raises:
|
||||
ValueError: If ticker is invalid or security doesn't exist
|
||||
"""
|
||||
# Get or create security
|
||||
security = self._get_or_create_security(ticker)
|
||||
|
||||
# Default date range: last year
|
||||
if end_date is None:
|
||||
end_date = date.today()
|
||||
if start_date is None:
|
||||
start_date = end_date - timedelta(days=365)
|
||||
|
||||
# Check existing data unless force_refresh
|
||||
if not force_refresh:
|
||||
start_date = self._get_missing_date_range_start(security.id, start_date, end_date)
|
||||
if start_date > end_date:
|
||||
logger.info(f"No missing data for {ticker} in range, skipping fetch")
|
||||
return 0
|
||||
|
||||
logger.info(f"Fetching prices for {ticker} from {start_date} to {end_date}")
|
||||
|
||||
# Fetch from yfinance
|
||||
try:
|
||||
df = self._fetch_yfinance_data(ticker, start_date, end_date)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch data for {ticker}: {e}")
|
||||
raise
|
||||
|
||||
if df.empty:
|
||||
logger.warning(f"No data returned for {ticker}")
|
||||
return 0
|
||||
|
||||
# Store in database
|
||||
count = self._store_prices(security.id, df)
|
||||
logger.info(f"Stored {count} price records for {ticker}")
|
||||
return count
|
||||
|
||||
def _get_or_create_security(self, ticker: str) -> Security:
|
||||
"""Get existing security or create a new one."""
|
||||
stmt = select(Security).where(Security.ticker == ticker.upper())
|
||||
security = self.session.scalars(stmt).first()
|
||||
|
||||
if not security:
|
||||
security = Security(ticker=ticker.upper(), name=ticker, asset_type="stock")
|
||||
self.session.add(security)
|
||||
self.session.commit()
|
||||
logger.info(f"Created new security: {ticker}")
|
||||
|
||||
return security
|
||||
|
||||
def _get_missing_date_range_start(
|
||||
self, security_id: int, start_date: date, end_date: date
|
||||
) -> date:
|
||||
"""
|
||||
Find the earliest date we need to fetch (to avoid re-fetching existing data).
|
||||
Returns start_date if no data exists, or the day after the latest existing date.
|
||||
"""
|
||||
stmt = (
|
||||
select(Price.date)
|
||||
.where(Price.security_id == security_id)
|
||||
.where(Price.date >= start_date)
|
||||
.where(Price.date <= end_date)
|
||||
.order_by(Price.date.desc())
|
||||
.limit(1)
|
||||
)
|
||||
latest = self.session.scalars(stmt).first()
|
||||
|
||||
if latest:
|
||||
# Resume from the day after latest
|
||||
return latest + timedelta(days=1)
|
||||
return start_date
|
||||
|
||||
def _fetch_yfinance_data(self, ticker: str, start_date: date, end_date: date) -> pd.DataFrame:
|
||||
"""Fetch OHLCV data from yfinance."""
|
||||
stock = yf.Ticker(ticker)
|
||||
df = stock.history(
|
||||
start=start_date.isoformat(),
|
||||
end=(end_date + timedelta(days=1)).isoformat(), # yfinance end is exclusive
|
||||
auto_adjust=False, # Keep raw prices
|
||||
)
|
||||
|
||||
if df.empty:
|
||||
return df
|
||||
|
||||
# Reset index to get date as a column
|
||||
df = df.reset_index()
|
||||
|
||||
# Normalize column names
|
||||
df.columns = df.columns.str.lower()
|
||||
|
||||
# Keep only columns we need
|
||||
required_cols = ["date", "open", "high", "low", "close", "volume"]
|
||||
df = df[[col for col in required_cols if col in df.columns]]
|
||||
|
||||
# Convert date to date (not datetime)
|
||||
df["date"] = pd.to_datetime(df["date"]).dt.date
|
||||
|
||||
return df
|
||||
|
||||
def _store_prices(self, security_id: int, df: pd.DataFrame) -> int:
|
||||
"""
|
||||
Store price data in the database using upsert (insert or update).
|
||||
"""
|
||||
records = []
|
||||
for _, row in df.iterrows():
|
||||
record = {
|
||||
"security_id": security_id,
|
||||
"date": row["date"],
|
||||
"open": Decimal(str(row.get("open"))) if pd.notna(row.get("open")) else None,
|
||||
"high": Decimal(str(row.get("high"))) if pd.notna(row.get("high")) else None,
|
||||
"low": Decimal(str(row.get("low"))) if pd.notna(row.get("low")) else None,
|
||||
"close": Decimal(str(row["close"])),
|
||||
"volume": int(row["volume"]) if pd.notna(row.get("volume")) else None,
|
||||
"adjusted_close": None, # We'll compute this later if needed
|
||||
"source": "yfinance",
|
||||
"created_at": datetime.now(timezone.utc),
|
||||
}
|
||||
records.append(record)
|
||||
|
||||
if not records:
|
||||
return 0
|
||||
|
||||
# SQLite upsert: insert or replace on conflict
|
||||
stmt = sqlite_insert(Price).values(records)
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
index_elements=["security_id", "date"],
|
||||
set_={
|
||||
"open": stmt.excluded.open,
|
||||
"high": stmt.excluded.high,
|
||||
"low": stmt.excluded.low,
|
||||
"close": stmt.excluded.close,
|
||||
"volume": stmt.excluded.volume,
|
||||
"source": stmt.excluded.source,
|
||||
},
|
||||
)
|
||||
|
||||
self.session.execute(stmt)
|
||||
self.session.commit()
|
||||
|
||||
return len(records)
|
||||
|
||||
def bulk_fetch_prices(
|
||||
self,
|
||||
tickers: list[str],
|
||||
start_date: date | None = None,
|
||||
end_date: date | None = None,
|
||||
force_refresh: bool = False,
|
||||
) -> dict[str, int]:
|
||||
"""
|
||||
Fetch prices for multiple tickers.
|
||||
|
||||
Returns:
|
||||
Dict mapping ticker -> count of records inserted
|
||||
"""
|
||||
results = {}
|
||||
for ticker in tickers:
|
||||
try:
|
||||
count = self.fetch_and_store_prices(ticker, start_date, end_date, force_refresh)
|
||||
results[ticker] = count
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch {ticker}: {e}")
|
||||
results[ticker] = 0
|
||||
|
||||
return results
|
||||
@@ -0,0 +1,133 @@
|
||||
"""
|
||||
Security enrichment using yfinance.
|
||||
Fetches company names, sectors, industries, and exchanges for securities.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
import yfinance as yf
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from pote.db.models import Security
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SecurityEnricher:
|
||||
"""Enriches securities table with data from yfinance."""
|
||||
|
||||
def __init__(self, session: Session):
|
||||
self.session = session
|
||||
|
||||
def enrich_security(self, security: Security, force: bool = False) -> bool:
|
||||
"""
|
||||
Enrich a single security with yfinance data.
|
||||
|
||||
Args:
|
||||
security: Security model instance
|
||||
force: If True, re-fetch even if already enriched
|
||||
|
||||
Returns:
|
||||
True if enriched, False if skipped or failed
|
||||
"""
|
||||
# Skip if already enriched (unless force)
|
||||
if not force and security.name and security.name != security.ticker:
|
||||
logger.debug(f"Skipping {security.ticker} (already enriched)")
|
||||
return False
|
||||
|
||||
logger.info(f"Enriching {security.ticker}")
|
||||
|
||||
try:
|
||||
ticker_obj = yf.Ticker(security.ticker)
|
||||
info = ticker_obj.info
|
||||
|
||||
if not info or "symbol" not in info:
|
||||
logger.warning(f"No data found for {security.ticker}")
|
||||
return False
|
||||
|
||||
# Update fields
|
||||
security.name = info.get("longName") or info.get("shortName") or security.ticker
|
||||
security.sector = info.get("sector")
|
||||
security.industry = info.get("industry")
|
||||
security.exchange = info.get("exchange") or info.get("exchangeShortName")
|
||||
|
||||
# Determine asset type
|
||||
quote_type = info.get("quoteType", "").lower()
|
||||
if "etf" in quote_type:
|
||||
security.asset_type = "etf"
|
||||
elif "mutualfund" in quote_type:
|
||||
security.asset_type = "mutual_fund"
|
||||
elif "index" in quote_type:
|
||||
security.asset_type = "index"
|
||||
else:
|
||||
security.asset_type = "stock"
|
||||
|
||||
self.session.commit()
|
||||
logger.info(f"Enriched {security.ticker}: {security.name} ({security.sector})")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to enrich {security.ticker}: {e}")
|
||||
self.session.rollback()
|
||||
return False
|
||||
|
||||
def enrich_all_securities(
|
||||
self, limit: int | None = None, force: bool = False
|
||||
) -> dict[str, int]:
|
||||
"""
|
||||
Enrich all securities in the database.
|
||||
|
||||
Args:
|
||||
limit: Maximum number to enrich (None = all)
|
||||
force: If True, re-enrich already enriched securities
|
||||
|
||||
Returns:
|
||||
Dict with counts: {"total": N, "enriched": M, "failed": K}
|
||||
"""
|
||||
# Get securities to enrich
|
||||
stmt = select(Security)
|
||||
if not force:
|
||||
# Only enrich those with name == ticker (not yet enriched)
|
||||
stmt = stmt.where(Security.name == Security.ticker)
|
||||
|
||||
if limit:
|
||||
stmt = stmt.limit(limit)
|
||||
|
||||
securities = self.session.scalars(stmt).all()
|
||||
|
||||
if not securities:
|
||||
logger.info("No securities to enrich")
|
||||
return {"total": 0, "enriched": 0, "failed": 0}
|
||||
|
||||
logger.info(f"Enriching {len(securities)} securities")
|
||||
|
||||
enriched = 0
|
||||
failed = 0
|
||||
|
||||
for security in securities:
|
||||
if self.enrich_security(security, force=force):
|
||||
enriched += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
return {"total": len(securities), "enriched": enriched, "failed": failed}
|
||||
|
||||
def enrich_by_ticker(self, ticker: str) -> bool:
|
||||
"""
|
||||
Enrich a specific security by ticker.
|
||||
|
||||
Args:
|
||||
ticker: Stock ticker symbol
|
||||
|
||||
Returns:
|
||||
True if enriched, False if not found or failed
|
||||
"""
|
||||
stmt = select(Security).where(Security.ticker == ticker.upper())
|
||||
security = self.session.scalars(stmt).first()
|
||||
|
||||
if not security:
|
||||
logger.warning(f"Security {ticker} not found in database")
|
||||
return False
|
||||
|
||||
return self.enrich_security(security, force=True)
|
||||
@@ -0,0 +1,212 @@
|
||||
"""
|
||||
ETL for loading congressional trade data into the database.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from pote.db.models import Official, Security, Trade
|
||||
from pote.ingestion.house_watcher import (
|
||||
normalize_transaction_type,
|
||||
parse_amount_range,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TradeLoader:
|
||||
"""Loads congressional trade data into the database."""
|
||||
|
||||
def __init__(self, session: Session):
|
||||
self.session = session
|
||||
|
||||
def ingest_transactions(
|
||||
self, transactions: list[dict], source: str = "house_watcher"
|
||||
) -> dict[str, int]:
|
||||
"""
|
||||
Ingest a list of transactions from House Stock Watcher format.
|
||||
|
||||
Args:
|
||||
transactions: List of transaction dicts from HouseWatcherClient
|
||||
source: Source identifier (default: "house_watcher")
|
||||
|
||||
Returns:
|
||||
Dict with counts: {"officials": N, "securities": N, "trades": N}
|
||||
"""
|
||||
logger.info(f"Ingesting {len(transactions)} transactions from {source}")
|
||||
|
||||
officials_created = 0
|
||||
securities_created = 0
|
||||
trades_created = 0
|
||||
|
||||
for txn in transactions:
|
||||
try:
|
||||
# Get or create official
|
||||
official, is_new_official = self._get_or_create_official(txn)
|
||||
if is_new_official:
|
||||
officials_created += 1
|
||||
|
||||
# Get or create security
|
||||
ticker = txn.get("ticker", "").strip().upper()
|
||||
if not ticker or ticker in ("N/A", "--", ""):
|
||||
logger.debug(
|
||||
f"Skipping transaction with no ticker: {txn.get('representative')}"
|
||||
)
|
||||
continue
|
||||
|
||||
security, is_new_security = self._get_or_create_security(ticker)
|
||||
if is_new_security:
|
||||
securities_created += 1
|
||||
|
||||
# Create trade (upsert)
|
||||
trade_created = self._upsert_trade(txn, official.id, security.id, source)
|
||||
if trade_created:
|
||||
trades_created += 1
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to ingest transaction {txn}: {e}")
|
||||
continue
|
||||
|
||||
self.session.commit()
|
||||
|
||||
logger.info(
|
||||
f"Ingestion complete: {officials_created} officials, "
|
||||
f"{securities_created} securities, {trades_created} trades"
|
||||
)
|
||||
|
||||
return {
|
||||
"officials": officials_created,
|
||||
"securities": securities_created,
|
||||
"trades": trades_created,
|
||||
}
|
||||
|
||||
def _get_or_create_official(self, txn: dict) -> tuple[Official, bool]:
|
||||
"""
|
||||
Get or create an official from transaction data.
|
||||
|
||||
Returns:
|
||||
Tuple of (official, is_new)
|
||||
"""
|
||||
name = txn.get("representative", "").strip()
|
||||
if not name:
|
||||
raise ValueError("Transaction missing representative name")
|
||||
|
||||
# Try to find existing by name (simple for now)
|
||||
stmt = select(Official).where(Official.name == name)
|
||||
official = self.session.scalars(stmt).first()
|
||||
|
||||
if official:
|
||||
return (official, False)
|
||||
|
||||
# Create new official
|
||||
chamber = "Senate" if txn.get("house") == "Senate" else "House"
|
||||
party = txn.get("party", "").strip() or None
|
||||
state = None # House Watcher doesn't always provide state cleanly
|
||||
district = txn.get("district", "").strip() or None
|
||||
|
||||
official = Official(
|
||||
name=name,
|
||||
chamber=chamber,
|
||||
party=party,
|
||||
state=state,
|
||||
external_ids=f'{{"district": "{district}"}}' if district else None,
|
||||
)
|
||||
self.session.add(official)
|
||||
self.session.flush() # Get ID without committing
|
||||
|
||||
logger.info(f"Created new official: {name} ({chamber}, {party})")
|
||||
return (official, True)
|
||||
|
||||
def _get_or_create_security(self, ticker: str) -> tuple[Security, bool]:
|
||||
"""
|
||||
Get or create a security by ticker.
|
||||
|
||||
Returns:
|
||||
Tuple of (security, is_new)
|
||||
"""
|
||||
stmt = select(Security).where(Security.ticker == ticker)
|
||||
security = self.session.scalars(stmt).first()
|
||||
|
||||
if security:
|
||||
return (security, False)
|
||||
|
||||
# Create new security (minimal info for now)
|
||||
security = Security(
|
||||
ticker=ticker,
|
||||
name=ticker, # We'll enrich with yfinance later
|
||||
asset_type="stock",
|
||||
)
|
||||
self.session.add(security)
|
||||
self.session.flush()
|
||||
|
||||
logger.debug(f"Created new security: {ticker}")
|
||||
return (security, True)
|
||||
|
||||
def _upsert_trade(self, txn: dict, official_id: int, security_id: int, source: str) -> bool:
|
||||
"""
|
||||
Insert or update a trade record.
|
||||
|
||||
Returns:
|
||||
True if a new trade was created, False if updated
|
||||
"""
|
||||
# Parse dates
|
||||
try:
|
||||
txn_date_str = txn.get("transaction_date", "")
|
||||
filing_date_str = txn.get("disclosure_date", "")
|
||||
|
||||
if "/" in txn_date_str:
|
||||
transaction_date = datetime.strptime(txn_date_str, "%m/%d/%Y").date()
|
||||
else:
|
||||
transaction_date = datetime.strptime(txn_date_str, "%Y-%m-%d").date()
|
||||
|
||||
if "/" in filing_date_str:
|
||||
filing_date = datetime.strptime(filing_date_str, "%m/%d/%Y").date()
|
||||
else:
|
||||
filing_date = datetime.strptime(filing_date_str, "%Y-%m-%d").date()
|
||||
|
||||
except (ValueError, TypeError) as e:
|
||||
logger.warning(f"Failed to parse dates for transaction: {e}")
|
||||
return False
|
||||
|
||||
# Parse amount
|
||||
amount_str = txn.get("amount", "")
|
||||
value_min, value_max = parse_amount_range(amount_str)
|
||||
|
||||
# Normalize side
|
||||
side = normalize_transaction_type(txn.get("transaction", ""))
|
||||
|
||||
# Build external ID for deduplication
|
||||
external_id = f"{official_id}_{security_id}_{transaction_date}_{side}"
|
||||
|
||||
# Check if exists
|
||||
stmt = select(Trade).where(Trade.source == source, Trade.external_id == external_id)
|
||||
existing = self.session.scalars(stmt).first()
|
||||
|
||||
if existing:
|
||||
# Update (in case data changed)
|
||||
existing.filing_date = filing_date
|
||||
existing.value_min = Decimal(str(value_min)) if value_min else None
|
||||
existing.value_max = Decimal(str(value_max)) if value_max else None
|
||||
return False
|
||||
|
||||
# Create new trade
|
||||
trade = Trade(
|
||||
official_id=official_id,
|
||||
security_id=security_id,
|
||||
source=source,
|
||||
external_id=external_id,
|
||||
transaction_date=transaction_date,
|
||||
filing_date=filing_date,
|
||||
side=side,
|
||||
value_min=Decimal(str(value_min)) if value_min else None,
|
||||
value_max=Decimal(str(value_max)) if value_max else None,
|
||||
currency="USD",
|
||||
quality_flags=None, # Can add flags like "range_only" later
|
||||
)
|
||||
|
||||
self.session.add(trade)
|
||||
return True
|
||||
Reference in New Issue
Block a user