Vote Up Down
The (Unofficial) Guide to Charles Schwab’s Trader APIs
Issue
Body
I would like to thank Carsten Savage and Tyler Bowers for his fantastic YouTube video and the GitHub repo which are another tremendous resource for getting started with Schwab’s Trader APIs.
import os import base64 import requests import webbrowser from loguru import logger def construct_init_auth_url() -> tuple[str, str, str]: app_key = "your-app-key" app_secret = "your-app-secret" auth_url = f"https://api.schwabapi.com/v1/oauth/authorize?client_id={app_key}&redirect_uri=https://127.0.0.1" logger.info("Click to authenticate:") logger.info(auth_url) return app_key, app_secret, auth_url def construct_headers_and_payload(returned_url, app_key, app_secret): response_code = f"{returned_url[returned_url.index('code=') + 5: returned_url.index('%40')]}@" credentials = f"{app_key}:{app_secret}" base64_credentials = base64.b64encode(credentials.encode("utf-8")).decode( "utf-8" ) headers = { "Authorization": f"Basic {base64_credentials}", "Content-Type": "application/x-www-form-urlencoded", } payload = { "grant_type": "authorization_code", "code": response_code, "redirect_uri": "https://127.0.0.1", } return headers, payload def retrieve_tokens(headers, payload) -> dict: init_token_response = requests.post( url="https://api.schwabapi.com/v1/oauth/token", headers=headers, data=payload, ) init_tokens_dict = init_token_response.json() return init_tokens_dict def main(): app_key, app_secret, cs_auth_url = construct_init_auth_url() webbrowser.open(cs_auth_url) logger.info("Paste Returned URL:") returned_url = input() init_token_headers, init_token_payload = construct_headers_and_payload( returned_url, app_key, app_secret ) init_tokens_dict = retrieve_tokens( headers=init_token_headers, payload=init_token_payload ) logger.debug(init_tokens_dict) return "Done!" if __name__ == "__main__": main()
When you run the code above, the code will open a Charles Schwab login screen in your browser. From there, you’ll be routed to an empty page, which contains a URL in the search bar with an access code embedded in it. You’ll need to copy the entire URL. The code will prompt you to paste it in terminal, and once you do so, the code will make a request for the first batch of authentication tokens, and logger will print the tokens out for you in terminal. Then:
import base64 import requests from loguru import logger def refresh_tokens(): logger.info("Initializing...") app_key = "your-app-key" app_secret = "your-app-secret" # You can pull this from a local file, # Google Cloud Firestore/Secret Manager, etc. refresh_token_value = "your-current-refresh-token" payload = { "grant_type": "refresh_token", "refresh_token": refresh_token_value, } headers = { "Authorization": f'Basic {base64.b64encode(f"{app_key}:{app_secret}".encode()).decode()}', "Content-Type": "application/x-www-form-urlencoded", } refresh_token_response = requests.post( url="https://api.schwabapi.com/v1/oauth/token", headers=headers, data=payload, ) if refresh_token_response.status_code == 200: logger.info("Retrieved new tokens successfully using refresh token.") else: logger.error( f"Error refreshing access token: {refresh_token_response.text}" ) return None refresh_token_dict = refresh_token_response.json() logger.debug(refresh_token_dict) logger.info("Token dict refreshed.") return "Done!" if __name__ == "__main__": refresh_tokens()
Finally:
class AccountsTrading: def __init__(self): # Initialize access token during class instantiation self.access_token = None self.account_hash_value = None self.refresh_access_token() self.base_url = "https://api.schwabapi.com/trader/v1" self.headers = {"Authorization": f"Bearer {self.access_token}"} self.get_account_number_hash_value() def refresh_access_token(self): # Custom function to retrieve access token from Firestore self.access_token = retrieve_firestore_value( collection_id="your-collection-id", document_id="your-doc-id", key="your-access-token", ) def get_account_number_hash_value(self): response = requests.get( self.base_url + f"/accounts/accountNumbers", headers=self.headers ) response_frame = pandas.json_normalize(response.json()) self.account_hash_value = response_frame["hashValue"].iloc[0] def create_order(self, order_payload): order_payload = json.dumps(order_payload) response = requests.post( self.base_url + f"/accounts/{self.account_hash_value}/orders", data=order_payload, headers={ "Authorization": f"Bearer {self.access_token}", "accept": "*/*", "Content-Type": "application/json", }, ) # Custom function to create dataframe from response response_frame = return_dataframe_from_response(response) return response_frame
That's it!