You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
941 lines
38 KiB
Python
941 lines
38 KiB
Python
"""
|
|
Service for fetching and processing price data from Yahoo Finance using yfinance-plus
|
|
"""
|
|
|
|
from datetime import datetime, timezone, timedelta, date
|
|
from typing import Dict, List, Optional, Tuple, Union
|
|
import logging
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, and_, desc
|
|
import asyncio
|
|
import sys
|
|
import os
|
|
|
|
# Add parent directory to path for imports
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
|
|
|
from app.models.financial import PriceData
|
|
from app.schemas.financial import DataSource, ErrorType
|
|
from app.utils.date_utils import parse_period, quarters_to_date_range, resolve_time_parameters
|
|
from app.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Import yfinance-plus for price data only
|
|
try:
|
|
import yfinance_plus as yf
|
|
YFINANCE_AVAILABLE = True
|
|
logger.info("yfinance-plus imported successfully for price data")
|
|
except ImportError:
|
|
logger.error("yfinance-plus not available for price data")
|
|
YFINANCE_AVAILABLE = False
|
|
|
|
class PriceDataService:
|
|
def __init__(self):
|
|
self.yf_available = YFINANCE_AVAILABLE
|
|
if not self.yf_available:
|
|
logger.warning("Yahoo Finance (yfinance-plus) data will not be available")
|
|
|
|
async def get_or_update_price_data(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str,
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str = "1d",
|
|
force_refresh: bool = False
|
|
) -> List[PriceData]:
|
|
"""
|
|
Get price data from database or fetch from Yahoo Finance if needed
|
|
|
|
Args:
|
|
db: Database session
|
|
ticker: Stock ticker symbol
|
|
start_date: Start date for data retrieval
|
|
end_date: End date for data retrieval
|
|
interval: Data interval (1d, 1w, 1m, 1h, etc.)
|
|
force_refresh: Force refresh data from Yahoo Finance
|
|
|
|
Returns:
|
|
List of PriceData objects
|
|
"""
|
|
ticker = ticker.upper()
|
|
|
|
# Check if we need to fetch new data
|
|
missing_periods = await self._check_missing_periods(
|
|
db, ticker, start_date, end_date, interval
|
|
)
|
|
|
|
if missing_periods or force_refresh:
|
|
if not self.yf_available:
|
|
raise ValueError("Yahoo Finance (yfinance-plus) data source not available")
|
|
|
|
# Fetch data from Yahoo Finance using yfinance-plus
|
|
await self._fetch_and_store_price_data(
|
|
db, ticker, start_date, end_date, interval
|
|
)
|
|
|
|
# Retrieve data from database
|
|
price_data = await self._get_price_data_from_db(
|
|
db, ticker, start_date, end_date, interval
|
|
)
|
|
|
|
return price_data
|
|
|
|
async def _check_missing_periods(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str,
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str
|
|
) -> List[datetime]:
|
|
"""Check which periods are missing in the database"""
|
|
# 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
|
|
|
|
# Check if we have any data for this ticker and interval
|
|
result = await db.execute(
|
|
select(PriceData.date)
|
|
.where(
|
|
and_(
|
|
PriceData.ticker == ticker,
|
|
PriceData.date >= start_date,
|
|
PriceData.date <= end_date
|
|
)
|
|
)
|
|
.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(
|
|
self,
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str
|
|
) -> List[datetime]:
|
|
"""Generate expected trading dates based on interval"""
|
|
expected_dates = []
|
|
current_date = start_date
|
|
|
|
# Simple date generation (doesn't account for market holidays)
|
|
if interval == "1d":
|
|
while current_date <= end_date:
|
|
# Skip weekends for daily data
|
|
if current_date.weekday() < 5: # Monday = 0, Friday = 4
|
|
expected_dates.append(current_date)
|
|
current_date += timedelta(days=1)
|
|
elif interval == "1w":
|
|
while current_date <= end_date:
|
|
expected_dates.append(current_date)
|
|
current_date += timedelta(weeks=1)
|
|
elif interval == "1m":
|
|
# Monthly data - first day of each month
|
|
while current_date <= end_date:
|
|
expected_dates.append(current_date)
|
|
# Move to next month
|
|
if current_date.month == 12:
|
|
current_date = current_date.replace(year=current_date.year + 1, month=1)
|
|
else:
|
|
current_date = current_date.replace(month=current_date.month + 1)
|
|
else:
|
|
# For other intervals, just return the date range
|
|
expected_dates = [start_date, end_date]
|
|
|
|
return expected_dates
|
|
|
|
async def _fetch_and_store_price_data(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str,
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str
|
|
):
|
|
"""Fetch price data from Yahoo Finance using yfinance-plus and store in database"""
|
|
try:
|
|
logger.info(f"Fetching price data for {ticker} from {start_date} to {end_date}")
|
|
|
|
# Create yfinance-plus ticker object
|
|
yf_ticker = yf.Ticker(ticker)
|
|
|
|
# Fetch historical data
|
|
# Convert dates to strings in YYYY-MM-DD format
|
|
start_str = start_date.strftime('%Y-%m-%d')
|
|
# yfinance's `end` parameter is exclusive for daily data when using date strings.
|
|
# Add +1 day to include the intended end_date day in the results.
|
|
from datetime import timedelta
|
|
end_inclusive = end_date + timedelta(days=1)
|
|
end_str = end_inclusive.strftime('%Y-%m-%d')
|
|
|
|
# Run yfinance-plus in executor to avoid blocking
|
|
loop = asyncio.get_event_loop()
|
|
hist_data = await loop.run_in_executor(
|
|
None,
|
|
lambda: yf_ticker.history(
|
|
start=start_str,
|
|
end=end_str,
|
|
interval=interval,
|
|
auto_adjust=True,
|
|
prepost=False,
|
|
period=None # Explicitly set period to None when using start/end dates
|
|
)
|
|
)
|
|
|
|
if hist_data.empty:
|
|
logger.warning(f"No price data returned for {ticker}")
|
|
return
|
|
|
|
# Store data in database
|
|
await self._store_price_data(db, ticker, hist_data, interval)
|
|
|
|
await db.commit()
|
|
|
|
logger.info(f"Successfully stored {len(hist_data)} price records for {ticker}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching price data for {ticker}: {str(e)}")
|
|
await db.rollback()
|
|
raise
|
|
|
|
async def get_quote(self, ticker: str, use_prepost: bool = True) -> Dict:
|
|
"""Get latest quote using yfinance-plus .info fields with fallback to fast history last row."""
|
|
if not self.yf_available:
|
|
raise ValueError("Yahoo Finance (yfinance-plus) data source not available")
|
|
try:
|
|
yf_ticker = yf.Ticker(ticker)
|
|
loop = asyncio.get_event_loop()
|
|
info = await loop.run_in_executor(None, lambda: yf_ticker.info)
|
|
# Prefer regular/post/pre values
|
|
regular = info.get("regularMarketPrice")
|
|
post = info.get("postMarketPrice") if use_prepost else None
|
|
pre = info.get("preMarketPrice") if use_prepost else None
|
|
price = post or pre or regular
|
|
currency = info.get("currency")
|
|
exchange = info.get("exchange") or info.get("fullExchangeName")
|
|
market_state = info.get("marketState")
|
|
ts = info.get("regularMarketTime") or info.get("postMarketTime") or info.get("preMarketTime")
|
|
if isinstance(ts, (int, float)):
|
|
ts = datetime.fromtimestamp(ts, tz=timezone.utc)
|
|
elif isinstance(ts, datetime):
|
|
if ts.tzinfo is None:
|
|
ts = ts.replace(tzinfo=timezone.utc)
|
|
else:
|
|
ts = datetime.now(timezone.utc)
|
|
return {
|
|
"ticker": ticker.upper(),
|
|
"price": float(price) if price is not None else None,
|
|
"regular_price": float(regular) if regular is not None else None,
|
|
"pre_market_price": float(pre) if pre is not None else None,
|
|
"post_market_price": float(post) if post is not None else None,
|
|
"currency": currency,
|
|
"exchange": exchange,
|
|
"market_state": market_state,
|
|
"timestamp": ts,
|
|
"source": DataSource.YAHOO_FINANCE,
|
|
"delayed": True,
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error fetching quote for {ticker}: {str(e)}")
|
|
raise
|
|
|
|
async def get_intraday(self, ticker: str, interval: str = "1m", period: str = "1d") -> List[Dict]:
|
|
"""Get intraday candles using yfinance-plus history with period/interval."""
|
|
if not self.yf_available:
|
|
raise ValueError("Yahoo Finance (yfinance-plus) data source not available")
|
|
try:
|
|
yf_ticker = yf.Ticker(ticker)
|
|
loop = asyncio.get_event_loop()
|
|
df = await loop.run_in_executor(
|
|
None,
|
|
lambda: yf_ticker.history(period=period, interval=interval, auto_adjust=True, prepost=True)
|
|
)
|
|
candles = []
|
|
if not df.empty:
|
|
for ts, row in df.iterrows():
|
|
dt = ts.to_pydatetime()
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
candles.append({
|
|
"timestamp": dt,
|
|
"open": float(row.get("Open", 0)) if not pd.isna(row.get("Open")) else None,
|
|
"high": float(row.get("High", 0)) if not pd.isna(row.get("High")) else None,
|
|
"low": float(row.get("Low", 0)) if not pd.isna(row.get("Low")) else None,
|
|
"close": float(row.get("Close", 0)) if not pd.isna(row.get("Close")) else 0.0,
|
|
"volume": float(row.get("Volume", 0)) if not pd.isna(row.get("Volume")) else None,
|
|
})
|
|
return candles
|
|
except Exception as e:
|
|
logger.error(f"Error fetching intraday for {ticker}: {str(e)}")
|
|
raise
|
|
|
|
async def get_today_ohlc(self, ticker: str) -> Dict:
|
|
"""Get today's OHLC. If daily not yet finalized, aggregate from intraday 1m."""
|
|
if not self.yf_available:
|
|
raise ValueError("Yahoo Finance (yfinance-plus) data source not available")
|
|
try:
|
|
# First try daily with period=1d
|
|
yf_ticker = yf.Ticker(ticker)
|
|
loop = asyncio.get_event_loop()
|
|
daily = await loop.run_in_executor(
|
|
None, lambda: yf_ticker.history(period="1d", interval="1d", auto_adjust=True, prepost=False)
|
|
)
|
|
if daily is not None and not daily.empty:
|
|
ts, row = list(daily.iterrows())[-1]
|
|
d = ts.to_pydatetime().date()
|
|
return {
|
|
"ticker": ticker.upper(),
|
|
"date": d,
|
|
"open": float(row.get("Open", 0)) if not pd.isna(row.get("Open")) else None,
|
|
"high": float(row.get("High", 0)) if not pd.isna(row.get("High")) else None,
|
|
"low": float(row.get("Low", 0)) if not pd.isna(row.get("Low")) else None,
|
|
"close": float(row.get("Close", 0)) if not pd.isna(row.get("Close")) else 0.0,
|
|
"volume": float(row.get("Volume", 0)) if not pd.isna(row.get("Volume")) else None,
|
|
"source": DataSource.YAHOO_FINANCE,
|
|
"method": "daily",
|
|
}
|
|
# Fallback to intraday aggregation
|
|
intraday = await self.get_intraday(ticker, interval="1m", period="1d")
|
|
if not intraday:
|
|
raise ValueError("No intraday data available for today")
|
|
o = next((c["open"] for c in intraday if c.get("open") is not None), None)
|
|
h = max((c.get("high") or c.get("close") or 0.0) for c in intraday)
|
|
l = min((c.get("low") or c.get("close") or float("inf")) for c in intraday)
|
|
c = next((candle.get("close") for candle in reversed(intraday) if candle.get("close") is not None), 0.0)
|
|
v = sum((c.get("volume") or 0.0) for c in intraday)
|
|
today_date = intraday[0]["timestamp"].date()
|
|
return {
|
|
"ticker": ticker.upper(),
|
|
"date": today_date,
|
|
"open": o,
|
|
"high": h if h != 0.0 else None,
|
|
"low": l if l != float("inf") else None,
|
|
"close": c,
|
|
"volume": v or None,
|
|
"source": DataSource.YAHOO_FINANCE,
|
|
"method": "intraday_aggregate",
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error fetching today OHLC for {ticker}: {str(e)}")
|
|
raise
|
|
|
|
async def _store_price_data(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str,
|
|
hist_data,
|
|
interval: str
|
|
):
|
|
"""Store price data in database"""
|
|
for date, row in hist_data.iterrows():
|
|
# Convert pandas timestamp to datetime
|
|
price_date = date.to_pydatetime()
|
|
if price_date.tzinfo is None:
|
|
price_date = price_date.replace(tzinfo=timezone.utc)
|
|
|
|
# Check if record already exists
|
|
existing = await db.execute(
|
|
select(PriceData).where(
|
|
and_(
|
|
PriceData.ticker == ticker,
|
|
PriceData.date == price_date
|
|
)
|
|
)
|
|
)
|
|
|
|
if existing.first():
|
|
continue # Skip if already exists
|
|
|
|
# Create new price data record
|
|
price_record = PriceData(
|
|
ticker=ticker,
|
|
date=price_date,
|
|
open=float(row.get('Open', 0)) if not pd.isna(row.get('Open')) else None,
|
|
high=float(row.get('High', 0)) if not pd.isna(row.get('High')) else None,
|
|
low=float(row.get('Low', 0)) if not pd.isna(row.get('Low')) else None,
|
|
close=float(row.get('Close', 0)) if not pd.isna(row.get('Close')) else 0,
|
|
volume=float(row.get('Volume', 0)) if not pd.isna(row.get('Volume')) else None,
|
|
adjusted_close=float(row.get('Close', 0)) if not pd.isna(row.get('Close')) else None, # Auto-adjusted
|
|
data_source=DataSource.YAHOO_FINANCE
|
|
)
|
|
|
|
db.add(price_record)
|
|
|
|
async def _get_price_data_from_db(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str,
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str
|
|
) -> List[PriceData]:
|
|
"""Get price data from database"""
|
|
# Build query conditions
|
|
conditions = [PriceData.ticker == ticker]
|
|
|
|
if start_date is not None:
|
|
conditions.append(PriceData.date >= start_date)
|
|
if end_date is not None:
|
|
conditions.append(PriceData.date <= end_date)
|
|
|
|
result = await db.execute(
|
|
select(PriceData)
|
|
.where(and_(*conditions))
|
|
.order_by(PriceData.date)
|
|
)
|
|
|
|
return result.scalars().all()
|
|
|
|
async def get_latest_price(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str
|
|
) -> Optional[PriceData]:
|
|
"""Get the latest price for a ticker"""
|
|
result = await db.execute(
|
|
select(PriceData)
|
|
.where(PriceData.ticker == ticker.upper())
|
|
.order_by(desc(PriceData.date))
|
|
.limit(1)
|
|
)
|
|
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_ticker_info(self, ticker: str) -> Dict:
|
|
"""Get ticker information from Yahoo Finance using yfinance-plus"""
|
|
if not self.yf_available:
|
|
raise ValueError("Yahoo Finance (yfinance-plus) data source not available")
|
|
|
|
try:
|
|
yf_ticker = yf.Ticker(ticker)
|
|
|
|
# Run in executor to avoid blocking
|
|
loop = asyncio.get_event_loop()
|
|
info = await loop.run_in_executor(None, lambda: yf_ticker.info)
|
|
|
|
return info
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error fetching ticker info for {ticker}: {str(e)}")
|
|
raise
|
|
|
|
async def get_multiple_tickers_data(
|
|
self,
|
|
db: AsyncSession,
|
|
tickers: List[str],
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str = "1d",
|
|
force_refresh: bool = False
|
|
) -> Dict[str, List[PriceData]]:
|
|
"""Get price data for multiple tickers (legacy method)"""
|
|
results = {}
|
|
|
|
for ticker in tickers:
|
|
try:
|
|
data = await self.get_or_update_price_data(
|
|
db, ticker, start_date, end_date, interval, force_refresh
|
|
)
|
|
results[ticker] = data
|
|
except Exception as e:
|
|
logger.error(f"Error fetching data for {ticker}: {str(e)}")
|
|
results[ticker] = []
|
|
|
|
return results
|
|
|
|
async def get_multiple_tickers_data_optimized(
|
|
self,
|
|
db: AsyncSession,
|
|
tickers: List[str],
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str = "1d",
|
|
force_refresh: bool = False
|
|
) -> Tuple[List, int, int]:
|
|
"""
|
|
Optimized bulk processing for multiple tickers with chunking for 100+ tickers:
|
|
1. Smart chunking to handle 100+ tickers efficiently
|
|
2. Parallel processing using asyncio with concurrency limits
|
|
3. Bulk yfinance queries using yfinance-plus bulk features
|
|
4. Optimized database operations with batch processing
|
|
5. Progress tracking for large requests
|
|
|
|
Supports unlimited ticker count with intelligent chunking:
|
|
- Small batches (≤50): Process in single chunk
|
|
- Medium batches (51-200): Process in 2-4 chunks
|
|
- Large batches (200+): Process in optimal chunks with progress tracking
|
|
|
|
Returns:
|
|
Tuple of (results, successful_count, failed_count) for API response
|
|
"""
|
|
from app.schemas.financial import BulkPriceDataItem, PriceDataResponse, PriceDataPoint
|
|
|
|
results = []
|
|
successful_count = 0
|
|
failed_count = 0
|
|
|
|
# Normalize tickers and validate
|
|
tickers = [t.upper().strip() for t in tickers if t.strip()]
|
|
total_tickers = len(tickers)
|
|
|
|
logger.info(f"Starting bulk processing for {total_tickers} tickers")
|
|
|
|
# Determine optimal chunking strategy based on ticker count
|
|
if total_tickers <= 50:
|
|
chunk_size = total_tickers # Single chunk for small requests
|
|
max_concurrent = 1
|
|
elif total_tickers <= 200:
|
|
chunk_size = 50 # Moderate chunks for medium requests
|
|
max_concurrent = 4
|
|
else:
|
|
chunk_size = 75 # Larger chunks for big requests
|
|
max_concurrent = 6
|
|
|
|
try:
|
|
# Process tickers in chunks to avoid overwhelming APIs and memory
|
|
all_results = []
|
|
all_successful = 0
|
|
all_failed = 0
|
|
|
|
for chunk_start in range(0, total_tickers, chunk_size):
|
|
chunk_end = min(chunk_start + chunk_size, total_tickers)
|
|
chunk_tickers = tickers[chunk_start:chunk_end]
|
|
chunk_num = (chunk_start // chunk_size) + 1
|
|
total_chunks = (total_tickers + chunk_size - 1) // chunk_size
|
|
|
|
logger.info(f"Processing chunk {chunk_num}/{total_chunks}: {len(chunk_tickers)} tickers")
|
|
|
|
# Step 1: Batch check missing periods for chunk
|
|
missing_tickers = []
|
|
if force_refresh:
|
|
missing_tickers = chunk_tickers.copy()
|
|
else:
|
|
missing_tickers = await self._batch_check_missing_periods(
|
|
db, chunk_tickers, start_date, end_date, interval
|
|
)
|
|
|
|
# Step 2: If we have missing data, use bulk yfinance fetch
|
|
if missing_tickers and self.yf_available:
|
|
logger.info(f"Bulk fetching price data for {len(missing_tickers)} tickers in chunk {chunk_num}")
|
|
await self._bulk_fetch_and_store_price_data(
|
|
db, missing_tickers, start_date, end_date, interval
|
|
)
|
|
|
|
# Step 3: Batch retrieve all data from database for this chunk
|
|
ticker_data_map = await self._batch_get_price_data_from_db(
|
|
db, chunk_tickers, start_date, end_date, interval
|
|
)
|
|
|
|
# Step 4: Process results for this chunk
|
|
chunk_results = []
|
|
chunk_successful = 0
|
|
chunk_failed = 0
|
|
|
|
for ticker in chunk_tickers:
|
|
try:
|
|
price_data = ticker_data_map.get(ticker, [])
|
|
|
|
# Convert to response models
|
|
price_points = [
|
|
PriceDataPoint.model_validate(pd) for pd in price_data
|
|
]
|
|
|
|
# Calculate actual date range from returned data
|
|
actual_start_date = start_date
|
|
actual_end_date = end_date
|
|
|
|
if price_points:
|
|
# Get actual start and end dates from the data
|
|
actual_start_date = min(point.date for point in price_points)
|
|
actual_end_date = max(point.date for point in price_points)
|
|
|
|
response = PriceDataResponse(
|
|
ticker=ticker,
|
|
interval=interval,
|
|
data=price_points,
|
|
metadata={
|
|
"request_id": str(ticker),
|
|
"data_points": len(price_points),
|
|
"interval": interval,
|
|
"date_range": {
|
|
"start": actual_start_date.isoformat(),
|
|
"end": actual_end_date.isoformat()
|
|
},
|
|
"last_updated": datetime.now(timezone.utc).isoformat()
|
|
}
|
|
)
|
|
|
|
chunk_results.append(BulkPriceDataItem(
|
|
ticker=ticker,
|
|
success=True,
|
|
data=response,
|
|
error=None
|
|
))
|
|
chunk_successful += 1
|
|
|
|
except Exception as e:
|
|
# Handle individual ticker failure
|
|
error_message = str(e)
|
|
if "No price data found" in error_message or "No data returned" in error_message:
|
|
error_message = f"No price data found for ticker {ticker}"
|
|
elif "Invalid ticker" in error_message:
|
|
error_message = f"Invalid or unknown ticker: {ticker}"
|
|
elif "Yahoo Finance data source not available" in error_message:
|
|
error_message = "Yahoo Finance data source not available"
|
|
|
|
chunk_results.append(BulkPriceDataItem(
|
|
ticker=ticker,
|
|
success=False,
|
|
data=None,
|
|
error=error_message
|
|
))
|
|
chunk_failed += 1
|
|
|
|
# Aggregate chunk results
|
|
all_results.extend(chunk_results)
|
|
all_successful += chunk_successful
|
|
all_failed += chunk_failed
|
|
|
|
logger.info(f"Chunk {chunk_num} completed: {chunk_successful} successful, {chunk_failed} failed")
|
|
|
|
# Small delay between chunks to avoid overwhelming APIs
|
|
if chunk_num < total_chunks:
|
|
await asyncio.sleep(0.2)
|
|
|
|
logger.info(f"Bulk processing completed: {all_successful} successful, {all_failed} failed out of {total_tickers} total")
|
|
return all_results, all_successful, all_failed
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in optimized bulk processing: {str(e)}")
|
|
# Fallback to individual processing
|
|
return await self._fallback_individual_processing(
|
|
db, tickers, start_date, end_date, interval, force_refresh
|
|
)
|
|
|
|
async def _batch_check_missing_periods(
|
|
self,
|
|
db: AsyncSession,
|
|
tickers: List[str],
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str
|
|
) -> List[str]:
|
|
"""Batch check which tickers have missing periods"""
|
|
# Single query to check all tickers at once
|
|
from sqlalchemy import func, case
|
|
|
|
result = await db.execute(
|
|
select(
|
|
PriceData.ticker,
|
|
func.count(PriceData.date).label('count'),
|
|
func.min(PriceData.date).label('min_date'),
|
|
func.max(PriceData.date).label('max_date')
|
|
)
|
|
.where(
|
|
and_(
|
|
PriceData.ticker.in_(tickers),
|
|
PriceData.date >= start_date,
|
|
PriceData.date <= end_date
|
|
)
|
|
)
|
|
.group_by(PriceData.ticker)
|
|
)
|
|
|
|
existing_tickers = {}
|
|
for row in result.fetchall():
|
|
ticker, count, min_date, max_date = row
|
|
existing_tickers[ticker] = {
|
|
'count': count,
|
|
'min_date': min_date,
|
|
'max_date': max_date
|
|
}
|
|
|
|
# Determine expected count based on interval
|
|
expected_days = (end_date - start_date).days
|
|
if interval == "1d":
|
|
expected_count = expected_days * 0.7 # Rough estimate for trading days
|
|
elif interval == "1w":
|
|
expected_count = expected_days / 7
|
|
else:
|
|
expected_count = 1
|
|
|
|
missing_tickers = []
|
|
for ticker in tickers:
|
|
ticker_data = existing_tickers.get(ticker)
|
|
if not ticker_data or ticker_data['count'] < expected_count * 0.8:
|
|
missing_tickers.append(ticker)
|
|
|
|
logger.info(f"Found {len(missing_tickers)} tickers needing data refresh out of {len(tickers)}")
|
|
return missing_tickers
|
|
|
|
async def _bulk_fetch_and_store_price_data(
|
|
self,
|
|
db: AsyncSession,
|
|
tickers: List[str],
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str
|
|
):
|
|
"""Optimized bulk fetch using yfinance-plus bulk features"""
|
|
try:
|
|
logger.info(f"Starting bulk fetch for {len(tickers)} tickers")
|
|
|
|
# Convert dates to strings
|
|
start_str = start_date.strftime('%Y-%m-%d')
|
|
end_str = end_date.strftime('%Y-%m-%d')
|
|
|
|
# Use yfinance-plus bulk download feature
|
|
loop = asyncio.get_event_loop()
|
|
|
|
# Use adaptive chunk size for yfinance API calls based on ticker count
|
|
# Smaller chunks for yfinance API calls to avoid overwhelming the service
|
|
total_tickers = len(tickers)
|
|
if total_tickers <= 10:
|
|
chunk_size = total_tickers # Single chunk for very small batches
|
|
elif total_tickers <= 50:
|
|
chunk_size = 15 # Small chunks for moderate batches
|
|
else:
|
|
chunk_size = 20 # Standard chunks for large batches
|
|
for i in range(0, len(tickers), chunk_size):
|
|
chunk_tickers = tickers[i:i + chunk_size]
|
|
|
|
logger.info(f"Processing chunk {i//chunk_size + 1}: {len(chunk_tickers)} tickers")
|
|
|
|
# Use yfinance-plus bulk download
|
|
bulk_data = await loop.run_in_executor(
|
|
None,
|
|
lambda: yf.download(
|
|
tickers=' '.join(chunk_tickers),
|
|
start=start_str,
|
|
end=end_str,
|
|
interval=interval,
|
|
auto_adjust=True,
|
|
prepost=False,
|
|
group_by='ticker',
|
|
threads=True # Enable multi-threading
|
|
)
|
|
)
|
|
|
|
# Process and store data for each ticker in the chunk
|
|
await self._process_bulk_data(db, chunk_tickers, bulk_data, interval)
|
|
|
|
# Small delay to be nice to the API
|
|
await asyncio.sleep(0.1)
|
|
|
|
await db.commit()
|
|
logger.info(f"Successfully completed bulk fetch for {len(tickers)} tickers")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in bulk fetch: {str(e)}")
|
|
await db.rollback()
|
|
raise
|
|
|
|
async def _process_bulk_data(
|
|
self,
|
|
db: AsyncSession,
|
|
tickers: List[str],
|
|
bulk_data,
|
|
interval: str
|
|
):
|
|
"""Process bulk data returned from yfinance and store in database"""
|
|
if bulk_data.empty:
|
|
logger.warning("No bulk data returned from yfinance")
|
|
return
|
|
|
|
# Handle different data structures from yfinance bulk download
|
|
if len(tickers) == 1:
|
|
# Single ticker - data is a simple DataFrame
|
|
await self._store_ticker_data(db, tickers[0], bulk_data, interval)
|
|
else:
|
|
# Multiple tickers - data is grouped by ticker
|
|
for ticker in tickers:
|
|
try:
|
|
if ticker in bulk_data.columns.get_level_values(0):
|
|
ticker_data = bulk_data[ticker]
|
|
if not ticker_data.empty:
|
|
await self._store_ticker_data(db, ticker, ticker_data, interval)
|
|
except Exception as e:
|
|
logger.error(f"Error processing data for {ticker}: {str(e)}")
|
|
continue
|
|
|
|
async def _store_ticker_data(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str,
|
|
ticker_data,
|
|
interval: str
|
|
):
|
|
"""Store individual ticker data with optimized batch operations"""
|
|
# Batch check existing dates to avoid individual DB queries
|
|
existing_dates = await self._get_existing_dates_for_ticker(db, ticker)
|
|
|
|
new_records = []
|
|
for date, row in ticker_data.iterrows():
|
|
# Convert pandas timestamp to datetime
|
|
price_date = date.to_pydatetime()
|
|
if price_date.tzinfo is None:
|
|
price_date = price_date.replace(tzinfo=timezone.utc)
|
|
|
|
# Skip if already exists
|
|
if price_date.date() in existing_dates:
|
|
continue
|
|
|
|
# Prepare new record
|
|
price_record = PriceData(
|
|
ticker=ticker,
|
|
date=price_date,
|
|
open=float(row.get('Open', 0)) if not pd.isna(row.get('Open')) else None,
|
|
high=float(row.get('High', 0)) if not pd.isna(row.get('High')) else None,
|
|
low=float(row.get('Low', 0)) if not pd.isna(row.get('Low')) else None,
|
|
close=float(row.get('Close', 0)) if not pd.isna(row.get('Close')) else 0,
|
|
volume=float(row.get('Volume', 0)) if not pd.isna(row.get('Volume')) else None,
|
|
adjusted_close=float(row.get('Close', 0)) if not pd.isna(row.get('Close')) else None,
|
|
data_source=DataSource.YAHOO_FINANCE
|
|
)
|
|
new_records.append(price_record)
|
|
|
|
# Batch insert new records
|
|
if new_records:
|
|
db.add_all(new_records)
|
|
logger.info(f"Added {len(new_records)} new price records for {ticker}")
|
|
|
|
async def _get_existing_dates_for_ticker(
|
|
self,
|
|
db: AsyncSession,
|
|
ticker: str
|
|
) -> set:
|
|
"""Get existing dates for a ticker to avoid duplicates"""
|
|
result = await db.execute(
|
|
select(PriceData.date)
|
|
.where(PriceData.ticker == ticker)
|
|
)
|
|
return {row[0].date() for row in result.fetchall()}
|
|
|
|
async def _batch_get_price_data_from_db(
|
|
self,
|
|
db: AsyncSession,
|
|
tickers: List[str],
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str
|
|
) -> Dict[str, List[PriceData]]:
|
|
"""Batch retrieve price data for multiple tickers"""
|
|
# Single query to get data for all tickers
|
|
result = await db.execute(
|
|
select(PriceData)
|
|
.where(
|
|
and_(
|
|
PriceData.ticker.in_(tickers),
|
|
PriceData.date >= start_date,
|
|
PriceData.date <= end_date
|
|
)
|
|
)
|
|
.order_by(PriceData.ticker, PriceData.date)
|
|
)
|
|
|
|
# Group results by ticker
|
|
ticker_data_map = {}
|
|
for ticker in tickers:
|
|
ticker_data_map[ticker] = []
|
|
|
|
for record in result.scalars().all():
|
|
if record.ticker in ticker_data_map:
|
|
ticker_data_map[record.ticker].append(record)
|
|
|
|
return ticker_data_map
|
|
|
|
async def _fallback_individual_processing(
|
|
self,
|
|
db: AsyncSession,
|
|
tickers: List[str],
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
interval: str,
|
|
force_refresh: bool
|
|
) -> Tuple[List, int, int]:
|
|
"""Fallback to individual processing if bulk processing fails"""
|
|
from app.schemas.financial import BulkPriceDataItem, PriceDataResponse, PriceDataPoint
|
|
|
|
logger.warning("Falling back to individual ticker processing")
|
|
|
|
results = []
|
|
successful_count = 0
|
|
failed_count = 0
|
|
|
|
for ticker in tickers:
|
|
try:
|
|
# Get price data
|
|
price_data = await self.get_or_update_price_data(
|
|
db, ticker, start_date, end_date, interval, force_refresh
|
|
)
|
|
|
|
# Convert to response models
|
|
price_points = [
|
|
PriceDataPoint.model_validate(pd) for pd in price_data
|
|
]
|
|
|
|
# Calculate actual date range from returned data
|
|
actual_start_date = start_date
|
|
actual_end_date = end_date
|
|
|
|
if price_points:
|
|
actual_start_date = min(point.date for point in price_points)
|
|
actual_end_date = max(point.date for point in price_points)
|
|
|
|
response = PriceDataResponse(
|
|
ticker=ticker,
|
|
interval=interval,
|
|
data=price_points,
|
|
metadata={
|
|
"request_id": str(ticker),
|
|
"data_points": len(price_points),
|
|
"interval": interval,
|
|
"date_range": {
|
|
"start": actual_start_date.isoformat(),
|
|
"end": actual_end_date.isoformat()
|
|
},
|
|
"last_updated": datetime.now(timezone.utc).isoformat()
|
|
}
|
|
)
|
|
|
|
results.append(BulkPriceDataItem(
|
|
ticker=ticker,
|
|
success=True,
|
|
data=response,
|
|
error=None
|
|
))
|
|
successful_count += 1
|
|
|
|
except Exception as e:
|
|
error_message = str(e)
|
|
results.append(BulkPriceDataItem(
|
|
ticker=ticker,
|
|
success=False,
|
|
data=None,
|
|
error=error_message
|
|
))
|
|
failed_count += 1
|
|
|
|
return results, successful_count, failed_count
|
|
|
|
|
|
# Import pandas for data processing
|
|
try:
|
|
import pandas as pd
|
|
except ImportError:
|
|
logger.error("pandas not available - price data service will not work")
|
|
pd = None |