Fetch Polarity Digital Data

fetch_pd_data.py

"""
This script fetches cryptocurrency data from the Polarity Digital API. It retrieves historical and 
additional data for a list of cryptocurrencies and saves them as JSON files.

The script uses the requests library to make API calls and argparse for 
command line argument parsing. 
Data is fetched for each coin and metric, then saved in files named with the 
format YYYYMMDD-{coin}-{metric}.json.

SETUP:
$ pip install requests tenacity

!!! THIS IS FOR PERSONAL USE ONLY, AND REQUIRES PROPER AUTHENTICATION 
(bearer token) TO WORK. DO NOT SHARE YOUR BEARER TOKEN WITH ANYONE. !!!

As with any API, please be mindful of rate limits and other restrictions.
"""

import argparse
import requests
import time
import random
import json
import concurrent.futures
from datetime import datetime
from typing import Dict, Any, Optional, List
from concurrent.futures import ThreadPoolExecutor
from tenacity import retry, stop_after_attempt, wait_fixed


@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def fetch_data_base(url: str, bearer_token: str) -> Dict[str, Any]:
    """
    Fetch raw data from pre-specified URL using a token.

    Args:
        url: URL to call
        bearer_token: token to use

    Returns:
      A dictionary containing the response data
    """
    try:
        headers = {
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "en-US,en;q=0.7",
            "Authorization": f"Bearer {bearer_token}",
            "Connection": "keep-alive",
            "Origin": "https://www.polaritydigital.io",
            "Referer": "https://www.polaritydigital.io/",
            "Sec-Fetch-Dest": "empty",
            "Sec-Fetch-Mode": "cors",
            "Sec-Fetch-Site": "same-site",
            "Sec-GPC": "1",
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            "sec-ch-ua": '"Not_A Brand";v="8", "Chromium";v="120", "Brave";v="120"',
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": '"macOS"',
        }
        response = requests.get(url, headers=headers)
        return response.json()
    except requests.RequestException as e:
        print(f"Failed to fetch data for {url}: {e}")
        print(f"\t{response}")
        print(f"\t{response.text}")
        return None


def fetch_data(coin: str, metric: str, bearer_token: str) -> Dict[str, Any]:
    """
    Fetch data from the proHistoricalData API for a given coin and metric.

    Args:
        coin: The cryptocurrency symbol (e.g., 'btc').
        metric: The specific metric to fetch (e.g., 'mbi').
        bearer_token: The authorization bearer token.

    Returns:
        A dictionary containing the response data.
    """
    url = f"https://api.polaritydigital.io/api/coin/proHistoricalData?coin={coin}&metric={metric}"
    data = fetch_data_base(url, bearer_token)
    return data


def fetch_additional_data(coin: str, metric: str, bearer_token: str) -> Dict[str, Any]:
    """
    Fetch additional data from the historicalData API for a given coin and metric.

    Args:
        coin: The cryptocurrency symbol (e.g., 'btc').
        metric: The specific metric to fetch (e.g., 'closeprice').
        bearer_token: The authorization bearer token.

    Returns:
        A dictionary containing the response data.
    """
    url = f"https://api.polaritydigital.io/api/coin/historicalData?coin={coin}&metric={metric}"
    data = fetch_data_base(url, bearer_token)
    return data


def save_data(data: Dict[str, Any], coin: str, metric: str) -> None:
    """
    Save the provided data as a JSON file.

    Args:
        data: The data to save.
        coin: The cryptocurrency symbol (e.g., 'btc').
        metric: The specific metric (e.g., 'mbi').

    Returns:
        None. The function saves the data to a file.
    """
    date_str = datetime.now().strftime("%Y%m%d")
    file_name = f"{date_str}-{coin}-{metric}.json"
    with open(file_name, "w") as file:
        json.dump(data, file)


def fetch_data_parallel(
    coin: str, metrics: List[str], bearer_token: str, fetch_function
) -> List[Dict[str, Any]]:
    """
    Fetch data for a given coin and list of metrics in parallel.

    Args:
        coin: The cryptocurrency symbol (e.g., 'btc').
        metrics: The list of metrics to fetch (e.g., ['mbi', 'mcm']).
        bearer_token: The authorization bearer token.
        fetch_function: The function to use to fetch data.

    Returns:
        A list of dictionaries containing the response data.
    """

    def fetch_task(metric):
        return fetch_function(coin, metric, bearer_token)

    with concurrent.futures.ThreadPoolExecutor() as executor:
        responses = list(executor.map(fetch_task, metrics))

    return responses


def main() -> None:
    """
    Main function to orchestrate data fetching and saving for various cryptocurrencies and metrics.

    The function fetches data for predefined coins and metrics, including additional metrics,
    and saves them as JSON files. It uses command line arguments for bearer token.

    Returns:
        None. The function orchestrates data fetching and saving.
    """
    parser = argparse.ArgumentParser(description="Fetch and save crypto data.")
    parser.add_argument("bearer_token", type=str, help="Bearer token for authorization")
    args = parser.parse_args()
    bearer_token = args.bearer_token

    #
    # Some constants for the script
    #

    sleep_min = 5
    sleep_max = 20

    coins = [
        "btc",
        "eth",
        "bnb",
        "sol",
        "xrp",
        "ada",
        "avax",
        "dot",
        "matic",
        "link",
        "ltc",
        "near",
        "atom",
        "xlm",
        "hbar",
        "etc",
        "xmr",
        "vet",
        "stx",
        "qnt",
        "mkr",
        "rndr",
        "algo",
        "egld",
        "ftm",
        "theta",
        "sand",
        "xtz",
        "hnt",
        "mana",
        "fet",
        "ar",
        "zil",
        "ksm",
        "enj",
        "lrc",
        "one",
    ]
    metrics = [
        "mdccv",
        "tcicv",
        "udpis",
        "udpim",
        "udpil",
        "mdc",
        "mcm",
        "mbi",
        "mtm",
        "upprob",
        "tci",
    ]
    additional_metrics = ["closeprice", "mc", "volume", "cs"]

    for i, coin in enumerate(coins, 1):
        cur_metrics = metrics.copy()
        if coin != "btc":
            # drop mcm and mtm for non btc
            cur_metrics.remove("mcm")
            cur_metrics.remove("mtm")

        print(f"[{i}/{len(coins)}] Fetching data for {coin} ...")

        # Fetch pro metrics in parallel
        pro_responses = fetch_data_parallel(coin, cur_metrics, bearer_token, fetch_data)

        # Fetch the additional metrics in parallel
        additional_responses = fetch_data_parallel(
            coin, additional_metrics, bearer_token, fetch_additional_data
        )

        # Save results
        for metric, response in zip(cur_metrics, pro_responses):
            save_data(response, coin, metric)

        for metric, response in zip(additional_metrics, additional_responses):
            save_data(response, coin, metric)

        print(
            f"[{i}/{len(coins)}] Done fetching data for {coin}. "
            f"Respectfully sleeping ..."
        )
        time.sleep(random.randint(sleep_min, sleep_max))


if __name__ == "__main__":
    main()