Markets

polygon/rest/__init__.py

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

polygon/rest/models/__init__.py

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

polygon/rest/models/markets.py

@@ -0,0 +1,31 @@
+from typing import Optional, Dict
+from dataclasses import dataclass
+
+
+@dataclass
+class MarketHoliday:
+    "MarketHoliday contains data for upcoming market holidays and their open/close times."
+    close: Optional[str] = None
+    date: Optional[str] = None
+    exchange: Optional[str] = None
+    name: Optional[str] = None
+    open: Optional[str] = None
+    status: Optional[str] = None
+
+    @staticmethod
+    def from_dict(d):
+        return MarketHoliday(**d)
+
+@dataclass
+class MarketStatus:
+    "MarketStatus contains data for the current trading status of the exchanges and overall financial markets."
+    after_hours: Optional[bool] = None
+    currencies: Optional[Dict[str, str]] = None
+    early_hours: Optional[bool] = None
+    exchanges: Optional[Dict[str, str]] = None
+    market: Optional[str] = None
+    server_time: Optional[str] = None
+
+    @staticmethod
+    def from_dict(d):
+        return MarketStatus(**d)

polygon/rest/reference.py

@@ -0,0 +1,40 @@
+from .base import BaseClient
+from typing import Optional, Any, Dict, List, Union
+from .models import MarketHoliday, MarketStatus
+from urllib3 import HTTPResponse
+
+# https://polygon.io/docs/stocks
+class MarketsClient(BaseClient):
+    def list_market_holidays(
+        self,
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False
+    ) -> Union[List[MarketHoliday], HTTPResponse]:
+        """
+        Get upcoming market holidays and their open/close times.
+
+        :param params: Any additional query params
+        :param raw: Return HTTPResponse object instead of results object
+        :return: List of quotes
+        :rtype: List[Quote]
+        """
+        url = "/v1/marketstatus/upcoming"
+
+        return self._get(path=url, params=params, deserializer=MarketHoliday.from_dict, raw=raw)
+
+    def get_market_status(
+        self,
+        params: Optional[Dict[str, Any]] = None,
+        raw: bool = False
+    ) -> Union[MarketStatus, HTTPResponse]:
+        """
+        Get the current trading status of the exchanges and overall financial markets.
+
+        :param params: Any additional query params
+        :param raw: Return HTTPResponse object instead of results object
+        :return: List of quotes
+        :rtype: List[Quote]
+        """
+        url = "/v1/marketstatus/now"
+
+        return self._get(path=url, params=params, deserializer=MarketStatus.from_dict, raw=raw)