Merge branch 'main' into dev

Author: Cfomodz

.gitignore

@@ -2,4 +2,5 @@
 __pycache__/
 .env
 token_data.json
-custom
+custom/*
+tmp/*

README.md

@@ -43,4 +43,4 @@ This Python client simplifies interactions with the Schwab API by providing a us
 Credit to [Tyler Bowers](https://github.com/tylerebowers) for inspiring this work. This is a blatant rip-off of [his work](https://github.com/tylerebowers/Schwab-API-Python) formatted in a more pythonic way. We started as a fork but felt this would likely end up going in a whole different direction as far as our intended purpose and scope. Thank you, Tyler, for your work. We will continue to steal his work and publish it as our own and hope he does the same if it is ever advantageous for him to return the favor.
 
 ## License (MIT)
-This software is provided "as is", without warranty of any kind, express or implied, including but not limited to the warranties of merchantability, fitness for a particular purpose, and non-infringement. In no event shall the authors or copyright holders be responsible for any claim, damages, or other liabilities, whether in an action of contract, tort, or otherwise, arising from, out of, or in connection with the software or the use or other dealings in the software.
+Pythonic-Schwab-API is an unofficial API wrapper. It is in no way endorsed by or affiliated with Charles Schwab or any associated organization. Make sure to read and understand the terms of service of the underlying API before using this package. This software is provided "as is", without warranty of any kind, express or implied, including but not limited to the warranties of merchantability, fitness for a particular purpose, and non-infringement. In no event shall the authors or copyright holders be responsible for any claim, damages, or other liabilities, whether in an action of contract, tort, or otherwise, arising from, out of, or in connection with the software or the use or other dealings in the software.

api_client.py

@@ -11,9 +11,10 @@
 
 
 class APIClient:
-    def __init__(self):
+    def __init__(self, initials):
+        self.initials = initials
         self.account_numbers = None
-        self.config = APIConfig
+        self.config = APIConfig(self.initials)
         self.session = requests.Session()
         self.setup_logging()
         self.token_info = self.load_token()
@@ -23,7 +24,7 @@ def __init__(self):
             self.manual_authorization_flow()
 
     def setup_logging(self):
-        logging.basicConfig(**APIConfig.LOGGING_CONFIG)
+        logging.basicConfig(**self.config.LOGGING_CONFIG)
         self.logger = logging.getLogger(__name__)
 
     def ensure_valid_token(self):
@@ -42,7 +43,7 @@ def ensure_valid_token(self):
     def manual_authorization_flow(self):
         """ Handle the manual steps required to get the authorization code from the user. """
         self.logger.info("Starting manual authorization flow.")
-        auth_url = f"{APIConfig.API_BASE_URL}/v1/oauth/authorize?client_id={APIConfig.APP_KEY}&redirect_uri={APIConfig.CALLBACK_URL}&response_type=code"
+        auth_url = f"{self.config.API_BASE_URL}/v1/oauth/authorize?client_id={self.config.APP_KEY}&redirect_uri={self.config.CALLBACK_URL}&response_type=code"
         webbrowser.open(auth_url)
         self.logger.info(f"Please authorize the application by visiting: {auth_url}")
         response_url = ColorPrint.input(
@@ -67,9 +68,8 @@ def post_token_request(self, data):
             'Content-Type': 'application/x-www-form-urlencoded'
         }
         response = self.session.post(f"{self.config.API_BASE_URL}/v1/oauth/token", headers=headers, data=data)
-        if response.ok:
+        if response.status_code == 200:
             self.save_token(response.json())
-            self.load_token()
             self.logger.info("Tokens successfully updated.")
             return True
         else:
@@ -86,31 +86,37 @@ 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):
         """ Save token data securely. """
         token_data['expires_at'] = (datetime.now() + timedelta(seconds=token_data['expires_in'])).isoformat()
-        with open('token_data.json', 'w') as f:
+        with open(f'schwab_token_data_{self.initials}.json', 'w') as f:
             json.dump(token_data, f)
         self.logger.info("Token data saved successfully.")
 
     def load_token(self):
         """ Load token data. """
         try:
-            with open('token_data.json', 'r') as f:
+            with open(f'schwab_token_data_{self.initials}.json', 'r') as f:
                 token_data = json.load(f)
                 return token_data
         except Exception as e:
             self.logger.warning(f"Loading token failed: {e}")
         return None
 
-    def validate_token(self):
-        """ Validate the current token's validity. """
+    def validate_token(self, force=False):
+        """ Validate the current token. """
+        # 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 +152,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

@@ -2,25 +2,37 @@
 from dotenv import load_dotenv
 
 load_dotenv()
+SANDBOX = False
 
 
 class APIConfig:
-    API_BASE_URL = "https://api.schwabapi.com"
-    TRADER_BASE_URL = f"{API_BASE_URL}/trader/v1"
-    ACCOUNTS_BASE_URL = f"{TRADER_BASE_URL}/accounts"
-    MARKET_DATA_BASE_URL = f"{API_BASE_URL}/marketdata/v1"
-    ORDERS_BASE_URL = ACCOUNTS_BASE_URL
-    REQUEST_TIMEOUT = 30  # Timeout for API requests in seconds
-    RETRY_STRATEGY = {
-        'total': 3,  # Total number of retries to allow
-        'backoff_factor': 1  # Factor by which the delay between retries will increase
-    }
-    TOKEN_REFRESH_THRESHOLD_SECONDS = 300  # Time in seconds before token expiration to attempt refresh
-    DEBUG_MODE = False
-    LOGGING_CONFIG = {
-        'level': 'INFO',
-        'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-    }
-    APP_KEY = os.getenv('APP_KEY')
-    APP_SECRET = os.getenv('APP_SECRET')
-    CALLBACK_URL = os.getenv('CALLBACK_URL')
+    def __init__(self, initials):
+        self.initials = initials
+        if SANDBOX:
+            self.API_BASE_URL = "http://localhost:4020"
+            self.TRADER_BASE_URL = self.API_BASE_URL
+            self.ACCOUNTS_BASE_URL = f"{self.API_BASE_URL}/accounts"
+            self.MARKET_DATA_BASE_URL = f"{self.API_BASE_URL}/marketdata"
+            self.ORDERS_BASE_URL = self.ACCOUNTS_BASE_URL
+            self.STREAMER_INFO_URL = f"{self.API_BASE_URL}/streamer-info"
+        else:
+            self.API_BASE_URL = "https://api.schwabapi.com"
+            self.TRADER_BASE_URL = f"{self.API_BASE_URL}/trader/v1"
+            self.ACCOUNTS_BASE_URL = f"{self.TRADER_BASE_URL}/accounts"
+            self.MARKET_DATA_BASE_URL = f"{self.API_BASE_URL}/marketdata/v1"
+            self.ORDERS_BASE_URL = self.ACCOUNTS_BASE_URL
+            self.STREAMER_INFO_URL = f"{self.API_BASE_URL}/streamer-info"
+        self.REQUEST_TIMEOUT = 30  # Timeout for API requests in seconds
+        self. RETRY_STRATEGY = {
+            'total': 3,  # Total number of retries to allow
+            'backoff_factor': 1  # Factor by which the delay between retries will increase
+        }
+        self.TOKEN_REFRESH_THRESHOLD_SECONDS = 300  # Time in seconds before token expiration to attempt refresh
+        self.DEBUG_MODE = False
+        self.LOGGING_CONFIG = {
+            'level': 'INFO',
+            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+        }
+        self.APP_KEY = os.getenv(f'SCHWAB_APP_KEY_{self.initials}')
+        self.APP_SECRET = os.getenv(f'SCHWAB_APP_SECRET_{self.initials}')
+        self.CALLBACK_URL = os.getenv('CALLBACK_URL')

main.py

@@ -1,29 +1,68 @@
+import asyncio
 from datetime import datetime, timedelta
 from api_client import APIClient
 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
+import stream_utilities
+
+
+async def main_stream():
+    initials="AB"
+    client = APIClient(initials=initials)  # Initialize the API client
+    stream_client = StreamClient(client)
+    await stream_client.start()  # Start and connect
+
+    while stream_client.active:
+        # Construct and send a subscription request
+        request = stream_utilities.basic_request(
+            "LEVELONE_EQUITIES",
+            request_id=stream_client.request_id,
+            command="SUBS",
+            customer_id=stream_client.streamer_info.get("schwabClientCustomerId"),
+            correl_id=stream_client.streamer_info.get("schwabClientCorrelId"),
+            parameters={
+                "keys": "TSLA,AMZN,AAPL,NFLX,BABA",
+                "fields": "0,1,2,3,4,5,8,9,12,13,15,24,28,29,30,31,48"
+            }
+        )
+        await stream_client.send(request)
+        message = await stream_client.receive()
+        print(f"Received: {message}")
+        await asyncio.sleep(1)  # Delay between messages
+
+    stream_client.stop()
 
 
 def main():
-    client = APIClient()  # Initialize the API client
+    initials="AB"
+    client = APIClient(initials=initials)  # Initialize the API client
     accounts_api = Accounts(client)
     orders_api = Orders(client)
 
     # Get account numbers for linked accounts
-    # print(accounts_api.get_account_numbers())  # working
+    client.account_numbers = accounts_api.get_account_numbers()  # working
+    print(client.account_numbers)
 
     # Get positions for linked accounts
-    # print(accounts_api.get_all_accounts())  # working
+    print(accounts_api.get_all_accounts())  # working
 
-    sample_account = client.account_numbers[0]
-    account_hash = sample_account['accountHash']
+    sample_account = client.account_numbers[0]  # working
+    print(sample_account)
+    account_hash = sample_account['hashValue']  # working
+    print(account_hash)
 
     # Get specific account positions
-    # print(accounts_api.get_account(fields="positions"))
+    print(accounts_api.get_account(account_hash=account_hash, fields="positions"))
 
     # Get up to 3000 orders for an account for the past week
-    print(orders_api.get_orders(3000, datetime.now() - timedelta(days=7), datetime.now()).json())
+    print(orders_api.get_orders(account_hash=account_hash,
+                                max_results=3000,
+                                from_entered_time=datetime.now() - timedelta(days=7),
+                                to_entered_time=datetime.now())
+          )
 
     # Example to place an order (commented out for safety)
     """
@@ -46,11 +85,16 @@ def main():
     # print(orders_api.get_order('account_hash', order_id).json())
 
     # Get up to 3000 orders for all accounts for the past week
-    print(orders_api.get_orders(account_hash=account_hash, max_results=3000, from_entered_time=datetime.now() - timedelta(days=7), to_entered_time=datetime.now()))
+    for account in client.account_numbers:
+        account_hash = account['hashValue']
+        print(orders_api.get_orders(account_hash=account_hash, max_results=3000, from_entered_time=datetime.now() - timedelta(days=7), to_entered_time=datetime.now()))
 
     # Get all transactions for an account
-    print(accounts_api.get_account_transactions('account_hash', datetime.now() - timedelta(days=7), datetime.now(),
-                                                "TRADE").json())
+    print(accounts_api.get_account_transactions(account_hash=account_hash,
+                                                start_date=datetime.now() - timedelta(days=7),
+                                                end_date=datetime.now(),
+                                                types="TRADE")
+          )
 
     # Market-data-related requests
     quotes = Quotes(client)
@@ -69,6 +113,9 @@ def main():
     # Get an option expiration chain
     print(options.get_chains("AAPL").json())
 
+    # Get price history for a symbol
+    print(price_history.by_symbol("AAPL", period_type="day", period=1, frequency_type="minute", frequency=5).json())
+
     # Get movers for an index
     print(movers.get_movers("$DJI").json())
 
@@ -86,6 +133,8 @@ def main():
 
 
 if __name__ == '__main__':
-    print(
-        "Welcome to the unofficial Schwab API interface!\nGitHub: https://github.com/Patch-Code-Prosperity/Pythonic-Schwab-API")
+    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()

multi_terminal.py

@@ -15,7 +15,7 @@ def __init__(self, title="Terminal", height=20, width=200, font=("Courier New",
         self.textColor = textColor
         self.allowClosing = allowClosing
         self.ignoreClosedPrints = ignoreClosedPrints
-        self.isOpen = False
+        self.is_open = False
         self.start()
 
     def run(self):
@@ -25,16 +25,16 @@ def run(self):
         self.text_box = tk.Text(self.root, height=self.height, width=self.width, font=self.font,
                                 bg=self.backgroundColor, fg=self.textColor, state='disabled')
         self.text_box.pack(side="left", fill="both", expand=True)
-        self.isOpen = True
+        self.is_open = True
         self.root.mainloop()
 
     def close(self):
-        if self.isOpen:
-            self.isOpen = False
+        if self.is_open:
+            self.is_open = False
             self.root.destroy()
 
     def print(self, text, end="\n"):
-        if not self.isOpen:
+        if not self.is_open:
             if not self.ignoreClosedPrints:
                 raise Exception(f"Terminal '{self.title}' is closed.")
             return

orders.py

@@ -1,14 +1,21 @@
+from datetime import datetime, timezone, timedelta
+
+
 class Orders:
     def __init__(self, client):
         self.client = client
         self.base_url = client.config.ORDERS_BASE_URL
 
     def get_orders(self, account_hash, max_results=100, from_entered_time=None, to_entered_time=None, status=None):
         """Retrieve a list of orders for a specified account."""
+        if from_entered_time is None:
+            from_entered_time = (datetime.now(timezone.utc) - timedelta(days=364)).isoformat(timespec='seconds')
+        if to_entered_time is None:
+            to_entered_time = datetime.now(timezone.utc).isoformat(timespec='seconds')
         params = {
             'maxResults': max_results,
-            'fromEnteredTime': from_entered_time.isoformat() if from_entered_time else None,
-            'toEnteredTime': to_entered_time.isoformat() if to_entered_time else None,
+            'fromEnteredTime': from_entered_time,
+            'toEnteredTime': to_entered_time,
             'status': status
         }
         endpoint = f"{self.base_url}/{account_hash}/orders"
@@ -27,9 +34,9 @@ def get_order(self, account_hash, order_id):
     def cancel_order(self, account_hash, order_id):
         """Cancel a specific order."""
         endpoint = f"{self.base_url}/{account_hash}/orders/{order_id}"
-        return self.client.delete(endpoint)
+        return self.client.make_request(endpoint, method='DELETE')
 
     def replace_order(self, account_hash, order_id, new_order_details):
         """Replace an existing order with new details."""
         endpoint = f"{self.base_url}/{account_hash}/orders/{order_id}"
-        return self.client.put(endpoint, data=new_order_details)
+        return self.client.make_request(endpoint, method='PUT', data=new_order_details)