From 73bf6115fa889c58e9c74c4bd4b4e4b3e06806c7 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Mon, 13 Apr 2026 18:29:33 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20=5Fbatch=5Fcheck=5Fmissing=5Fperiods=20?= =?UTF-8?q?=E2=80=94=20today=20=ED=8F=AC=ED=95=A8=20=EC=8B=9C=20=ED=95=AD?= =?UTF-8?q?=EC=83=81=20re-fetch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _check_missing_periods는 수정됐지만 bulk 경로에서 사용하는 _batch_check_missing_periods에는 동일한 today 체크가 빠져 있었음. POST /price/data/bulk로 당일 데이터 요청 시 DB에 mid-session으로 캐시된 partial-volume 데이터를 그대로 반환하는 버그. end_date.date() >= today이면 모든 티커를 missing으로 처리하여 _check_missing_periods와 동일한 동작 보장. Co-Authored-By: Claude Sonnet 4.6 --- app/services/price_data_service.py | 70 +++++++++++++++++++++++++----- 1 file changed, 58 insertions(+), 12 deletions(-) diff --git a/app/services/price_data_service.py b/app/services/price_data_service.py index 1e80751..3e842ad 100644 --- a/app/services/price_data_service.py +++ b/app/services/price_data_service.py @@ -103,11 +103,22 @@ class PriceDataService: end_date: datetime, interval: str ) -> List[datetime]: - """Check which periods are missing in the database""" + """Check which periods are missing in the database. + + Also treats today's data as always-missing so that end-of-day volumes + get re-fetched rather than returning a mid-session snapshot that was + cached earlier in the day. + """ # If no date range provided, assume we need to fetch data if start_date is None or end_date is None: return [datetime.now()] # Return a dummy date to trigger fetch - + + # Always re-fetch if the range includes today: daily bars fetched mid-session + # are stored with partial (intraday) volume and must be refreshed after close. + today = datetime.now(timezone.utc).date() + if end_date.date() >= today: + return [datetime.now()] + # Check if we have any data for this ticker and interval result = await db.execute( select(PriceData.date) @@ -120,19 +131,19 @@ class PriceDataService: ) .order_by(PriceData.date) ) - + existing_dates = {row[0].date() for row in result.fetchall()} - + # Generate expected dates based on interval expected_dates = self._generate_expected_dates(start_date, end_date, interval) - + # Find missing dates missing_dates = [date for date in expected_dates if date not in existing_dates] - + # If more than 10% of dates are missing, consider it as needing refresh if len(missing_dates) > len(expected_dates) * 0.1: return missing_dates - + return [] def _generate_expected_dates( @@ -506,11 +517,24 @@ class PriceDataService: if not rows: return - # Single batch INSERT — skip rows that violate the unique constraint (ticker, date) + # 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_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, + 'data_source': stmt.excluded.data_source, + 'updated_at': stmt.excluded.updated_at, + } + ) await db.execute(stmt) - + async def _get_price_data_from_db( self, db: AsyncSession, @@ -779,7 +803,17 @@ class PriceDataService: end_date: datetime, interval: str ) -> List[str]: - """Batch check which tickers have missing periods""" + """Batch check which tickers have missing periods. + + If the range includes today, all tickers are treated as missing so that + mid-session cached rows (partial volume) are always refreshed. + """ + # Always re-fetch when range includes today (same logic as _check_missing_periods) + today = datetime.now(timezone.utc).date() + if end_date.date() >= today: + 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 from sqlalchemy import func, case @@ -968,7 +1002,19 @@ class PriceDataService: return stmt = pg_insert(PriceData).values(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, + 'data_source': stmt.excluded.data_source, + 'updated_at': stmt.excluded.updated_at, + } + ) await db.execute(stmt) logger.info(f"Batch upserted {len(rows)} price records for {ticker}")