Update the market data streams (#473)

* Update the market data streams

* address review comments

Makefile

@@ -0,0 +1,22 @@
+# The default "help" goal nicely prints all the available goals based on the funny looking ## comments.
+# Source: https://marmelab.com/blog/2016/02/29/auto-documented-makefile.html
+.DEFAULT_GOAL := help
+.PHONY: help
+help:  ## Display this help
+	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
+
+.PHONY: install
+install: ## Install the SDK and its dependencies using poetry
+	poetry install
+
+.PHONY: lint
+lint: ## Run the linters
+	poetry run black --check alpaca/ tests/
+
+.PHONY: generate
+generate: ## Generate the documentation
+	./tools/scripts/generate-docs.sh
+
+.PHONY: test
+test: ## Run the unit tests
+	poetry run pytest

alpaca/data/historical/__init__.py

@@ -1,7 +1,13 @@
-from alpaca.data.historical.stock import StockHistoricalDataClient
 from alpaca.data.historical.crypto import CryptoHistoricalDataClient
+from alpaca.data.historical.news import NewsClient
+from alpaca.data.historical.option import OptionHistoricalDataClient
+from alpaca.data.historical.screener import ScreenerClient
+from alpaca.data.historical.stock import StockHistoricalDataClient
 
 __all__ = [
-    "StockHistoricalDataClient",
     "CryptoHistoricalDataClient",
+    "StockHistoricalDataClient",
+    "NewsClient",
+    "OptionHistoricalDataClient",
+    "ScreenerClient",
 ]

alpaca/data/historical/crypto.py

@@ -1,29 +1,28 @@
 from collections import defaultdict
-from typing import Union, Optional, List, Dict
+from typing import Dict, List, Optional, Union
 
 from alpaca.common.constants import DATA_V2_MAX_LIMIT
-from alpaca.common.types import RawData
 from alpaca.common.enums import BaseURL
 from alpaca.common.rest import RESTClient
-from alpaca.common.types import Credentials
-from alpaca.data import Snapshot, Bar
+from alpaca.common.types import Credentials, RawData
+from alpaca.data import Bar, Snapshot
+from alpaca.data.enums import CryptoFeed
+from alpaca.data.historical.stock import DataExtensionType
 from alpaca.data.historical.utils import (
-    parse_obj_as_symbol_dict,
-    format_latest_data_response,
     format_dataset_response,
+    format_latest_data_response,
+    parse_obj_as_symbol_dict,
 )
-from alpaca.data.models import BarSet, QuoteSet, TradeSet, Orderbook, Trade, Quote
-from alpaca.data.historical.stock import DataExtensionType
+from alpaca.data.models import BarSet, Orderbook, Quote, Trade, TradeSet
 from alpaca.data.requests import (
     CryptoBarsRequest,
-    CryptoTradesRequest,
-    CryptoLatestTradeRequest,
+    CryptoLatestBarRequest,
+    CryptoLatestOrderbookRequest,
     CryptoLatestQuoteRequest,
+    CryptoLatestTradeRequest,
     CryptoSnapshotRequest,
-    CryptoLatestOrderbookRequest,
-    CryptoLatestBarRequest,
+    CryptoTradesRequest,
 )
-from alpaca.data.enums import CryptoFeed
 
 
 class CryptoHistoricalDataClient(RESTClient):

alpaca/data/historical/screener.py

@@ -1,14 +1,10 @@
 from typing import Optional, Union
 
-from alpaca.common.rest import RESTClient
-
 from alpaca.common.enums import BaseURL
-
-from alpaca.data.requests import MarketMoversRequest, MostActivesRequest
-
-from alpaca.data.models.screener import MostActives, Movers
-
+from alpaca.common.rest import RESTClient
 from alpaca.common.types import RawData
+from alpaca.data.models.screener import MostActives, Movers
+from alpaca.data.requests import MarketMoversRequest, MostActivesRequest
 
 
 class ScreenerClient(RESTClient):

alpaca/data/historical/stock.py

@@ -1,29 +1,28 @@
 from collections import defaultdict
 from enum import Enum
-from typing import List, Optional, Union, Type, Dict
+from typing import Dict, List, Optional, Type, Union
 
+from alpaca.common.constants import DATA_V2_MAX_LIMIT
 from alpaca.common.enums import BaseURL
-from alpaca.common.rest import RESTClient, HTTPResult
+from alpaca.common.rest import HTTPResult, RESTClient
 from alpaca.common.types import RawData
-from alpaca.data import Quote, Trade, Snapshot, Bar
+from alpaca.data import Bar, Quote, Snapshot, Trade
 from alpaca.data.historical.utils import (
-    parse_obj_as_symbol_dict,
-    format_latest_data_response,
     format_dataset_response,
+    format_latest_data_response,
     format_snapshot_data,
+    parse_obj_as_symbol_dict,
 )
-
 from alpaca.data.models import BarSet, QuoteSet, TradeSet
 from alpaca.data.requests import (
     StockBarsRequest,
-    StockQuotesRequest,
-    StockTradesRequest,
-    StockLatestTradeRequest,
+    StockLatestBarRequest,
     StockLatestQuoteRequest,
+    StockLatestTradeRequest,
+    StockQuotesRequest,
     StockSnapshotRequest,
-    StockLatestBarRequest,
+    StockTradesRequest,
 )
-from alpaca.common.constants import DATA_V2_MAX_LIMIT
 
 
 class DataExtensionType(Enum):

alpaca/data/historical/utils.py

@@ -1,5 +1,5 @@
 from collections import defaultdict
-from typing import Type, Dict
+from typing import Dict, Type
 
 from alpaca.common import HTTPResult, RawData
 

alpaca/data/live/__init__.py

@@ -1,7 +1,11 @@
-from alpaca.data.live.stock import StockDataStream
 from alpaca.data.live.crypto import CryptoDataStream
+from alpaca.data.live.news import NewsDataStream
+from alpaca.data.live.option import OptionDataStream
+from alpaca.data.live.stock import StockDataStream
 
 __all__ = [
-    "StockDataStream",
     "CryptoDataStream",
+    "NewsDataStream",
+    "OptionDataStream",
+    "StockDataStream",
 ]

alpaca/data/live/crypto.py

@@ -1,14 +1,17 @@
-from alpaca.common.websocket import BaseStream
-from typing import Optional, Dict
+from typing import Awaitable, Callable, Dict, Optional, Union
+
 from alpaca.common.enums import BaseURL
 from alpaca.data.enums import CryptoFeed
+from alpaca.data.live.websocket import DataStream
+from alpaca.data.models.bars import Bar
+from alpaca.data.models.orderbooks import Orderbook
+from alpaca.data.models.quotes import Quote
+from alpaca.data.models.trades import Trade
 
 
-class CryptoDataStream(BaseStream):
+class CryptoDataStream(DataStream):
     """
     A WebSocket client for streaming live crypto data.
-
-    See BaseStream for more information on implementation and the methods available.
     """
 
     def __init__(
@@ -27,6 +30,7 @@ def __init__(
             api_key (str): Alpaca API key.
             secret_key (str): Alpaca API secret key.
             raw_data (bool, optional): Whether to return wrapped data or raw API data. Defaults to False.
+            feed (CryptoFeed, optional): Which crypto feed to use. Defaults to US.
             websocket_params (Optional[Dict], optional): Any parameters for configuring websocket connection. Defaults to None.
             url_override (Optional[str]): If specified allows you to override the base url the client
               points to for proxy/testing. Defaults to None.
@@ -42,3 +46,123 @@ def __init__(
             raw_data=raw_data,
             websocket_params=websocket_params,
         )
+
+    def subscribe_trades(
+        self, handler: Callable[[Union[Trade, Dict]], Awaitable[None]], *symbols: str
+    ) -> None:
+        """Subscribe to trades.
+
+        Args:
+            handler (Callable[[Union[Trade, Dict]], Awaitable[None]): The coroutine callback
+                function to handle the incoming data.
+            *symbols: List of ticker symbols to subscribe to. "*" for everything.
+        """
+        self._subscribe(handler, symbols, self._handlers["trades"])
+
+    def subscribe_quotes(
+        self, handler: Callable[[Union[Quote, Dict]], Awaitable[None]], *symbols: str
+    ) -> None:
+        """Subscribe to quotes
+
+        Args:
+            handler (Callable[[Union[Quote, Dict]], Awaitable[None]]): The coroutine callback
+                function to handle the incoming data.
+            *symbols: List of ticker symbols to subscribe to. "*" for everything.
+        """
+        self._subscribe(handler, symbols, self._handlers["quotes"])
+
+    def subscribe_bars(
+        self, handler: Callable[[Union[Quote, Dict]], Awaitable[None]], *symbols: str
+    ) -> None:
+        """Subscribe to minute bars
+
+        Args:
+            handler (Callable[[Union[Quote, Dict]], Awaitable[None]]): The coroutine callback
+                function to handle the incoming data.
+            *symbols: List of ticker symbols to subscribe to. "*" for everything.
+        """
+        self._subscribe(handler, symbols, self._handlers["bars"])
+
+    def subscribe_updated_bars(
+        self, handler: Callable[[Union[Bar, Dict]], Awaitable[None]], *symbols: str
+    ) -> None:
+        """Subscribe to updated minute bars
+
+        Args:
+            handler (Callable[[Union[Bar, Dict]], Awaitable[None]]): The coroutine callback
+                function to handle the incoming data.
+            *symbols: List of ticker symbols to subscribe to. "*" for everything.
+        """
+        self._subscribe(handler, symbols, self._handlers["updatedBars"])
+
+    def subscribe_daily_bars(
+        self, handler: Callable[[Union[Bar, Dict]], Awaitable[None]], *symbols: str
+    ) -> None:
+        """Subscribe to daily bars
+
+        Args:
+            handler (Callable[[Union[Bar, Dict]], Awaitable[None]]): The coroutine callback
+                function to handle the incoming data.
+            *symbols: List of ticker symbols to subscribe to. "*" for everything.
+        """
+        self._subscribe(handler, symbols, self._handlers["dailyBars"])
+
+    def subscribe_orderbooks(
+        self, handler: Callable[[Union[Orderbook, Dict]], Awaitable[None]], *symbols
+    ) -> None:
+        """Subscribe to orderbooks
+
+        Args:
+            handler (Callable[[Union[Bar, Dict]], Awaitable[None]]): The coroutine callback
+                function to handle the incoming data.
+            *symbols: List of ticker symbols to subscribe to. "*" for everything.
+        """
+        self._subscribe(handler, symbols, self._handlers["orderbooks"])
+
+    def unsubscribe_trades(self, *symbols: str) -> None:
+        """Unsubscribe from trades
+
+        Args:
+            *symbols (str): List of ticker symbols to unsubscribe from. "*" for everything.
+        """
+        self._unsubscribe("trades", symbols)
+
+    def unsubscribe_quotes(self, *symbols: str) -> None:
+        """Unsubscribe from quotes
+
+        Args:
+            *symbols (str): List of ticker symbols to unsubscribe from. "*" for everything.
+        """
+        self._unsubscribe("quotes", symbols)
+
+    def unsubscribe_bars(self, *symbols: str) -> None:
+        """Unsubscribe from minute bars
+
+        Args:
+            *symbols (str): List of ticker symbols to unsubscribe from. "*" for everything.
+        """
+        self._unsubscribe("bars", symbols)
+
+    def unsubscribe_updated_bars(self, *symbols: str) -> None:
+        """Unsubscribe from updated bars
+
+        Args:
+            *symbols (str): List of ticker symbols to unsubscribe from. "*" for everything.
+        """
+        self._unsubscribe("updatedBars", symbols)
+
+    def unsubscribe_daily_bars(self, *symbols: str) -> None:
+        """Unsubscribe from daily bars
+
+        Args:
+            *symbols (str): List of ticker symbols to unsubscribe from. "*" for everything.
+        """
+        self._unsubscribe("dailyBars", symbols)
+
+    def unsubscribe_orderbooks(self, *symbols: str) -> None:
+        """Unsubscribe from orderbooks
+
+        Args:
+            *symbols (str): List of ticker symbols to unsubscribe from. "*" for everything.
+        """
+        self._unsubscribe("orderbooks", symbols)

alpaca/data/live/news.py

@@ -1,10 +1,11 @@
-from typing import Optional, Dict
+from typing import Awaitable, Callable, Dict, Optional, Union
 
 from alpaca.common.enums import BaseURL
-from alpaca.common.websocket import BaseStream
+from alpaca.data.live.websocket import DataStream
+from alpaca.data.models.news import News
 
 
-class NewsDataStream(BaseStream):
+class NewsDataStream(DataStream):
     """
     A WebSocket client for streaming news.
     """
@@ -39,3 +40,23 @@ def __init__(
             raw_data=raw_data,
             websocket_params=websocket_params,
         )
+
+    def subscribe_news(
+        self, handler: Callable[[Union[News, Dict]], Awaitable[None]], *symbols: str
+    ) -> None:
+        """Subscribe to news
+
+        Args:
+            handler (Callable[[Union[News, Dict]], Awaitable[None]]): The coroutine callback
+                function to handle the incoming data.
+            *symbols: List of ticker symbols to subscribe to. "*" for everything.
+        """
+        self._subscribe(handler, symbols, self._handlers["news"])
+
+    def unsubscribe_news(self, *symbols: str) -> None:
+        """Unsubscribe from news
+
+        Args:
+            *symbols (str): List of ticker symbols to unsubscribe from. "*" for everything.
+        """
+        self._unsubscribe("news", symbols)