discord-stock-bot/polygon_api.py

"""Polygon.io API client for historical stock data."""
import requests
import logging
from typing import Optional, Dict, List
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)


class PolygonAPI:
    """Polygon.io implementation for historical stock data."""

    def __init__(self, api_key: str):
        """
        Initialize Polygon API client.

        Args:
            api_key: Polygon.io API key
        """
        self.api_key = api_key
        self.base_url = "https://api.polygon.io"

    def get_candles(self, ticker: str, days: int = 30, resolution: str = 'D') -> Optional[Dict[str, List]]:
        """
        Retrieve historical candlestick data from Polygon.io.

        Args:
            ticker: Stock ticker symbol
            days: Number of days of historical data to fetch (ignored for intraday)
            resolution: Candle resolution ('D' for daily, '5min' requires paid tier)

        Returns:
            Dictionary with OHLCV data or None if unavailable
        """
        try:
            if resolution == '5min':
                # For 5-minute bars, get current day data
                # Start at market open (9:30 AM ET)
                import pytz
                et_tz = pytz.timezone('America/New_York')
                now_et = datetime.now(et_tz)

                # Set to today at 9:30 AM ET
                market_open = now_et.replace(hour=9, minute=30, second=0, microsecond=0)

                # If current time is before market open, use yesterday
                if now_et < market_open:
                    market_open = market_open - timedelta(days=1)
                    now_et = market_open.replace(hour=16, minute=0)  # Use previous day's close

                from_date = market_open.strftime('%Y-%m-%d')
                to_date = now_et.strftime('%Y-%m-%d')

                # Build API URL for 5-minute bars
                url = f"{self.base_url}/v2/aggs/ticker/{ticker}/range/5/minute/{from_date}/{to_date}"
            else:
                # Daily bars - Calculate date range (end yesterday to avoid missing today's data)
                end_date = datetime.now() - timedelta(days=1)
                start_date = end_date - timedelta(days=days)

                # Format dates as YYYY-MM-DD
                from_date = start_date.strftime('%Y-%m-%d')
                to_date = end_date.strftime('%Y-%m-%d')

                # Build API URL for daily bars
                url = f"{self.base_url}/v2/aggs/ticker/{ticker}/range/1/day/{from_date}/{to_date}"

            # Make request
            params = {
                'adjusted': 'true',
                'sort': 'asc',
                'apiKey': self.api_key
            }

            logger.info(f"Fetching Polygon data for {ticker} from {from_date} to {to_date}")
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()

            data = response.json()

            # Log response for debugging
            logger.info(f"Polygon API response status: {data.get('status')}, results count: {len(data.get('results', []))}")

            # Check if we got results
            if data.get('status') != 'OK' or not data.get('results'):
                logger.warning(f"No candle data available for ticker: {ticker}. Response: {data}")
                return None

            results = data['results']

            # Convert to expected format
            timestamps = []
            opens = []
            highs = []
            lows = []
            closes = []
            volumes = []

            for candle in results:
                # Polygon returns timestamp in milliseconds
                timestamps.append(int(candle['t'] / 1000))
                opens.append(float(candle['o']))
                highs.append(float(candle['h']))
                lows.append(float(candle['l']))
                closes.append(float(candle['c']))
                volumes.append(int(candle['v']))

            logger.info(f"Successfully fetched {len(timestamps)} candles for {ticker}")

            return {
                'timestamps': timestamps,
                'open': opens,
                'high': highs,
                'low': lows,
                'close': closes,
                'volume': volumes
            }

        except requests.exceptions.RequestException as e:
            logger.error(f"HTTP error fetching candle data for {ticker}: {e}")
            return None
        except Exception as e:
            logger.error(f"Error fetching candle data for {ticker}: {e}")
            return None

    def is_available(self) -> bool:
        """
        Check if Polygon API is accessible.

        Returns:
            True if accessible, False otherwise
        """
        try:
            # Test with a simple request
            test_data = self.get_candles("AAPL", days=5)
            return test_data is not None
        except Exception as e:
            logger.error(f"Polygon API unavailable: {e}")
            return False