@ -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 } " )