"""Finnhub data provider for fundamentals and financial statements."""
import httpx
from typing import Optional
from datetime import datetime
from ..schemas import (
SymbolSearchResult,
Quote,
HistoryRecord,
Fundamentals,
FinancialStatements,
FinancialStatementItem,
)
from ..util import ProviderError, now_iso, timestamp_to_date, get_env
class FinnhubProvider:
"""Finnhub API provider."""
BASE_URL = "https://finnhub.io/api/v1"
def __init__(self, api_key: Optional[str] = None, timeout: int = 10):
self.api_key = api_key or get_env("FINNHUB_API_KEY")
if not self.api_key:
raise ProviderError(
code="MISSING_API_KEY",
message="Finnhub API key is required. Set FINNHUB_API_KEY environment variable.",
provider="finnhub"
)
self.timeout = timeout
self.client = httpx.Client(timeout=timeout)
def _request(self, endpoint: str, params: Optional[dict] = None) -> dict:
"""Make a request to Finnhub API."""
url = f"{self.BASE_URL}/{endpoint}"
params = params or {}
params["token"] = self.api_key
try:
response = self.client.get(url, params=params)
response.raise_for_status()
return response.json()
except httpx.TimeoutException:
raise ProviderError(
code="PROVIDER_TIMEOUT",
message="Finnhub API request timed out",
provider="finnhub"
)
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
raise ProviderError(
code="PROVIDER_RATE_LIMIT",
message="Finnhub API rate limit exceeded",
provider="finnhub"
)
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Finnhub API error: {e.response.status_code}",
details={"status_code": e.response.status_code},
provider="finnhub"
)
except Exception as e:
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Finnhub request failed: {str(e)}",
provider="finnhub"
)
def search_symbol(self, query: str) -> list[SymbolSearchResult]:
"""
Search for symbols using Finnhub.
"""
try:
data = self._request("search", {"q": query})
results = []
for item in data.get("result", [])[:10]:
results.append(SymbolSearchResult(
symbol=item.get("symbol", ""),
name=item.get("description", ""),
market=None, # Finnhub doesn't always provide clear market info
exchange=item.get("type", ""),
source="finnhub"
))
return results
except ProviderError:
raise
except Exception as e:
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Symbol search failed: {str(e)}",
provider="finnhub"
)
def get_quote(self, symbol: str) -> Quote:
"""
Get real-time quote from Finnhub.
"""
try:
data = self._request("quote", {"symbol": symbol})
# Check if we got valid data
if not data or data.get("c") == 0:
raise ProviderError(
code="SYMBOL_NOT_FOUND",
message=f"No quote data found for {symbol}",
details={"symbol": symbol},
provider="finnhub"
)
current_price = data.get("c")
prev_close = data.get("pc")
change = current_price - prev_close if (current_price and prev_close) else None
change_pct = (change / prev_close * 100) if (change and prev_close) else None
return Quote(
symbol=symbol,
price=current_price,
change=change,
change_pct=change_pct,
open=data.get("o"),
high=data.get("h"),
low=data.get("l"),
prev_close=prev_close,
timestamp=now_iso(),
source="finnhub",
source_detail={"provider": "finnhub", "raw": data}
)
except ProviderError:
raise
except Exception as e:
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Quote fetch failed: {str(e)}",
details={"symbol": symbol},
provider="finnhub"
)
def get_fundamentals(self, symbol: str, period: str = "ttm") -> Fundamentals:
"""
Get fundamental metrics from Finnhub.
Uses company-basic-financials endpoint.
"""
try:
# Get basic financials
data = self._request("stock/metric", {
"symbol": symbol,
"metric": "all"
})
metric = data.get("metric", {})
if not metric:
raise ProviderError(
code="NO_DATA",
message=f"No fundamental data found for {symbol}",
details={"symbol": symbol},
provider="finnhub"
)
# Map Finnhub fields to our unified schema
# Note: Field names may vary, using common ones
fundamentals = Fundamentals(
symbol=symbol,
currency=None, # Finnhub doesn't always provide currency in metrics
# Valuation
market_cap=metric.get("marketCapitalization"),
pe=metric.get("peNormalizedAnnual") or metric.get("peTTM"),
pb=metric.get("pbAnnual") or metric.get("pbQuarterly"),
ps=metric.get("psAnnual") or metric.get("psTTM"),
dividend_yield=metric.get("dividendYieldIndicatedAnnual"),
# Profitability
roe=metric.get("roeRfy") or metric.get("roeTTM"),
roa=metric.get("roaRfy") or metric.get("roaTTM"),
gross_margin=metric.get("grossMarginTTM") or metric.get("grossMarginAnnual"),
operating_margin=metric.get("operatingMarginTTM") or metric.get("operatingMarginAnnual"),
net_margin=metric.get("netProfitMarginTTM") or metric.get("netProfitMarginAnnual"),
# Financial Health
debt_to_equity=metric.get("totalDebt/totalEquityQuarterly"),
current_ratio=metric.get("currentRatioQuarterly"),
quick_ratio=metric.get("quickRatioQuarterly"),
# Performance (from series data if available)
revenue_ttm=metric.get("revenueTTM"),
net_income_ttm=metric.get("netIncomeTTM"),
free_cash_flow_ttm=metric.get("freeCashFlowTTM"),
updated_at=now_iso(),
source="finnhub",
raw=metric
)
return fundamentals
except ProviderError:
raise
except Exception as e:
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Fundamentals fetch failed: {str(e)}",
details={"symbol": symbol},
provider="finnhub"
)
def get_financial_statements(
self,
symbol: str,
statement: str,
period: str = "annual"
) -> FinancialStatements:
"""
Get financial statements from Finnhub.
Uses financials-reported endpoint.
"""
try:
# Map our statement types to Finnhub's
statement_map = {
"income": "ic", # Income Statement
"balance": "bs", # Balance Sheet
"cashflow": "cf" # Cash Flow
}
finnhub_statement = statement_map.get(statement)
if not finnhub_statement:
raise ProviderError(
code="INVALID_ARGUMENT",
message=f"Invalid statement type: {statement}",
details={"statement": statement},
provider="finnhub"
)
# Get financials
freq = "annual" if period == "annual" else "quarterly"
data = self._request("stock/financials-reported", {
"symbol": symbol,
"freq": freq
})
statements_data = data.get("data", [])
if not statements_data:
raise ProviderError(
code="NO_DATA",
message=f"No {statement} statement data found for {symbol}",
details={"symbol": symbol, "statement": statement},
provider="finnhub"
)
items = []
for stmt_data in statements_data[:8]: # Last 8 periods
report = stmt_data.get("report", {})
# Extract based on statement type
item_data = {
"period_end": stmt_data.get("endDate", ""),
"raw": report
}
# Map common fields based on statement type
if statement == "income":
ic_data = report.get(finnhub_statement, [])
for entry in ic_data:
label = entry.get("label", "").lower()
value = entry.get("value")
if "revenue" in label or "sales" in label:
item_data["revenue"] = value
elif "gross profit" in label:
item_data["gross_profit"] = value
elif "operating income" in label:
item_data["operating_income"] = value
elif "net income" in label:
item_data["net_income"] = value
elif "earnings per share" in label or "eps" in label:
item_data["eps"] = value
elif statement == "balance":
bs_data = report.get(finnhub_statement, [])
for entry in bs_data:
label = entry.get("label", "").lower()
value = entry.get("value")
if "total assets" in label:
item_data["total_assets"] = value
elif "total liabilities" in label:
item_data["total_liabilities"] = value
elif "total equity" in label or "stockholders' equity" in label:
item_data["total_equity"] = value
elif "cash" in label and "equivalents" in label:
item_data["cash"] = value
elif statement == "cashflow":
cf_data = report.get(finnhub_statement, [])
for entry in cf_data:
label = entry.get("label", "").lower()
value = entry.get("value")
if "operating" in label and "cash flow" in label:
item_data["operating_cash_flow"] = value
elif "investing" in label and "cash flow" in label:
item_data["investing_cash_flow"] = value
elif "financing" in label and "cash flow" in label:
item_data["financing_cash_flow"] = value
elif "free cash flow" in label:
item_data["free_cash_flow"] = value
items.append(FinancialStatementItem(**item_data))
return FinancialStatements(
symbol=symbol,
statement=statement, # type: ignore
period=period, # type: ignore
items=items,
source="finnhub"
)
except ProviderError:
raise
except Exception as e:
raise ProviderError(
code="PROVIDER_ERROR",
message=f"Financial statements fetch failed: {str(e)}",
details={"symbol": symbol, "statement": statement},
provider="finnhub"
)
def close(self):
"""Close the HTTP client."""
self.client.close()
