Creating a Python API Wrapper (Ally Invest API)

Navigation:

• URL Handling
• AllyAPI – The wrapper class
  • “Full” API Wrapper Implementation
• Example Usage

For my investment accounts, I use Ally Invest which has a nice API that lets you query ticker and account data as well as create orders, manage watchlists, and stream data. The API documentation has some example programs written in Java, Node.JS, PHP, R, and Ruby. However, I had a couple of ideas for applications that I would like to implement in Python.

There are already a few open-source projects on GitHub that implement wrappers for the Ally Invest API but none of them accomplish everything I was wanting to do. Inspired by Discord.py and PyAlly I set out to create an API wrapper for Ally Invest that simply allows access to all of the API calls and returns the raw JSON or XML response with little fluff, as is done with Discord API calls in Discord.py.

The requirements were simple. I wanted a Python class that takes as input my OAuth keys and desired response format, provides functions to each of the API calls and returns the raw data to be processed by the wrapper’s user. Below I document the structure of my Ally Invest API wrapper (available on GitHub).

Storing the URLs

One thing I wanted to avoid was having all of the URLs running around in the main API wrapping functionality. I needed to store them somewhere they could easily be updated if the API were to be changed by Ally. This led to the creation of a URLs class which, essentially, holds the URL strings and serves them to the API wrapper. I wanted this to be ‘black-box’ so that, when calling the function to retrieve a URL, we don’t care or know what the URL is but we can assume it’s right. Below is the implementation of this object.

class URLs:
    def __init__(self, response_format="json"):
        self.format = response_format

        self.base_url = "https://api.tradeking.com/v1/"

        # account
        self.accounts = "accounts.{format}".format(format=self.format)
        self.accounts_balances = "accounts/balances.{format}".format(format=self.format)
        self.account = "accounts/{id}.{format}".format(format=self.format, id="{id}")
        self.account_balances = "accounts/{id}/balances.{format}".format(format=self.format, id="{id}")
        self.account_history = "accounts/{id}/history.{format}".format(format=self.format, id="{id}")
        self.account_holdings = "accounts/{id}/holdings.{format}".format(format=self.format, id="{id}")

        # market
        self.clock = "market/clock.{format}".format(format=self.format)
        self.quote = "market/ext/quotes{format}".format(format=self.format)

    def base_url(self):
        return self.base_url

    def accounts_url(self):
        return self.base_url + self.accounts

    def accounts_balances_url(self):
        return self.base_url + self.accounts_balances

    def account_url(self):
        return self.base_url + self.account

    def account_balances_url(self):
        return self.base_url + self.account_balances

    def account_history_url(self):
        return self.base_url + self.account_history

    def account_holdings_url(self):
        return self.base_url + self.account_holdings

    def clock_url(self):
        return self.base_url + self.clock

    def quote_url(self):
        return self.base_url + self.quote

Here, since this is only an example, about half of the functionality currently implemented is displayed, to save space. (I’ve also removed most of the Doxygen comments.) As shown here, the base_url or API endpoint is stored once and re-used to construct every specific URL. This will allow easy updates if the endpoint is ever changed, e.g. the version is incremented from v1 to v2. The same can be said for the remainder of the URLs. Holding them in this class and initializing them in the constructor makes them easy to find and maintain.

The API Wrapper Class

The next class is the actual API wrapper class. This class depends on the requests_oauthlib, datetime, requests, xml.eltree, and json Python modules. This class has a few private helper functions that make the rest of the implementation trivial, these are listed in the following sections and a subset of the currently implemented features is given below.

__create_auth()

def __create_auth(self):
    """A private method to create the OAuth1 object, if necessary."""
    now = datetime.datetime.now()
    if self.auth == None or self.auth_time + self.valid_auth_dt < now:
        self.auth_time = now
        self.auth = OAuth1(self.client_key, self.client_secret, self.oauth_token, self.oauth_secret, signature_type='auth_header')

This method is used to create/recreate the OAuth object belonging to the AllyAPI class. This is called before making any GET, POST, or DELETE API calls.

__get_symbol_string(symbols)

def __get_symbol_string(self, symbols):
    if not isinstance(symbols, str): # list
        symbols = ",".join(symbols)
    return symbols

Many of the API calls use ticker symbols (e.g. AAPL or MSFT) as parameters. The method above accepts either a list of tickers or a string with a single ticker to be used as parameters to the API call. If a list is provided the list is converted into a comma-separated list of values and returned as a string. Otherwise, the symbol string is returned unchanged.

__to_format(response)

def __to_format(self, response):
    if self.format == "json":
        return response.json()
    else:
        return ElementTree.fromstring(response.content)

In the AllyAPI class, a format is specified on creation. Valid values are ‘json’ and ‘xml’ as these are the valid response types returned by the Ally Invest API. Rather than formatting then returning the response content every time we make an API call, i.e. repeating this code snippet 30+ times, this functionality was pulled into its own private function.

__get_data(url)

def __get_data(self, url):
   self.__create_auth()
   return self.__to_format(requests.get(url, auth=self.auth))

This private method actually makes GET requests to the API. As seen here the __to_format() function is already being called to format the response from the Ally server. Similarly, the __create_auth() method is called here as well so the OAuth object can be created/refreshed before attempting to make the API call. The URL in this function comes from the URLs class described above.

__submit_post(url, data)

def __submit_post(self, url, data):
    self.__create_auth()
    return self.__to_format(requests.post(url, data=data, auth=self.auth))

The final private method submits a POST request to Ally’s server. Once again, the OAuth object is created and the URL is from the URLs class. However, new to this function is the data parameter which is a dictionary containing the data to be sent to the server, i.e. the API call’s parameters.

Full AllyAPI Class

from requests_oauthlib import OAuth1
from xml.etree import ElementTree
import datetime
import requests
import json

from URLs import *

class AllyAPI:
    def __init__(self, oauth_secret, oauth_token, client_key,
                response_format="json"):
        self.format = response_format
        self.url = URLs(response_format=response_format)

        self.oauth_secret = oauth_secret
        self.oauth_token = oauth_token
        self.client_key = client_key
        self.client_secret = client_key

        self.auth_time = None
        self.auth = None
        self.valid_auth_dt = datetime.timedelta(seconds=10)

    def __create_auth(self):
        """A private method to create the OAuth1 object, if necessary."""
        now = datetime.datetime.now()

        if self.auth == None or self.auth_time + self.valid_auth_dt < now:
            self.auth_time = now
            self.auth = OAuth1(self.client_key, self.client_secret, self.oauth_token,
                          self.oauth_secret, signature_type='auth_header')

    def __get_symbol_string(self, symbols):
        if not isinstance(symbols, str): # list
            symbols = ",".join(symbols)
        return symbols

    def __to_format(self, response):
        if self.format == "json":
            return response.json()
        else:
            return ElementTree.fromstring(response.content)

    def __get_data(self, url):
        self.__create_auth()
        return self.__to_format(requests.get(url, auth=self.auth))

    def __submit_post(self, url, data):
        self.__create_auth()
        return self.__to_format(requests.post(url, data=data, auth=self.auth))

    def get_accounts(self):
        """Returns all of the user's accounts."""
        return self.__get_data(self.url.accounts_url())

    def get_accounts_balances(self):
        """Returns the balances of all of the user's accounts."""
        return self.__get_data(self.url.accounts_balances_url())

    def get_account(self, id):
        """Returns a specific account provided the account ID (account number)
        """
        return self.__get_data(self.url.account_url().format(id=str(id)))

    def get_account_balances(self, id):
        """Returns the balances of a specific account (ID = account number)
        """
        return self.__get_data(self.url.account_balances_url().format(id=str(id)))

    def get_account_history(self, id):
        """Returns the history of a specific account (ID = account number)
        """
        return self.__get_data(self.url.account_history_url().format(id=str(id)))

    def get_account_holdings(self, id):
        """Returns the holdings of a specific account (ID = account number)"""
        return self.__get_data(self.url.account_holdings_url().format(id=str(id)))

    def get_market_clock(self):
        """Returns the state of the market, the time until next state change, and current server timestamp."""
        return self.__get_data(self.url.clock_url())

    def get_quote(self, symbols):
        """Returns quote information for a single ticker or list of tickers. Note: this function does not implement selecting customer FIDs as described in the API documentation. These can be filtered from the return if need be."""
        url = self.url.quote_url()+"?symbols={symbols}"
        symbols = self.__get_symbol_string(symbols)
        return self.__get_data(url.format(symbols=symbols))

Above is about half of the functionality currently implemented in the AllyAPI class (which is a little over half of the entire Ally Invest API function library). As seen here, the typical API call is just a couple of lines of code given the private methods above. These private methods could be used in almost any API wrapper class making the rest of the API class easy to implement (provided the URLs class is reconstructed). You can also see here that URLs are pulled from the URLs class and formatted if necessary, depending on the API call’s parameters.

Example Usage of the API Wrapper

Below is a code snippet that uses the AllyAPI class with details of the functionality in the comments. As seen here, the API is straightforward to use and provides a blackbox interface to the API allowing developers to be somewhat oblivious to the details of the API calls.

from ally import *

## These values are from Ally Invest API Applications page.
CONSUMER_KEY = "CONSUMER KEY"
CONSUMER_SECRET = "CONSUMER SECRET"
OAUTH_TOKEN = "OAUTH TOKEN"
OAUTH_SECRET = "OAUTH TOKEN SECRET"

if __name__ == "__main__":
     ally = AllyAPI(OAUTH_SECRET, OAUTH_TOKEN, CONSUMER_KEY, response_format="json")
     print(ally.get_accounts())
     
     print(ally.get_quote("AAPL"))  # returns data about the AAPL ticker
     # returns data about the AAPL, MSFT, XLNX and NXPI tickers
     print(ally.get_quote(["AAPL", "MSFT", "XLNX", "NXPI"]))

     # search for news articules for a ticker or list of tickers
     print(ally.news_search("AAPL"))
     print(ally.news_search(["AAPL", "MSFT", "XLNX", "NXPI"]))

     # retrieve watchlist 
     print(ally.get_watchlists())

     # create a watchlist which holds the AAPL ticker
     print(ally.create_watchlist("API Test List 2", "AAPL"))
     # create an empty watchlist with the name "API Test List 3"
     print(ally.create_watchlist("API Test List 3"))