Instantly share code, notes, and snippets.

Embed
What would you like to do?
how to access the Empirasign abs/mbs API using Python in a functional manner
# -*- coding: utf-8 -*-
"""
api_examples_func.py
this small library illustrates how to access the Empirasign abs/mbs
API using Python in a functional manner
https://gist.github.com/empirasign/cbd204ae7f45bc4eadae
"""
from __future__ import absolute_import
from __future__ import print_function
from __future__ import unicode_literals
import hashlib #needed to compute request signatures
import datetime
import requests
# constants
API_HOST = "www.empirasign.com"
VERBOSE = False
# user-specific parameters
APP_ID = 'MY_APP_ID'
PW = 'MY_PASS'
PROXY_SERVER = '' # e.g. proxy.mycompany.net:8080
def _proxies_dict(proxy_server):
"""
return proxy dictionary as needed by requests library
if this helper function is not comprehensive enough for your use case, consult
http://docs.python-requests.org/en/latest/user/advanced/#proxies
"""
if proxy_server:
return {'http': 'http://' + PROXY_SERVER, 'https': 'https://' + PROXY_SERVER}
return {}
def _make_req_sig(args):
"""
generate request signature
as a convenience, datetime.date objects autoconvert to YYYYMMDD strings
"""
args = [arg.strftime("%Y%m%d") if isinstance(arg, datetime.date) else arg for arg in args]
sig_keys = [APP_ID] + args + [PW]
return hashlib.sha1("".join(sig_keys).encode('utf-8')).hexdigest()
def get_bond(uid, d0=None, d1=None, fmt='json', prepend_price32=True):
"""
get bwic and dealer runs data for a specific cusip/ISIN/BBG Ticker
d0 and d1 are not required, they indicated start and end (inclusive)
of date range to search, must be datetime.date type
fmt can be json or csv (if you ask us to support xml, we will think
less of you)
"""
api_url = 'http://{}/api/cusip/'.format(API_HOST)
#compute the request signature
req_sig = _make_req_sig([uid, datetime.date.today()])
url_params = {'cusip': uid, 'app_id': APP_ID, 'req_sig': req_sig, 'quote': prepend_price32}
if d0 and d1:
url_params['d0'] = d0.strftime("%Y%m%d")
url_params['d1'] = d1.strftime("%Y%m%d")
if fmt.lower() != 'json':
url_params['csv'] = True
resp = requests.get(api_url, url_params, proxies=_proxies_dict(PROXY_SERVER))
if VERBOSE:
print(resp.url)
if fmt.lower() == 'json':
return resp.json()
return resp.text
def get_bonds(uids, d0=None, d1=None, fmt='json'):
"""
get bwic and dealer runs data for a list of cusips/ISINs/BBG Tickers
the maxium length of uids processed by server is 200, items 201 and
higher will be ignored.
d0 and d1 are not required, they indicated start and end (inclusive)
of date range to search, must be datetime.date type
fmt can be json or csv
"""
api_url = 'http://{}/api/bonds/'.format(API_HOST)
#compute the request signature
req_sig = _make_req_sig([",".join(uids), datetime.date.today()])
req_params = {'app_id': APP_ID, 'req_sig': req_sig, 'bonds': ",".join(uids)}
if d0 and d1:
req_params['d0'] = d0.strftime("%Y%m%d")
req_params['d1'] = d1.strftime("%Y%m%d")
if fmt.lower() != 'json':
req_params['csv'] = True
resp = requests.post(api_url, data=req_params, proxies=_proxies_dict(PROXY_SERVER))
if VERBOSE:
print(resp.url)
if fmt.lower() == 'json':
return resp.json()
return resp.text
def get_bwics(sector, d0, d1=None, fmt='json', bsym_sector=False):
"""get summary level data for BWICs in a given sector. This
query does not return actual market data so it does not count
against the daily quota
d0: start of date range (max 30 day lookback)
d1: end of date range
when d0 = d1, this returns same results as get_bwics_old()
sector: must be on of the following or server will return an error
agency, cmbs, nonag, abs, naresi, conabs, clo, euro, eu-clo, eu-resi,
eu-abs, eu-cmbs, agarm, agcmo, snr, agcmbs, am, aj, io, mezz, unk,
alta-15, alta-30, alta-a, prime-15, prime-30, prime-a, auto, card,
equip, heloc, heq, mh, other, student, subprime, utility
sector descriptions can be found here: www.empirasign.com/api-mbs/
fmt can be json or csv
"""
api_url = 'http://{}/api/bwics/'.format(API_HOST)
# compute the request signature
# the trick here is the dt for the request signature and dt for the
# query parameter are the one and the same
if d0 and d1:
d0 = d0.strftime("%Y%m%d")
d1 = d1.strftime("%Y%m%d")
req_sig = _make_req_sig([sector, d0, d1])
url_params = {
'sector': sector,
'app_id': APP_ID,
'd0': d0,
'd1': d1,
'bsym_sec_type': bsym_sector,
'req_sig': req_sig
}
else:
# search on single date and pass dt as arg name
d0 = d0.strftime("%Y%m%d")
req_sig = _make_req_sig([sector, d0])
url_params = {
'sector': sector,
'app_id': APP_ID,
'dt': d0,
'bsym_sec_type': bsym_sector,
'req_sig': req_sig
}
if fmt.lower() != 'json':
url_params['csv'] = True
resp = requests.get(api_url, url_params, proxies=_proxies_dict(PROXY_SERVER))
if VERBOSE:
print(resp.url)
if fmt.lower() == 'json':
return resp.json()
return resp.text
def get_offers(dt, dealer=None, sector=None, constraints=(), fmt='json'):
"""
Get all offers for a specific date, or for a dealer/dector/dt combo
only dt parameter is required as a datetime.date type
if only dt provided, a list of dealer and sector combos of dealers
runs available for that date will be returned. This particular quiery
as it returns no market data has no query cost.
if dt, sector and dealer are provided, then the result set will be that
particular dealer's run for a the given sector and date
fmt can be json or csv
"""
api_url = 'http://{}/api/offers/'.format(API_HOST)
# compute the request signature
dt = dt.strftime("%Y%m%d")
if sector and dealer:
sig_keys = [dealer, sector, dt]
else:
sig_keys = [dt]
req_sig = _make_req_sig(sig_keys)
url_params = {'app_id': APP_ID, 'dt': dt, 'req_sig': req_sig}
for key in constraints:
url_params[key] = constraints[key]
if sector and dealer:
url_params.update({'sector': sector, 'dealer': dealer})
if fmt.lower() != 'json':
url_params['csv'] = True
resp = requests.get(api_url, url_params, proxies=_proxies_dict(PROXY_SERVER))
if VERBOSE:
print(resp.url)
if fmt.lower() == 'json':
return resp.json()
return resp.text
def get_collab_search(bonds, fmt='json'):
"""
return a list of similar bonds for each bond in bonds
based on Empirasign Collaborative Search Algorithm
https://www.empirasign.com/blog/Collaborative-Search/
this API METHOD requires HTTP POST
"""
api_url = 'http://{}/api/collab/'.format(API_HOST)
dt = datetime.date.today().strftime("%Y%m%d")
data = {
'app_id': APP_ID,
'dt': dt,
'req_sig': _make_req_sig([",".join(bonds), dt]),
'bonds': ",".join(bonds)
}
if fmt.lower() != 'json':
data['csv'] = True
resp = requests.post(api_url, data=data)
if VERBOSE:
print(resp.url)
if fmt.lower() == 'json':
return resp.json()
return resp.text
def get_events(n=15, fmt='json'):
"""
get the latest db events (new bwic, new price talk, or
new trade color) from the news feed
this function has no required parameters
n is the n latest events
fmt can be json or csv
"""
api_url = 'http://{}/api/mbsfeed/'.format(API_HOST)
# compute the request signature
req_sig = _make_req_sig([str(n), datetime.date.today()])
url_params = {'app_id': APP_ID, 'n': n, 'req_sig': req_sig}
if fmt.lower() != 'json':
url_params['csv'] = True
resp = requests.get(api_url, url_params, proxies=_proxies_dict(PROXY_SERVER))
if VERBOSE:
print(resp.url)
if fmt.lower() == 'json':
return resp.json()
return resp.text
def get_api_status(fmt='json'):
"""
check API status and see how many requests are left
fmt can be json or csv
"""
api_url = 'http://{}/api/mystatus/'.format(API_HOST)
req_sig = _make_req_sig([datetime.date.today()])
url_params = {'app_id': APP_ID, 'req_sig': req_sig}
if fmt.lower() != 'json':
url_params['csv'] = True
resp = requests.get(api_url, url_params, proxies=_proxies_dict(PROXY_SERVER))
if VERBOSE:
print(resp.url)
if fmt.lower() == 'json':
return resp.json()
return resp.text
def get_query_log(dt=None, fmt='json'):
"""
get a log of queries made on a given date
if provided, dt must be a datetime.date type
fmt can be json or csv
"""
api_url = 'http://{}/api/query_log/'.format(API_HOST)
# compute request signature
if not dt:
dt = datetime.date.today().strftime("%Y%m%d")
else:
dt = dt.strftime("%Y%m%d")
req_sig = _make_req_sig([dt])
url_params = {'app_id': APP_ID, 'dt': dt, 'req_sig': req_sig}
if fmt.lower() != 'json':
url_params['csv'] = True
resp = requests.get(api_url, url_params, proxies=_proxies_dict(PROXY_SERVER))
if VERBOSE:
print(resp.url)
if fmt.lower() == 'json':
return resp.json()
return resp.text
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment