Piousbox

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!