allow user to select data feed (iex/sip)

Author: camelpac

alpaca_backtrader_api/alpacadata.py

@@ -133,6 +133,7 @@ class AlpacaData(with_metaclass(MetaAlpacaData, DataBase)):
         ('reconnect', True),
         ('reconnections', -1),  # forever
         ('reconntimeout', 5.0),
+        ('data_feed', 'iex'),  # options iex/sip for pro
     )
 
     _store = alpacastore.AlpacaStore
@@ -233,7 +234,8 @@ def _st_start(self, instart=True, tmout=None):
             return True
         self.qlive = self.o.streaming_prices(self.p.dataname,
                                              self.p.timeframe,
-                                             tmout=tmout)
+                                             tmout=tmout,
+                                             data_feed=self.p.data_feed)
         if instart:
             self._statelivereconn = self.p.backfill_start
         else:

alpaca_backtrader_api/alpacastore.py

@@ -117,6 +117,7 @@ def __init__(
             method: StreamingMethod = StreamingMethod.AccountUpdate,
             base_url='',
             data_url='',
+            data_feed='iex',
             *args,
             **kwargs):
         try:
@@ -128,7 +129,7 @@ def __init__(
         self.conn = Stream(api_key,
                            api_secret,
                            base_url,
-                           data_feed='iex')
+                           data_feed=data_feed)
         self.instrument = instrument
         self.method = method
         self.q = q
@@ -610,18 +611,20 @@ def _resample(df):
         response = response[~response.index.duplicated()]
         return response
 
-    def streaming_prices(self, dataname, timeframe, tmout=None):
+    def streaming_prices(self,
+                         dataname, timeframe, tmout=None, data_feed='iex'):
         q = queue.Queue()
         kwargs = {'q': q,
                   'dataname': dataname,
                   'timeframe': timeframe,
+                  'data_feed': data_feed,
                   'tmout': tmout}
         t = threading.Thread(target=self._t_streaming_prices, kwargs=kwargs)
         t.daemon = True
         t.start()
         return q
 
-    def _t_streaming_prices(self, dataname, timeframe, q, tmout):
+    def _t_streaming_prices(self, dataname, timeframe, q, tmout, data_feed):
         if tmout is not None:
             _time.sleep(tmout)
 
@@ -639,7 +642,8 @@ def _t_streaming_prices(self, dataname, timeframe, q, tmout):
                             instrument=dataname,
                             method=method,
                             base_url=self.p.base_url,
-                            data_url=os.environ.get("DATA_PROXY_WS", ''))
+                            data_url=os.environ.get("DATA_PROXY_WS", ''),
+                            data_feed=data_feed)
 
         streamer.run()