fix: price_data_service null/zero close 자동 재수집 + Yahoo rate limit 429 + earnings 캐시

- price_data_service: null/zero close를 missing으로 처리 → 재수집 트리거
  UTC 자정 정규화로 yf.history / yf.download 중복 방지
  get_quote fallback: .info 실패 시 fast_info + history('2d') 체인
  historical upsert: close IS NULL/0인 경우만 덮어쓰기 (정상 데이터 보호)
- price.py: Yahoo rate limit → HTTP 429 + Retry-After: 30 응답
- earnings.py: bulk calendar 엔드포인트 캐시 1시간 추가

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alembic/versions/q8h9i0j1k2l3_add_pit_universe_membership_view.py

@@ -0,0 +1,71 @@
+"""add pit_universe_membership view
+
+Revision ID: q8h9i0j1k2l3
+Revises: p7g8h9i0j1k2
+Create Date: 2026-05-30
+
+PIT(Point-in-Time) 유니버스 멤버십 뷰.
+
+finra_short_volume에서 (date, symbol)을 그대로 노출하는 뷰.
+각 날짜에 실제 거래되던 종목 집합 = PIT 멤버십.
+
+생존편향-0 달성 근거:
+  - finra_short_volume은 2018-08-01 ~ 현재, 22,722개 심볼 보유
+  - 그 중 16,143개는 현 활성 유니버스(9,635)에 없는 상폐/합병/과거 종목
+  - SIVB(2023-03-09), SBNY(2023-03-10), FRC(2023-04-28), TWTR(2022-10-27) 등
+    실제 상폐일에 정확히 끊김 → 생존편향 없이 PIT 멤버십 표현
+
+사용 예:
+  -- 특정 날짜에 거래된 종목 집합 (PIT 유니버스)
+  SELECT DISTINCT symbol
+  FROM pit_universe_membership
+  WHERE d = '2023-03-09'
+  ORDER BY symbol;
+
+  -- 공매도비율 × PIT 멤버십 (생존편향-0 횡단면)
+  SELECT p.d, p.symbol, p.short_ratio, a.close
+  FROM pit_universe_membership p
+  LEFT JOIN alpaca_price_data a
+    ON a.ticker = p.symbol AND a.date::date = p.d AND a.interval = '1d'
+  WHERE p.d = '2023-03-09'
+  ORDER BY p.short_ratio;
+
+한계:
+  - 티커 재활용: BBBY(2023-05 파산 → 2025-08 다른 엔티티 재사용) 등
+    심볼 기준 PIT는 상폐 경계에서 두 회사를 혼동할 수 있음.
+    엄밀 해결은 CUSIP/PERMNO(유료) 매핑 필요.
+  - 2020-04-01 ~ 2020-10-31 데이터 결손 (COVID 갭).
+    backfill_finra_2020_gap.py 실행 후 해소 가능.
+  - NMS 슈퍼셋: ETF/ADR/워런트(/U, /WS) 포함.
+    분석단에서 심볼 패턴 필터 권장.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "q8h9i0j1k2l3"
+down_revision: Union[str, Sequence[str], None] = "p7g8h9i0j1k2"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # CREATE OR REPLACE VIEW: idempotent, safe to re-run
+    # 성능: idx_finra_date (date), idx_finra_symbol_date (symbol, date) 인덱스를
+    # Postgres가 베이스 테이블에서 자동 활용하므로 별도 뷰 인덱스 불필요
+    op.execute("""
+        CREATE OR REPLACE VIEW pit_universe_membership AS
+        SELECT
+            date::date              AS d,
+            symbol,
+            total_volume,
+            short_volume,
+            short_exempt_volume,
+            short_ratio,
+            market
+        FROM finra_short_volume
+    """)
+
+
+def downgrade() -> None:
+    op.execute("DROP VIEW IF EXISTS pit_universe_membership")

app/api/v1/endpoints/alpaca.py

@@ -4,11 +4,17 @@ Alpaca Market Data endpoints — standalone price data via Alpaca API
 
 import asyncio
 import gc
+import logging
 from datetime import date, datetime, timezone, timedelta
 from typing import Optional
 from zoneinfo import ZoneInfo
 
-from fastapi import APIRouter, HTTPException, Query
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Query
+from sqlalchemy import text
+
+from app.core.database import AsyncSessionLocal
+
+logger = logging.getLogger(__name__)
 
 _ET = ZoneInfo("America/New_York")
 _MARKET_CLOSE_HOUR = 16  # 4:00 PM ET
@@ -325,6 +331,190 @@ def _parse_snapshot(ticker: str, raw: dict) -> AlpacaSnapshotResponse:
     )
 
 
+# ------------------------------------------------------------------
+# Admin: PIT price backfill
+# ------------------------------------------------------------------
+
+def _finra_to_alpaca(symbol: str) -> str:
+    """FINRA "/" → Alpaca "." (BRK/B → BRK.B, AAC/U → AAC.U)."""
+    return symbol.replace("/", ".").replace("-", ".")
+
+
+async def _run_pit_backfill(start_str: str, end_str: str, force: bool) -> None:
+    """Background task: backfill Alpaca 1d bars for all FINRA PIT symbols."""
+    from sqlalchemy.dialects.postgresql import insert as pg_insert
+    from app.models.alpaca_price import AlpacaPriceData
+
+    BATCH_SIZE = 100
+    CHUNK_SIZE = 2300  # asyncpg 32767 bind-param limit
+    ADJUSTMENT  = "all"
+    TOLERANCE_DAYS = 5
+
+    client = AlpacaClient()
+    if not client.is_configured():
+        logger.error("PIT backfill: Alpaca keys not configured")
+        return
+
+    try:
+        # Step 1: FINRA PIT 심볼 + 활동기간
+        async with AsyncSessionLocal() as db:
+            rows = (await db.execute(text("""
+                SELECT symbol,
+                       MIN(date)::date AS finra_first,
+                       MAX(date)::date AS finra_last,
+                       COUNT(DISTINCT date::date) AS finra_days
+                FROM finra_short_volume
+                GROUP BY symbol
+            """))).fetchall()
+        finra_info = {r.symbol: (r.finra_first, r.finra_last, r.finra_days) for r in rows}
+        total_obs = sum(v[2] for v in finra_info.values())
+        logger.info(f"PIT backfill: {len(finra_info):,} FINRA symbols")
+
+        # Step 2: 기존 Alpaca 1d 커버리지 (정규화 키로 저장)
+        async with AsyncSessionLocal() as db:
+            cov_rows = (await db.execute(text("""
+                SELECT ticker, MAX(date)::date AS alpaca_max
+                FROM alpaca_price_data WHERE interval = '1d'
+                GROUP BY ticker
+            """))).fetchall()
+        alpaca_cov: dict = {}
+        for r in cov_rows:
+            key = _finra_to_alpaca(r.ticker).upper()
+            if key not in alpaca_cov or r.alpaca_max > alpaca_cov[key]:
+                alpaca_cov[key] = r.alpaca_max
+
+        # Step 3: 수집 필요 심볼 결정
+        if force:
+            need_fetch = list(finra_info.keys())
+        else:
+            need_fetch = [
+                sym for sym, (_, finra_last, _) in finra_info.items()
+                if (am := alpaca_cov.get(_finra_to_alpaca(sym).upper())) is None
+                or (finra_last - am).days > TOLERANCE_DAYS
+            ]
+        logger.info(f"PIT backfill: fetching {len(need_fetch):,} / {len(finra_info):,} symbols")
+
+        # Step 4: 배치 수집 + upsert
+        total_inserted = 0
+        n_batches = (len(need_fetch) + BATCH_SIZE - 1) // BATCH_SIZE
+
+        for batch_idx in range(0, len(need_fetch), BATCH_SIZE):
+            batch = need_fetch[batch_idx: batch_idx + BATCH_SIZE]
+            batch_num = batch_idx // BATCH_SIZE + 1
+            if batch_num == 1 or batch_num % 50 == 0:
+                logger.info(f"PIT backfill batch {batch_num}/{n_batches} | inserted={total_inserted:,}")
+
+            # FINRA "/" → Alpaca "." 변환 후 전송 (미변환 시 Alpaca 400)
+            # reverse_map: 정규화된 Alpaca 심볼 → FINRA 원본 심볼
+            reverse_map = {_finra_to_alpaca(s).upper(): s for s in batch}
+            alpaca_batch = [_finra_to_alpaca(s) for s in batch]
+            try:
+                raw = await client.get_multi_bars(
+                    symbols=alpaca_batch, timeframe="1d",
+                    start=start_str, end=end_str,
+                    adjustment=ADJUSTMENT,
+                )
+            except Exception as exc:
+                logger.warning(f"PIT backfill batch {batch_num} error: {exc}")
+                continue
+
+            rows_to_insert = []
+            for alpaca_sym, bar_list in raw.items():
+                if not bar_list:
+                    continue
+                original = reverse_map.get(alpaca_sym.upper(), alpaca_sym)
+                for bar in bar_list:
+                    dt = datetime.fromisoformat(bar["t"].replace("Z", "+00:00"))
+                    if dt.tzinfo is None:
+                        dt = dt.replace(tzinfo=timezone.utc)
+                    rows_to_insert.append({
+                        "ticker":      original,
+                        "date":        dt,
+                        "interval":    "1d",
+                        "open":        float(bar.get("o") or 0),
+                        "high":        float(bar.get("h") or 0),
+                        "low":         float(bar.get("l") or 0),
+                        "close":       float(bar.get("c") or 0),
+                        "volume":      float(bar.get("v") or 0),
+                        "vwap":        float(bar["vw"]) if bar.get("vw") else None,
+                        "trade_count": int(bar["n"])    if bar.get("n")  else None,
+                        "data_source": "ALPACA",
+                    })
+
+            if rows_to_insert:
+                async with AsyncSessionLocal() as db:
+                    for i in range(0, len(rows_to_insert), CHUNK_SIZE):
+                        stmt = pg_insert(AlpacaPriceData).values(
+                            rows_to_insert[i: i + CHUNK_SIZE]
+                        )
+                        stmt = stmt.on_conflict_do_nothing(constraint="uq_alpaca_price_data")
+                        result = await db.execute(stmt)
+                        total_inserted += result.rowcount
+                        await asyncio.sleep(0)
+                    await db.commit()
+
+        # Step 5: 완료 리포트
+        async with AsyncSessionLocal() as db:
+            new_cov = {
+                _finra_to_alpaca(r.ticker).upper()
+                for r in (await db.execute(text(
+                    "SELECT DISTINCT ticker FROM alpaca_price_data WHERE interval='1d'"
+                ))).fetchall()
+            }
+        covered_obs  = sum(fd for sym, (_, _, fd) in finra_info.items() if _finra_to_alpaca(sym).upper() in new_cov)
+        missing_obs  = total_obs - covered_obs
+        bias_pct     = 100.0 * missing_obs / total_obs if total_obs else 0
+        logger.info(
+            f"PIT backfill complete — inserted={total_inserted:,} | "
+            f"covered={len(new_cov):,} symbols | "
+            f"residual bias={bias_pct:.2f}% (row-weighted)"
+        )
+    finally:
+        await client.close()
+
+
+@router.post(
+    "/admin/backfill-pit",
+    summary="PIT 가격 백필 — 상폐 종목 포함 전체 FINRA 심볼",
+    description=(
+        "FINRA short-volume DB에 등장한 모든 심볼(현 활성 유니버스 + 상폐/합병 과거 심볼)의 "
+        "Alpaca SIP 일봉(1d)을 백필합니다. 생존편향-0 수익 계산에 필요.\n\n"
+        "**특성**:\n"
+        "- `adjustment=all` (분할+배당 조정) — 상폐 종목은 future-proof\n"
+        "- DB-first, idempotent (`on_conflict_do_nothing`) — 재실행 안전\n"
+        "- Alpaca SIP 일봉은 무료 플랜에서 2016-01-04부터 제공\n"
+        "- 기본 시작일: 2018-08-01 (FINRA DB 시작일)\n\n"
+        "**백그라운드 실행**: 즉시 `started` 응답, 1-3시간 소요.\n"
+        "진행 상황: `GET /alpaca/status` 또는 DB `SELECT COUNT(DISTINCT ticker) FROM alpaca_price_data WHERE interval='1d';`\n\n"
+        "**PIT 뷰** (백필 후): `SELECT DISTINCT symbol FROM pit_universe_membership WHERE d='2023-03-09';`"
+    ),
+)
+async def backfill_pit_prices(
+    background_tasks: BackgroundTasks,
+    start_date: date = Query(date(2018, 8, 1), description="백필 시작일 (기본: 2018-08-01)"),
+    force: bool = Query(False, description="이미 커버된 심볼도 재수집"),
+):
+    svc = _require_alpaca()  # API 키 확인
+    _ = svc  # 키 확인용
+
+    end_date = datetime.now(timezone.utc).date() - timedelta(days=1)
+    start_str = start_date.isoformat()
+    end_str   = end_date.isoformat()
+
+    background_tasks.add_task(_run_pit_backfill, start_str, end_str, force)
+
+    return {
+        "status":     "started",
+        "start_date": start_str,
+        "end_date":   end_str,
+        "adjustment": "all",
+        "note": (
+            "Backfilling Alpaca 1d bars for all FINRA PIT symbols in background. "
+            "Typically 1-3 hours for ~22k symbols. Check container logs for progress."
+        ),
+    }
+
+
 @router.get(
     "/snapshot",
     response_model=AlpacaMultiSnapshotResponse,

app/api/v1/endpoints/attention.py

@@ -15,25 +15,27 @@ Admin endpoints:
 
 import asyncio
 import logging
-from datetime import date, timedelta
+from datetime import date, datetime, timedelta, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db
-from app.utils.cache import with_cache
+from app.utils.cache import build_cache_key, get_cached_response, set_cached_response, with_cache
 from app.models.attention import AttentionFeaturesDaily, CompanyEntityMap, GdeltArticleRaw
 from app.services.attention.gdelt_collector import GDELT_EARLIEST_DATE
 from app.schemas.attention import (
     CollectionStatusResponse,
     EntityInfo,
+    EntityOverrideResponse,
     EntityResolveResponse,
     EventAttentionResponse,
     NewsFeatures,
     WikiFeatures,
 )
 from app.services.attention.entity_resolver import resolve_entity
+from app.services.attention.entity_resolver import _build_gdelt_query, _normalize_name
 from app.services.attention.feature_materializer import materialize_features
 from app.services.attention.gdelt_collector import collect_gdelt_articles
 from app.services.attention.wiki_collector import collect_wiki_pageviews
@@ -46,6 +48,22 @@ router = APIRouter()
 _EVENT_SEMAPHORE = asyncio.Semaphore(8)
 _EVENT_SEMAPHORE_WAIT = 10
 
+# Re-resolve a present-but-unresolved entity at most once per this window.
+# Bounds the Wikipedia query rate when fithia2 re-runs its bulk upcoming-
+# earnings scan (at semaphore=8) against still-NULL rows — prevents the 429
+# storm that originally left 1423/1697 mappings broken.
+_RESOLVE_RETRY_WINDOW = timedelta(hours=1)
+
+
+def _resolve_retry_due(updated_at) -> bool:
+    """True if enough time has elapsed since the last resolution attempt to
+    retry a still-unresolved entity. A missing timestamp counts as due."""
+    if updated_at is None:
+        return True
+    if updated_at.tzinfo is None:
+        updated_at = updated_at.replace(tzinfo=timezone.utc)
+    return datetime.now(timezone.utc) - updated_at >= _RESOLVE_RETRY_WINDOW
+
 
 def _entity_to_info(entity: CompanyEntityMap) -> EntityInfo:
     return EntityInfo(
@@ -226,6 +244,62 @@ async def admin_collect_gdelt(
     )
 
 
+@router.post(
+    "/admin/entity/{ticker}/override",
+    response_model=EntityOverrideResponse,
+    summary="Manually set wiki_title for a ticker (override automatic resolver)",
+    description=(
+        "Directly sets the Wikipedia article title for a ticker, bypassing the automatic resolver. "
+        "Sets `is_manual_override=True` so the resolver will never overwrite this mapping.\n\n"
+        "Use when the resolver persistently picks the wrong article "
+        "(e.g. a lawsuit page or an acquired subsidiary instead of the company itself).\n\n"
+        "**Example**: `POST /admin/entity/CSCO/override?wiki_title=Cisco%20Systems`"
+    ),
+    tags=["attention-admin"],
+)
+async def admin_override_entity(
+    ticker: str,
+    wiki_title: str = Query(..., description="Exact Wikipedia article title to use"),
+    db: AsyncSession = Depends(get_db),
+) -> EntityOverrideResponse:
+    ticker = ticker.upper()
+
+    entity_result = await db.execute(
+        select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker)
+    )
+    entity = entity_result.scalars().first()
+    if not entity:
+        raise HTTPException(
+            status_code=404,
+            detail=f"No entity mapping found for {ticker}. POST /admin/resolve/{ticker} first.",
+        )
+
+    canonical_name = entity.canonical_name
+    aliases = entity.aliases_json or []
+    gdelt_query = _build_gdelt_query(canonical_name, aliases)
+
+    from sqlalchemy import update as sa_update
+    await db.execute(
+        sa_update(CompanyEntityMap)
+        .where(CompanyEntityMap.ticker == ticker)
+        .values(
+            wiki_title=wiki_title,
+            gdelt_query=gdelt_query,
+            is_manual_override=True,
+            resolver_confidence=1.0,
+        )
+    )
+    await db.commit()
+    logger.info("Manual override applied: %s → wiki_title=%r", ticker, wiki_title)
+
+    return EntityOverrideResponse(
+        ticker=ticker,
+        wiki_title=wiki_title,
+        gdelt_query=gdelt_query,
+        message=f"Manual override set: {ticker} → {wiki_title!r}. is_manual_override=True.",
+    )
+
+
 # ===========================================================================
 # Parameterized entity route (before /event/ to avoid shadowing)
 # ===========================================================================
@@ -273,6 +347,10 @@ async def get_entity(
 # ===========================================================================
 
 
+_EVENT_CACHE_TTL_COMPLETE = 3600   # full response with wiki data → 1 h
+_EVENT_CACHE_TTL_INCOMPLETE = 300  # wiki_views still null → 5 min, retry sooner
+
+
 @router.get(
     "/event/{ticker}",
     response_model=EventAttentionResponse,
@@ -306,7 +384,6 @@ async def get_entity(
         500: {"description": "Feature materialization or collection error"},
     },
 )
-@with_cache(namespace="attention:event", ttl=3600, key_params=["ticker", "event_date"])
 async def get_event_attention(
     ticker: str,
     event_date: date = Query(..., description="Event date in YYYY-MM-DD format"),
@@ -314,6 +391,17 @@ async def get_event_attention(
     db: AsyncSession = Depends(get_db),
 ) -> EventAttentionResponse:
     ticker = ticker.upper()
+    cache_key = build_cache_key("attention:event", ticker, str(event_date))
+
+    cached = await get_cached_response(cache_key)
+    if cached:
+        cached_body, etag = cached
+        if response is not None:
+            response.headers["X-Cache"] = "HIT"
+            response.headers["Cache-Control"] = f"public, max-age={_EVENT_CACHE_TTL_COMPLETE}"
+            response.headers["ETag"] = etag
+            response.headers["X-Data-Source"] = "redis-cache"
+        return cached_body
 
     try:
         await asyncio.wait_for(_EVENT_SEMAPHORE.acquire(), timeout=_EVENT_SEMAPHORE_WAIT)
@@ -323,10 +411,22 @@ async def get_event_attention(
             detail="서버가 바빠서 요청을 처리할 수 없습니다. 잠시 후 다시 시도하세요.",
         )
     try:
-        return await _get_event_attention_impl(ticker, event_date, db)
+        result = await _get_event_attention_impl(ticker, event_date, db)
     finally:
         _EVENT_SEMAPHORE.release()
 
+    # Incomplete responses (wiki_views still null) cache for only 5 min so they
+    # re-trigger collection sooner rather than serving stale nulls for up to 1 h.
+    wiki_complete = result.wiki.views is not None
+    ttl = _EVENT_CACHE_TTL_COMPLETE if wiki_complete else _EVENT_CACHE_TTL_INCOMPLETE
+    etag = await set_cached_response(cache_key, result.model_dump(), ttl_seconds=ttl)
+    if response is not None:
+        response.headers["X-Cache"] = "MISS"
+        response.headers["Cache-Control"] = f"public, max-age={ttl}"
+        response.headers["ETag"] = etag
+
+    return result
+
 
 async def _get_event_attention_impl(
     ticker: str,
@@ -348,6 +448,23 @@ async def _get_event_attention_impl(
         except Exception as exc:
             logger.error("Auto entity resolution failed for %s: %s", ticker, exc)
             raise HTTPException(status_code=500, detail=f"Entity resolution failed: {exc}")
+    elif (
+        not entity.wiki_title
+        and (entity.resolver_confidence or 0.0) < 0.5
+        and not entity.is_manual_override
+        and _resolve_retry_due(entity.updated_at)
+    ):
+        # Present-but-unresolved row — e.g. a transient Wikipedia 429 during a
+        # prior bulk run persisted wiki_title=NULL. Retry resolution, but at
+        # most once per _RESOLVE_RETRY_WINDOW so a bulk scan over still-NULL
+        # tickers can't re-create the 429 storm. Best-effort: on failure keep
+        # serving the (empty) row rather than 500-ing. The resolver's
+        # no-downgrade guard ensures a failed retry can't worsen the row.
+        logger.info("Entity %s unresolved — attempting on-demand re-resolve", ticker)
+        try:
+            entity = await resolve_entity(db, ticker)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("On-demand re-resolve failed for %s: %s", ticker, exc)
 
     # 2. Check if features already exist in DB
     features_result = await db.execute(
@@ -358,12 +475,19 @@ async def _get_event_attention_impl(
     )
     features = features_result.scalars().first()
 
-    if features is None:
-        # 3. On-demand collection + materialization
+    needs_wiki = entity.wiki_title and (
+        features is None
+        or features.wiki_views is None
+        or features.wiki_zscore_20d is None
+    )
+    if features is None or needs_wiki:
+        # 3. On-demand collection + materialization.
+        # Re-triggers when features are missing OR wiki_views / wiki_zscore_20d is null:
+        # - wiki_views null: wiki_title was unset at collection time, or API failed
+        # - wiki_zscore_20d null: insufficient lookback data when previously materialized
         # NOTE: GDELT is intentionally excluded here — it must be collected via
         # the scheduler (POST /admin/collect/gdelt/{ticker}) to avoid IP rate bans.
-        # This endpoint only collects Wikipedia data on-demand.
-        logger.info("No features for %s on %s — collecting wiki on-demand", ticker, event_date)
+        logger.info("No features (or incomplete wiki) for %s on %s — collecting wiki on-demand", ticker, event_date)
 
         if entity.wiki_title:
             try:

app/api/v1/endpoints/earnings.py

@@ -102,6 +102,7 @@ async def get_earnings_calendar(
         "Useful for checking upcoming earnings of sector peers or candidates."
     ),
 )
+@with_cache(namespace="earnings:calendar_bulk", ttl=3600, key_params=["request"])
 async def get_bulk_earnings_calendar(
     request: BulkEarningsCalendarRequest,
     response: Response,

app/api/v1/endpoints/finra.py

@@ -3,10 +3,11 @@ FINRA Short Sale Volume endpoints
 """
 
 from datetime import date, datetime, timedelta, timezone
-from typing import Optional
+from typing import List, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 from fastapi.responses import Response
+from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.config import settings
@@ -104,6 +105,85 @@ async def get_short_ratio(
     return body
 
 
+@router.get(
+    "/pit-panel",
+    summary="PIT 횡단면 패널 — 특정 날짜 전체 심볼 (생존편향-0)",
+    description=(
+        "특정 날짜(또는 날짜 범위)에 실제 거래되던 모든 종목의 공매도량 + 종가를 반환.\n\n"
+        "`pit_universe_membership` 뷰 ⋈ `alpaca_price_data(interval='1d', adjustment='all')` 조인.\n\n"
+        "**생존편향-0**: 상폐/합병 종목(SIVB, FRC, TWTR 등)도 그날 거래됐으면 포함됨.\n\n"
+        "**날짜 범위**: `date_from`/`date_to` 둘 다 지정 시 최대 `limit`일치 반환 (기본 1일).\n\n"
+        "**주의**: 전체 패널(22k×8yr)은 날짜별 반복 호출로 조합. "
+        "단일 날짜 응답은 ~11k rows."
+    ),
+)
+async def get_pit_panel(
+    date_from: date = Query(..., description="조회 시작일 (YYYY-MM-DD)"),
+    date_to: Optional[date] = Query(None, description="조회 종료일 — 생략 시 date_from 단일 날짜"),
+    limit: int = Query(50000, ge=1, le=200000, description="최대 반환 행 수"),
+    db: AsyncSession = Depends(get_db),
+):
+    if date_to is None:
+        date_to = date_from
+    if date_from > date_to:
+        raise HTTPException(status_code=400, detail="date_from must be <= date_to")
+
+    rows = (await db.execute(
+        text("""
+            SELECT
+                p.d                   AS date,
+                p.symbol,
+                p.short_volume,
+                p.short_exempt_volume,
+                p.total_volume,
+                p.short_ratio,
+                a.open,
+                a.high,
+                a.low,
+                a.close,
+                a.volume              AS price_volume,
+                a.vwap
+            FROM pit_universe_membership p
+            LEFT JOIN alpaca_price_data a
+                ON a.ticker    = p.symbol
+               AND a.date::date = p.d
+               AND a.interval  = '1d'
+            WHERE p.d BETWEEN :d_from AND :d_to
+            ORDER BY p.d, p.symbol
+            LIMIT :lim
+        """),
+        {"d_from": date_from, "d_to": date_to, "lim": limit},
+    )).fetchall()
+
+    data = [
+        {
+            "date":               str(r.date),
+            "symbol":             r.symbol,
+            "short_volume":       r.short_volume,
+            "short_exempt_volume":r.short_exempt_volume,
+            "total_volume":       r.total_volume,
+            "short_ratio":        r.short_ratio,
+            "open":               r.open,
+            "high":               r.high,
+            "low":                r.low,
+            "close":              r.close,
+            "price_volume":       r.price_volume,
+            "vwap":               r.vwap,
+        }
+        for r in rows
+    ]
+
+    price_matched = sum(1 for d in data if d["close"] is not None)
+    return {
+        "date_from":     date_from.isoformat(),
+        "date_to":       date_to.isoformat(),
+        "count":         len(data),
+        "price_matched": price_matched,
+        "price_coverage_pct": round(100 * price_matched / len(data), 1) if data else 0,
+        "data":          data,
+    }
+
+
 @router.post(
     "/admin/ingest",
     response_model=IngestResponse,

app/api/v1/endpoints/price.py

@@ -641,8 +641,18 @@ async def get_quote(
     use_prepost: bool = Query(True, description="Include pre/post market prices if available"),
 ):
     svc = PriceDataService()
-    data = await svc.get_quote(ticker, use_prepost=use_prepost)
-    return QuoteResponse(**data)
+    try:
+        data = await svc.get_quote(ticker, use_prepost=use_prepost)
+        return QuoteResponse(**data)
+    except Exception as e:
+        err = str(e).lower()
+        if "rate limit" in err or "too many requests" in err or "429" in err or "ratelimit" in err:
+            raise HTTPException(
+                status_code=429,
+                detail={"error_type": "RATE_LIMIT_ERROR", "message": "Yahoo Finance rate limit. Retry after a short delay."},
+                headers={"Retry-After": "30"},
+            )
+        raise HTTPException(status_code=500, detail={"error_type": "SERVER_ERROR", "message": str(e)})
 
 @router.get(
     "/intraday",
@@ -729,14 +739,20 @@ async def get_intraday(
     period: str = Query("1d"),
 ):
     svc = PriceDataService()
-    candles = await svc.get_intraday(ticker, interval=interval, period=period)
-    return IntradayResponse(
-        ticker=ticker.upper(),
-        interval=interval,
-        period=period,
-        candles=[IntradayCandle(**c) for c in candles],
-        metadata={"count": len(candles)}
-    )
+    try:
+        candles = await svc.get_intraday(ticker, interval=interval, period=period)
+        return IntradayResponse(
+            ticker=ticker.upper(),
+            interval=interval,
+            period=period,
+            candles=[IntradayCandle(**c) for c in candles],
+            metadata={"count": len(candles)}
+        )
+    except Exception as e:
+        err = str(e).lower()
+        if "rate limit" in err or "too many requests" in err or "429" in err or "ratelimit" in err:
+            raise HTTPException(status_code=429, detail={"error_type": "RATE_LIMIT_ERROR", "message": "Yahoo Finance rate limit. Retry after a short delay."}, headers={"Retry-After": "30"})
+        raise HTTPException(status_code=500, detail={"error_type": "SERVER_ERROR", "message": str(e)})
 
 @router.get(
     "/today/{ticker}",
@@ -748,5 +764,11 @@ async def get_today_ohlc(
     ticker: str,
 ):
     svc = PriceDataService()
-    data = await svc.get_today_ohlc(ticker)
-    return TodayOHLCResponse(**data)
+    try:
+        data = await svc.get_today_ohlc(ticker)
+        return TodayOHLCResponse(**data)
+    except Exception as e:
+        err = str(e).lower()
+        if "rate limit" in err or "too many requests" in err or "429" in err or "ratelimit" in err:
+            raise HTTPException(status_code=429, detail={"error_type": "RATE_LIMIT_ERROR", "message": "Yahoo Finance rate limit. Retry after a short delay."}, headers={"Retry-After": "30"})
+        raise HTTPException(status_code=500, detail={"error_type": "SERVER_ERROR", "message": str(e)})

app/schemas/attention.py

@@ -153,3 +153,10 @@ class CollectionStatusResponse(BaseModel):
     records_collected: int
     date_range: Dict[str, Any] = Field(default_factory=dict)
     status: str
+
+
+class EntityOverrideResponse(BaseModel):
+    ticker: str
+    wiki_title: str
+    gdelt_query: Optional[str]
+    message: str

app/services/alpaca_client.py

@@ -166,6 +166,7 @@ class AlpacaClient:
         end: Optional[str] = None,
         limit: int = 10000,
         feed: Optional[str] = None,
+        adjustment: Optional[str] = None,
     ) -> List[Dict]:
         """
         Fetch bars for a single symbol with automatic pagination.
@@ -178,6 +179,9 @@ class AlpacaClient:
             limit: Max bars per page (Alpaca max 10000)
             feed: Data feed ("iex" = free real-time, "sip" = paid consolidated).
                   Defaults to "iex" for intraday, no feed param for daily+.
+            adjustment: Price adjustment ("raw", "split", "dividend", "all").
+                        None = Alpaca default (raw). Use "all" for
+                        split+dividend adjusted bars (recommended for backtests).
 
         Returns:
             List of bar dicts with keys: t, o, h, l, c, v, n, vw
@@ -192,6 +196,8 @@ class AlpacaClient:
         effective_feed = feed or (_default_feed(timeframe))
         if effective_feed:
             params["feed"] = effective_feed
+        if adjustment:
+            params["adjustment"] = adjustment
 
         all_bars: List[Dict] = []
         path = f"/v2/stocks/{normalize_ticker(symbol).upper()}/bars"
@@ -217,6 +223,7 @@ class AlpacaClient:
         limit: int = 10000,
         batch_size: int = 100,
         feed: Optional[str] = None,
+        adjustment: Optional[str] = None,
     ) -> Dict[str, List[Dict]]:
         """
         Fetch bars for multiple symbols with auto-pagination and transparent batching.
@@ -232,6 +239,9 @@ class AlpacaClient:
             end: RFC-3339 date/datetime string
             limit: Max bars per page (Alpaca max 10000)
             batch_size: Max symbols per Alpaca request (default 100, conservative safe limit)
+            adjustment: Price adjustment ("raw", "split", "dividend", "all").
+                        None = Alpaca default (raw). Use "all" for
+                        split+dividend adjusted bars (recommended for backtests).
 
         Returns:
             Dict mapping normalized Alpaca symbol → list of bar dicts
@@ -259,6 +269,8 @@ class AlpacaClient:
                 params["end"] = end
             if effective_feed:
                 params["feed"] = effective_feed
+            if adjustment:
+                params["adjustment"] = adjustment
 
             while True:
                 data = await self._request("GET", path, params=params)

app/services/alpaca_price_service.py

@@ -32,10 +32,15 @@ class AlpacaPriceService:
         start_date: datetime,
         end_date: datetime,
         interval: str = "1d",
+        adjustment: Optional[str] = None,
     ) -> int:
         """
         Fetch bars from Alpaca and upsert into AlpacaPriceData table.
 
+        Args:
+            adjustment: Price adjustment ("raw", "split", "dividend", "all").
+                        None = Alpaca default (raw). Use "all" for backtests.
+
         Returns:
             Number of newly inserted records.
         """
@@ -49,6 +54,7 @@ class AlpacaPriceService:
             timeframe=interval,
             start=start_str,
             end=end_str,
+            adjustment=adjustment,
         )
 
         if not bars:
@@ -101,6 +107,7 @@ class AlpacaPriceService:
         interval: str = "1d",
         force_refresh: bool = False,
         feed: Optional[str] = None,
+        adjustment: Optional[str] = None,
     ) -> Dict[str, List[AlpacaPriceData]]:
         """
         DB-first multi-ticker daily bars.
@@ -112,6 +119,10 @@ class AlpacaPriceService:
         2. Fetch only missing tickers from Alpaca (no session held).
         3. Short session: upsert fetched rows.
         4. Short session: read back and return as Dict[ticker → rows].
+
+        Args:
+            adjustment: Price adjustment ("raw", "split", "dividend", "all").
+                        None = Alpaca default (raw). Use "all" for backtests.
         """
         upper_tickers = [t.upper() for t in tickers]
 
@@ -155,6 +166,7 @@ class AlpacaPriceService:
                 start=start_str,
                 end=end_str,
                 feed=feed,
+                adjustment=adjustment,
             )
 
             rows = []

app/services/attention/entity_resolver.py

@@ -11,10 +11,11 @@ Resolution pipeline:
 
 import logging
 import re
+from datetime import datetime, timezone
 from typing import Optional
 
 import httpx
-from sqlalchemy import select, update
+from sqlalchemy import or_, select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -26,13 +27,28 @@ logger = logging.getLogger(__name__)
 # Legal suffixes to strip for canonical name derivation
 _SUFFIX_PATTERN = re.compile(
     r",?\s+\b(Inc\.?|Corp\.?|Corporation|Holdings?|Ltd\.?|Limited|LLC|L\.L\.C\.|"
-    r"Group|Co\.?|Company|Technologies|Technology|International|Industries|"
+    r"L\.P\.?|LP|"  # Limited Partnership (e.g. "Enterprise Products Partners L.P.")
+    r"Group|Co\.?|Compan(?:y|ies)|Technologies|Technology|International|Industries|"
     r"Pharmaceuticals?|Therapeutics?|Sciences?|Bancorp|Financial|Holding|"
     r"Acquisition|Acquisitions|Capital|Partners|Trust|"
+    r"Nv|N\.V\.?|Plc\.?|p\.l\.c\.?|"  # Dutch (N.V.) / British (Plc / p.l.c.)
+    r"S\.A\.?B?(?:\s+de\s+C\.V\.)?|"  # Spanish (S.A., S.A.B., S.A.B. de C.V.)
+    r"Aktiengesellschaft|GmbH|AG|SE|"  # German/Swiss/European
     r"Com)\s*$",  # "Com" catches SEC-style ".com" artifacts (e.g. "AMAZON COM")
     re.IGNORECASE,
 )
 
+# SEC filing artifact: state-of-incorporation suffix after slash
+# e.g. "Costco Wholesale Corp /New", "Wells Fargo & Company/Mn", "Applied Materials Inc /DE",
+# "Canadian Imperial Bank Of Commerce /Can/"
+_SEC_NEW_PATTERN = re.compile(r"\s*/\s*(?:New|[A-Za-z]{2,4})\s*$", re.IGNORECASE)
+
+# Danish/Norwegian corporate designation: "NOVO NORDISK A/S" → "NOVO NORDISK"
+_AS_PATTERN = re.compile(r"\bA/S\s*$", re.IGNORECASE)
+
+# Belgian SA/NV: "Anheuser-Busch InBev SA/NV" → strip "/NV" first, then "SA" via suffix
+_SA_NV_PATTERN = re.compile(r"\s*SA/NV\s*$", re.IGNORECASE)
+
 _COMPANY_KEYWORDS = {
     "company", "corporation", "inc", "corp", "ltd", "llc", "holdings",
     "stock", "shares", "nasdaq", "nyse", "ticker", "finance", "financial",
@@ -47,7 +63,30 @@ def _normalize_name(raw_name: str) -> tuple[str, list[str]]:
     Preserves the original and intermediate forms as aliases.
     """
     aliases = []
-    current = raw_name.strip()
+    # Normalize path separators: some DB records use backslash (e.g. "US BANCORP \DE\")
+    current = raw_name.strip().replace("\\", "/").rstrip("/").strip()
+
+    # Belgian "SA/NV" corporate designation (e.g. "Anheuser-Busch InBev SA/NV")
+    stripped_sanv = _SA_NV_PATTERN.sub("", current).strip()
+    if stripped_sanv and stripped_sanv != current:
+        aliases.append(current)
+        current = stripped_sanv
+
+    # Strip Danish/Norwegian "A/S" corporate designation (e.g. "NOVO NORDISK A/S")
+    stripped_as = _AS_PATTERN.sub("", current).strip()
+    if stripped_as and stripped_as != current:
+        aliases.append(current)
+        current = stripped_as
+
+    # Strip SEC reincorporation artifact "/New" (and state/province codes like "/DE", "/Can")
+    stripped_new = _SEC_NEW_PATTERN.sub("", current).strip()
+    if stripped_new and stripped_new != current:
+        aliases.append(current)
+        current = stripped_new
+
+    # Normalize SEC dot-com artifact: "Amazon.Com" / "Amazon.com" → "Amazon Com"
+    # so the iterative loop can strip "Com" as a regular suffix.
+    current = re.sub(r"\.com\b", " Com", current, flags=re.IGNORECASE)
 
     for _ in range(5):  # max 5 iterations to avoid infinite loops
         stripped = _SUFFIX_PATTERN.sub("", current).strip().rstrip(",").strip()
@@ -56,7 +95,9 @@ def _normalize_name(raw_name: str) -> tuple[str, list[str]]:
         aliases.append(current)
         current = stripped
 
-    canonical = current
+    # Strip trailing punctuation artifacts left by suffix removal (e.g. "&" from
+    # "JPMorgan Chase & Co" → strip "Co" → "JPMorgan Chase &").
+    canonical = current.rstrip(" &/,").strip()
     # Also add the fully original name if not already captured
     if raw_name.strip() != canonical and raw_name.strip() not in aliases:
         aliases.insert(0, raw_name.strip())
@@ -154,7 +195,11 @@ def _score_wiki_result(result: dict, canonical_name: str, aliases: list[str]) ->
     """
     raw_title = result.get("title", "")
     title = raw_title.lower()
-    snippet = result.get("snippet", "").lower()
+    # Wikipedia snippets contain <span class="searchmatch"> HTML — strip before matching.
+    # Replace each tag with a space then collapse runs so "Costco</span> <span>Wholesale"
+    # becomes "Costco Wholesale" rather than "Costco  Wholesale" (breaking string match).
+    raw_snippet = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", result.get("snippet", ""))).strip()
+    snippet = raw_snippet.lower()
     combined = title + " " + snippet
 
     # Penalize obvious non-company pages immediately
@@ -165,31 +210,56 @@ def _score_wiki_result(result: dict, canonical_name: str, aliases: list[str]) ->
     if any(sig in combined for sig in non_company_signals):
         return 0.1
 
-    # Require snippet to have at least one finance keyword for non-exact-match titles
-    finance_signals = ["company", "corporation", "stock", "nasdaq", "nyse", "shares",
-                       "investor", "business", "enterprise", "holdings", "inc."]
-    if not any(sig in combined for sig in finance_signals):
-        return 0.15
+    # Penalize legal case titles: "X v. Y" format (e.g. "FSF v. Cisco Systems, Inc.")
+    if re.search(r'\w v\. \w', raw_title):
+        return 0.05
 
     canonical_lower = canonical_name.lower()
     all_names = [canonical_name] + aliases
     all_names_lower = [n.lower() for n in all_names]
 
-    # Check for exact title match (e.g. "Apple Inc." == alias "Apple Inc.")
+    # Exact title match — checked BEFORE the finance-signal filter so that
+    # valid company pages whose snippet focuses on technical/product details
+    # (e.g. TSMC → fabs, JPMorgan → banking operations) still score 0.95.
     raw_title_stripped = raw_title.strip()
     for name in all_names:
         if raw_title_stripped.lower() == name.lower():
-            return 0.95  # exact match
+            return 0.95
+
+    # Space-collapsed match: handles merged brand names like "ExxonMobil" vs "Exxon Mobil"
+    title_no_space = raw_title_stripped.lower().replace(" ", "")
+    for name in all_names:
+        name_no_space = name.lower().replace(" ", "")
+        if len(name_no_space) > 4 and title_no_space == name_no_space:
+            return 0.90
+
+    # Require snippet to have at least one finance keyword for non-exact-match titles
+    finance_signals = ["company", "corporation", "stock", "nasdaq", "nyse", "shares",
+                       "investor", "business", "enterprise", "holdings", "inc."]
+    if not any(sig in combined for sig in finance_signals):
+        return 0.15
+
+    # Hyphen-normalized forms: "COCA COLA" matches "The Coca-Cola Company"
+    # because "coca cola" is in "the coca cola company" after replacing hyphens with spaces.
+    title_norm = title.replace("-", " ")
+    canonical_norm = canonical_lower.replace("-", " ")
+    all_names_norm = [n.replace("-", " ") for n in all_names_lower]
 
     # Check if title starts with canonical name
-    title_starts_with_canonical = title.startswith(canonical_lower)
-    title_contains_canonical = canonical_lower in title
+    title_starts_with_canonical = title_norm.startswith(canonical_norm)
+    title_contains_canonical = canonical_norm in title_norm
 
     # Also check aliases
-    title_starts_with_alias = any(title.startswith(n) for n in all_names_lower)
-    title_contains_alias = any(n in title for n in all_names_lower)
+    title_starts_with_alias = any(title_norm.startswith(n) for n in all_names_norm)
+    title_contains_alias = any(n in title_norm for n in all_names_norm)
 
     if not (title_contains_canonical or title_contains_alias):
+        # Snippet-contains fallback: handles acronym titles (e.g. "TSMC" article whose
+        # snippet reads "Taiwan Semiconductor Manufacturing Company Limited (TSMC)...")
+        # and short-title articles (e.g. "Costco" snippet contains "Costco Wholesale").
+        snippet_norm = snippet.replace("-", " ")
+        if canonical_norm in snippet_norm or any(n in snippet_norm for n in all_names_norm):
+            return 0.6
         return 0.0
 
     # Check for pages that are ABOUT the company (vs. lists, histories, etc.)
@@ -207,11 +277,13 @@ def _score_wiki_result(result: dict, canonical_name: str, aliases: list[str]) ->
     else:
         base = 0.0
 
-    # Reward company/finance keywords in title or snippet
+    # Reward company/finance keywords in title or snippet.
+    # Cap at 0.89 so title-starts-with matches never outrank exact-title (0.95)
+    # or space-collapsed exact matches (0.90), regardless of keyword density.
     keyword_hits = sum(1 for kw in _COMPANY_KEYWORDS if kw in combined)
     keyword_score = min(keyword_hits / 3, 1.0)
 
-    return min(base + keyword_score * 0.3, 1.0)
+    return min(base + keyword_score * 0.3, 0.89)
 
 
 async def _resolve_wiki(
@@ -311,41 +383,60 @@ async def resolve_entity(
         ticker, wiki_title, confidence,
     )
 
-    # Upsert into company_entity_map
-    stmt = (
-        pg_insert(CompanyEntityMap)
-        .values(
-            ticker=ticker,
+    # Upsert into company_entity_map.
+    #
+    # No-downgrade guard: a transient Wikipedia failure (429/network) is
+    # swallowed by _search_wikipedia → returns [] → (wiki_title=None,
+    # confidence=0.0). Without this guard, re-running resolution while
+    # rate-limited would overwrite a previously-good mapping with NULL —
+    # which is exactly how the 2026-03-17 bulk run left 1423/1697 rows
+    # broken. The upsert therefore only updates when the new result is
+    # itself good (wiki_title not NULL) OR the existing row was already
+    # unresolved (wiki_title NULL). Manual overrides are never touched.
+    insert_stmt = pg_insert(CompanyEntityMap).values(
+        ticker=ticker,
+        canonical_name=canonical_name,
+        wiki_title=wiki_title,
+        gdelt_query=gdelt_query,
+        aliases_json=aliases,
+        resolver_confidence=confidence,
+        is_manual_override=False,
+    )
+    stmt = insert_stmt.on_conflict_do_update(
+        index_elements=["ticker"],
+        set_=dict(
             canonical_name=canonical_name,
             wiki_title=wiki_title,
             gdelt_query=gdelt_query,
             aliases_json=aliases,
             resolver_confidence=confidence,
-            is_manual_override=False,
-        )
-        .on_conflict_do_update(
-            index_elements=["ticker"],
-            set_=dict(
-                canonical_name=canonical_name,
-                wiki_title=wiki_title,
-                gdelt_query=gdelt_query,
-                aliases_json=aliases,
-                resolver_confidence=confidence,
-            ),
-            where=CompanyEntityMap.is_manual_override == False,  # noqa: E712
-        )
-        .returning(CompanyEntityMap)
-    )
+            # Explicit: ORM `onupdate` does NOT fire for INSERT...ON CONFLICT,
+            # so stamp it here. The on-demand re-resolve recency guard in the
+            # /event endpoint relies on this reflecting the last attempt.
+            updated_at=datetime.now(timezone.utc),
+        ),
+        where=(
+            (CompanyEntityMap.is_manual_override == False)  # noqa: E712
+            & or_(
+                insert_stmt.excluded.wiki_title.isnot(None),
+                CompanyEntityMap.wiki_title.is_(None),
+            )
+        ),
+    ).returning(CompanyEntityMap)
 
-    result = await db.execute(stmt)
+    await db.execute(stmt)
     await db.commit()
 
-    row = result.scalars().first()
-    if row is None:
-        # Manual override prevented update — return existing
-        existing_result2 = await db.execute(
-            select(CompanyEntityMap).where(CompanyEntityMap.ticker == ticker)
-        )
-        row = existing_result2.scalars().first()
+    # Read back via fresh SELECT with populate_existing=True.
+    # After commit, SQLAlchemy expires identity-map entries but does NOT evict them.
+    # A plain SELECT in the same session can return the expired (stale) cached object
+    # instead of reading the committed DB state. populate_existing forces the ORM to
+    # overwrite the identity-map entry with the fresh DB row.
+    fresh_result = await db.execute(
+        select(CompanyEntityMap)
+        .where(CompanyEntityMap.ticker == ticker)
+        .execution_options(populate_existing=True)
+    )
+    row = fresh_result.scalars().first()
 
     return row

app/services/attention/feature_materializer.py

@@ -169,19 +169,20 @@ async def materialize_features(
         .returning(AttentionFeaturesDaily)
     )
 
-    result = await db.execute(stmt)
+    await db.execute(stmt)
     await db.commit()
 
-    row = result.scalars().first()
-    if row is None:
-        # Fetch after upsert
-        row_result = await db.execute(
-            select(AttentionFeaturesDaily).where(
-                AttentionFeaturesDaily.ticker == ticker,
-                AttentionFeaturesDaily.date == event_date,
-            )
+    # Always re-fetch with populate_existing=True to avoid SQLAlchemy identity-map
+    # returning stale pre-upsert values (same pattern as resolve_entity).
+    row_result = await db.execute(
+        select(AttentionFeaturesDaily)
+        .where(
+            AttentionFeaturesDaily.ticker == ticker,
+            AttentionFeaturesDaily.date == event_date,
         )
-        row = row_result.scalars().first()
+        .execution_options(populate_existing=True)
+    )
+    row = row_result.scalars().first()
 
     logger.info(
         "Materialized features for %s on %s: wiki=%s gdelt_1d=%d gdelt_3d=%d",

app/services/price_data_service.py

@@ -6,7 +6,7 @@ from datetime import datetime, timezone, timedelta, date
 from typing import Dict, List, Optional, Tuple, Union
 import logging
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy import select, and_, desc
+from sqlalchemy import select, and_, desc, or_
 import asyncio
 import sys
 import os
@@ -120,14 +120,16 @@ class PriceDataService:
         if end_date.date() >= today:
             return [datetime.now()]
 
-        # Check if we have any data for this ticker and interval
+        # Count only rows with valid close (null/0 rows are treated as missing)
         result = await db.execute(
             select(PriceData.date)
             .where(
                 and_(
                     PriceData.ticker == ticker,
                     PriceData.date >= start_date,
-                    PriceData.date <= end_date
+                    PriceData.date <= end_date,
+                    PriceData.close.is_not(None),
+                    PriceData.close > 0,
                 )
             )
             .order_by(PriceData.date)
@@ -228,22 +230,28 @@ class PriceDataService:
         return hist_data
 
     async def get_quote(self, ticker: str, use_prepost: bool = True) -> Dict:
-        """Get latest quote using yfinance-plus .info fields with fallback to fast history last row."""
+        """Get latest quote using yfinance-plus .info fields with fallback to fast_info + history."""
         if not self.yf_available:
             raise ValueError("Yahoo Finance (yfinance-plus) data source not available")
+
+        yf_ticker = yf.Ticker(ticker)
+        loop = asyncio.get_event_loop()
+
+        # Try .info first; fall back to fast_info + history on timeout/error
+        info = None
         try:
-            yf_ticker = yf.Ticker(ticker)
-            loop = asyncio.get_event_loop()
             info = await _run_with_timeout(
                 loop.run_in_executor(None, lambda: yf_ticker.info),
                 timeout_seconds=20,
                 description=f"info {ticker}"
             )
-            # Prefer regular/post/pre values
+        except Exception as e:
+            logger.warning(f"get_quote .info failed for {ticker}: {e}, falling back to fast_info")
+
+        if info:
             regular = info.get("regularMarketPrice")
             post = info.get("postMarketPrice") if use_prepost else None
             pre = info.get("preMarketPrice") if use_prepost else None
-            price = post or pre or regular
             currency = info.get("currency")
             exchange = info.get("exchange") or info.get("fullExchangeName")
             market_state = info.get("marketState")
@@ -255,22 +263,67 @@ class PriceDataService:
                     ts = ts.replace(tzinfo=timezone.utc)
             else:
                 ts = datetime.now(timezone.utc)
-            return {
-                "ticker": ticker.upper(),
-                "price": float(price) if price is not None else None,
-                "regular_price": float(regular) if regular is not None else None,
-                "pre_market_price": float(pre) if pre is not None else None,
-                "post_market_price": float(post) if post is not None else None,
-                "currency": currency,
-                "exchange": exchange,
-                "market_state": market_state,
-                "timestamp": ts,
-                "source": DataSource.YAHOO_FINANCE,
-                "delayed": True,
-            }
-        except Exception as e:
-            logger.error(f"Error fetching quote for {ticker}: {str(e)}")
-            raise
+            pre_val = float(pre) if pre is not None else None
+            post_val = float(post) if post is not None else None
+            regular_val = float(regular) if regular is not None else None
+            price_val = post_val or pre_val or regular_val
+        else:
+            # Fallback: fast_info for price/currency/exchange, history for timestamp
+            try:
+                fast = await _run_with_timeout(
+                    loop.run_in_executor(None, lambda: yf_ticker.fast_info),
+                    timeout_seconds=10,
+                    description=f"fast_info {ticker}"
+                )
+                regular_val = getattr(fast, "last_price", None)
+                if regular_val is None and isinstance(fast, dict):
+                    regular_val = fast.get("lastPrice") or fast.get("last_price")
+                currency = getattr(fast, "currency", None) or (fast.get("currency") if isinstance(fast, dict) else None)
+                exchange = getattr(fast, "exchange", None) or (fast.get("exchange") if isinstance(fast, dict) else None)
+            except Exception as e2:
+                logger.error(f"get_quote fast_info also failed for {ticker}: {e2}")
+                raise
+
+            # Try to get last close from recent history for timestamp
+            try:
+                df = await _run_with_timeout(
+                    loop.run_in_executor(
+                        None,
+                        lambda: yf_ticker.history(period="2d", interval="1d", auto_adjust=True)
+                    ),
+                    timeout_seconds=15,
+                    description=f"history fallback {ticker}"
+                )
+                if not df.empty:
+                    last_row = df.iloc[-1]
+                    if regular_val is None:
+                        regular_val = float(last_row.get("Close", 0)) or None
+                    ts = df.index[-1].to_pydatetime()
+                    if ts.tzinfo is None:
+                        ts = ts.replace(tzinfo=timezone.utc)
+                else:
+                    ts = datetime.now(timezone.utc)
+            except Exception:
+                ts = datetime.now(timezone.utc)
+
+            pre_val = None
+            post_val = None
+            price_val = regular_val
+            market_state = None
+
+        return {
+            "ticker": ticker.upper(),
+            "price": float(price_val) if price_val is not None else None,
+            "regular_price": float(regular_val) if regular_val is not None else None,
+            "pre_market_price": pre_val,
+            "post_market_price": post_val,
+            "currency": currency,
+            "exchange": exchange,
+            "market_state": market_state,
+            "timestamp": ts,
+            "source": DataSource.YAHOO_FINANCE,
+            "delayed": True,
+        }
 
     async def get_intraday(self, ticker: str, interval: str = "1m", period: str = "1d") -> List[Dict]:
         """Get intraday candles using yfinance-plus history with period/interval."""
@@ -487,6 +540,15 @@ class PriceDataService:
             price_date = date_idx.to_pydatetime()
             if price_date.tzinfo is None:
                 price_date = price_date.replace(tzinfo=timezone.utc)
+            # Normalize to UTC midnight so uq_price_data(ticker, date) deduplicates
+            # correctly regardless of whether data came from yf.Ticker().history()
+            # (returns Eastern midnight = UTC 04:00) or yf.download() (returns UTC 00:00).
+            price_date = price_date.replace(hour=0, minute=0, second=0, microsecond=0,
+                                            tzinfo=timezone.utc)
+
+            close_val = _safe(row.get('Close'))
+            if close_val is None:  # Skip rows with no valid close price
+                continue
 
             target_rows = live_rows if price_date.date() >= today else historical_rows
             target_rows.append({
@@ -496,9 +558,9 @@ class PriceDataService:
                 'open': _safe(row.get('Open')),
                 'high': _safe(row.get('High')),
                 'low': _safe(row.get('Low')),
-                'close': _safe(row.get('Close')) or 0.0,
+                'close': close_val,
                 'volume': _safe(row.get('Volume')),
-                'adjusted_close': _safe(row.get('Close')),
+                'adjusted_close': close_val,
                 'data_source': DataSource.YAHOO_FINANCE.value,
                 'created_at': now,
                 'updated_at': now,
@@ -507,13 +569,23 @@ class PriceDataService:
         if not historical_rows and not live_rows:
             return
 
-        # Yahoo-adjusted historical prices are not point-in-time stable: future
-        # dividends/splits can rewrite old OHLC values. Preserve existing
-        # historical rows and only update today's row, where intraday partials
-        # legitimately need EOD replacement.
+        # Historical rows: preserve valid data but overwrite null/zero-close garbage
+        # (yf.download MultiIndex parse failures leave close=0 rows that must self-heal).
         if historical_rows:
             stmt = pg_insert(PriceData).values(historical_rows)
-            stmt = stmt.on_conflict_do_nothing(constraint='uq_price_data')
+            stmt = stmt.on_conflict_do_update(
+                constraint='uq_price_data',
+                set_={
+                    'open': stmt.excluded.open,
+                    'high': stmt.excluded.high,
+                    'low': stmt.excluded.low,
+                    'close': stmt.excluded.close,
+                    'volume': stmt.excluded.volume,
+                    'adjusted_close': stmt.excluded.adjusted_close,
+                    'updated_at': stmt.excluded.updated_at,
+                },
+                where=or_(PriceData.close.is_(None), PriceData.close == 0.0),
+            )
             await db.execute(stmt)
 
         if live_rows:
@@ -794,9 +866,9 @@ class PriceDataService:
             logger.debug(f"_batch_check_missing_periods: end_date includes today — forcing re-fetch for all {len(tickers)} tickers")
             return list(tickers)
 
-        # Single query to check all tickers at once
+        # Single query to check all tickers at once — count only valid rows
         from sqlalchemy import func, case
-        
+
         result = await db.execute(
             select(
                 PriceData.ticker,
@@ -808,7 +880,9 @@ class PriceDataService:
                 and_(
                     PriceData.ticker.in_(tickers),
                     PriceData.date >= start_date,
-                    PriceData.date <= end_date
+                    PriceData.date <= end_date,
+                    PriceData.close.is_not(None),
+                    PriceData.close > 0,
                 )
             )
             .group_by(PriceData.ticker)
@@ -918,8 +992,15 @@ class PriceDataService:
         
         # Handle different data structures from yfinance bulk download
         if len(tickers) == 1:
-            # Single ticker - data is a simple DataFrame
-            await self._store_ticker_data(db, tickers[0], bulk_data, interval)
+            # yf.download with group_by='ticker' returns MultiIndex columns even for a
+            # single ticker: [('SGOV', 'Open'), ('SGOV', 'Close'), ...].  Extract the
+            # ticker slice so _store_ticker_data receives a flat DataFrame.
+            ticker = tickers[0]
+            if hasattr(bulk_data.columns, 'levels') and ticker in bulk_data.columns.get_level_values(0):
+                ticker_data = bulk_data[ticker]
+            else:
+                ticker_data = bulk_data
+            await self._store_ticker_data(db, ticker, ticker_data, interval)
         else:
             # Multiple tickers - data is grouped by ticker
             for ticker in tickers:
@@ -960,6 +1041,13 @@ class PriceDataService:
             price_date = date_idx.to_pydatetime()
             if price_date.tzinfo is None:
                 price_date = price_date.replace(tzinfo=timezone.utc)
+            # Normalize to UTC midnight (same logic as _store_price_data)
+            price_date = price_date.replace(hour=0, minute=0, second=0, microsecond=0,
+                                            tzinfo=timezone.utc)
+
+            close_val = _safe(row.get('Close'))
+            if close_val is None:  # Skip rows with no valid close price
+                continue
 
             target_rows = live_rows if price_date.date() >= today else historical_rows
             target_rows.append({
@@ -969,9 +1057,9 @@ class PriceDataService:
                 'open': _safe(row.get('Open')),
                 'high': _safe(row.get('High')),
                 'low': _safe(row.get('Low')),
-                'close': _safe(row.get('Close')) or 0.0,
+                'close': close_val,
                 'volume': _safe(row.get('Volume')),
-                'adjusted_close': _safe(row.get('Close')),
+                'adjusted_close': close_val,
                 'data_source': DataSource.YAHOO_FINANCE.value,
                 'created_at': now,
                 'updated_at': now,
@@ -982,7 +1070,19 @@ class PriceDataService:
 
         if historical_rows:
             stmt = pg_insert(PriceData).values(historical_rows)
-            stmt = stmt.on_conflict_do_nothing(constraint='uq_price_data')
+            stmt = stmt.on_conflict_do_update(
+                constraint='uq_price_data',
+                set_={
+                    'open': stmt.excluded.open,
+                    'high': stmt.excluded.high,
+                    'low': stmt.excluded.low,
+                    'close': stmt.excluded.close,
+                    'volume': stmt.excluded.volume,
+                    'adjusted_close': stmt.excluded.adjusted_close,
+                    'updated_at': stmt.excluded.updated_at,
+                },
+                where=or_(PriceData.close.is_(None), PriceData.close == 0.0),
+            )
             await db.execute(stmt)
 
         if live_rows:
@@ -1002,7 +1102,7 @@ class PriceDataService:
             )
             await db.execute(stmt)
         logger.info(
-            "Stored price records for %s: historical_insert_only=%d live_upsert=%d",
+            "Stored price records for %s: historical=%d live=%d",
             ticker,
             len(historical_rows),
             len(live_rows),

docker-compose.yml

@@ -63,6 +63,7 @@ services:
       - ./app:/app/app  # Mount app directory for development
       - ./alembic:/app/alembic  # Mount alembic for live migration access
       - ./alembic.ini:/app/alembic.ini
+      - ./scripts:/app/scripts  # Mount scripts for live access (backfill, export)
       - ./stock_oracle_analyzer.py:/app/stock_oracle_analyzer.py
       - ./API_DOCUMENTATION.md:/app/API_DOCUMENTATION.md  # API documentation
       - ./yfinance_plus:/app/yfinance_plus  # Mount yfinance_plus for development
@@ -70,7 +71,7 @@ services:
     mem_limit: 3g
     memswap_limit: 3g
     restart: unless-stopped
-    command: ["sh", "-c", "cp /app/yfinance_plus/yfinance_plus.py /usr/local/lib/python3.11/site-packages/yfinance_plus.py && python -m uvicorn app.main:app --host 0.0.0.0 --port 18000 --limit-concurrency 50"]
+    command: ["sh", "-c", "cp /app/yfinance_plus/yfinance_plus.py /usr/local/lib/python3.11/site-packages/yfinance_plus.py && python -m uvicorn app.main:app --host 0.0.0.0 --port 18000 --limit-concurrency 50 --timeout-keep-alive 120"]
 
   # Frontend Application
   frontend:

docs/DATA_COVERAGE.md

@@ -15,7 +15,7 @@
 | `/alpaca/intraday` (SIP, 과거) | Alpaca SIP | ✅ | 요청 기반 자동 누적 | 2016년~어제 | 요청 기반 자동 누적 |
 | `/alpaca/intraday/today` (IEX, 당일) | Alpaca IEX | ✅ | 당일만 | 오늘 장 중 | 해당 없음 |
 | `/alpaca/snapshot` | Alpaca IEX | ❌ | 실시간만 | 없음 | 해당 없음 |
-| `/finra/short-volume` | FINRA CDN | ✅ | **2026-02-10 ~ 현재 (28거래일)** | 수년치 | **⚠️ 백필 권장** |
+| `/finra/short-volume` | FINRA CDN | ✅ | **2018-08-01 ~ 현재 (1,829일, 22,722심볼)** | 2016년~ | 백필 완료 (2020-04~10 갭 제외) |
 | `/etf/holdings` | SEC EDGAR (NPORT) | ✅ (스냅샷) | 요청 기반 자동 누적 | 2019년~ | 요청 기반 자동 누적 |
 | `/filings/search` | SEC EDGAR | ✅ | 1994-01-05 ~ 현재 (1598 티커) | 1994년~ | 요청 기반 자동 누적 |
 | `/stocks/most-active` | Yahoo 실시간 스크래핑 | ❌ | 실시간만 | 없음 | 해당 없음 |
@@ -100,37 +100,55 @@ curl "http://localhost:18001/api/v1/alpaca/snapshot?tickers=AAPL,MSFT,SPY"
 
 ### `/api/v1/finra` — FINRA 공매도 (RegSHO)
 
-**현재 DB 보유**: 2026-02-10 ~ 2026-03-20, 28거래일 (서비스 가동 시점부터)
+**현재 DB 보유**: 2018-08-01 ~ 2026-05-29, **1,829 거래일, 22,722 심볼**
+- FINRA CDN(무료, API 키 없음): `cdn.finra.org/equity/regsho/daily/CNMSshvol{YYYYMMDD}.txt`
+- 롤링 ~7년 보유 (2018-08 이전 403)
+- **2020-04-01 ~ 2020-10-31 결손** (~138 평일 — COVID 갭): `backfill_finra_2020_gap.py`로 메울 수 있음
 
-> 데이터 소스인 FINRA CDN은 수년치 과거 파일을 보유하고 있으나, 현재 DB에는 최근 28일치만 있음.
+#### PIT(Point-in-Time) 유니버스 멤버십
 
-**조회 파라미터**:
-- `days`: 최근 N일 (기본 30, 최대 365)
-- `limit`: 반환 최대 건수 (기본 100, 최대 1000)
+`finra_short_volume`에 등장한 22,722 심볼 중 **16,143개는 현 활성 유니버스(9,635)에 없는 상폐/합병 과거 종목**. DB의 `pit_universe_membership` 뷰로 노출.
+
+```sql
+-- 특정 날짜에 실제 거래되던 종목 (PIT 유니버스, 생존편향 0)
+SELECT DISTINCT symbol FROM pit_universe_membership WHERE d = '2023-03-09';
+-- SIVB, SBNY 등 그날 마지막으로 거래된 종목 포함됨
+
+-- 공매도비율 횡단면 (상폐 종목 포함)
+SELECT p.d, p.symbol, p.short_ratio, a.close
+FROM pit_universe_membership p
+LEFT JOIN alpaca_price_data a
+  ON a.ticker = p.symbol AND a.date::date = p.d AND a.interval = '1d'
+WHERE p.d = '2023-03-09' ORDER BY p.short_ratio;
+```
 
-**⚠️ 백필 방법**:
+**⚠️ 한계**:
+- **티커 재활용**: BBBY(2023-05 파산 → 2년 공백 → 2025-08 다른 엔티티)처럼 동일 티커가 재사용될 수 있음. 심볼 기준 PIT에서 경계 날짜 부근 ±2주 윈도우 제외 권장.
+- **NMS 슈퍼셋**: ETF, ADR, 워런트(/U, /WS), 우선주 포함. 분석단에서 필터.
+- 엄밀한 티커 재활용 해결 = CUSIP/PERMNO 매핑 (유료 데이터, 현재 범위 외).
 
-```bash
-# 단일 날짜 백필
-POST /api/v1/finra/admin/ingest?date=2025-01-02&force=false
+#### API 조회 파라미터
 
-# 날짜 범위 백필 (권장)
-POST /api/v1/finra/admin/ingest?start_date=2024-01-01&end_date=2026-02-09&force=false
-```
+- `days`: 최근 N일 (기본 30, **최대 3650 ≈ 10년**)
+- `limit`: 반환 최대 건수 (기본 100, **최대 10000**)
+- **주의**: `limit=100` 기본값은 "100 거래일"이 아니라 "100행" 제한. 전체 히스토리 조회 시 반드시 지정:
 
-curl 예시:
 ```bash
-# 2024년 전체 백필 (~252 거래일 × ~11,000 심볼 = ~280만 행)
-curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?start_date=2024-01-01&end_date=2024-12-31"
+# SIVB 전체 히스토리 (상폐 전까지)
+curl "http://localhost:18001/api/v1/finra/short-volume/SIVB?days=3650&limit=10000"
 
-# 2025년 전체 백필
-curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?start_date=2025-01-01&end_date=2025-12-31"
+# 정규화된 일봉 공매도비율 (시장 통합)
+curl "http://localhost:18001/api/v1/finra/short-ratio/AAPL?days=3650"
+```
+
+#### 2020 갭 메우기
 
-# 운영 공백 구간 채우기 (2026-01-01 ~ 2026-02-09)
-curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?start_date=2026-01-01&end_date=2026-02-09"
+```bash
+docker exec stock_oracle_api python scripts/backfill_finra_2020_gap.py
+# 예상 소요: ~10-20분, idempotent (이미 있는 날짜 자동 스킵)
 ```
 
-> 주의: 1년치 백필 시 ~3000 HTTP 요청 + DB write. 수십 분 소요될 수 있음. FINRA CDN은 API 키 없이 사용 가능하나 과부하를 피하기 위해 범위를 분할해서 실행 권장.
+> FINRA CDN은 API 키 없이 사용 가능. 1년치 백필 시 ~3000 HTTP 요청 + DB write, 20-40분 소요.
 
 ---
 
@@ -651,22 +669,47 @@ docker exec stock_oracle_api python scripts/news_backfill.py \
 
 | 우선순위 | 대상 | 이유 | 예상 소요 시간 |
 |---|---|---|---|
-| 🔴 높음 | FINRA 1년치 (2025년) | z-score 계산 윈도우(30일)가 너무 짧아 신호 품질 저하 | 20-40분 |
+| 🔴 높음 | **상폐 가격 백필** (PIT 생존편향-0) | 숏볼륨 신호 검정 시 수익 측 생존편향 제거 필수 | 1-3시간 |
+| 🔴 높음 | **FINRA 2020 갭** (2020-04~10) | COVID 약세장/회복 레짐 없으면 멀티레짐 검정 불가 | 10-20분 |
 | 🔴 높음 | Universe 스냅샷 빌드 | 백테스팅 유니버스 기능 사용 전 필수 1회 실행 | 30-60분 (4000 종목 × 10년) |
-| 🟡 중간 | FINRA 2년치 (2024년) | 더 긴 추세 분석 가능 | 1-2시간 |
-| 🟢 낮음 | Alpaca 데이터 | Yahoo Finance와 중복, API 키 필요 | 필요시 |
+| 🟢 낮음 | 추가 FINRA 구간 | 이미 2018-08 ~ 현재 수집 완료 | — |
 
-### FINRA 권장 백필 스크립트
+### 생존편향-0 데이터셋 구축 (권장 실행 순서)
 
 ```bash
-# 1단계: 2025년 (가장 중요)
-curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?start_date=2025-01-02&end_date=2025-12-31"
+# Step 1: FINRA 2020 갭 메우기 (10-20분, idempotent)
+docker exec stock_oracle_api python scripts/backfill_finra_2020_gap.py
+
+# Step 2: 상폐 가격 백필 — PIT 생존편향-0 핵심 (1-3시간, idempotent)
+# FINRA 22,722 심볼 전체에 대해 Alpaca SIP 일봉 수집 (무료 플랜 포함)
+# adjustment='all' (분할+배당 조정), 2018-08-01부터
+docker exec stock_oracle_api python scripts/backfill_alpaca_daily_pit.py
+
+# Step 3 (옵션): 리서치 레이어용 parquet export
+# 외부 환경에서 직접 DB 연결 (port 15433) 후 pandas로 export:
+# python -c "
+# import pandas as pd
+# from sqlalchemy import create_engine
+# eng = create_engine('postgresql+psycopg2://stockoracle:stockoracle2024@localhost:15433/stock_oracle')
+# df = pd.read_sql('SELECT p.d, p.symbol, p.short_ratio, a.close FROM pit_universe_membership p LEFT JOIN alpaca_price_data a ON a.ticker=p.symbol AND a.date::date=p.d AND a.interval=\'1d\' WHERE p.d BETWEEN \'2018-08-01\' AND NOW()', eng)
+# df.to_parquet('pit_panel.parquet', index=False)
+# print(df.shape)
+# "
+```
 
-# 2단계: 2026년 공백 구간
-curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?start_date=2026-01-02&end_date=2026-02-09"
+### PIT 뷰 (DB 직접 쿼리 시)
 
-# 3단계 (선택): 2024년
-curl -X POST "http://localhost:18001/api/v1/finra/admin/ingest?start_date=2024-01-02&end_date=2024-12-31"
+```sql
+-- 날짜별 PIT 유니버스 (상폐 종목 포함)
+SELECT DISTINCT symbol FROM pit_universe_membership WHERE d = '2023-03-09';
+
+-- 공매도비율 × 가격 패널 (생존편향-0, backfill_alpaca_daily_pit.py 실행 후)
+SELECT p.d, p.symbol, p.short_ratio, a.close
+FROM pit_universe_membership p
+LEFT JOIN alpaca_price_data a
+  ON a.ticker = p.symbol AND a.date::date = p.d AND a.interval = '1d'
+WHERE p.d BETWEEN '2022-01-01' AND '2023-12-31'
+ORDER BY p.d, p.short_ratio;
 ```
 
 ---