"""Yahoo Finance provider for quotes and historical data."""
import yfinance as yf
import pandas as pd
from datetime import datetime
from typing import Optional
from ..schemas import (
SymbolSearchResult,
Quote,
HistoryRecord,
)
from ..util import ProviderError, now_iso
class YahooProvider:
"""Yahoo Finance data provider."""
def search_symbol(self, query: str) -> list[SymbolSearchResult]:
"""
Search for symbols using Yahoo Finance.
Note: Basic exact-match lookup.
"""
try:
ticker = yf.Ticker(query.upper())
info = ticker.info
if info and 'symbol' in info:
return [SymbolSearchResult(
symbol=info.get('symbol', query.upper()),
name=info.get('longName') or info.get('shortName', ''),
exchange=info.get('exchange'),
source="yahoo"
)]
return []
except Exception as e:
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Yahoo symbol search failed: {str(e)}",
details={"query": query},
provider="yahoo"
)
def get_quote(self, symbol: str) -> Quote:
"""
Get current quote from Yahoo Finance.
"""
try:
ticker = yf.Ticker(symbol)
quote_data = {
"symbol": symbol,
"timestamp": now_iso(),
"source": "yahoo"
}
# Try fast_info first (faster but less data)
try:
fast_info = ticker.fast_info
quote_data["price"] = getattr(fast_info, 'last_price', None)
quote_data["currency"] = getattr(fast_info, 'currency', None)
quote_data["exchange"] = getattr(fast_info, 'exchange', None)
quote_data["prev_close"] = getattr(fast_info, 'previous_close', None)
quote_data["open"] = getattr(fast_info, 'open', None)
# Calculate change if we have price and prev_close
if quote_data["price"] and quote_data["prev_close"]:
quote_data["change"] = quote_data["price"] - quote_data["prev_close"]
quote_data["change_pct"] = (quote_data["change"] / quote_data["prev_close"]) * 100
except Exception:
# fast_info failed, will try info below
pass
# Fallback to info if we don't have price yet
if quote_data.get("price") is None:
try:
info = ticker.info
quote_data["price"] = info.get('currentPrice') or info.get('regularMarketPrice')
quote_data["name"] = info.get('longName') or info.get('shortName')
quote_data["currency"] = info.get('currency')
quote_data["exchange"] = info.get('exchange')
quote_data["open"] = info.get('regularMarketOpen')
quote_data["high"] = info.get('regularMarketDayHigh')
quote_data["low"] = info.get('regularMarketDayLow')
quote_data["prev_close"] = info.get('previousClose') or info.get('regularMarketPreviousClose')
quote_data["volume"] = info.get('volume') or info.get('regularMarketVolume')
# Calculate change
if quote_data["price"] and quote_data["prev_close"]:
quote_data["change"] = quote_data["price"] - quote_data["prev_close"]
quote_data["change_pct"] = (quote_data["change"] / quote_data["prev_close"]) * 100
except Exception as e:
raise ProviderError(
code="QUOTE_FETCH_FAILED",
message=f"Failed to fetch quote data: {str(e)}",
details={"symbol": symbol},
provider="yahoo"
)
# Check if we got any valid price
if quote_data.get("price") is None:
raise ProviderError(
code="SYMBOL_NOT_FOUND",
message=f"No quote data found for {symbol}",
details={"symbol": symbol},
provider="yahoo"
)
return Quote(**quote_data)
except ProviderError:
raise
except Exception as e:
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Yahoo quote fetch failed: {str(e)}",
details={"symbol": symbol},
provider="yahoo"
)
def get_history(
self,
symbol: str,
start_date: str,
end_date: str,
interval: str = "1d"
) -> list[HistoryRecord]:
"""
Get historical OHLCV data from Yahoo Finance.
"""
try:
# Validate date format
try:
datetime.strptime(start_date, "%Y-%m-%d")
datetime.strptime(end_date, "%Y-%m-%d")
except ValueError as e:
raise ProviderError(
code="INVALID_ARGUMENT",
message=f"Invalid date format: {str(e)}",
details={"start_date": start_date, "end_date": end_date},
provider="yahoo"
)
# Download data
df = yf.download(
symbol,
start=start_date,
end=end_date,
interval=interval,
progress=False
)
if df.empty:
raise ProviderError(
code="NO_DATA",
message=f"No historical data available for {symbol}",
details={"symbol": symbol, "start_date": start_date, "end_date": end_date},
provider="yahoo"
)
# Convert to records
records = []
for date, row in df.iterrows():
try:
# Handle both single and multi-level column names
if isinstance(df.columns, pd.MultiIndex):
open_val = row[('Open', symbol)]
high_val = row[('High', symbol)]
low_val = row[('Low', symbol)]
close_val = row[('Close', symbol)]
volume_val = row[('Volume', symbol)]
else:
open_val = row['Open']
high_val = row['High']
low_val = row['Low']
close_val = row['Close']
volume_val = row['Volume']
records.append(HistoryRecord(
date=date.strftime("%Y-%m-%d"),
open=float(open_val),
high=float(high_val),
low=float(low_val),
close=float(close_val),
volume=int(volume_val)
))
except (KeyError, ValueError):
# Skip records with missing data
continue
if not records:
raise ProviderError(
code="NO_DATA",
message=f"No valid historical data could be parsed for {symbol}",
details={"symbol": symbol},
provider="yahoo"
)
return records
except ProviderError:
raise
except Exception as e:
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Yahoo history fetch failed: {str(e)}",
details={"symbol": symbol, "start_date": start_date, "end_date": end_date},
provider="yahoo"
)
