Schwab API: Example for placing order using schwab-py wrapper

schwab-py-place-order.py

import json
import logging
import os
import sys

from schwab import auth
from schwab.orders.common import *
from schwab.orders.generic import OrderBuilder

# Include local lib code from the project
local_lib_dir = os.path.join(os.path.dirname(__file__), '..', '..', 'lib')
if local_lib_dir not in sys.path:
    sys.path.append(local_lib_dir)
from environment.instance_settings import schwabSettings

client_id = schwabSettings.SCHWAB_APP_ID
client_secret = schwabSettings.SCHWAB_APP_SECRET
redirect_uri = schwabSettings.SCHWAB_REDIRECT_URI
token_path = schwabSettings.SCHWAB_TOKEN_PATH
account_id = schwabSettings.SCHWAB_ACCOUNT_ID

# Logging
logger = logging.getLogger(__name__)

#=====================================================================================
class SchwabBroker():
    #=====================================================================================
    def __init__(self):
        logger.info("Initializing Schwab API client...")

        # Basic Info
        self._broker_name = "Schwab"

        # Get the schwab-py client
        # self._schwab_client = auth.easy_client(
        #     api_key=client_id,
        #     app_secret=client_secret,
        #     callback_url=redirect_uri,
        #     token_path=token_path
        # )
        self._schwab_client = auth.client_from_token_file(
            api_key=client_id,
            app_secret=client_secret,
            token_path=token_path
        )

        # Account related
        self._account_numbers_map = None

    #=====================================================================================
    def broker_name(self):
        return self._broker_name

    #=====================================================================================
    def _acquire__account_numbers(self, force=False):
        # If the account numbers were already retrieved from the api, then just use the cached data.
        if not force and self._account_numbers_map is not None:
            return

        logger.info(f"SchwabBroker::_acquire__account_numbers: Calling get_account_numbers")
        r = self._schwab_client.get_account_numbers()
        logger.info(f"SchwabBroker::_acquire__account_numbers: status_code = {r.status_code}")
        logger.info(f"r.text = {r.text}")
        assert (r.status_code == 200 or r.status_code == 201), r.raise_for_status()
        account_num_objs = r.json()

        self._account_numbers_map = {}
        for account_num_obj in account_num_objs:
            account_id = account_num_obj['accountNumber']
            hashValue = account_num_obj['hashValue']
            self._account_numbers_map[account_id] = hashValue

    #=====================================================================================
    def get_account_numbers(self):
        # Make sure we have the account numbers
        self._acquire__account_numbers()
        return self._account_numbers_map.keys()

    #=====================================================================================
    def get_quote(self, symbol):
        logger.info(f"SchwabBroker::get_quote: Calling get_quotes for symbol = {symbol}")
        r = self._schwab_client.get_quotes([symbol])
        logger.info(f"SchwabBroker::get_quote: status_code = {r.status_code}")
        logger.info(f"r.text = {r.text}")
        assert (r.status_code == 200 or r.status_code == 201), r.raise_for_status()
        return r.json()


    #=====================================================================================
    def place_order(self, order,  account_id) -> str:
        """Returns order id of the newly placed order."""
        # Is real time trading enabled?
        # if not settings.TRADING_ENABLED:
        #     raise Exception("Trading is disabled.")

        # Make sure we have the account numbers
        self._acquire__account_numbers()

        account_hash = self._account_numbers_map.get(account_id, None)
        assert account_hash is not None, f"Account number not found for account_id = {account_id}"

        logger.info(f"SchwabBroker::place_order: Calling place order for account_id = {account_id}, order = {order}")
        r = self._schwab_client.place_order(account_hash, order)
        logger.info(f"SchwabBroker::place_order: status_code = {r.status_code}")
        logger.info(f"r.text = {r.text}")
        logger.info(f"r.headers = \n{json.dumps(dict(r.headers), indent=4)}")
        assert (r.status_code == 200 or r.status_code == 201), r.raise_for_status()
        # 'Location':  'https://api.schwabapi.com/trader/v1/accounts/ACCABCDEF0123456789ABCDEF0123456789/orders/123456789'
        location = r.headers.get('location', None)
        if location is not None:
            tokens = location.split('/')
            if len(tokens) > 4 and tokens[-2] == "orders":
                order_id = tokens[-1]
                logger.info(f"order_id = {order_id}")
                return order_id
        return None


#===============================================================================
# T E S T I N G
#===============================================================================
def test():
    broker = SchwabBroker()

    print(f"Account Lists: {broker.get_account_numbers()}")

    print(f"Quote for AAPL: {broker.get_quote('AAPL')}")

    # Test placing an order
    buy_price = 150.0
    buy_shares = 1
    symbol = 'AAPL'
    (entry_order) = (OrderBuilder()
                     .set_order_strategy_type(OrderStrategyType.SINGLE)
                     .set_session(Session.NORMAL)
                     .set_duration(Duration.DAY)
                     .set_order_type(OrderType.LIMIT)
                     .set_price(buy_price)
                     .add_equity_leg(EquityInstruction.BUY, symbol, buy_shares))

    order_id = broker.place_order(entry_order, account_id)
    print(f"Order ID: {order_id}")

#===============================================================================
if __name__ == '__main__':
    from oslib.debug_utils import setup_root_logger_for_console
    setup_root_logger_for_console()

    print("Running tests...")
    test()
    print("Completed tests.")