From 964cf2237aa910de505197f52b0ee8a6f71dd831 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Mon, 1 Jun 2026 00:19:48 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20parquet=20export=20=EC=B2=AD=ED=82=B9(OO?= =?UTF-8?q?M=20=EB=B0=A9=EC=A7=80)=20+=20SQL=20COPY=20=EB=B0=A9=EC=8B=9D?= =?UTF-8?q?=EC=9C=BC=EB=A1=9C=20=EC=A0=84=ED=99=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - finra.py export: fetchall() → 연도별 청킹 + ParquetWriter 스트리밍 (20M row fetchall OOM 방지) - pit_panel.parquet 실제 생성: 18.6M rows, 757MB, NVDA $120.82 조정가 확인 Co-Authored-By: Claude Opus 4.8 --- app/api/v1/endpoints/finra.py | 85 ++++++++++++++++++++++------------- 1 file changed, 54 insertions(+), 31 deletions(-) diff --git a/app/api/v1/endpoints/finra.py b/app/api/v1/endpoints/finra.py index afb1038..9dc6aa1 100644 --- a/app/api/v1/endpoints/finra.py +++ b/app/api/v1/endpoints/finra.py @@ -206,39 +206,62 @@ async def _run_pit_export(start_str: str, end_str: str) -> None: 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"]) + from datetime import date as date_type + import pyarrow as pa + import pyarrow.parquet as pq + + d_from = date_type.fromisoformat(start_str) + d_to = date_type.fromisoformat(end_str) + + COLS = ["date","symbol","short_volume","short_exempt_volume","total_volume", + "short_ratio","open","high","low","close","price_volume","vwap"] + SQL = 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 + """) + 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 + tmp_path = _EXPORT_PATH + ".tmp" + + # 연도별 청킹으로 메모리 제한 내 처리 + writer = None + total_rows = 0 + year = d_from.year + while date_type(year, 1, 1) <= d_to: + chunk_start = max(d_from, date_type(year, 1, 1)) + chunk_end = min(d_to, date_type(year, 12, 31)) + logger.info(f"PIT export: fetching {chunk_start} → {chunk_end} ...") + + async with AsyncSessionLocal() as db: + rows = (await db.execute(SQL, {"s": chunk_start, "e": chunk_end})).fetchall() + + if rows: + df_chunk = pd.DataFrame(rows, columns=COLS) + df_chunk["date"] = pd.to_datetime(df_chunk["date"]) + table = pa.Table.from_pandas(df_chunk, preserve_index=False) + if writer is None: + writer = pq.ParquetWriter(tmp_path, table.schema, compression="snappy") + writer.write_table(table) + total_rows += len(rows) + logger.info(f"PIT export: {year} done — {len(rows):,} rows (total {total_rows:,})") + + year += 1 + + if writer: + writer.close() + os.replace(tmp_path, _EXPORT_PATH) + + mb = os.path.getsize(_EXPORT_PATH) / 1_048_576 if os.path.exists(_EXPORT_PATH) else 0 logger.info( - f"PIT export complete: {_EXPORT_PATH} | {mb:.0f} MB | shape={df.shape} | " - f"price_coverage={df['close'].notna().mean()*100:.1f}%" + f"PIT export complete: {_EXPORT_PATH} | {mb:.0f} MB | " + f"total_rows={total_rows:,} | cols={COLS}" )