Python script for fetching multiple feed data from chainlink's data streams api

chainlink_data_streams.py

import hmac
import hashlib
import binascii
import time
from dataclasses import dataclass
from urllib.parse import urlparse, urlunparse, urljoin

import requests
from eth_abi import decode
from eth_utils import to_hex

API_URL = 'https://api.testnet-dataengine.chain.link'
# replace <CHAINLINK_CLIENT_ID>
CLIENT_ID = '<CHAINLINK_CLIENT_ID>'
# replace <CHAINLINK_SECRET>
USER_SECRET = '<CHAINLINK_SECRET>'

MEEM_FEED_IDS = {
    'DOGE': '0x00032057c7f224d0266b4311a81cdc3e38145e36442713350d3300fb12e85c99',
    'SHIB': '0x0003e5dfcb9c8fda20285c89ab6b289926b17dde5e704a1c228a8b468913f56a',
    'PEPE': '0x0003c8541bf8517752d466f6efc2057b54a9e189e45f08fa37a4db73452df4d4',
    'FLOKI': '0x0003e35ba9242261003459b8c97933127cdff9b727b9492ab017de9484fdc6b9',
    'ETH': '0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782',
}



@dataclass
class FeedData:
    feed_id: str
    valid_from_timestamp: int
    observations_timestamp: int
    native_fee: int
    link_fee: int
    expires_at: int
    benchmark_price: int
    bid: int
    ask: int


def extract_full_path_with_params(url):
    # Parse the URL
    parsed_url = urlparse(url)

    # Reconstruct the URL with all components
    full_path_with_params = urlunparse((
        '',  # scheme (empty for path only)
        '',  # netloc (empty for path only)
        parsed_url.path,  # path
        parsed_url.params,  # params
        parsed_url.query,  # query
        parsed_url.fragment  # fragment
    ))

    return full_path_with_params


def create_url(api_url, path, params):
    return requests.Request(
        'GET',
        urljoin(api_url, path),
        params=params
    ).prepare().url


def generate_hmac(method: str, path: str, body: bytes, client_id: str, timestamp: int, user_secret: str) -> str:
    # Calculate the SHA256 hash of the body
    body_hash = hashlib.sha256(body).digest()
    body_hash_hex = binascii.hexlify(body_hash).decode('utf-8')

    # Construct the string to be signed
    server_body_hash_string = f"{method} {path} {body_hash_hex} {client_id} {timestamp}"

    # Create HMAC signature
    signed_message = hmac.new(user_secret.encode(), server_body_hash_string.encode(), hashlib.sha256)
    user_hmac = binascii.hexlify(signed_message.digest()).decode('utf-8')

    return user_hmac


def generate_auth_headers(
        method: str,
        path: str,
        body: bytes,
        client_id: str,
        user_secret: str,
        timestamp: int
) -> dict:
    headers = dict()
    # Generate HMAC signature
    hmac_string = generate_hmac(method, path, body, client_id, timestamp, user_secret)

    # Add headers
    headers['Authorization'] = client_id
    headers['X-Authorization-Timestamp'] = str(timestamp)
    headers['X-Authorization-Signature-Sha256'] = hmac_string
    return headers


def prepare_bulk_url(feed_ids: list, timestamp):
    path = "/api/v1/reports/bulk"
    return create_url(
        API_URL,
        path,
        dict(
            feedIDs=','.join(feed_ids),
            timestamp=int(timestamp/1000))
        )


def decode_feed_data(data):
    feed_data = []
    for report in data['reports']:
        intermediate_decoded_data = decode(
            ["bytes32[3]", "bytes", "bytes32[]", "bytes32[]", "bytes32"],
            bytes.fromhex(report['fullReport'][2:])
        )

        decoded_data = decode(
            ["bytes32", "uint32", "uint32", "uint192", "uint192", "uint32", "int192", "int192", "int192"],
            intermediate_decoded_data[1]
        )
        feed_data.append(FeedData(
            feed_id=to_hex(decoded_data[0]),
            valid_from_timestamp=decoded_data[1],
            observations_timestamp=decoded_data[2],
            native_fee=decoded_data[3],
            link_fee=decoded_data[4],
            expires_at=decoded_data[5],
            benchmark_price=decoded_data[6],
            bid=decoded_data[7],
            ask=decoded_data[8],
        ))
    return feed_data


def fetch_latest_meem_index_feeds():
    method = "GET"
    body = b""
    timestamp = int(time.time() * 1000)
    request_url = prepare_bulk_url(
        [MEEM_FEED_IDS['DOGE'], MEEM_FEED_IDS['SHIB'], MEEM_FEED_IDS['PEPE'], MEEM_FEED_IDS['FLOKI'], MEEM_FEED_IDS['ETH']],
        timestamp
    )
    path = extract_full_path_with_params(request_url)
    response = requests.get(
        request_url,
        headers=generate_auth_headers(
            method,
            path,
            body,
            CLIENT_ID,
            USER_SECRET,
            timestamp
        )
    )
    response.raise_for_status()
    data = response.json()
    return decode_feed_data(data), response

def main():
    while True:
        print('requesting feed data...')
        feed_data, response = fetch_latest_meem_index_feeds()
        if len(feed_data) != 5:
            request = response.request
            print(f'Oh no.. received only {len(feed_data)} feeds, expected 5')
            print('Below are the details of the request')
            print(f'request url: {request.url}')
            print(f'request method: {request.method}')
            print(f'request headers: {request.headers}')
            print(f'request body: {request.body}')
            return
        print('received correct feed data...')
        time.sleep(1)

if __name__ == "__main__":
    main()

replace <CHAINLINK_CLIENT_ID> and <CHAINLINK_SECRET> with appropriate secrets acquired from chainlink.