feat: 백테스팅 유니버스 과거 시점 주식 스크리닝 (Historical Stock Universe)

## 새 기능
- 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 <noreply@anthropic.com>

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")

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()
 
@@ -29,3 +29,4 @@ 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"])
+api_router.include_router(universe.router, prefix="/universe", tags=["universe"])

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."
+        ),
+    }

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",
 ]

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"),
+    )

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",
+    )

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