Python Liberator: Advanced - Query data using request.post RESTFul API

request.post from Python to access Liberator Data

The request.post is an internet request method to request that the liberator server accept the data enclosed arguments and process the request.

Post Arguments:

• 'https://api.cloudquant.ai/liberator/query'
  • This is the url of the query interface. This must be this exact string to query.
• data=json.dumps({"compress":compressed_transfer, "json_xfer":False, "user":user, "token":token, "name":name, "as_of":as_of, "back_to":back_to, "symbols":symbols, "system":"API"})
  • This json.dumps section provides the body of the query.
• headers={'Content-Type':'application/json'},
• cert=cert,
  • cert comes from with pfx_to_pem('liberator.pfx','') as cert: ro
• stream=True

Query the Last Known Value(s)

To get the last known value in any dataset simply do not provide the as_of or back_to arguments. This will give you the last known value for the given dataset and symbols.

Query for a Time Series Result

Adding the back_to argument for any query will give you the time series data all the way back to the specified back to date.

Every dataset has different data frequencies. We suggest using short time frames until you become familiar with the datasets, some of our datasets are quite large and thus an over-extended request may result in an extremely large amount of data returned.

json.dumps components

Example Calling the Liberator REST API with Python

Example Code for calling query interface in liberator using POST can be found in under_the_hood.py.

import sys

import zlib

import json

import base64

import requests

import tempfile

import contextlib

import pyarrow as pa

import OpenSSL.crypto




# from a gist

@contextlib.contextmanager

def pfx_to_pem(pfx_path, pfx_password):

   ''' Decrypts the .pfx file to be used with requests. '''

   with tempfile.NamedTemporaryFile(suffix='.pem') as t_pem:

       t_pem.close()

       f_pem = open(t_pem.name, 'wb')

       pfx = open(pfx_path, 'rb').read()

       p12 = OpenSSL.crypto.load_pkcs12(pfx, pfx_password)

       f_pem.write(OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM, p12.get_privatekey()))

       f_pem.write(OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, p12.get_certificate()))

       ca = p12.get_ca_certificates()

       if ca is not None:

           for cert in ca:

               f_pem.write(OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, cert))

       f_pem.close()

       yield t_pem.name




compressed_transfer=True




# POST the query and prepare for a stream of single-line JSON replies

with pfx_to_pem('liberator.pfx','') as cert:

   r=requests.post('https://api.cloudquant.ai/liberator/query',

           data=json.dumps({"compress":compressed_transfer,"json_xfer":False,

               "user":user,"token":token,

               "name":name,"as_of":as_of,"back_to":back_to,"symbols":symbols,

               "system":"API"}),

           headers={'Content-Type':'application/json'},

           cert=cert,

           stream=True)




# ensure that the request was successful

if r.status_code!=200:

   print(r.json())

else:

   batches=[]

   for line in r.iter_lines(4096):

       # Show progress

       print('.',end='')

       sys.stdout.flush()

       # The response stream contains informational messages other than "batch"

       if line[0:14] == b'{"exception":"':

           print('Error: '+json.loads(''.join([chr(c) for c in line]), encoding='latin-1')['value'])

           break

       if line[0:10] != b'{"batch":"':

           continue

       # cut out the payload from the JSON line w/o json module overhead

       decoded=base64.b64decode(line[10:-3]) # "},

       # interpret the payload as an Apache Arrow IPC stream

       reader=pa.ipc.open_stream(zlib.decompress(decoded) if compressed_transfer else decoded)

       # accumulate RecordBatch objects in a list

       batches.extend([batch for batch in reader])

   print('')

   if batches:

       # create an Arrow Table view on the batches and convert (e.g.) to a pandas DataFrame

       df=pa.Table.from_batches(batches).to_pandas()

       if 'muts' in df and '_seq' in df:

           # by default, the data is sorted by symbol... re-sort by time

           df=df.sort_values(['muts', '_seq'])

           df.reset_index(drop=True,inplace=True)

       print(df)