fix: replace concurrent yfinance fetches with single batch download to avoid rate limiting

This commit is contained in:
Johan Martensson
2026-03-22 05:31:28 +00:00
parent 803a296133
commit 98a9293166
2 changed files with 99 additions and 139 deletions
+73 -34
View File
@@ -1,9 +1,8 @@
"""Financial data fetchers — defense stocks and oil prices.
Uses yfinance for ticker data with concurrent execution for performance.
Uses yfinance batch download to minimise Yahoo Finance requests and avoid rate limiting.
"""
import logging
import concurrent.futures
import yfinance as yf
from services.fetchers._store import latest_data, _data_lock, _mark_fresh
from services.fetchers.retry import with_retry
@@ -11,48 +10,88 @@ from services.fetchers.retry import with_retry
logger = logging.getLogger(__name__)
def _fetch_single_ticker(symbol: str, period: str = "2d"):
"""Fetch a single yfinance ticker. Returns (symbol, data_dict) or (symbol, None)."""
def _batch_fetch(symbols: list[str], period: str = "5d") -> dict:
"""Fetch multiple tickers in a single yfinance request. Returns {symbol: {price, change_percent, up}}."""
try:
ticker = yf.Ticker(symbol)
hist = ticker.history(period=period)
if len(hist) >= 1:
current_price = hist['Close'].iloc[-1]
prev_close = hist['Close'].iloc[0] if len(hist) > 1 else current_price
change_percent = ((current_price - prev_close) / prev_close) * 100 if prev_close else 0
return symbol, {
"price": round(float(current_price), 2),
"change_percent": round(float(change_percent), 2),
"up": bool(change_percent >= 0)
}
hist = yf.download(symbols, period=period, auto_adjust=True, progress=False)
if hist.empty:
return {}
close = hist["Close"]
result = {}
for sym in symbols:
try:
col = close[sym] if len(symbols) > 1 else close
col = col.dropna()
if len(col) < 1:
continue
current = float(col.iloc[-1])
prev = float(col.iloc[0]) if len(col) > 1 else current
change = ((current - prev) / prev * 100) if prev else 0
result[sym] = {
"price": round(current, 2),
"change_percent": round(change, 2),
"up": bool(change >= 0),
}
except Exception as e:
logger.warning(f"Could not parse {sym}: {e}")
return result
except Exception as e:
logger.warning(f"Could not fetch data for {symbol}: {e}")
return symbol, None
logger.warning(f"Batch fetch failed: {e}")
return {}
@with_retry(max_retries=1, base_delay=1)
_STOCK_TICKERS = ["RTX", "LMT", "NOC", "GD", "BA", "PLTR"]
_OIL_MAP = {"WTI Crude": "CL=F", "Brent Crude": "BZ=F"}
_ALL_TICKERS = _STOCK_TICKERS + list(_OIL_MAP.values())
_MARKET_COOLDOWN_SECONDS = 1800 # fetch at most once every 30 minutes
_last_market_fetch: float = 0.0
def _fetch_all_market_data():
"""Single yfinance download for all market tickers to avoid rate limiting."""
raw = _batch_fetch(_ALL_TICKERS, period="5d")
stocks = {sym: raw[sym] for sym in _STOCK_TICKERS if sym in raw}
oil = {name: raw[sym] for name, sym in _OIL_MAP.items() if sym in raw}
return stocks, oil
@with_retry(max_retries=2, base_delay=10)
def fetch_defense_stocks():
tickers = ["RTX", "LMT", "NOC", "GD", "BA", "PLTR"]
global _last_market_fetch
import time
if time.time() - _last_market_fetch < _MARKET_COOLDOWN_SECONDS:
return
try:
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
results = pool.map(lambda t: _fetch_single_ticker(t, "2d"), tickers)
stocks_data = {sym: data for sym, data in results if data}
with _data_lock:
latest_data['stocks'] = stocks_data
_mark_fresh("stocks")
stocks, oil = _fetch_all_market_data()
if stocks:
_last_market_fetch = time.time()
with _data_lock:
latest_data['stocks'] = stocks
if oil:
latest_data['oil'] = oil
_mark_fresh("stocks")
if oil:
_mark_fresh("oil")
logger.info(f"Markets: {len(stocks)} stocks, {len(oil)} oil tickers")
else:
logger.warning("Markets: empty result from yfinance (rate limited?)")
except Exception as e:
logger.error(f"Error fetching stocks: {e}")
logger.error(f"Error fetching market data: {e}")
@with_retry(max_retries=1, base_delay=1)
@with_retry(max_retries=1, base_delay=10)
def fetch_oil_prices():
tickers = {"WTI Crude": "CL=F", "Brent Crude": "BZ=F"}
# Oil is now fetched together with stocks in fetch_defense_stocks to use a single request.
# This function is kept for scheduler compatibility but is a no-op if stocks already ran.
with _data_lock:
if latest_data.get('oil'):
return # Already populated by fetch_defense_stocks
try:
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
results = pool.map(lambda item: (_fetch_single_ticker(item[1], "5d")[1], item[0]), tickers.items())
oil_data = {name: data for data, name in results if data}
with _data_lock:
latest_data['oil'] = oil_data
_mark_fresh("oil")
_, oil = _fetch_all_market_data()
if oil:
with _data_lock:
latest_data['oil'] = oil
_mark_fresh("oil")
except Exception as e:
logger.error(f"Error fetching oil: {e}")