Quotes (#117)

Author: Darcy-Linde

polygon/rest/__init__.py

@@ -1,6 +1,7 @@
 from .aggs import AggsClient
 from .trades import TradesClient
+from .quotes import QuotesClient
 
 
-class RESTClient(AggsClient, TradesClient):
+class RESTClient(AggsClient, TradesClient, QuotesClient):
     pass

polygon/rest/models/__init__.py

@@ -1,5 +1,6 @@
 from .aggs import *
 from .trades import *
+from .quotes import *
 from enum import Enum
 
 

polygon/rest/models/quotes.py

@@ -0,0 +1,61 @@
+from typing import Optional, List
+from dataclasses import dataclass
+
+
+@dataclass
+class Quote:
+    "Quote contains quote data for a specified ticker symbol."
+    ask_exchange: Optional[int] = None
+    ask_price: Optional[float] = None
+    ask_size: Optional[float] = None
+    bid_exchange: Optional[int] = None
+    bid_price: Optional[float] = None
+    bid_size: Optional[float] = None
+    conditions: Optional[List[int]] = None
+    indicators: Optional[List[int]] = None
+    participant_timestamp: Optional[int] = None
+    sequence_number: Optional[int] = None
+    sip_timestamp: Optional[int] = None
+    tape: Optional[int] = None
+    trf_timestamp: Optional[int] = None
+
+    @staticmethod
+    def from_dict(d):
+        return Quote(**d)
+
+@dataclass
+class LastQuote:
+    "LastQuote contains data for the most recent NBBO (Quote) tick for a given stock."
+    ticker: Optional[str] = None
+    trf_timestamp: Optional[int] = None
+    sequence_number: Optional[int] = None
+    sip_timestamp: Optional[int] = None
+    participant_timestamp: Optional[int] = None
+    ask_price: Optional[float] = None
+    ask_size: Optional[int] = None
+    ask_exchange: Optional[int] = None
+    conditions: Optional[List[int]] = None
+    indicators: Optional[List[int]] = None
+    bid_price: Optional[float] = None
+    bid_size: Optional[int] = None
+    bid_exchange: Optional[int] = None
+    tape: Optional[int] = None
+
+    @staticmethod
+    def from_dict(d):
+        return LastQuote(
+            ticker=d.get("T", None),
+            trf_timestamp=d.get("f", None),
+            sequence_number=d.get("q", None),
+            sip_timestamp=d.get("t", None),
+            participant_timestamp=d.get("y", None),
+            ask_price=d.get("P", None),
+            ask_size=d.get("S", None),
+            ask_exchange=d.get("X", None),
+            conditions=d.get("c", None),
+            indicators=d.get("i", None),
+            bid_price=d.get("p", None),
+            bid_size=d.get("s", None),
+            bid_exchange=d.get("x", None),
+            tape=d.get("z", None)
+        )

polygon/rest/quotes.py

@@ -0,0 +1,67 @@
+from .base import BaseClient
+from typing import Optional, Any, Dict, List, Union
+from .models import Quote, LastQuote, Sort, Order
+from urllib3 import HTTPResponse
+
+# https://polygon.io/docs/stocks
+class QuotesClient(BaseClient):
+    def list_quotes(
+        self,
+        ticker: str,
+        timestamp: Optional[str] = None,
+        timestamp_lt: Optional[str] = None,
+        timestamp_lte: Optional[str] = None,
+        timestamp_gt: Optional[str] = None,
+        timestamp_gte: Optional[str] = None,
+        limit: Optional[int] = None,
+        sort: Optional[Union[str, Sort]] = None,
+        order: Optional[Union[str, Order]] = None,
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False
+    ) -> Union[List[Quote], HTTPResponse]:
+        """
+        Get quotes for a ticker symbol in a given time range.
+
+        :param ticker: The ticker symbol to get quotes for.
+        :param timestamp: Query by timestamp. Either a date with the format YYYY-MM-DD or a nanosecond timestamp.
+        :param timestamp_lt: Timestamp less than
+        :param timestamp_lte: Timestamp less than or equal to
+        :param timestamp_gt: Timestamp greater than
+        :param timestamp_gte: Timestamp greater than or equal to
+        :param limit: Limit the number of results returned, default is 10 and max is 50000.
+        :param sort: Sort field used for ordering.
+        :param order: Order results based on the sort field.
+        :param params: Any additional query params
+        :param raw: Return HTTPResponse object instead of results object
+        :return: List of quotes
+        :rtype: List[Quote]
+        """
+        url = f"/v3/quotes/{ticker}"
+
+        return self._paginate(
+            path=url,
+            params=self._get_params(self.list_quotes, locals()),
+            raw=raw,
+            deserializer=Quote.from_dict,
+        )
+    
+    def get_last_quote(
+        self,
+        ticker: str,
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False
+    ):
+        """
+        Get the most recent NBBO (Quote) tick for a given stock.
+
+        :param ticker: The ticker symbol of the stock/equity.
+        :param params: Any additional query params
+        :param raw: Return HTTPResponse object instead of results object
+        :return: Last Quote
+        :rtype: LastQuote
+        """
+        url = f"/v2/last/nbbo/{ticker}"
+
+        return self._get(path=url, params=params, deserializer=LastQuote.from_dict, raw=raw)
+
+