add snapshot (#141)

* add snapshot

* style

Author: clickingbuttons

polygon/rest/aggs.py

@@ -4,7 +4,7 @@
 from urllib3 import HTTPResponse
 from datetime import datetime, date
 
-# https://polygon.io/docs/stocks
+
 class AggsClient(BaseClient):
     def get_aggs(
         self,

polygon/rest/models/__init__.py

@@ -1,3 +1,4 @@
+from .shared import *
 from .aggs import *
 from .trades import *
 from .quotes import *
@@ -7,4 +8,4 @@
 from .dividends import *
 from .conditions import *
 from .exchanges import *
-from .shared import *
+from .snapshot import *

polygon/rest/models/aggs.py

@@ -4,11 +4,11 @@
 
 @dataclass
 class Agg:
-    open: float
-    high: float
-    low: float
-    close: float
-    volume: float
+    open: Optional[float]
+    high: Optional[float]
+    low: Optional[float]
+    close: Optional[float]
+    volume: Optional[float]
     vwap: Optional[float]
     timestamp: Optional[int]
     transactions: Optional[int]

polygon/rest/models/shared.py

@@ -60,3 +60,14 @@ class ExchangeType(Enum):
     EXCHANGE = "exchange"
     TRF = "TRF"
     SIP = "SIP"
+
+
+class Direction(Enum):
+    GAINERS = "gainers"
+    LOSERS = "losers"
+
+
+class SnapshotMarketType(Enum):
+    STOCKS = "stocks"
+    FOREX = "forex"
+    CRYPTO = "crypto"

polygon/rest/models/snapshot.py

@@ -0,0 +1,177 @@
+from dataclasses import dataclass
+from typing import Optional, List, Dict
+from .aggs import Agg
+from .quotes import LastQuote
+from .trades import LastTrade
+
+
+@dataclass
+class SnapshotMin:
+    "Most recent minute bar"
+    accumulated_volume: Optional[float]
+    open: Optional[float]
+    high: Optional[float]
+    low: Optional[float]
+    close: Optional[float]
+    volume: Optional[float]
+    vwap: Optional[float]
+
+    @staticmethod
+    def from_dict(d):
+        return SnapshotMin(
+            d.get("ac", None),
+            d.get("o", None),
+            d.get("h", None),
+            d.get("l", None),
+            d.get("c", None),
+            d.get("v", None),
+            d.get("vw", None),
+        )
+
+
+@dataclass
+class Snapshot:
+    day: Optional[Agg]
+    last_quote: Optional[LastQuote]
+    last_trade: Optional[LastTrade]
+    min: Optional[SnapshotMin]
+    prev_day: Optional[Agg]
+    ticker: str
+    todays_change: float
+    todays_change_percent: float
+    updated: int
+
+    @staticmethod
+    def from_dict(d):
+        return Snapshot(
+            d.get("day", None),
+            d.get("lastQuote", None),
+            d.get("lastTrade", None),
+            d.get("min", None),
+            d.get("prevDay", None),
+            d.get("ticker", None),
+            d.get("todaysChange", None),
+            d.get("todaysChangePercent", None),
+            d.get("updated", None),
+        )
+
+
+@dataclass
+class DayOptionContractSnapshot:
+    change: Optional[float]
+    change_percent: Optional[float]
+    close: Optional[float]
+    high: Optional[float]
+    last_updated: Optional[int]
+    low: Optional[float]
+    open: Optional[float]
+    previous_close: Optional[float]
+    volume: Optional[float]
+    vwap: Optional[float]
+
+    @staticmethod
+    def from_dict(d):
+        return DayOptionContractSnapshot(**d)
+
+
+@dataclass
+class OptionDetails:
+    contract_type: str
+    exercise_style: str
+    expiration_date: str
+    shares_per_contract: float
+    strike_price: float
+    ticker: str
+
+    @staticmethod
+    def from_dict(d):
+        return OptionDetails(**d)
+
+
+@dataclass
+class OptionLastQuote:
+    ask: Optional[float]
+    ask_size: Optional[float]
+    bid: Optional[float]
+    bid_size: Optional[float]
+    last_updated: Optional[int]
+    midpoint: Optional[float]
+    timeframe: Optional[str]
+
+    @staticmethod
+    def from_dict(d):
+        return OptionLastQuote(**d)
+
+
+@dataclass
+class OptionGreeks:
+    delta: Optional[float]
+    gamma: Optional[float]
+    theta: Optional[float]
+    vega: Optional[float]
+
+    @staticmethod
+    def from_dict(d):
+        return OptionGreeks(**d)
+
+
+@dataclass
+class UnderlyingAsset:
+    change_to_break_even: Optional[float]
+    last_updated: Optional[int]
+    price: Optional[float]
+    ticker: Optional[str]
+    timeframe: Optional[str]
+
+    @staticmethod
+    def from_dict(d):
+        return UnderlyingAsset(**d)
+
+
+@dataclass
+class OptionContractSnapshot:
+    break_even_price: Optional[float]
+    day: Optional[Agg]
+    details: Optional[OptionDetails]
+    greeks: Optional[OptionGreeks]
+    implied_volatility: Optional[float]
+    last_quote: Optional[OptionLastQuote]
+    open_interest: Optional[float]
+    underlying_asset: Optional[float]
+
+    @staticmethod
+    def from_dict(d):
+        return OptionContractSnapshot(**d)
+
+
+@dataclass
+class OrderBookQuote:
+    price: Optional[float]
+    exchange_shares: Dict[str, float]
+
+    @staticmethod
+    def from_dict(d):
+        return OrderBookQuote(**d)
+
+
+@dataclass
+class SnapshotTickerFullBook:
+    ticker: Optional[str]
+    bids: Optional[List[OrderBookQuote]]
+    asks: Optional[List[OrderBookQuote]]
+    bid_count: Optional[float]
+    ask_count: Optional[float]
+    spread: Optional[float]
+    updated: int
+
+    @staticmethod
+    def from_dict(d):
+        return SnapshotTickerFullBook(
+            d.get("ticker", None),
+            d.get("bids", None),
+            d.get("asks", None),
+            d.get("bidCount", None),
+            d.get("askCount", None),
+            d.get("spread", None),
+            d.get("updated", None),
+        )

polygon/rest/snapshot.py

@@ -0,0 +1,136 @@
+from .base import BaseClient
+from typing import Optional, Any, Dict, List, Union
+from .models import (
+    Snapshot,
+    Direction,
+    OptionContractSnapshot,
+    SnapshotMarketType,
+    SnapshotTickerFullBook,
+)
+from urllib3 import HTTPResponse
+
+
+class SnapshotClient(BaseClient):
+    def get_snapshot_all(
+        self,
+        market_type: Optional[Union[str, SnapshotMarketType]] = "stocks",
+        tickers: Optional[Union[str, List[str]]] = None,
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False,
+    ) -> Union[List[Snapshot], HTTPResponse]:
+        """
+        Get the most up-to-date market data for all traded stock symbols.
+
+        Note: Snapshot data is cleared at 12am EST and gets populated as data is received from the exchanges. This can happen as early as 4am EST.
+
+        :param market_type: Which market to get a snapshot of.
+        :param tickers: A comma separated list of tickers to get snapshots for.
+        :return: List of Snapshots
+        """
+        url = f"/v2/snapshot/locale/us/markets/{market_type}/tickers"
+        if type(tickers) is list:
+            tickers = ",".join(tickers)
+        return self._get(
+            path=url,
+            params=self._get_params(self.get_snapshot_all, locals()),
+            deserializer=Snapshot.from_dict,
+            raw=raw,
+        )
+
+    def get_snapshot_direction(
+        self,
+        direction: Union[str, Direction],
+        market_type: Optional[Union[str, SnapshotMarketType]] = "stocks",
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False,
+    ) -> Union[List[Snapshot], HTTPResponse]:
+        """
+        Get the most up-to-date market data for the current top 20 gainers or losers of the day in the stocks/equities markets.
+
+        Top gainers are those tickers whose price has increased by the highest percentage since the previous day's close. Top losers are those tickers whose price has decreased by the highest percentage since the previous day's close.
+
+        Note: Snapshot data is cleared at 12am EST and gets populated as data is received from the exchanges.
+
+        :param market_type: Which market to get a snapshot of.
+        :param direction: The direction ("gainers" or "losers")
+        :return: List of Snapshots
+        """
+        url = f"/v2/snapshot/locale/us/markets/{market_type}/{direction}"
+        return self._get(
+            path=url,
+            params=self._get_params(self.get_snapshot_direction, locals()),
+            result_key="tickers",
+            deserializer=Snapshot.from_dict,
+            raw=raw,
+        )
+
+    def get_snapshot_ticker(
+        self,
+        ticker: str,
+        market_type: Optional[Union[str, SnapshotMarketType]] = "stocks",
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False,
+    ) -> Union[Snapshot, HTTPResponse]:
+        """
+        Get the most up-to-date market data for all traded stock symbols.
+
+        Note: Snapshot data is cleared at 12am EST and gets populated as data is received from the exchanges. This can happen as early as 4am EST.
+
+        :param market_type: Which market to get a snapshot of.
+        :param ticker: The ticker symbol.
+        :return: List of Snapshots
+        """
+        url = f"/v2/snapshot/locale/us/markets/{market_type}/tickers/{ticker}"
+        return self._get(
+            path=url,
+            params=self._get_params(self.get_snapshot_ticker, locals()),
+            result_key="ticker",
+            deserializer=Snapshot.from_dict,
+            raw=raw,
+        )
+
+    def get_snapshot_option(
+        self,
+        underlying_asset: str,
+        option_contract: str,
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False,
+    ) -> Union[OptionContractSnapshot, HTTPResponse]:
+        """
+        Get the snapshot of an option contract for a stock equity.
+
+        :param underlying_asset: The underlying ticker symbol of the option contract.
+        :param option_contract: The option contract identifier.
+        :return: List of Snapshots
+        """
+        url = f"/v2/snapshot/options/{underlying_asset}/{option_contract}"
+        return self._get(
+            path=url,
+            params=self._get_params(self.get_snapshot_option, locals()),
+            result_key="results",
+            deserializer=OptionContractSnapshot.from_dict,
+            raw=raw,
+        )
+
+    def get_snapshot_crypto_book(
+        self,
+        ticker: str,
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False,
+    ) -> Union[SnapshotTickerFullBook, HTTPResponse]:
+        """
+        Get the current level 2 book of a single ticker. This is the combined book from all of the exchanges.
+
+        Note: Snapshot data is cleared at 12am EST and gets populated as data is received from the exchanges.
+
+        :param ticker: The ticker symbol.
+        :return: List of Snapshots
+        """
+        url = f" /v2/snapshot/locale/global/markets/crypto/tickers/{ticker}/book"
+        return self._get(
+            path=url,
+            params=self._get_params(self.get_snapshot_crypto_book, locals()),
+            result_key="data",
+            deserializer=SnapshotTickerFullBook.from_dict,
+            raw=raw,
+        )