discord-stock-bot/stock_api/finnhub_api.py

import finnhub
import logging
from typing import Optional, Dict, Any, List
from .base import StockAPIBase
from datetime import datetime, timedelta
import pytz
import urllib.parse
import json


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 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

            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

            # Use MarketHours utility for accurate market status
            from market_hours import MarketHours
            market_open = MarketHours.is_market_open()

            return {
                'ticker': ticker.upper(),
                '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_intraday_candles(self, ticker: str, resolution: int = 5, days_back: int = 1) -> Optional[List[Dict[str, Any]]]:
        """
        Retrieve intraday candlestick data from Finnhub.

        Args:
            ticker: Stock ticker symbol
            resolution: Candle resolution in minutes (1, 5, 15, 30, 60)
            days_back: Number of days of historical data to fetch

        Returns:
            List of candle dictionaries or None if unavailable
        """
        try:
            # Calculate time range (Unix timestamps)
            now = datetime.now(pytz.timezone('America/New_York'))
            end_time = int(now.timestamp())
            start_time = int((now - timedelta(days=days_back)).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}")
                return None

            # Format candles for chart consumption
            formatted_candles = []
            for i in range(len(candles['t'])):
                formatted_candles.append({
                    'x': candles['t'][i] * 1000,  # Convert to milliseconds
                    'o': round(candles['o'][i], 2),
                    'h': round(candles['h'][i], 2),
                    'l': round(candles['l'][i], 2),
                    'c': round(candles['c'][i], 2)
                })

            return formatted_candles

        except Exception as e:
            logger.error(f"Error fetching candles for {ticker}: {e}")
            return None

    def generate_chart_url(self, ticker: str, candles: List[Dict[str, Any]]) -> str:
        """
        Generate QuickChart URL for candlestick chart.

        Args:
            ticker: Stock ticker symbol
            candles: List of candle dictionaries with x, o, h, l, c keys

        Returns:
            QuickChart URL string
        """
        # Limit to last 50 candles for readability
        chart_data = candles[-50:] if len(candles) > 50 else candles

        chart_config = {
            'type': 'candlestick',
            'data': {
                'datasets': [{
                    'label': f'{ticker} 5-Minute',
                    'data': chart_data
                }]
            },
            'options': {
                'plugins': {
                    'title': {
                        'display': True,
                        'text': f'{ticker} - 5 Minute Chart'
                    }
                },
                'scales': {
                    'x': {
                        'type': 'time',
                        'time': {
                            'unit': 'minute',
                            'displayFormats': {
                                'minute': 'HH:mm'
                            }
                        }
                    }
                }
            }
        }

        # Encode config for URL
        config_json = json.dumps(chart_config)
        encoded_config = urllib.parse.quote(config_json)

        return f"https://quickchart.io/chart?v=3&c={encoded_config}&width=800&height=400"