/api_examples_func.py Secret
Last active
March 15, 2024 16:58
Star
You must be signed in to star a gist
how to access the Empirasign abs/mbs API using Python in a functional manner
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# -*- 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 | |
""" | |
import datetime | |
import hashlib #needed to compute request signatures | |
import requests | |
# constants | |
API_HOST = "www.empirasign.com" | |
API_SCHEME = 'https' | |
API_VERSION = 0 | |
VERBOSE = False | |
# user-specific parameters | |
APP_ID = 'MY_ID' | |
PW = 'MY_PASS' | |
PROXY_SERVER = '' # e.g. proxy.mycompany.net:8080 | |
# if you get a 407 Proxy Authentication Required error, you need to set | |
# PROXY_SERVER to something like username:password@proxy.mycompany.net:8080 | |
if API_VERSION: | |
API_VERSION = f'/v{API_VERSION}' | |
else: | |
API_VERSION = '' | |
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 {'https': 'http://' + 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_bonds(uids, d0=None, d1=None, fmt='json', nport=False): | |
""" | |
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 = '{}://{}/api{}/bonds/'.format(API_SCHEME, API_HOST, API_VERSION) | |
#compute the request signature | |
req_sig = _make_req_sig([",".join(uids), datetime.date.today()]) | |
req_params = {'app_id': APP_ID, 'req_sig': req_sig, 'nport': nport, '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_nport(uids, d0=None, d1=None, fmt='json'): | |
""" | |
get nport data for a list of cusips/ISINs/BBG Tickers | |
the maxium length of uids processed by server is 500, items 501 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 = '{}://{}/api{}/nport/'.format(API_SCHEME, API_HOST, API_VERSION) | |
#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_deal(uid, fmt='json'): | |
""" | |
get all tranches on a deal - indicated by a ticker, cusip, or isin for a deal | |
ambiguous deal ticker will return maximum results of 500 | |
fmt can be json or csv | |
""" | |
api_url = '{}://{}/api{}/deal-classes/'.format(API_SCHEME, API_HOST, API_VERSION) | |
#compute the request signature | |
req_sig = _make_req_sig([uid]) | |
req_params = {'app_id': APP_ID, 'req_sig': req_sig, 'bond': uid} | |
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 = '{}://{}/api{}/bwics/'.format(API_SCHEME, API_HOST, API_VERSION) | |
# 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, | |
'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, | |
'req_sig': req_sig | |
} | |
if bsym_sector: | |
url_params['bsym_sec_type'] = bsym_sector | |
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 = '{}://{}/api{}/offers/'.format(API_SCHEME, API_HOST, API_VERSION) | |
# 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_all_bonds(dt, figi_marketsector, kind=None, fmt='json'): | |
""" | |
return a list of all bonds that appeared on BWICs or Dealer Runs for a given date / sector | |
figi_marketsector is exactly equivalent to marketSecDes as defined by OpenFIGI | |
https://www.openfigi.com/api#openapi-schema | |
kind = bwics (show only list of all BWIC bonds for a given date) | |
kind = runs (show only list of all runs bonds for a given date) | |
fmt can be json or csv | |
""" | |
api_url = '{}://{}/api{}/all-bonds/'.format(API_SCHEME, API_HOST, API_VERSION) | |
dt = dt.strftime("%Y%m%d") | |
# compute the request signature | |
req_sig = _make_req_sig([figi_marketsector, dt]) | |
url_params = { | |
'app_id': APP_ID, | |
'figi_marketsector': figi_marketsector, | |
'dt': dt, | |
'req_sig': req_sig | |
} | |
if kind is not None: | |
url_params["kind"] = kind | |
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_all_matchers(dt, fmt='json'): | |
""" | |
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. | |
fmt can be json or csv | |
""" | |
api_url = '{}://{}/api{}/all-matchers/'.format(API_SCHEME, API_HOST, API_VERSION) | |
dt = dt.strftime("%Y%m%d") | |
# compute the request signature | |
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 | |
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 = '{}://{}/api{}/collab/'.format(API_SCHEME, API_HOST, API_VERSION) | |
dt = datetime.date.today().strftime("%Y%m%d") | |
data = { | |
'app_id': APP_ID, | |
'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 = '{}://{}/api{}/mbsfeed/'.format(API_SCHEME, API_HOST, API_VERSION) | |
# 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 = '{}://{}/api{}/mystatus/'.format(API_SCHEME, API_HOST, API_VERSION) | |
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 = '{}://{}/api{}/query_log/'.format(API_SCHEME, API_HOST, API_VERSION) | |
# 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