streaming

Author: Cfomodz

api_client.py

@@ -69,7 +69,6 @@ def post_token_request(self, data):
         response = self.session.post(f"{self.config.API_BASE_URL}/v1/oauth/token", headers=headers, data=data)
         if response.ok:
             self.save_token(response.json())
-            self.load_token()
             self.logger.info("Tokens successfully updated.")
             return True
         else:
@@ -86,7 +85,7 @@ def refresh_access_token(self):
         if not self.post_token_request(data):
             self.logger.error("Failed to refresh access token.")
             return False
-
+        self.token_info = self.load_token()
         return self.validate_token()
 
     def save_token(self, token_data):
@@ -106,11 +105,17 @@ def load_token(self):
             self.logger.warning(f"Loading token failed: {e}")
         return None
 
-    def validate_token(self):
+    def validate_token(self, force=False):
         """ Validate the current token's validity. """
+        print(self.token_info['expires_at'])
+        print(datetime.now())
+        print(datetime.fromisoformat(self.token_info['expires_at']))
+        print(datetime.now() < datetime.fromisoformat(self.token_info['expires_at']))
         if self.token_info and datetime.now() < datetime.fromisoformat(self.token_info['expires_at']):
+            print(f"Token expires in {datetime.fromisoformat(self.token_info['expires_at']) - datetime.now()} seconds")
             return True
-        else:
+        elif force:
+            print("Token expired or invalid.")
             # get AAPL to validate token
             params = {'symbol': 'AAPL'}
             response = self.make_request(endpoint=f"{self.config.MARKET_DATA_BASE_URL}/chains", params=params, validating=True)
@@ -146,3 +151,11 @@ def make_request(self, endpoint, method="GET", **kwargs):
             response = self.session.request(method, url, headers=headers, **kwargs)
         response.raise_for_status()
         return response.json()
+
+    def get_user_preferences(self):
+        """Retrieve user preferences."""
+        try:
+            return self.make_request(f'{self.config.TRADER_BASE_URL}/userPreference')
+        except Exception as e:
+            self.logger.error(f"Failed to get user preferences: {e}")
+            return None

color_print.py

@@ -15,6 +15,11 @@ def print(message_type, message, end="\n"):
     def input(message):
         return input(f"{ColorPrint.COLORS['input']}{message}")
 
+    @staticmethod
+    def user_input(message):
+        return input(f"{ColorPrint.COLORS['user']}{message}")
+
+
 
 if __name__ == '__main__':
     ColorPrint.print('info', 'This is an informational message')

config.py

@@ -10,6 +10,7 @@ class APIConfig:
     ACCOUNTS_BASE_URL = f"{TRADER_BASE_URL}/accounts"
     MARKET_DATA_BASE_URL = f"{API_BASE_URL}/marketdata/v1"
     ORDERS_BASE_URL = ACCOUNTS_BASE_URL
+    STREAMER_INFO_URL = f"{API_BASE_URL}/streamer-info"
     REQUEST_TIMEOUT = 30  # Timeout for API requests in seconds
     RETRY_STRATEGY = {
         'total': 3,  # Total number of retries to allow

main.py

@@ -3,6 +3,14 @@
 from accounts import Accounts
 from market_data import Quotes, Options, PriceHistory, Movers, MarketHours, Instruments
 from orders import Orders
+from stream_client import StreamClient
+from asyncio import get_event_loop
+
+
+async def main_stream():
+    client = APIClient()  # Initialize the API client
+    stream_client = StreamClient(client)
+    await stream_client.start()
 
 
 def main():
@@ -86,6 +94,8 @@ def main():
 
 
 if __name__ == '__main__':
-    print(
-        "Welcome to the unofficial Schwab API interface!\nGitHub: https://github.com/Patch-Code-Prosperity/Pythonic-Schwab-API")
-    main()
+    print("Welcome to the unofficial Schwab API interface!\n"
+          "GitHub: https://github.com/Patch-Code-Prosperity/Pythonic-Schwab-API")
+    loop = get_event_loop()
+    loop.run_until_complete(main_stream())
+    # main()

stream_client.py

@@ -0,0 +1,88 @@
+import json
+import asyncio
+import threading
+import websockets
+from datetime import datetime, time
+from multi_terminal import MultiTerminal
+from api_client import APIClient
+from stream_utilities import basic_request  # Importing utility functions
+from color_print import ColorPrint
+
+
+class StreamClient:
+    def __init__(self, client: APIClient):
+        self.websocket = None
+        self.streamer_info = None
+        self.start_timestamp = None
+        self.terminal = MultiTerminal(title="Stream Output")
+        self.color_print = ColorPrint()
+        self.active = False
+        self.request_id = 0
+        self.client = client
+
+    async def start(self):
+        response = self.client.get_user_preferences()
+        if not response:
+            self.color_print.print("error", f"Failed to get streamer info: {response.status_code}")
+            exit(1)
+        self.streamer_info = response['streamerInfo'][0]
+        login = self._construct_login_message()
+
+        while True:
+            try:
+                await self._connect_and_stream(login)
+            except websockets.exceptions.ConnectionClosedOK:
+                self.color_print.print("info", "Stream has closed.")
+                break
+            except Exception as e:
+                self.color_print.print("error", f"{e}")
+                self._handle_stream_error(e)
+
+    def _construct_login_message(self):
+        self.request_id += 1
+        return basic_request("ADMIN", "LOGIN", self.request_id, {
+            "Authorization": self.client.token_info.get("access_token"),
+            "SchwabClientChannel": self.streamer_info.get("schwabClientChannel"),
+            "SchwabClientFunctionId": self.streamer_info.get("schwabClientFunctionId")
+        })
+
+    async def _connect_and_stream(self, login):
+        self.start_timestamp = datetime.now()
+        self.color_print.print("info", "Connecting to server...")
+        self.color_print.print("info", f"Start timestamp: {self.start_timestamp}")
+        self.color_print.print("info", f"Streamer socket URL: {self.streamer_info.get('streamerSocketUrl')}")
+        async with websockets.connect(self.streamer_info.get('streamerSocketUrl'),
+                                      ping_interval=None) as self.websocket:
+            self.terminal.print("[INFO]: Connecting to server...")
+            await self.websocket.send(json.dumps(login))
+            self.terminal.print(f"[Login]: {await self.websocket.recv()}")
+            self.active = True
+            while True:
+                received = await self.websocket.recv()
+                self.terminal.print(received)
+
+    def _handle_stream_error(self, error):
+        self.active = False
+        if isinstance(error, RuntimeError) and str(error) == "Streaming window has been closed":
+            self.color_print.print("warning", "Streaming window has been closed.")
+        else:
+            if (datetime.now() - self.start_timestamp).seconds < 70:
+                self.color_print.print("error", "Stream not alive for more than 1 minute, exiting...")
+            else:
+                self.terminal.print("[WARNING]: Connection lost to server, reconnecting...")
+
+    def send(self, listOfRequests):
+        async def _send(to_send):
+            await self.websocket.send(to_send)
+
+        if not isinstance(listOfRequests, list):
+            listOfRequests = [listOfRequests]
+        if self.active:
+            to_send = json.dumps({"requests": listOfRequests})
+            asyncio.run(_send(to_send))
+        else:
+            self.color_print.print("warning", "Stream is not active, nothing sent.")
+
+    def stop(self):
+        self.send(basic_request("ADMIN", "LOGOUT", self.request_id))
+        self.active = False

stream_utilities.py

@@ -0,0 +1,7 @@
+def basic_request(service, command, request_id, parameters=None):
+    return {
+        "service": service.upper(),
+        "command": command.upper(),
+        "requestid": request_id,
+        "parameters": parameters if parameters else {}
+    }