From e2ac9fa4d410ea84c5ccdb326a8f293e863d0ea1 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sun, 17 May 2026 09:53:21 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20=EA=B3=BC=EA=B1=B0=20=EA=B0=80=EA=B2=A9?= =?UTF-8?q?=20=EB=8D=B0=EC=9D=B4=ED=84=B0=20upsert=20=E2=86=92=20insert-on?= =?UTF-8?q?ly=EB=A1=9C=20=EB=B3=80=EA=B2=BD=20(Yahoo=20adjusted=20price=20?= =?UTF-8?q?=EB=8D=AE=EC=96=B4=EC=93=B0=EA=B8=B0=20=EB=B0=A9=EC=A7=80)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 과거 OHLC 행은 on_conflict_do_nothing, 오늘 행만 on_conflict_do_update. Yahoo Finance의 배당·분할 소급 조정으로 인한 기존 PIT 데이터 훼손 방지. Co-Authored-By: Claude Sonnet 4.6 --- app/services/price_data_service.py | 103 ++++++++++++++++++----------- yfinance_plus | 2 +- 2 files changed, 65 insertions(+), 40 deletions(-) diff --git a/app/services/price_data_service.py b/app/services/price_data_service.py index a8f9afe..cb9b273 100644 --- a/app/services/price_data_service.py +++ b/app/services/price_data_service.py @@ -480,13 +480,16 @@ class PriceDataService: return None now = datetime.now(timezone.utc) - rows = [] + today = now.date() + historical_rows = [] + live_rows = [] for date_idx, row in hist_data.iterrows(): price_date = date_idx.to_pydatetime() if price_date.tzinfo is None: price_date = price_date.replace(tzinfo=timezone.utc) - rows.append({ + target_rows = live_rows if price_date.date() >= today else historical_rows + target_rows.append({ 'id': _uuid.uuid4(), 'ticker': ticker, 'date': price_date, @@ -501,26 +504,34 @@ class PriceDataService: 'updated_at': now, }) - if not rows: + if not historical_rows and not live_rows: return - # Upsert: on conflict update OHLCV + updated_at so that stale intraday-cached rows - # get overwritten when force_refresh=True re-fetches final EOD data. - stmt = pg_insert(PriceData).values(rows) - 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, - 'data_source': stmt.excluded.data_source, - 'updated_at': stmt.excluded.updated_at, - } - ) - await db.execute(stmt) + # 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. + if historical_rows: + stmt = pg_insert(PriceData).values(historical_rows) + stmt = stmt.on_conflict_do_nothing(constraint='uq_price_data') + await db.execute(stmt) + + if live_rows: + stmt = pg_insert(PriceData).values(live_rows) + 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, + 'data_source': stmt.excluded.data_source, + 'updated_at': stmt.excluded.updated_at, + } + ) + await db.execute(stmt) async def _get_price_data_from_db( self, @@ -942,13 +953,16 @@ class PriceDataService: return None now = datetime.now(timezone.utc) - rows = [] + today = now.date() + historical_rows = [] + live_rows = [] for date_idx, row in ticker_data.iterrows(): price_date = date_idx.to_pydatetime() if price_date.tzinfo is None: price_date = price_date.replace(tzinfo=timezone.utc) - rows.append({ + target_rows = live_rows if price_date.date() >= today else historical_rows + target_rows.append({ 'id': _uuid.uuid4(), 'ticker': ticker, 'date': price_date, @@ -963,25 +977,36 @@ class PriceDataService: 'updated_at': now, }) - if not rows: + if not historical_rows and not live_rows: return - stmt = pg_insert(PriceData).values(rows) - 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, - 'data_source': stmt.excluded.data_source, - 'updated_at': stmt.excluded.updated_at, - } + if historical_rows: + stmt = pg_insert(PriceData).values(historical_rows) + stmt = stmt.on_conflict_do_nothing(constraint='uq_price_data') + await db.execute(stmt) + + if live_rows: + stmt = pg_insert(PriceData).values(live_rows) + 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, + 'data_source': stmt.excluded.data_source, + 'updated_at': stmt.excluded.updated_at, + } + ) + await db.execute(stmt) + logger.info( + "Stored price records for %s: historical_insert_only=%d live_upsert=%d", + ticker, + len(historical_rows), + len(live_rows), ) - await db.execute(stmt) - logger.info(f"Batch upserted {len(rows)} price records for {ticker}") async def _get_existing_dates_for_ticker( self, @@ -1112,4 +1137,4 @@ try: import pandas as pd except ImportError: logger.error("pandas not available - price data service will not work") - pd = None \ No newline at end of file + pd = None diff --git a/yfinance_plus b/yfinance_plus index 674b7e4..e7011ca 160000 --- a/yfinance_plus +++ b/yfinance_plus @@ -1 +1 @@ -Subproject commit 674b7e45c8c79f3617dee9cffdd3f1438e4852cf +Subproject commit e7011cabd7cd7853c3bc0fbd41601e004cc99ff4