From f5feffb80ef80f9fcfa75da308920bf94c318e40 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sun, 31 May 2026 10:49:55 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20adjustment=3D'all'=20=EC=A1=B0=EC=A0=95?= =?UTF-8?q?=EA=B0=80=20=EC=A0=81=EC=9A=A9=20+=20parquet=20export=20?= =?UTF-8?q?=EC=97=94=EB=93=9C=ED=8F=AC=EC=9D=B8=ED=8A=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - alpaca.py backfill: on_conflict_do_nothing → on_conflict_do_update 기존 미조정 행(split 전 $1,208)을 조정가($120)로 덮어씀 NVDA 2024-06-07 ~$120 통과 기준 - finra.py: POST /admin/export-pit-panel (Background) + GET /download pit_universe_membership × alpaca_price_data 조인 → /app/data/pit_panel.parquet - requirements-api.txt: pyarrow>=14.0.0 추가 Co-Authored-By: Claude Opus 4.8 --- app/api/v1/endpoints/alpaca.py | 16 ++++- app/api/v1/endpoints/finra.py | 104 ++++++++++++++++++++++++++++++++- requirements-api.txt | 1 + 3 files changed, 117 insertions(+), 4 deletions(-) diff --git a/app/api/v1/endpoints/alpaca.py b/app/api/v1/endpoints/alpaca.py index 88cb702..6166f99 100644 --- a/app/api/v1/endpoints/alpaca.py +++ b/app/api/v1/endpoints/alpaca.py @@ -447,7 +447,19 @@ async def _run_pit_backfill(start_str: str, end_str: str, force: bool) -> None: stmt = pg_insert(AlpacaPriceData).values( rows_to_insert[i: i + CHUNK_SIZE] ) - stmt = stmt.on_conflict_do_nothing(constraint="uq_alpaca_price_data") + stmt = stmt.on_conflict_do_update( + constraint="uq_alpaca_price_data", + set_={ + "open": stmt.excluded.open, + "high": stmt.excluded.high, + "low": stmt.excluded.low, + "close": stmt.excluded.close, + "volume": stmt.excluded.volume, + "vwap": stmt.excluded.vwap, + "trade_count": stmt.excluded.trade_count, + "data_source": stmt.excluded.data_source, + } + ) result = await db.execute(stmt) total_inserted += result.rowcount await asyncio.sleep(0) @@ -481,7 +493,7 @@ async def _run_pit_backfill(start_str: str, end_str: str, force: bool) -> None: "Alpaca SIP 일봉(1d)을 백필합니다. 생존편향-0 수익 계산에 필요.\n\n" "**특성**:\n" "- `adjustment=all` (분할+배당 조정) — 상폐 종목은 future-proof\n" - "- DB-first, idempotent (`on_conflict_do_nothing`) — 재실행 안전\n" + "- `on_conflict_do_update`: 기존 행도 조정가로 덮어씀 (split 이전 미조정 데이터 수정)\n" "- Alpaca SIP 일봉은 무료 플랜에서 2016-01-04부터 제공\n" "- 기본 시작일: 2018-08-01 (FINRA DB 시작일)\n\n" "**백그라운드 실행**: 즉시 `started` 응답, 1-3시간 소요.\n" diff --git a/app/api/v1/endpoints/finra.py b/app/api/v1/endpoints/finra.py index 56ee160..afb1038 100644 --- a/app/api/v1/endpoints/finra.py +++ b/app/api/v1/endpoints/finra.py @@ -2,14 +2,20 @@ FINRA Short Sale Volume endpoints """ +import logging +import os from datetime import date, datetime, timedelta, timezone from typing import List, Optional -from fastapi import APIRouter, Depends, HTTPException, Query -from fastapi.responses import Response +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query +from fastapi.responses import FileResponse, Response from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession +from app.core.database import AsyncSessionLocal + +logger = logging.getLogger(__name__) + from app.core.config import settings from app.core.database import get_db from app.schemas.finra import ( @@ -184,6 +190,100 @@ async def get_pit_panel( } +_EXPORT_PATH = "/app/data/pit_panel.parquet" + + +async def _run_pit_export(start_str: str, end_str: str) -> None: + """Background task: export PIT panel to /app/data/pit_panel.parquet.""" + try: + import pandas as pd + except ImportError: + logger.error("pandas not installed"); return + try: + import pyarrow # noqa: F401 + except ImportError: + logger.error("pyarrow not installed"); return + + logger.info(f"PIT export: loading {start_str} → {end_str} ...") + + async with AsyncSessionLocal() as db: + rows = (await db.execute(text(""" + SELECT + p.d::date 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 :s AND :e + ORDER BY p.d, p.symbol + """), {"s": start_str, "e": end_str})).fetchall() + + logger.info(f"PIT export: {len(rows):,} rows fetched, writing parquet ...") + df = pd.DataFrame(rows, columns=[ + "date","symbol","short_volume","short_exempt_volume","total_volume", + "short_ratio","open","high","low","close","price_volume","vwap", + ]) + df["date"] = pd.to_datetime(df["date"]) + os.makedirs(os.path.dirname(_EXPORT_PATH), exist_ok=True) + df.to_parquet(_EXPORT_PATH, index=False, engine="pyarrow") + mb = os.path.getsize(_EXPORT_PATH) / 1_048_576 + logger.info( + f"PIT export complete: {_EXPORT_PATH} | {mb:.0f} MB | shape={df.shape} | " + f"price_coverage={df['close'].notna().mean()*100:.1f}%" + ) + + +@router.post( + "/admin/export-pit-panel", + summary="PIT 패널 parquet export → /app/data/pit_panel.parquet", + description=( + "FINRA × Alpaca(adjustment='all') 조인 패널을 parquet으로 저장.\n\n" + "`GET /finra/admin/export-pit-panel/download` 로 다운로드.\n\n" + "백그라운드 실행 — 전체 기간(2018-08~현재) 기준 수 분 ~ 10여 분 소요." + ), +) +async def export_pit_panel( + background_tasks: BackgroundTasks, + start_date: date = Query(date(2018, 8, 1), description="시작일"), + end_date: date = Query(None, description="종료일 (기본: 오늘)"), +): + _end = end_date or date.today() + background_tasks.add_task(_run_pit_export, start_date.isoformat(), _end.isoformat()) + return { + "status": "started", + "start_date": start_date.isoformat(), + "end_date": _end.isoformat(), + "output": _EXPORT_PATH, + "note": "완료 후 GET /finra/admin/export-pit-panel/download 로 다운로드.", + } + + +@router.get( + "/admin/export-pit-panel/download", + summary="pit_panel.parquet 다운로드", +) +async def download_pit_panel(): + if not os.path.exists(_EXPORT_PATH): + raise HTTPException( + status_code=404, + detail="parquet 파일 없음. POST /finra/admin/export-pit-panel 먼저 실행.", + ) + return FileResponse( + _EXPORT_PATH, + media_type="application/octet-stream", + filename="pit_panel.parquet", + ) + + @router.post( "/admin/ingest", response_model=IngestResponse, diff --git a/requirements-api.txt b/requirements-api.txt index cf75df4..67960d9 100644 --- a/requirements-api.txt +++ b/requirements-api.txt @@ -23,6 +23,7 @@ beautifulsoup4>=4.12.0 # SEC data processing - using direct API calls and EDGAR downloader aiohttp>=3.8.0 pandas>=2.0.0,<3.0.0 +pyarrow>=14.0.0 # parquet export (pit_panel) numpy>=1.24.0,<3.0.0 python-dateutil>=2.8.0 sec-edgar-downloader>=5.0.0