import finnhub import logging from typing import Optional, Dict, Any, List from .base import StockAPIBase from datetime import datetime, timedelta import pytz logger = logging.getLogger(__name__) class FinnhubAPI(StockAPIBase): """Finnhub implementation of stock price provider.""" def __init__(self, api_key: str): """ Initialize Finnhub API client. Args: api_key: Finnhub API key """ self.client = finnhub.Client(api_key=api_key) self.api_key = api_key def get_stock_price(self, ticker: str) -> Optional[Dict[str, Any]]: """ Retrieve stock price data from Finnhub. Args: ticker: Stock ticker symbol Returns: Dictionary with stock data or None if unavailable """ try: # Get company name from company profile company_name = None try: profile = self.client.company_profile2(symbol=ticker) company_name = profile.get('name') or ticker.upper() except Exception as e: logger.warning(f"Could not fetch company name for {ticker}: {e}") company_name = ticker.upper() # Get current quote data quote = self.client.quote(ticker) if not quote or quote.get('c') is None or quote.get('c') == 0: logger.warning(f"No data available for ticker: {ticker}") return None current_price = float(quote['c']) # Current price previous_close = float(quote['pc']) # Previous close open_price = float(quote.get('o', 0)) # Opening price # Use MarketHours utility for accurate market status from market_hours import MarketHours market_open = MarketHours.is_market_open() # At market open, if current price equals previous close and we have an open price, # use the opening price as the current price (trades may not have updated yet) if market_open and current_price == previous_close and open_price > 0: current_price = open_price logger.info(f"Using opening price for {ticker} at market open: ${open_price}") if current_price == 0 or previous_close == 0: logger.warning(f"Invalid price data for ticker: {ticker}") return None change_dollar = current_price - previous_close change_percent = (change_dollar / previous_close) * 100 return { 'ticker': ticker.upper(), 'company_name': company_name, 'current_price': round(current_price, 2), 'previous_close': round(previous_close, 2), 'change_dollar': round(change_dollar, 2), 'change_percent': round(change_percent, 2), 'market_open': market_open } except Exception as e: logger.error(f"Error fetching data for {ticker}: {e}") return None def is_available(self) -> bool: """ Check if Finnhub API is accessible. Returns: True if accessible, False otherwise """ try: test_quote = self.client.quote("AAPL") return test_quote is not None and test_quote.get('c') is not None except Exception as e: logger.error(f"Finnhub API unavailable: {e}") return False def get_earnings_calendar(self, from_date: str, to_date: str, min_market_cap: float = 10_000_000_000) -> Optional[list]: """ Retrieve earnings calendar for date range, filtered by market cap. Args: from_date: Start date (YYYY-MM-DD) to_date: End date (YYYY-MM-DD) min_market_cap: Minimum market cap threshold in dollars (default 10B) Returns: List of enriched earnings dictionaries, or None if unavailable """ try: # Fetch earnings calendar from Finnhub earnings_data = self.client.earnings_calendar(_from=from_date, to=to_date, symbol='', international=False) if not earnings_data or 'earningsCalendar' not in earnings_data: logger.warning(f"No earnings data available for {from_date} to {to_date}") return [] earnings_list = earnings_data['earningsCalendar'] if not earnings_list: logger.info(f"No earnings found for {from_date} to {to_date}") return [] # Enrich each earning with company profile data and filter by market cap enriched_earnings = [] for earning in earnings_list: symbol = earning.get('symbol') if not symbol: continue try: # Fetch company profile for market cap and name profile = self.client.company_profile2(symbol=symbol) if not profile: logger.debug(f"No profile data for {symbol}") continue # Market cap in Finnhub is in millions, convert to dollars market_cap = profile.get('marketCapitalization', 0) * 1_000_000 # Filter by minimum market cap if market_cap < min_market_cap: continue # Enrich earnings data enriched_earning = { 'symbol': symbol, 'company_name': profile.get('name', symbol), 'market_cap': market_cap, 'date': earning.get('date'), 'eps_estimate': earning.get('epsEstimate'), 'eps_actual': earning.get('epsActual'), 'revenue_estimate': earning.get('revenueEstimate'), 'revenue_actual': earning.get('revenueActual'), 'hour': earning.get('hour', 'TBD'), 'quarter': earning.get('quarter'), 'year': earning.get('year') } enriched_earnings.append(enriched_earning) except Exception as e: logger.warning(f"Could not fetch profile for {symbol}: {e}") continue # Sort by date, then by market cap descending enriched_earnings.sort(key=lambda x: (x['date'], -x['market_cap'])) logger.info(f"Found {len(enriched_earnings)} earnings (market cap >= ${min_market_cap/1_000_000_000:.1f}B)") return enriched_earnings except Exception as e: logger.error(f"Error fetching earnings calendar: {e}") return None def get_candles(self, ticker: str, days: int = 30, resolution: str = 'D') -> Optional[Dict[str, List]]: """ Retrieve historical candlestick data from Finnhub. Args: ticker: Stock ticker symbol days: Number of days of historical data to fetch resolution: Candle resolution (1, 5, 15, 30, 60, D, W, M) Returns: Dictionary with OHLCV data or None if unavailable """ try: # Calculate timestamps end_time = int(datetime.now().timestamp()) start_time = int((datetime.now() - timedelta(days=days)).timestamp()) # Fetch candle data candles = self.client.stock_candles(ticker, resolution, start_time, end_time) if not candles or candles.get('s') != 'ok': logger.warning(f"No candle data available for ticker: {ticker}") return None return { 'timestamps': candles['t'], 'open': candles['o'], 'high': candles['h'], 'low': candles['l'], 'close': candles['c'], 'volume': candles['v'] } except Exception as e: logger.error(f"Error fetching candle data for {ticker}: {e}") return None