Simple object-oriented library for Empirasign API Access

mbs_api_oo.py

# -*- coding: utf-8 -*-
"""
mbs_api_oo.py

This simple library illustrates how to access the Empirasign Market Data API
using Python in a Object-Oriented manner.
Access every public endpoint using a single class.

Latest version of this module: https://gist.github.com/empirasign/e671dc29b40720b77171b73b93e4d0d5
Full API Documentation: https://www.empirasign.com/api-mbs/

Endpoint Summary:
    HTTP METHOD     API ENDPOINT        QUOTA HIT           CLASS METHOD
    POST            /api/bonds/         1 per bond          get_bonds
    GET             /api/bwics/         None                get_bwics
    POST            /api/nport/         1 per bond          get_nport
    GET             /api/offers/        None                get_available_runs
    GET             /api/offers/        1 per bond          get_dealer_runs
    POST            /api/deal-classes/  None                get_deal
    GET             /api/all-bonds/     None                get_active_bonds
    POST            /api/collab/        None                get_suggested
    GET             /api/mbsfeed/       None                get_events
    GET             /api/query_log/     None                get_query_log
    GET             /api/mystatus/      None                get_status

Example Usage:
    api = EmpirasignAPI("APP_ID", "API_SECRET")  # initialize an authorized API object

    ## grab market data
    api.get_bonds(["05543DBT0", "36242DL68", "43739EBN6"])

    ## grab market data for a specific date, datetime object or "YYYYMMDD" formated str accepted
    api.get_bonds(["05543DBT0", "36242DL68", "43739EBN6"], datetime.date(2021, 01, 01))

    ## use tuple to specify a date range
    api.get_bonds(["05543DBT0", "36242DL68", "43739EBN6"], (d0, d1))

    api.quota  # check your remaining queries
    api.format = 'csv'  # set your preferred response format, default is JSON
"""

import logging  # descriptive logging in verbose mode
import hashlib  # needed to compute request signatures
import datetime
import requests

logger = logging.getLogger(__name__)


class EmpirasignAPI:
    """
    Single class for accessing all endpoints of the Empirasign Market Data API
    https://www.empirasign.com/api-mbs/

    Properties:
        format      (str): sets the format of the API response objects, CSV or JSON
        verbose     (bool): indicates if methods should print descriptive messages
        quota       (obj): current status of API quota
    """
    _api_scheme = 'https'
    _api_host = 'www.empirasign.com'
    _response_format = 'json'
    _verbose = False
    _ua_string_suffix = " mbs-api-oo"

    def __init__(self, app_id, api_secret, proxy_server=None):
        """
        Initialize the class.
        Args:
            app_id       (str): Your assigned App ID
            api_secret   (str): Your assigned API Secret
            proxy_server (str, Optional): https://www.empirasign.com/api-mbs/#proxy-server
        """
        self.app_id = app_id
        self.api_secret = api_secret
        self.proxy_dict = self.__proxies_dict(proxy_server)
        self.headers = requests.utils.default_headers()
        self.headers["User-Agent"] += self._ua_string_suffix

    @property
    def format(self):
        return self._response_format

    @property
    def verbose(self):
        return self._verbose

    @property
    def quota(self):
        """
        Returns your current API quota.
        """
        tmp_format = self.format
        self.format = 'json'
        quota = self.get_status()['requests_left']
        self.format = tmp_format
        return quota

    @format.setter
    def format(self, value):
        if value.lower() not in ('json', 'csv'):
            raise ValueError("Response format must be CSV or JSON")
        self._response_format = value.lower()

    @verbose.setter
    def verbose(self, value):
        if not isinstance(value, bool):
            raise ValueError("Verbose must be a boolean value")
        self._verbose = value


    @staticmethod
    def __proxies_dict(proxy_server):
        """
        Returns proxy dictionary required by requests library.
        http://docs.python-requests.org/en/latest/user/advanced/#proxies
        """
        if proxy_server:
            return {'https': f'http://{proxy_server}'}
        return {}


    @staticmethod
    def __chunker(lst, chunk_size):
        """
        Break down large lists into managable chunks
        """
        for i in range(0, len(lst), chunk_size):
            yield lst[i:i + chunk_size]


    @staticmethod
    def __handle_date_args(req_params, date_args, default_today=False):
        """
        handle single date, date range, or no date situations
        """
        args = {}
        if isinstance(date_args, (list, tuple)) and len(date_args) == 2:
            start, end = date_args[0], date_args[1]
            args['d0'] = start.strftime("%Y%m%d") if isinstance(start, datetime.date) else start
            args['d1'] = end.strftime("%Y%m%d") if isinstance(end, datetime.date) else end
        elif isinstance(date_args, (str, datetime.date)):
            single = date_args
            args['dt'] = single.strftime("%Y%m%d") if isinstance(single, datetime.date) else single
        elif default_today:
            args['dt'] = datetime.date.today().strftime("%Y%m%d")
        elif date_args:
            raise ValueError("Invalid date arguments")
        req_params.update(args)
        return req_params


    @staticmethod
    def __handle_single_dt(dt, default_today=False):
        """
        handle single date formatting and default
        """
        if default_today:
            dt = dt or datetime.date.today()
        return dt.strftime('%Y%m%d') if isinstance(dt, datetime.date) else dt


    def __make_req_sig(self, args):
        """
        Generate valid request signature.
        """
        args = [arg.strftime("%Y%m%d") if isinstance(arg, datetime.date) else arg for arg in args]
        sig_keys = [self.app_id] + args + [self.api_secret]
        return hashlib.sha1("".join(sig_keys).encode('utf-8')).hexdigest()


    def __request(self, endpoint, params, sig_args, method='GET'):
        """
        Construct standard request format with valid signature.
        """
        api_url = f'{self._api_scheme}://{self._api_host}/api/{endpoint}/'
        params.update({'app_id': self.app_id, 'req_sig': self.__make_req_sig(sig_args)})
        if self.format == 'csv':
            params['csv'] = True
        if method == 'GET':
            resp = requests.get(api_url, params, headers=self.headers, proxies=self.proxy_dict)
        else:
            resp = requests.post(api_url, data=params, headers=self.headers,
                                 proxies=self.proxy_dict)
        if self.verbose:
            logger.info('Making request... %s', endpoint)
            logger.info('\tURL: %s', resp.url)
            logger.info('\tHTTPS Method: %s', method)
            logger.info('\tQuery Parameters: %s', params)
            logger.info('\tFormat: %s', self.format)
            logger.info('\tResponse Code: %s', resp.status_code)
        if self.format == 'json':
            resp_json = resp.json()
            if self.verbose:
                logger.info('\tRequests Left: %s', resp_json['meta']['requests_left'])
            return resp_json
        return resp.text


    def __bulk_request(self, endpoint, bulk_items, params, chunk_size,
                       req_key="bonds",
                       res_key="data"):
        """
        Break up bulk requests that exceed per-query limits, consolidate results
        """
        if self.format == 'json':
            results = {res_key: [], 'meta': {'errors': [], 'warnings': []}}
        else:
            results = ''
        for items_chunk in self.__chunker(bulk_items, chunk_size):
            sig_args = [",".join(items_chunk), datetime.date.today()]
            params[req_key] = ",".join(items_chunk)
            res = self.__request(endpoint, params, sig_args, method='POST')
            if self.format == 'json':
                results[res_key].extend(res.get(res_key, []))
                results['meta']['errors'].extend(res['meta'].get('errors', []))
                results['meta']['warnings'].extend(res['meta'].get('warnings', []))
                results['meta']['requests_left'] = res['meta']['requests_left']
            else:  # handle bulk CSV requests
                if results:
                    results += "\n" + "\n".join(res.splitlines()[1:])
                else:
                    results = res
        return results

    #--------------------- Public API Endpoint Methods -------------------------

    def get_bonds(self, uids, date_args=None, nport=False):
        """
        Get all market data for a list of bonds.
        Args:
            uids    (list of str): List of unique bond identifiers (CUSIP, ISIN, or BBG Ticker).
            date_args   (datetime.date or str:YYYYMMDD, Optional): Single search date.
                        (tuple, Optional): Start and end of date range (inclusive).
                        NOTE: Defaults to no date constraint
            nport   (boolean, Optional): Include NPORT records if True, default is False.
        """
        req_params = self.__handle_date_args({'nport': nport}, date_args)
        return self.__bulk_request('bonds', uids, req_params, 200)


    def get_bwics(self, sector, date_args=None):
        """
        Get summary level data for BWICs in a given sector. NOTE: Maximum 60 day lookback
        Args:
            sector      (str): BWIC sector defined in www.empirasign.com/api-mbs/
            date_args   (datetime.date or str:YYYYMMDD, Optional): Single search date.
                        (tuple, Optional): Start and end of date range (inclusive).
                        NOTE: Defaults to no date constraint
        """
        req_params = self.__handle_date_args({'sector': sector}, date_args, default_today=True)
        if req_params.get('dt'):
            sig_args = [sector, req_params['dt']]
        else:
            sig_args = [sector, req_params['d0'], req_params['d1']]
        return self.__request('bwics', req_params, sig_args)


    def get_nport(self, uids, date_args=None):
        """
        Get NPORT filing data for a list of bonds.
        Args:
            uids         (list of str): List of bond identifiers (CUSIPs or ISINs only).
            date_args    (datetime.date or str:YYYYMMDD, Optional): Single search date.
                         (tuple, Optional): Start and end of date range (inclusive).
                         NOTE: Defaults to no date constraint
        """
        req_params = self.__handle_date_args({}, date_args)
        return self.__bulk_request('nport', uids, req_params, 750)


    def get_deal(self, uid):
        """
        Give all tranches on the deal.
        Args:
            uid    (str): A bond identifier (CUSIP, ISIN, BBG Ticker, or Deal Series).
        """
        return self.__request('deal-classes', {'bond': uid}, [uid], method='POST')


    def get_available_runs(self, dt=None):
        """
        Get all dealer & sector combinations for Dealer Runs on a given date.
        Args:
            dt  (datetime.date or str:YYYYMMDD, Optional): The query date, defaults to today.
        """
        dt = self.__handle_single_dt(dt, default_today=True)
        return self.__request('offers', {'dt': dt}, [dt])


    def get_dealer_runs(self, dealer, sector, dt=None, min_cf=None):
        """
        Get all Dealer Runs records for a particular dealer, sector, & date combination.
        Args:
            dealer      (str): The dealer.
            sector      (str): The sector.
            dt          (datetime.date or str:YYYYMMDD, Optional): The query date, default today.
            min_cf      (float, Optional): NOTE: Minimum CF, applies to 'spec' sector ONLY.
        """

        dt = self.__handle_single_dt(dt, default_today=True)
        sig_args = [dealer, sector, dt]
        req_params = {'dealer': dealer, 'sector': sector, 'dt': dt }
        if min_cf is not None:
            req_params['min_cf'] = min_cf
        return self.__request('offers', req_params, sig_args)


    def get_active_bonds(self, dt, figi_marketsector='Mtge', kind=None):
        """
        Get all bonds that appeared on BWICs or Dealer Runs for a given date & sector.
        NOTE: figi_marketsector is exactly equivalent to marketSecDes as defined by OpenFIGI
        https://www.openfigi.com/api#openapi-schema
        Args:
            dt                  (datetime.date or str:YYYYMMDD): The query date.
            figi_marketsector   (str, Optional): The market sector "Mtge" or "Corp", default Mtge.
            kind                (str, Optional): Kind of market activity, "bwics" or "runs".
        """
        dt = self.__handle_single_dt(dt)
        sig_args = [figi_marketsector, dt]
        req_params = {'figi_marketsector': figi_marketsector, 'dt': dt}
        if kind and kind in ('bwics', 'runs'):
            req_params['kind'] = kind
        elif kind:
            raise ValueError("If specified, kind must be either 'bwics' or 'runs'")
        return self.__request('all-bonds', req_params, sig_args)


    def get_all_matchers(self, dt=None):
        """
        Get all bonds that appeared on BWICs or Dealer Runs for a given date that also appeared on
        recent Fund Holdings (N-PORT) or Insurance Transactions (NAIC) filings.
        https://www.openfigi.com/api#openapi-schema
        Args:
            dt                  (datetime.date or str:YYYYMMDD): The query date.
        """
        dt = self.__handle_single_dt(dt, default_today=True)
        return self.__request('all-matchers', {'dt': dt}, [dt])


    def get_suggested(self, uids):
        """
        Get a list of similar bonds for each bond provided using Empirasign's
        Collaborative Search Algorithm: https://www.empirasign.com/blog/Collaborative-Search/
        NOTE: Maximum 50 bonds per request
        Args:
            uids  (list of str): List of unique bond identifiers (CUSIP, ISIN, BBG Ticker)
        """
        return self.__bulk_request('collab', uids, {}, 50, res_key='rec_list')


    def get_events(self, n=15):
        """
        Get the latest market events from the news feed (new bwic, price talk, or trade color).
        Args:
            n   (int, Optional): Number of events, defaults to 15. n must be between 1 and 200.
        """
        return self.__request('mbsfeed', {'n': n}, [str(n), datetime.date.today()])


    def get_query_log(self, dt=None):
        """
        Get a log of queries made on a given date. 30 day maximum lookback.
        Args:
            dt  (datetime.date or str:YYYYMMDD, Optional): The query date, defaults to today.
        """
        dt = self.__handle_single_dt(dt, default_today=True)
        return self.__request('query_log', {'dt': dt}, [dt])


    def get_status(self):
        """
        Check API status and see how many queries are remaining in your quota.
        """
        return self.__request('mystatus', {}, [datetime.date.today()])