Skip to content

Instantly share code, notes, and snippets.

@empirasign
Last active Jun 2, 2022
Embed
What would you like to do?
Simple object-oriented library for Empirasign API Access
# -*- 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 {'http': f'http://{proxy_server}', 'https': f'https://{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:
if sector != 'spec':
logger.info('NOTE: min_cf constraint only applies to spec sector at this time')
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()])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment