Charles Schwab API – Authentication & Sample Calls

# Constants
# CLIENT_ID = 'Enter Client ID'
# CLIENT_SECRET = 'Enter Client Secret'
# REDIRECT_URI = 'https://127.0.0.1'
# SCOPE = 'scope_if_applicable'  # if needed
# STATE = 'random_string_for_csrf_protection'  # if needed

def get_authorization_url():
    """Generate and open the authorization URL."""
    params = {
        'response_type': 'code',
        'client_id': CLIENT_ID,
        'redirect_uri': REDIRECT_URI,
        'scope': SCOPE,
        'state': STATE
    }
    query_string = urllib.parse.urlencode(params)
    authorization_url = f"https://api.schwabapi.com/v1/oauth/authorize?{query_string}"
    webbrowser.open(authorization_url)
    return input("Paste the URL you were redirected to here: ")

def extract_code_from_url(url):
    """Extract the authorization code from the redirected URL."""
    parsed_url = urllib.parse.urlparse(url)
    query_params = urllib.parse.parse_qs(parsed_url.query)
    return query_params.get('code', [None])[0]

def get_access_token(authorization_code):
    """Exchange the authorization code for an access token."""
    TOKEN_URL = "https://api.schwabapi.com/v1/oauth/token"
    credentials = f"{CLIENT_ID}:{CLIENT_SECRET}"
    encoded_credentials = base64.b64encode(credentials.encode('utf-8')).decode('utf-8')
    headers = {
        'Authorization': f'Basic {encoded_credentials}',
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    payload = {
        'grant_type': 'authorization_code',
        'code': authorization_code,
        'redirect_uri': REDIRECT_URI
    }
    response = requests.post(TOKEN_URL, headers=headers, data=payload)
    if response.status_code == 200:
        return response.json()['access_token']
    else:
        raise Exception(f"Failed to obtain token: {response.text}")

def api_request(access_token, base_url, endpoint, params=None):
    """Generic function to make API requests."""
    headers = {'Authorization': f'Bearer {access_token}'}
    url = f"{base_url}{endpoint}"
    if params:
        url += '?' + urllib.parse.urlencode(params)
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Failed to retrieve {endpoint}: {response.text}")

# Account details
def get_accounts(access_token):
    return api_request(access_token, "https://api.schwabapi.com/trader/v1/", 'accounts')

# User preferences
def get_user_preferences(access_token):
    return api_request(access_token, "https://api.schwabapi.com/trader/v1/", 'userPreference')

# Market data
def get_quotes(access_token, symbol):
    return api_request(access_token, "https://api.schwabapi.com/marketdata/v1/", f"{symbol}/quotes")

def get_expiration_chain(access_token, symbol):
    return api_request(access_token, "https://api.schwabapi.com/marketdata/v1/", 'expirationchain', {'symbol': symbol})

def get_price_history(access_token, symbol, period_type='month', period=3, frequency_type='daily', need_previous_close=False):
    params = {
        'symbol': symbol,
        'periodType': period_type,
        'period': period,
        'frequencyType': frequency_type,
        'needPreviousClose': need_previous_close
    }
    return api_request(access_token, "https://api.schwabapi.com/marketdata/v1/", 'pricehistory', params)

# Authorization flow and example usage
url = get_authorization_url()
authorization_code = extract_code_from_url(url)
access_token = get_access_token(authorization_code)

# Retrieve data using the obtained access token
accounts = get_accounts(access_token)
user_preferences = get_user_preferences(access_token)
symbol = 'AAPL'
quotes = get_quotes(access_token, symbol)
expiration_chain = get_expiration_chain(access_token, symbol)
price_history = get_price_history(access_token, symbol)

# Display results
print("Accounts:", accounts)
print("User Preferences:", user_preferences)
print("Quotes for AAPL:", quotes)
print("Expiration Chain for AAPL:", expiration_chain)
print("Price History for AAPL:", price_history)