From b3092a0d5e1abe177e6e7468e7802788f500726b Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sun, 29 Mar 2026 14:22:18 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20=EB=B0=B1=ED=85=8C=EC=8A=A4=ED=8C=85=20?= =?UTF-8?q?=EC=9C=A0=EB=8B=88=EB=B2=84=EC=8A=A4=20=EA=B3=BC=EA=B1=B0=20?= =?UTF-8?q?=EC=8B=9C=EC=A0=90=20=EC=A3=BC=EC=8B=9D=20=EC=8A=A4=ED=81=AC?= =?UTF-8?q?=EB=A6=AC=EB=8B=9D=20(Historical=20Stock=20Universe)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 새 기능 - GET /universe/screen — 과거 날짜 기준 시총/섹터/거래소 필터링 - GET /universe/registry — 추적 종목 목록 조회 - POST /universe/admin/discover — yfinance screener로 US 주식 자동 등록 - POST /universe/admin/build-snapshots — SEC EDGAR × yfinance 월별 시총 스냅샷 생성 ## 데이터 모델 - universe_ticker_registry: 종목 마스터 (ticker, name, cik, sector, industry, exchange) - universe_snapshot: 월별 스냅샷 (ticker, snapshot_date, market_cap, close_price, shares_outstanding) - 인덱스: (snapshot_date, market_cap) — 핵심 스크리닝 쿼리 최적화 - ~4000종목 × 120개월 ≈ 480K 행 예상 ## 데이터 흐름 1. SEC EDGAR companyfacts → shares_outstanding (최신, 주가분할 반영) 2. yfinance bulk download 1mo interval → 월별 종가 3. market_cap = latest_shares × close_price (yfinance 분할조정 가격과 일관성) ## 제한사항 - Survivorship bias: 현재 상장 종목만 (상폐 종목 미포함) - 자사주 매입으로 과거 시총 ~20% 오차 가능 (분할 오차 방지가 주목적) Co-Authored-By: Claude Sonnet 4.6 --- .../e6f7a8b9c0d1_add_universe_snapshot.py | 65 ++ app/api/v1/api.py | 5 +- app/api/v1/endpoints/universe.py | 262 ++++++++ app/models/__init__.py | 4 + app/models/universe_snapshot.py | 64 ++ app/schemas/universe.py | 66 ++ app/services/universe_service.py | 620 ++++++++++++++++++ 7 files changed, 1084 insertions(+), 2 deletions(-) create mode 100644 alembic/versions/e6f7a8b9c0d1_add_universe_snapshot.py create mode 100644 app/api/v1/endpoints/universe.py create mode 100644 app/models/universe_snapshot.py create mode 100644 app/schemas/universe.py create mode 100644 app/services/universe_service.py diff --git a/alembic/versions/e6f7a8b9c0d1_add_universe_snapshot.py b/alembic/versions/e6f7a8b9c0d1_add_universe_snapshot.py new file mode 100644 index 0000000..ef23e7c --- /dev/null +++ b/alembic/versions/e6f7a8b9c0d1_add_universe_snapshot.py @@ -0,0 +1,65 @@ +"""add universe_snapshot tables + +Revision ID: e6f7a8b9c0d1 +Revises: d5e6f7a8b9c0 +Create Date: 2026-03-29 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision: str = "e6f7a8b9c0d1" +down_revision: Union[str, Sequence[str], None] = "d5e6f7a8b9c0" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + conn = op.get_bind() + + if not conn.dialect.has_table(conn, "universe_ticker_registry"): + op.create_table( + "universe_ticker_registry", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("ticker", sa.String(10), nullable=False), + sa.Column("name", sa.String(255), nullable=True), + sa.Column("cik", sa.String(20), nullable=True), + sa.Column("sector", sa.String(100), nullable=True), + sa.Column("industry", sa.String(200), nullable=True), + sa.Column("exchange", sa.String(20), nullable=True), + sa.Column("is_active", sa.Boolean(), nullable=False, server_default="true"), + sa.Column("created_at", postgresql.TIMESTAMP(timezone=True)), + sa.Column("updated_at", postgresql.TIMESTAMP(timezone=True)), + sa.UniqueConstraint("ticker", name="uq_universe_ticker_registry"), + ) + op.create_index("idx_registry_ticker", "universe_ticker_registry", ["ticker"]) + op.create_index("idx_registry_sector", "universe_ticker_registry", ["sector"]) + op.create_index("idx_registry_exchange", "universe_ticker_registry", ["exchange"]) + op.create_index("idx_registry_active", "universe_ticker_registry", ["is_active"]) + + if not conn.dialect.has_table(conn, "universe_snapshot"): + op.create_table( + "universe_snapshot", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("ticker", sa.String(10), nullable=False), + sa.Column("snapshot_date", postgresql.TIMESTAMP(timezone=True), nullable=False), + sa.Column("close_price", sa.Float(), nullable=True), + sa.Column("shares_outstanding", sa.Float(), nullable=True), + sa.Column("market_cap", sa.Float(), nullable=True), + sa.Column("sector", sa.String(100), nullable=True), + sa.Column("industry", sa.String(200), nullable=True), + sa.Column("exchange", sa.String(20), nullable=True), + sa.Column("created_at", postgresql.TIMESTAMP(timezone=True)), + sa.UniqueConstraint("ticker", "snapshot_date", name="uq_universe_snapshot"), + ) + op.create_index("idx_snapshot_ticker_date", "universe_snapshot", ["ticker", "snapshot_date"]) + op.create_index("idx_snapshot_date_mcap", "universe_snapshot", ["snapshot_date", "market_cap"]) + op.create_index("idx_snapshot_sector", "universe_snapshot", ["sector"]) + op.create_index("idx_snapshot_date", "universe_snapshot", ["snapshot_date"]) + + +def downgrade() -> None: + op.drop_table("universe_snapshot") + op.drop_table("universe_ticker_registry") diff --git a/app/api/v1/api.py b/app/api/v1/api.py index 4a49062..8d85e77 100644 --- a/app/api/v1/api.py +++ b/app/api/v1/api.py @@ -3,7 +3,7 @@ API v1 router """ from fastapi import APIRouter -from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra, overlay, screener, attention, insider, earnings +from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra, overlay, screener, attention, insider, earnings, universe api_router = APIRouter() @@ -28,4 +28,5 @@ api_router.include_router(overlay.router, prefix="/overlay", tags=["overlay"]) api_router.include_router(screener.router, prefix="/screener", tags=["screener"]) api_router.include_router(attention.router, prefix="/attention", tags=["attention"]) api_router.include_router(insider.router, prefix="/insider", tags=["insider"]) -api_router.include_router(earnings.router, prefix="/earnings", tags=["earnings"]) \ No newline at end of file +api_router.include_router(earnings.router, prefix="/earnings", tags=["earnings"]) +api_router.include_router(universe.router, prefix="/universe", tags=["universe"]) \ No newline at end of file diff --git a/app/api/v1/endpoints/universe.py b/app/api/v1/endpoints/universe.py new file mode 100644 index 0000000..34280a7 --- /dev/null +++ b/app/api/v1/endpoints/universe.py @@ -0,0 +1,262 @@ +""" +Historical Stock Universe endpoints — backtesting universe construction + +Endpoints: + GET /universe/screen — Screen stocks at a historical date by market_cap, sector, etc. + GET /universe/registry — Browse the registered ticker universe + POST /universe/admin/discover — Discover and register US tickers via yfinance screener + POST /universe/admin/build-snapshots — Build monthly market_cap snapshots (long-running) +""" + +import logging +from typing import Optional + +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query +from fastapi.responses import Response +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.database import get_db +from app.schemas.universe import ( + RegistryResponse, + SnapshotBuildRequest, + TickerRegistryItem, + UniverseScreenResponse, + UniverseSnapshotItem, +) +from app.services.universe_service import UniverseService +from app.utils.cache import with_cache + +router = APIRouter() +logger = logging.getLogger("app.api.v1.universe") + + +@router.get( + "/screen", + response_model=UniverseScreenResponse, + summary="Screen stocks at a historical date", + description=( + "Query monthly market_cap snapshots to find stocks matching criteria at a past date.\\n\\n" + "**용도**: 백테스팅 전략 유니버스 구성 — 특정 시점 시총/섹터 기준 종목 필터링.\\n\\n" + "**데이터 소스**: SEC EDGAR shares_outstanding × yfinance monthly close.\\n" + "**제한**: 현재 상장 종목만 포함 (survivorship bias). 상폐 종목 미포함.\\n\\n" + "**사전 조건**: `/universe/admin/discover` 후 `/universe/admin/build-snapshots` 실행 필요." + ), +) +@with_cache(namespace="universe:screen", ttl=3600, key_params=["date", "market_cap_min", "market_cap_max", "sector", "exchange", "page", "page_size", "sort_by", "sort_ascending"]) +async def screen_historical( + response: Response, + date: str = Query(..., description="Historical date YYYY-MM-DD (rounded to month start)"), + market_cap_min: Optional[float] = Query(None, description="Min market cap (USD), e.g. 2e9"), + market_cap_max: Optional[float] = Query(None, description="Max market cap (USD), e.g. 20e9"), + sector: Optional[str] = Query(None, description="Sector filter (e.g. Technology, Healthcare)"), + exchange: Optional[str] = Query(None, description="Exchange filter (NYSE, NASDAQ, AMEX)"), + page: int = Query(1, ge=1, description="Page number"), + page_size: int = Query(100, ge=1, le=500, description="Results per page"), + sort_by: str = Query("market_cap", description="Sort field: market_cap or ticker"), + sort_ascending: bool = Query(False, description="Sort direction"), + db: AsyncSession = Depends(get_db), +): + svc = UniverseService() + try: + items, total, snapshot_date = await svc.screen_historical( + db, + date_str=date, + market_cap_min=market_cap_min, + market_cap_max=market_cap_max, + sector=sector, + exchange=exchange, + page=page, + page_size=page_size, + sort_by=sort_by, + sort_ascending=sort_ascending, + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except Exception as e: + logger.error(f"Universe screen error: {e}") + raise HTTPException(status_code=502, detail=f"Universe screen failed: {e}") + + total_pages = max(1, (total + page_size - 1) // page_size) + filters: dict = {} + if market_cap_min is not None: + filters["market_cap_min"] = market_cap_min + if market_cap_max is not None: + filters["market_cap_max"] = market_cap_max + if sector: + filters["sector"] = sector + if exchange: + filters["exchange"] = exchange + + if total == 0: + note = ( + "No snapshots found for this date. " + "Run POST /universe/admin/build-snapshots to populate data." + ) + else: + note = "Survivorship bias: currently-listed stocks only. Delisted companies excluded." + + return UniverseScreenResponse( + stocks=[UniverseSnapshotItem(**item) for item in items], + total_count=total, + page=page, + page_size=page_size, + total_pages=total_pages, + snapshot_date=snapshot_date, + filters_applied=filters, + metadata={ + "sort_by": sort_by, + "sort_ascending": sort_ascending, + "note": note, + }, + ) + + +@router.get( + "/registry", + response_model=RegistryResponse, + summary="Browse registered ticker universe", + description="List tickers registered in the universe (populated via /admin/discover).", +) +async def get_registry( + sector: Optional[str] = Query(None, description="Filter by sector"), + exchange: Optional[str] = Query(None, description="Filter by exchange"), + is_active: Optional[bool] = Query(None, description="Filter by active status"), + page: int = Query(1, ge=1), + page_size: int = Query(100, ge=1, le=1000), + db: AsyncSession = Depends(get_db), +): + svc = UniverseService() + try: + rows, total = await svc.get_registry( + db, + sector=sector, + exchange=exchange, + is_active=is_active, + page=page, + page_size=page_size, + ) + except Exception as e: + logger.error(f"Registry fetch error: {e}") + raise HTTPException(status_code=502, detail=str(e)) + + total_pages = max(1, (total + page_size - 1) // page_size) + return RegistryResponse( + tickers=[TickerRegistryItem.model_validate(r) for r in rows], + total_count=total, + page=page, + page_size=page_size, + total_pages=total_pages, + ) + + +@router.post( + "/admin/discover", + summary="Discover and register US tickers", + description=( + "Scrapes US-listed stocks via yfinance screener and registers them in the universe.\\n\\n" + "**소요 시간**: 약 1~5분 (시총 기준에 따라 다름).\\n" + "**권장**: `market_cap_min=100000000` ($100M) → ~3000~5000 종목." + ), +) +async def discover_tickers( + market_cap_min: float = Query( + 1e8, + description="Min market cap for inclusion (USD). Default $100M.", + ), + db: AsyncSession = Depends(get_db), +): + svc = UniverseService() + try: + result = await svc.discover_tickers(db, market_cap_min=market_cap_min) + return { + "status": "completed", + **result, + "note": "Run POST /universe/admin/build-snapshots to compute historical market_cap snapshots.", + } + except Exception as e: + logger.error(f"Universe discover error: {e}") + raise HTTPException(status_code=502, detail=f"Discover failed: {e}") + + +async def _run_build_snapshots( + tickers, start_date: str, end_date: str, force_rebuild: bool +): + """Background task wrapper for build_snapshots (needs its own DB session).""" + from app.core.database import AsyncSessionLocal + + async with AsyncSessionLocal() as db: + svc = UniverseService() + try: + result = await svc.build_snapshots( + db, + tickers=tickers, + start_date=start_date, + end_date=end_date, + force_rebuild=force_rebuild, + ) + logger.info(f"Universe background build complete: {result}") + except Exception as e: + logger.error(f"Universe background build failed: {e}") + + +@router.post( + "/admin/build-snapshots", + summary="Build monthly market_cap snapshots", + description=( + "Computes monthly market_cap snapshots for registered tickers and stores them " + "in `universe_snapshot`.\\n\\n" + "**데이터 소스**: SEC EDGAR companyfacts (shares_outstanding) + yfinance monthly close.\\n\\n" + "**소요 시간**: 전체 유니버스(~4000 종목) × 10년 기준 30~60분. " + "백그라운드에서 실행되므로 응답은 즉시 반환됩니다.\\n\\n" + "**권장 시작점**: `tickers=[AAPL,MSFT,GOOGL]`로 소규모 테스트 후 전체 빌드." + ), +) +async def build_snapshots( + body: SnapshotBuildRequest, + background_tasks: BackgroundTasks, + db: AsyncSession = Depends(get_db), +): + tickers = [t.upper() for t in body.tickers] if body.tickers else None + + # Validate dates + try: + from datetime import date + date.fromisoformat(body.start_date) + date.fromisoformat(body.end_date) + except ValueError as e: + raise HTTPException(status_code=400, detail=f"Invalid date: {e}") + + # For small ticker lists (≤20), run synchronously for immediate feedback + if tickers and len(tickers) <= 20: + svc = UniverseService() + try: + result = await svc.build_snapshots( + db, + tickers=tickers, + start_date=body.start_date, + end_date=body.end_date, + force_rebuild=body.force_rebuild, + ) + return {"status": "completed", **result} + except Exception as e: + logger.error(f"Snapshot build error: {e}") + raise HTTPException(status_code=502, detail=str(e)) + + # Large jobs → background + ticker_count = len(tickers) if tickers else "all registry" + background_tasks.add_task( + _run_build_snapshots, + tickers, + body.start_date, + body.end_date, + body.force_rebuild, + ) + return { + "status": "started", + "tickers_queued": ticker_count, + "date_range": f"{body.start_date} → {body.end_date}", + "note": ( + "Building in background. " + "Query GET /universe/screen after a few minutes to verify data." + ), + } diff --git a/app/models/__init__.py b/app/models/__init__.py index 437097c..cc11285 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -8,6 +8,7 @@ from app.models.overlay_raw_event import OverlayHeadlineEvent, OverlayVideoEvent from app.models.overlay_feature import OverlayFeatureRecord, OverlayJobLog from app.models.insider_transaction import InsiderTransaction from app.models.earnings_surprise import EarningsSurprise +from app.models.universe_snapshot import UniverseTickerRegistry, UniverseSnapshot __all__ = [ "Company", @@ -37,4 +38,7 @@ __all__ = [ # Insider "InsiderTransaction", "EarningsSurprise", + # Universe + "UniverseTickerRegistry", + "UniverseSnapshot", ] \ No newline at end of file diff --git a/app/models/universe_snapshot.py b/app/models/universe_snapshot.py new file mode 100644 index 0000000..db45d15 --- /dev/null +++ b/app/models/universe_snapshot.py @@ -0,0 +1,64 @@ +""" +Universe Snapshot models for historical stock screening (backtesting universe) +""" + +from sqlalchemy import Column, String, Float, Index, UniqueConstraint, Boolean +from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP +from datetime import datetime, timezone +import uuid + +from app.core.database import Base + + +class UniverseTickerRegistry(Base): + __tablename__ = "universe_ticker_registry" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + ticker = Column(String(10), nullable=False) + name = Column(String(255), nullable=True) + cik = Column(String(20), nullable=True) + sector = Column(String(100), nullable=True) + industry = Column(String(200), nullable=True) + exchange = Column(String(20), nullable=True) + is_active = Column(Boolean, default=True, nullable=False) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + updated_at = Column( + TIMESTAMP(timezone=True), + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + ) + + __table_args__ = ( + UniqueConstraint("ticker", name="uq_universe_ticker_registry"), + Index("idx_registry_ticker", "ticker"), + Index("idx_registry_sector", "sector"), + Index("idx_registry_exchange", "exchange"), + Index("idx_registry_active", "is_active"), + ) + + +class UniverseSnapshot(Base): + __tablename__ = "universe_snapshot" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + ticker = Column(String(10), nullable=False, index=True) + snapshot_date = Column(TIMESTAMP(timezone=True), nullable=False) + close_price = Column(Float, nullable=True) + shares_outstanding = Column(Float, nullable=True) + market_cap = Column(Float, nullable=True) + sector = Column(String(100), nullable=True) + industry = Column(String(200), nullable=True) + exchange = Column(String(20), nullable=True) + created_at = Column( + TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + + __table_args__ = ( + UniqueConstraint("ticker", "snapshot_date", name="uq_universe_snapshot"), + Index("idx_snapshot_ticker_date", "ticker", "snapshot_date"), + Index("idx_snapshot_date_mcap", "snapshot_date", "market_cap"), + Index("idx_snapshot_sector", "sector"), + Index("idx_snapshot_date", "snapshot_date"), + ) diff --git a/app/schemas/universe.py b/app/schemas/universe.py new file mode 100644 index 0000000..3ceb4a2 --- /dev/null +++ b/app/schemas/universe.py @@ -0,0 +1,66 @@ +""" +Universe screening schemas for backtesting universe construction +""" + +from datetime import datetime +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + + +class UniverseSnapshotItem(BaseModel): + ticker: str + name: Optional[str] = None + market_cap: Optional[float] = None + close_price: Optional[float] = None + shares_outstanding: Optional[float] = None + sector: Optional[str] = None + industry: Optional[str] = None + exchange: Optional[str] = None + snapshot_date: Optional[datetime] = None + + model_config = {"from_attributes": True} + + +class UniverseScreenResponse(BaseModel): + stocks: List[UniverseSnapshotItem] + total_count: int + page: int + page_size: int + total_pages: int + snapshot_date: Optional[str] = None + filters_applied: Dict[str, Any] = {} + metadata: Dict[str, Any] = {} + + +class TickerRegistryItem(BaseModel): + ticker: str + name: Optional[str] = None + sector: Optional[str] = None + industry: Optional[str] = None + exchange: Optional[str] = None + cik: Optional[str] = None + is_active: bool = True + + model_config = {"from_attributes": True} + + +class RegistryResponse(BaseModel): + tickers: List[TickerRegistryItem] + total_count: int + page: int + page_size: int + total_pages: int + + +class SnapshotBuildRequest(BaseModel): + tickers: Optional[List[str]] = Field( + None, + description="Specific tickers to build. Omit for all registry tickers.", + ) + start_date: str = Field(..., description="Start date YYYY-MM-DD (e.g. 2015-01-01)") + end_date: str = Field(..., description="End date YYYY-MM-DD (e.g. 2025-12-01)") + force_rebuild: bool = Field( + False, + description="Delete existing snapshots for these tickers before rebuilding", + ) diff --git a/app/services/universe_service.py b/app/services/universe_service.py new file mode 100644 index 0000000..f58753b --- /dev/null +++ b/app/services/universe_service.py @@ -0,0 +1,620 @@ +""" +Universe Service — Historical stock universe construction for backtesting + +Builds monthly market_cap snapshots by combining: + - SEC EDGAR companyfacts → shares_outstanding (quarterly, carried forward) + - yfinance monthly close prices + +Data flow: + 1. discover_tickers() → universe_ticker_registry (current US stocks via yf.screen) + 2. build_snapshots() → universe_snapshot (monthly market_cap per ticker) + 3. screen_historical() → filtered results for a given historical date + +Survivorship bias note: v1 universe is based on currently-listed stocks only. +Delisted companies are not included. +""" + +import asyncio +import logging +import math +from datetime import date, datetime, timezone +from typing import Dict, List, Optional, Tuple + +from sqlalchemy import and_, desc, func, select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.universe_snapshot import UniverseSnapshot, UniverseTickerRegistry +from app.services.sec_http_client import SECHttpClient + +logger = logging.getLogger(__name__) + +_CHUNK = 3000 # asyncpg 32767-param limit (10 cols × 3000 = 30000) +_PRICE_BATCH = 100 # tickers per yfinance bulk download +_SEC_CONCURRENCY = 3 # concurrent SEC EDGAR companyfacts requests +_SCREEN_PAGE = 250 # max per yf.screen() call + + +class UniverseService: + + def __init__(self): + self._http = SECHttpClient("Stock Oracle Universe Service") + + # ------------------------------------------------------------------ + # Step 1: Discover and register tickers + # ------------------------------------------------------------------ + + def _screen_page_sync(self, query, offset: int) -> dict: + """Synchronous yfinance screen() — runs in executor.""" + import yfinance as yf + return yf.screen( + query, offset=offset, size=_SCREEN_PAGE, + sortField="intradaymarketcap", sortAsc=False, + ) + + async def discover_tickers(self, db: AsyncSession, market_cap_min: float = 1e8) -> Dict: + """ + Discover US-listed stocks via yfinance screener and store in registry. + Filters out ETFs, mutual funds, and foreign-listed stocks. + + Returns: { tickers_found, tickers_registered } + """ + from yfinance import EquityQuery + + query = EquityQuery("and", [ + EquityQuery("eq", ["region", "us"]), + EquityQuery("gt", ["intradaymarketcap", market_cap_min]), + ]) + + loop = asyncio.get_event_loop() + all_quotes: List[dict] = [] + offset = 0 + + while True: + raw = await loop.run_in_executor( + None, self._screen_page_sync, query, offset + ) + quotes = raw.get("quotes", []) + if not quotes: + break + all_quotes.extend(quotes) + total = raw.get("count") or raw.get("total") or len(all_quotes) + offset += len(quotes) + if offset >= total: + break + await asyncio.sleep(0.3) + + logger.info(f"Universe: screener returned {len(all_quotes)} quotes") + + # Fetch CIK map from SEC once (disk-cached after first call) + cik_map = await self._fetch_cik_map() + + _REVERSE_EXCHANGE = { + "NYQ": "NYSE", "NMS": "NASDAQ", "NGM": "NASDAQ", + "NCM": "NASDAQ", "ASE": "AMEX", "PCX": "NYSE_ARCA", + } + _SKIP_TYPES = {"ETF", "MUTUALFUND", "INDEX", "CURRENCY", "FUTURE", "OPTION"} + + rows = [] + for q in all_quotes: + ticker = (q.get("symbol") or "").upper() + if not ticker or len(ticker) > 10: + continue + if (q.get("quoteType") or "").upper() in _SKIP_TYPES: + continue + exchange_code = q.get("exchange", "") + rows.append({ + "ticker": ticker, + "name": q.get("shortName") or q.get("longName"), + "cik": cik_map.get(ticker), + "sector": q.get("sector"), + "industry": q.get("industry"), + "exchange": _REVERSE_EXCHANGE.get(exchange_code, exchange_code) or None, + "is_active": True, + }) + + if not rows: + return {"tickers_found": len(all_quotes), "tickers_registered": 0} + + for i in range(0, len(rows), _CHUNK): + chunk = rows[i:i + _CHUNK] + stmt = pg_insert(UniverseTickerRegistry).values(chunk) + stmt = stmt.on_conflict_do_update( + constraint="uq_universe_ticker_registry", + set_={ + "name": stmt.excluded.name, + "cik": stmt.excluded.cik, + "sector": stmt.excluded.sector, + "industry": stmt.excluded.industry, + "exchange": stmt.excluded.exchange, + "is_active": stmt.excluded.is_active, + "updated_at": func.now(), + }, + ) + await db.execute(stmt) + await db.commit() + + logger.info(f"Universe: upserted {len(rows)} tickers into registry") + return {"tickers_found": len(all_quotes), "tickers_registered": len(rows)} + + async def _fetch_cik_map(self) -> Dict[str, str]: + """Fetch SEC company_tickers.json once → {TICKER: padded_cik}.""" + try: + data = await self._http.fetch_json( + "https://www.sec.gov/files/company_tickers.json" + ) + return { + entry["ticker"].upper(): str(entry["cik_str"]).zfill(10) + for entry in data.values() + if "ticker" in entry and "cik_str" in entry + } + except Exception as e: + logger.warning(f"Universe: CIK map fetch failed: {e}") + return {} + + # ------------------------------------------------------------------ + # Step 2: Build monthly snapshots + # ------------------------------------------------------------------ + + async def build_snapshots( + self, + db: AsyncSession, + tickers: Optional[List[str]], + start_date: str, + end_date: str, + force_rebuild: bool = False, + ) -> Dict: + """ + Build monthly market_cap snapshots for given tickers (or all registry). + + For each ticker: + 1. Fetch shares_outstanding history from SEC EDGAR companyfacts + 2. Fetch monthly close prices via yfinance bulk download + 3. Compute market_cap = close × shares per month + 4. Upsert into universe_snapshot + + Returns: { tickers_processed, tickers_failed, snapshots_created } + """ + # Resolve ticker list + if tickers: + ticker_list = [t.upper() for t in tickers] + else: + result = await db.execute( + select(UniverseTickerRegistry.ticker) + .where(UniverseTickerRegistry.is_active == True) + .order_by(UniverseTickerRegistry.ticker) + ) + ticker_list = [r[0] for r in result.fetchall()] + + if not ticker_list: + logger.warning("Universe: no tickers to process") + return {"tickers_processed": 0, "tickers_failed": 0, "snapshots_created": 0} + + logger.info( + f"Universe: building snapshots for {len(ticker_list)} tickers " + f"({start_date} → {end_date})" + ) + + if force_rebuild: + await db.execute( + UniverseSnapshot.__table__.delete().where( + UniverseSnapshot.ticker.in_(ticker_list) + ) + ) + await db.flush() + + # Load registry metadata (sector/industry/exchange/CIK) + reg_result = await db.execute( + select(UniverseTickerRegistry).where( + UniverseTickerRegistry.ticker.in_(ticker_list) + ) + ) + registry_map: Dict[str, UniverseTickerRegistry] = { + r.ticker: r for r in reg_result.scalars().all() + } + + total_snapshots = 0 + total_failed = 0 + + # Process in batches of _PRICE_BATCH for yfinance bulk download + for batch_start in range(0, len(ticker_list), _PRICE_BATCH): + batch = ticker_list[batch_start:batch_start + _PRICE_BATCH] + + # ---- Fetch monthly prices (synchronous, in thread) ---- + price_data = await asyncio.to_thread( + self._fetch_bulk_monthly_prices, batch, start_date, end_date + ) + + # ---- Fetch shares_outstanding from SEC EDGAR (concurrent) ---- + sem = asyncio.Semaphore(_SEC_CONCURRENCY) + + async def _fetch_one(tkr: str) -> Tuple[str, List]: + async with sem: + history = await self._fetch_shares_history( + tkr, registry_map.get(tkr) + ) + return tkr, history + + results = await asyncio.gather( + *[_fetch_one(t) for t in batch], return_exceptions=True + ) + shares_map: Dict[str, List[Tuple[date, float]]] = {} + for item in results: + if isinstance(item, Exception): + continue + tkr, history = item + shares_map[tkr] = history + + # ---- Build snapshot rows ---- + batch_rows = [] + for ticker in batch: + shares_history = shares_map.get(ticker, []) + ticker_prices = price_data.get(ticker, {}) + reg = registry_map.get(ticker) + + if not ticker_prices: + total_failed += 1 + continue + + # Use the LATEST available shares_outstanding (most recent SEC filing). + # yfinance returns split-adjusted prices retroactively, so using the + # post-split shares count gives correct market_cap across all periods. + # Small error (~10-20%) may occur for companies with large buyback programs, + # but this is acceptable for screening purposes. + latest_shares = shares_history[-1][1] if shares_history else None + + for snap_date_key, close in ticker_prices.items(): + if isinstance(snap_date_key, datetime): + snap_date = snap_date_key.date() + else: + snap_date = snap_date_key + + shares = latest_shares + if shares is None or close is None: + continue + + market_cap = shares * close + if math.isnan(market_cap) or market_cap <= 0: + continue + + # Normalize to first of month + snapshot_dt = datetime( + snap_date.year, snap_date.month, 1, tzinfo=timezone.utc + ) + batch_rows.append({ + "ticker": ticker, + "snapshot_date": snapshot_dt, + "close_price": round(close, 4), + "shares_outstanding": shares, + "market_cap": round(market_cap, 0), + "sector": reg.sector if reg else None, + "industry": reg.industry if reg else None, + "exchange": reg.exchange if reg else None, + }) + + # ---- Batch upsert ---- + if batch_rows: + for i in range(0, len(batch_rows), _CHUNK): + chunk = batch_rows[i:i + _CHUNK] + stmt = pg_insert(UniverseSnapshot).values(chunk) + stmt = stmt.on_conflict_do_update( + constraint="uq_universe_snapshot", + set_={ + "close_price": stmt.excluded.close_price, + "shares_outstanding": stmt.excluded.shares_outstanding, + "market_cap": stmt.excluded.market_cap, + "sector": stmt.excluded.sector, + "industry": stmt.excluded.industry, + "exchange": stmt.excluded.exchange, + }, + ) + result = await db.execute(stmt) + total_snapshots += result.rowcount + await db.commit() + + batch_num = batch_start // _PRICE_BATCH + 1 + total_batches = (len(ticker_list) + _PRICE_BATCH - 1) // _PRICE_BATCH + logger.info( + f"Universe: batch {batch_num}/{total_batches} — " + f"{len(batch)} tickers, {len(batch_rows)} snapshot rows" + ) + + logger.info( + f"Universe: build complete — {len(ticker_list)} tickers, " + f"{total_snapshots} snapshots created, {total_failed} failed" + ) + return { + "tickers_processed": len(ticker_list), + "tickers_failed": total_failed, + "snapshots_created": total_snapshots, + } + + async def _fetch_shares_history( + self, + ticker: str, + registry_entry: Optional[UniverseTickerRegistry], + ) -> List[Tuple[date, float]]: + """Fetch shares_outstanding history from SEC EDGAR companyfacts.""" + cik = None + if registry_entry and registry_entry.cik: + cik = registry_entry.cik + else: + cik = await self._http.get_company_cik(ticker) + + if not cik: + return [] + + try: + url = ( + f"{self._http.sec_base_data}" + f"/api/xbrl/companyfacts/CIK{str(cik).zfill(10)}.json" + ) + facts = await self._http.fetch_json(url) + return _extract_shares_history(facts) + except Exception as e: + logger.debug(f"Universe: companyfacts failed for {ticker}: {e}") + return [] + + def _fetch_bulk_monthly_prices( + self, + tickers: List[str], + start: str, + end: str, + ) -> Dict[str, Dict[date, float]]: + """ + Synchronous bulk monthly price fetch via yfinance. + Returns {ticker: {date: close_price}}. + """ + try: + import math as _math + import yfinance as yf + except ImportError: + logger.error("yfinance not available for price download") + return {} + + if not tickers: + return {} + + # Always use group_by='ticker' for consistent MultiIndex column structure: + # data[ticker] -> DataFrame with ['Close', 'Adj Close', ...] + try: + data = yf.download( + tickers=tickers, + start=start, + end=end, + interval="1mo", + auto_adjust=False, + progress=False, + threads=True, + group_by="ticker", + ) + except Exception as e: + logger.warning(f"Universe: bulk price download failed: {e}") + return {} + + if data is None or data.empty: + return {} + + result: Dict[str, Dict[date, float]] = {} + + def _safe_close(val) -> Optional[float]: + if val is None: + return None + try: + f = float(val) + return None if _math.isnan(f) or f <= 0 else f + except (TypeError, ValueError): + return None + + def _to_date(dt) -> Optional[date]: + if hasattr(dt, "date"): + return dt.date() + if isinstance(dt, date): + return dt + return None + + for ticker in tickers: + result[ticker] = {} + try: + # With group_by='ticker', columns are MultiIndex (ticker, price_type) + # data[ticker] gives a flat DataFrame with price columns + lvl0 = data.columns.get_level_values(0) + if ticker not in lvl0: + continue + ticker_df = data[ticker] + close_col = ticker_df.get("Close") + if close_col is None: + close_col = ticker_df.get("Adj Close") + if close_col is None: + continue + for dt, val in close_col.items(): + d = _to_date(dt) + c = _safe_close(val) + if d and c: + result[ticker][d] = c + except Exception as e: + logger.debug(f"Universe: price parse error for {ticker}: {e}") + + return result + + # ------------------------------------------------------------------ + # Step 3: Historical screening + # ------------------------------------------------------------------ + + async def screen_historical( + self, + db: AsyncSession, + date_str: str, + market_cap_min: Optional[float], + market_cap_max: Optional[float], + sector: Optional[str], + exchange: Optional[str], + page: int, + page_size: int, + sort_by: str, + sort_ascending: bool, + ) -> Tuple[List[Dict], int, str]: + """ + Screen stocks at a historical date based on market_cap and other criteria. + + Returns: (items, total_count, actual_snapshot_date_str) + """ + try: + target = date.fromisoformat(date_str) + except ValueError: + raise ValueError(f"Invalid date format: {date_str!r} — use YYYY-MM-DD") + + snapshot_dt = datetime(target.year, target.month, 1, tzinfo=timezone.utc) + + conditions = [UniverseSnapshot.snapshot_date == snapshot_dt] + if market_cap_min is not None: + conditions.append(UniverseSnapshot.market_cap >= market_cap_min) + if market_cap_max is not None: + conditions.append(UniverseSnapshot.market_cap <= market_cap_max) + if sector: + conditions.append(UniverseSnapshot.sector == sector) + if exchange: + conditions.append(UniverseSnapshot.exchange == exchange.upper()) + + filter_clause = and_(*conditions) + + # Total count + count_q = await db.execute( + select(func.count(UniverseSnapshot.id)).where(filter_clause) + ) + total = count_q.scalar() or 0 + + # Sort + sort_col = { + "market_cap": UniverseSnapshot.market_cap, + "ticker": UniverseSnapshot.ticker, + }.get(sort_by, UniverseSnapshot.market_cap) + order = sort_col.asc() if sort_ascending else desc(sort_col) + + # Paginated fetch + rows_result = await db.execute( + select(UniverseSnapshot) + .where(filter_clause) + .order_by(order) + .limit(page_size) + .offset((page - 1) * page_size) + ) + rows = rows_result.scalars().all() + + # Enrich with names from registry + tickers_in_page = [r.ticker for r in rows] + name_map: Dict[str, Optional[str]] = {} + if tickers_in_page: + name_result = await db.execute( + select(UniverseTickerRegistry.ticker, UniverseTickerRegistry.name) + .where(UniverseTickerRegistry.ticker.in_(tickers_in_page)) + ) + name_map = {r[0]: r[1] for r in name_result.fetchall()} + + items = [ + { + "ticker": r.ticker, + "name": name_map.get(r.ticker), + "market_cap": r.market_cap, + "close_price": r.close_price, + "shares_outstanding": r.shares_outstanding, + "sector": r.sector, + "industry": r.industry, + "exchange": r.exchange, + "snapshot_date": r.snapshot_date, + } + for r in rows + ] + return items, total, snapshot_dt.strftime("%Y-%m-%d") + + async def get_registry( + self, + db: AsyncSession, + sector: Optional[str], + exchange: Optional[str], + is_active: Optional[bool], + page: int, + page_size: int, + ) -> Tuple[List, int]: + """Browse the ticker registry with optional filters.""" + conditions = [] + if sector: + conditions.append(UniverseTickerRegistry.sector == sector) + if exchange: + conditions.append(UniverseTickerRegistry.exchange == exchange.upper()) + if is_active is not None: + conditions.append(UniverseTickerRegistry.is_active == is_active) + + where = and_(*conditions) if conditions else True + + count_q = await db.execute( + select(func.count(UniverseTickerRegistry.id)).where(where) + ) + total = count_q.scalar() or 0 + + result = await db.execute( + select(UniverseTickerRegistry) + .where(where) + .order_by(UniverseTickerRegistry.ticker) + .limit(page_size) + .offset((page - 1) * page_size) + ) + return result.scalars().all(), total + + +# ------------------------------------------------------------------ +# Module-level helpers +# ------------------------------------------------------------------ + +def _extract_shares_history( + facts_data: dict, +) -> List[Tuple[date, float]]: + """ + Extract sorted (period_end_date, shares_outstanding) from SEC companyfacts JSON. + Prefers CommonStockSharesOutstanding over weighted average concepts. + """ + if not facts_data or "facts" not in facts_data: + return [] + + us_gaap = facts_data["facts"].get("us-gaap", {}) + _CONCEPTS = [ + "CommonStockSharesOutstanding", + "WeightedAverageNumberOfSharesOutstandingBasic", + "WeightedAverageNumberOfDilutedSharesOutstanding", + ] + _FORMS = {"10-K", "10-Q", "10-K/A", "10-Q/A"} + + all_points: Dict[date, float] = {} + for concept in _CONCEPTS: + if concept not in us_gaap: + continue + for unit_key, entries in us_gaap[concept].get("units", {}).items(): + if unit_key != "shares": + continue + for entry in entries: + if entry.get("form") not in _FORMS: + continue + end_str = entry.get("end") + val = entry.get("val") + if not end_str or val is None: + continue + try: + end_d = date.fromisoformat(end_str) + all_points[end_d] = float(val) + except (ValueError, TypeError): + continue + + return sorted(all_points.items()) + + +def _get_shares_at_date( + shares_history: List[Tuple[date, float]], target: date +) -> Optional[float]: + """Carry-forward: most recent shares_outstanding on or before target date.""" + result = None + for entry_date, shares in shares_history: + if entry_date <= target: + result = shares + else: + break + return result