Python 3.10 (#561)

* upgrade websockets version to 10.1

* use updated asyncio api methods

* flake8 format

* python 310 circle ci

Author: ccnlui

.circleci/config.yml

@@ -96,7 +96,7 @@ jobs:
           name: run tests
           command: |
             pip install pandas==1.3.4 numpy==1.21.4
-            python setup.py install
+            pip install .
             pip install flake8 && flake8 alpaca_trade_api tests
             python setup.py test
 
@@ -118,7 +118,7 @@ jobs:
       - run:
           name: make sure install works
           command: |
-            python setup.py install
+            pip install .
       - run:
           name: view installed packages
           command: |
@@ -133,11 +133,30 @@ jobs:
       - run:
           name: make sure install works
           command: |
-            python setup.py install
+            pip install .
+      - run:
+          name: view installed packages
+          command: |
+            pip freeze
+
+  build-test-python310:
+    docker:
+      - image: python:3.10
+    working_directory: ~/repo
+    steps:
+      - checkout
+      - run:
+          name: make sure install works
+          command: |
+            pip install .
       - run:
           name: view installed packages
           command: |
             pip freeze
+      - run:
+          name: unit test
+          command: |
+            python setup.py test
 
 workflows:
   version: 2
@@ -148,3 +167,4 @@ workflows:
       - build-test-python37
       - build-python38
       - build-python39
+      - build-test-python310

alpaca_trade_api/stream.py

@@ -51,6 +51,7 @@ def __init__(self,
         self._secret_key = secret_key
         self._ws = None
         self._running = False
+        self._loop = None
         self._raw_data = raw_data
         self._stop_stream_queue = queue.Queue()
         self._handlers = {
@@ -179,7 +180,9 @@ def _subscribe(self, handler, symbols, handlers):
         for symbol in symbols:
             handlers[symbol] = handler
         if self._running:
-            asyncio.get_event_loop().run_until_complete(self._subscribe_all())
+            asyncio.run_coroutine_threadsafe(
+                self._subscribe_all(), self._loop
+            ).result()
 
     async def _subscribe_all(self):
         if any(
@@ -210,6 +213,7 @@ async def _unsubscribe(self,
                 }))
 
     async def _run_forever(self):
+        self._loop = asyncio.get_running_loop()
         # do not start the websocket connection until we subscribe to something
         while not any(
             v for k, v in self._handlers.items()
@@ -262,32 +266,42 @@ def subscribe_daily_bars(self, handler, *symbols):
 
     def unsubscribe_trades(self, *symbols):
         if self._running:
-            asyncio.get_event_loop().run_until_complete(
-                self._unsubscribe(trades=symbols))
+            asyncio.run_coroutine_threadsafe(
+                self._unsubscribe(trades=symbols),
+                self._loop).result()
         for symbol in symbols:
             del self._handlers['trades'][symbol]
 
     def unsubscribe_quotes(self, *symbols):
         if self._running:
-            asyncio.get_event_loop().run_until_complete(
-                self._unsubscribe(quotes=symbols))
+            asyncio.run_coroutine_threadsafe(
+                self._unsubscribe(quotes=symbols),
+                self._loop).result()
         for symbol in symbols:
             del self._handlers['quotes'][symbol]
 
     def unsubscribe_bars(self, *symbols):
         if self._running:
-            asyncio.get_event_loop().run_until_complete(
-                self._unsubscribe(bars=symbols))
+            asyncio.run_coroutine_threadsafe(
+                self._unsubscribe(bars=symbols),
+                self._loop).result()
         for symbol in symbols:
             del self._handlers['bars'][symbol]
 
     def unsubscribe_daily_bars(self, *symbols):
         if self._running:
-            asyncio.get_event_loop().run_until_complete(
-                self._unsubscribe(daily_bars=symbols))
+            asyncio.run_coroutine_threadsafe(
+                self._unsubscribe(daily_bars=symbols),
+                self._loop).result()
         for symbol in symbols:
             del self._handlers['dailyBars'][symbol]
 
+    def stop(self):
+        if self._loop.is_running():
+            asyncio.run_coroutine_threadsafe(
+                self.stop_ws(),
+                self._loop).result()
+
 
 class DataStream(_DataStream):
     def __init__(self,
@@ -386,15 +400,17 @@ def subscribe_lulds(self, handler, *symbols):
 
     def unsubscribe_statuses(self, *symbols):
         if self._running:
-            asyncio.get_event_loop().run_until_complete(
-                self._unsubscribe(statuses=symbols))
+            asyncio.run_coroutine_threadsafe(
+                self._unsubscribe(statuses=symbols),
+                self._loop).result()
         for symbol in symbols:
             del self._handlers['statuses'][symbol]
 
     def unsubscribe_lulds(self, *symbols):
         if self._running:
-            asyncio.get_event_loop().run_until_complete(
-                self._unsubscribe(lulds=symbols))
+            asyncio.run_coroutine_threadsafe(
+                self._unsubscribe(lulds=symbols),
+                self._loop).result()
         for symbol in symbols:
             del self._handlers['lulds'][symbol]
 
@@ -484,8 +500,9 @@ def subscribe_news(self, handler, *symbols):
 
     def unsubscribe_news(self, *symbols):
         if self._running:
-            asyncio.get_event_loop().run_until_complete(
-                self._unsubscribe(news=symbols))
+            asyncio.run_coroutine_threadsafe(
+                self._unsubscribe(news=symbols),
+                self._loop).result()
         for symbol in symbols:
             del self._handlers['news'][symbol]
 
@@ -503,6 +520,7 @@ def __init__(self,
         self._trade_updates_handler = None
         self._ws = None
         self._running = False
+        self._loop = None
         self._raw_data = raw_data
         self._stop_stream_queue = queue.Queue()
         self._should_run = True
@@ -550,8 +568,9 @@ def subscribe_trade_updates(self, handler):
         _ensure_coroutine(handler)
         self._trade_updates_handler = handler
         if self._running:
-            asyncio.get_event_loop().run_until_complete(
-                self._subscribe_trade_updates())
+            asyncio.run_coroutine_threadsafe(
+                self._subscribe_trade_updates(),
+                self._loop).result()
 
     async def _start_ws(self):
         await self._connect()
@@ -577,6 +596,7 @@ async def _consume(self):
                     pass
 
     async def _run_forever(self):
+        self._loop = asyncio.get_running_loop()
         # do not start the websocket connection until we subscribe to something
         while not self._trade_updates_handler:
             if not self._stop_stream_queue.empty():
@@ -590,7 +610,7 @@ async def _run_forever(self):
             try:
                 if not self._should_run:
                     log.info("Trading stream stopped")
-                    break
+                    return
                 if not self._running:
                     log.info("starting trading websocket connection")
                     await self._start_ws()
@@ -618,6 +638,12 @@ async def stop_ws(self):
         if self._stop_stream_queue.empty():
             self._stop_stream_queue.put_nowait({"should_stop": True})
 
+    def stop(self):
+        if self._loop.is_running():
+            asyncio.run_coroutine_threadsafe(
+                self.stop_ws(),
+                self._loop).result()
+
 
 class Stream:
     def __init__(self,
@@ -630,25 +656,25 @@ def __init__(self,
                  crypto_exchanges: Optional[List[str]] = None):
         self._key_id, self._secret_key, _ = get_credentials(key_id, secret_key)
         self._base_url = base_url or get_base_url()
-        self._data_steam_url = data_stream_url or get_data_stream_url()
+        self._data_stream_url = data_stream_url or get_data_stream_url()
 
         self._trading_ws = TradingStream(self._key_id,
                                          self._secret_key,
                                          self._base_url,
                                          raw_data)
         self._data_ws = DataStream(self._key_id,
                                    self._secret_key,
-                                   self._data_steam_url,
+                                   self._data_stream_url,
                                    raw_data,
                                    data_feed.lower())
         self._crypto_ws = CryptoDataStream(self._key_id,
                                            self._secret_key,
-                                           self._data_steam_url,
+                                           self._data_stream_url,
                                            raw_data,
                                            crypto_exchanges)
         self._news_ws = NewsDataStream(self._key_id,
                                        self._secret_key,
-                                       self._data_steam_url,
+                                       self._data_stream_url,
                                        raw_data)
 
     def subscribe_trade_updates(self, handler):
@@ -836,9 +862,8 @@ async def _run_forever(self):
                              self._news_ws._run_forever())
 
     def run(self):
-        loop = asyncio.get_event_loop()
         try:
-            loop.run_until_complete(self._run_forever())
+            asyncio.run(self._run_forever())
         except KeyboardInterrupt:
             print('keyboard interrupt, bye')
             pass
@@ -859,6 +884,16 @@ async def stop_ws(self):
         if self._news_ws:
             await self._news_ws.stop_ws()
 
+    def stop(self):
+        if self._trading_ws:
+            self._trading_ws.stop()
+        if self._data_ws:
+            self._data_ws.stop()
+        if self._crypto_ws:
+            self._crypto_ws.stop()
+        if self._news_ws:
+            self._news_ws.stop()
+
     def is_open(self):
         """
         Checks if either of the websockets is open

examples/historic_async.py

@@ -106,8 +106,7 @@ async def main(symbols):
                         base_url=URL(base_url))
 
     start_time = time.time()
-    loop = asyncio.get_event_loop()
     symbols = [el.symbol for el in api.list_assets(status='active')]
     symbols = symbols[:200]
-    loop.run_until_complete(main(symbols))
+    asyncio.run(main(symbols))
     print(f"took {time.time() - start_time} sec")

examples/websockets/dynamic_subscription_example.py

@@ -28,13 +28,6 @@ async def print_bar(bar):
 
 
 def consumer_thread():
-    try:
-        # make sure we have an event loop, if not create a new one
-        loop = asyncio.get_event_loop()
-        loop.set_debug(True)
-    except RuntimeError:
-        asyncio.set_event_loop(asyncio.new_event_loop())
-
     global conn
     conn = Stream(ALPACA_API_KEY,
                   ALPACA_SECRET_KEY,
@@ -50,9 +43,6 @@ def consumer_thread():
     logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
                         level=logging.INFO)
     threading.Thread(target=consumer_thread).start()
-
-    loop = asyncio.get_event_loop()
-
     time.sleep(5)  # give the initial connection time to be established
     subscriptions = {"BABA": print_quote,
                      "AAPL": print_quote,

examples/websockets/streamconn_on_and_off.py

@@ -27,13 +27,6 @@ async def print_bar(bar):
 
 
 def consumer_thread():
-    try:
-        # make sure we have an event loop, if not create a new one
-        loop = asyncio.get_event_loop()
-        loop.set_debug(True)
-    except RuntimeError:
-        asyncio.set_event_loop(asyncio.new_event_loop())
-
     global conn
     conn = Stream(ALPACA_API_KEY,
                   ALPACA_SECRET_KEY,
@@ -48,18 +41,17 @@ def consumer_thread():
     logging.basicConfig(format='%(asctime)s  %(levelname)s %(message)s',
                         level=logging.INFO)
 
-    loop = asyncio.get_event_loop()
     pool = ThreadPoolExecutor(1)
 
     while 1:
         try:
             pool.submit(consumer_thread)
             time.sleep(20)
-            loop.run_until_complete(conn.stop_ws())
+            conn.stop()
             time.sleep(20)
         except KeyboardInterrupt:
             print("Interrupted execution by user")
-            loop.run_until_complete(conn.stop_ws())
+            conn.stop()
             exit(0)
         except Exception as e:
             print("You goe an exception: {} during execution. continue "

requirements/requirements.txt

@@ -3,8 +3,8 @@ numpy>=1.11.1  # pyup: ignore - allow all versions above this
 requests>2,<3
 urllib3>1.24,<2
 websocket-client>=0.56.0,<2
-websockets>=8.0,<10
+websockets>=8.0,<11
 msgpack==1.0.2
 aiohttp==3.7.4
-PyYAML==5.4.1
+PyYAML==6.0
 deprecation==2.1.0