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