Merge pull request #384 from alpacahq/feature/data-v2

Data v2

Author: camelpac

README.md

@@ -1,4 +1,5 @@
 from .rest import REST  # noqa
+from .stream import Stream  # noqa
 from .stream2 import StreamConn  # noqa
 
-__version__ = '0.53.0'
+__version__ = '1.0.0'

alpaca_trade_api/__init__.py

@@ -65,6 +65,13 @@ def get_data_url() -> URL:
         'APCA_API_DATA_URL', 'https://data.alpaca.markets').rstrip('/'))
 
 
+def get_data_stream_url() -> URL:
+    return URL(os.environ.get(
+        'APCA_API_STREAM_URL',
+        'https://stream.data.alpaca.markets').rstrip('/')
+    )
+
+
 def get_credentials(key_id: str = None,
                     secret_key: str = None,
                     oauth: str = None) -> Credentials:

alpaca_trade_api/common.py

@@ -0,0 +1,81 @@
+from enum import Enum
+import pandas as pd
+from .entity import Bar, Trade, Quote
+
+
+trade_mapping_v2 = {
+    "i": "id",
+    "S": "symbol",
+    "c": "conditions",
+    "x": "exchange",
+    "p": "price",
+    "s": "size",
+    "t": "timestamp",
+    "z": "tape"
+}
+
+quote_mapping_v2 = {
+    "S": "symbol",
+    "ax": "ask_exchange",
+    "ap": "ask_price",
+    "as": "ask_size",
+    "bx": "bid_exchange",
+    "bp": "bid_price",
+    "bs": "bid_size",
+    "c": "conditions",
+    "t": "timestamp",
+    "z": "tape"
+}
+
+bar_mapping_v2 = {
+    "S": "symbol",
+    "o": "open",
+    "h": "high",
+    "l": "low",
+    "c": "close",
+    "v": "volume",
+    "t": "timestamp"
+}
+
+
+class EntityListType(Enum):
+    Trade = Trade, trade_mapping_v2
+    Quote = Quote, quote_mapping_v2
+    Bar = Bar, bar_mapping_v2
+
+
+class EntityList(list):
+    def __init__(self, entity_type: EntityListType, raw):
+        entity = entity_type.value[0]
+        super().__init__([entity(o) for o in raw])
+        self._raw = raw
+        self.mapping = entity_type.value[1]
+
+    @property
+    def df(self):
+        if not hasattr(self, '_df'):
+            df = pd.DataFrame(
+                self._raw,
+            )
+
+            df.columns = [self.mapping[c] for c in df.columns]
+            if not df.empty:
+                df.set_index('timestamp', inplace=True)
+                df.index = pd.DatetimeIndex(df.index)
+            self._df = df
+        return self._df
+
+
+class BarsV2(EntityList):
+    def __init__(self, raw):
+        super().__init__(EntityListType.Bar, raw)
+
+
+class TradesV2(EntityList):
+    def __init__(self, raw):
+        super().__init__(EntityListType.Trade, raw)
+
+
+class QuotesV2(EntityList):
+    def __init__(self, raw):
+        super().__init__(EntityListType.Quote, raw)

alpaca_trade_api/entity_v2.py

@@ -1,20 +1,22 @@
 import logging
 import os
-from typing import List
+from typing import Iterator, List, Union
 import requests
 from requests.exceptions import HTTPError
 import time
+from enum import Enum
 from .common import (
     get_base_url,
     get_data_url,
     get_credentials,
     get_api_version, URL, FLOAT,
 )
 from .entity import (
-    Entity, Account, AccountConfigurations, AccountActivity,
+    Bar, Entity, Account, AccountConfigurations, AccountActivity,
     Asset, Order, Position, BarSet, Clock, Calendar,
     Aggs, Trade, Quote, Watchlist, PortfolioHistory
 )
+from .entity_v2 import BarsV2, TradesV2, QuotesV2
 from . import polygon
 
 logger = logging.getLogger(__name__)
@@ -24,6 +26,11 @@
 AccountActivities = List[AccountActivity]
 Calendars = List[Calendar]
 Watchlists = List[Watchlist]
+TradeIterator = Iterator[Union[Trade, dict]]
+QuoteIterator = Iterator[Union[Quote, dict]]
+BarIterator = Iterator[Union[Bar, dict]]
+
+DATA_V2_MAX_LIMIT = 10000  # max items per api call
 
 
 class RetryException(Exception):
@@ -62,6 +69,13 @@ def response(self):
             return self._http_error.response
 
 
+class TimeFrame(Enum):
+    Day = "1Day"
+    Hour = "1Hour"
+    Minute = "1Min"
+    Sec = "1Sec"
+
+
 class REST(object):
     def __init__(self,
                  key_id: str = None,
@@ -173,10 +187,10 @@ def patch(self, path, data=None):
     def delete(self, path, data=None):
         return self._request('DELETE', path, data)
 
-    def data_get(self, path, data=None):
+    def data_get(self, path, data=None, api_version='v1'):
         base_url: URL = get_data_url()
         return self._request(
-            'GET', path, data, base_url=base_url, api_version='v1'
+            'GET', path, data, base_url=base_url, api_version=api_version,
         )
 
     def get_account(self) -> Account:
@@ -505,6 +519,118 @@ def get_last_quote(self, symbol: str) -> Quote:
         resp = self.data_get('/last_quote/stocks/{}'.format(symbol))
         return self.response_wrapper(resp['last'], Quote)
 
+    def _data_get_v2(self, endpoint: str, symbol: str, **kwargs):
+        page_token = None
+        total_items = 0
+        limit = kwargs.get('limit')
+        while True:
+            actual_limit = None
+            if limit:
+                actual_limit = min(int(limit) - total_items, DATA_V2_MAX_LIMIT)
+                if actual_limit < 1:
+                    break
+            data = kwargs
+            data['limit'] = actual_limit
+            data['page_token'] = page_token
+            resp = self.data_get('/stocks/{}/{}'.format(symbol, endpoint),
+                                 data=data, api_version='v2')
+            items = resp.get(endpoint, [])
+            for item in items:
+                yield item
+            total_items += len(items)
+            page_token = resp.get('next_page_token')
+            if not page_token:
+                break
+
+    def get_trades_iter(self,
+                        symbol: str,
+                        start: str,
+                        end: str,
+                        limit: int = None,
+                        raw=False) -> TradeIterator:
+        trades = self._data_get_v2('trades', symbol,
+                                   start=start, end=end, limit=limit)
+        for trade in trades:
+            if raw:
+                yield trade
+            else:
+                yield self.response_wrapper(trade, Trade)
+
+    def get_trades(self,
+                   symbol: str,
+                   start: str,
+                   end: str,
+                   limit: int = None,
+                   ) -> TradesV2:
+        trades = list(self.get_trades_iter(symbol,
+                                           start,
+                                           end,
+                                           limit,
+                                           raw=True))
+        return TradesV2(trades)
+
+    def get_quotes_iter(self,
+                        symbol: str,
+                        start: str,
+                        end: str,
+                        limit: int = None,
+                        raw=False) -> QuoteIterator:
+        quotes = self._data_get_v2('quotes', symbol,
+                                   start=start, end=end, limit=limit)
+        for quote in quotes:
+            if raw:
+                yield quote
+            else:
+                yield self.response_wrapper(quote, Quote)
+
+    def get_quotes(self,
+                   symbol: str,
+                   start: str,
+                   end: str,
+                   limit: int = None,
+                   ) -> QuotesV2:
+        quotes = list(self.get_quotes_iter(symbol,
+                                           start,
+                                           end,
+                                           limit,
+                                           raw=True))
+        return QuotesV2(quotes)
+
+    def get_bars_iter(self,
+                      symbol: str,
+                      timeframe: TimeFrame,
+                      start: str,
+                      end: str,
+                      adjustment: str = 'all',
+                      limit: int = None,
+                      raw=False) -> BarIterator:
+        bars = self._data_get_v2('bars', symbol,
+                                 timeframe=timeframe.value,
+                                 adjustment=adjustment,
+                                 start=start, end=end, limit=limit)
+        for bar in bars:
+            if raw:
+                yield bar
+            else:
+                yield self.response_wrapper(bar, Bar)
+
+    def get_bars(self,
+                 symbol: str,
+                 timeframe: TimeFrame,
+                 start: str,
+                 end: str,
+                 adjustment: str = 'all',
+                 limit: int = None,
+                 ) -> BarsV2:
+        bars = list(self.get_bars_iter(symbol,
+                                       timeframe,
+                                       start,
+                                       end,
+                                       adjustment,
+                                       limit,
+                                       raw=True))
+        return BarsV2(bars)
+
     def get_clock(self) -> Clock:
         resp = self.get('/clock')
         return self.response_wrapper(resp, Clock)