Plot Cypto Fear and Greed index with Python API

Fear and greed index with Python API

# STEP 1. get BTC price
from datetime import datetime
import pandas as pd
import requests
from typing import *
import time

class BinanceClient:
    def __init__(self, futures=False):
        self.exchange = "BINANCE"
        self.futures = futures

        if self.futures:
            self._base_url = "https://fapi.binance.com"
        else:
            self._base_url = "https://api.binance.com"

        self.symbols = self._get_symbols()

    def _make_request(self, endpoint: str, query_parameters: Dict):
        try:
            response = requests.get(self._base_url + endpoint, params=query_parameters)
        except Exception as e:
            print("Connection error while making request to %s: %s", endpoint, e)
            return None

        if response.status_code == 200:
            return response.json()
        else:
            print("Error while making request to %s: %s (status code = %s)",
                         endpoint, response.json(), response.status_code)
            return None

    def _get_symbols(self) -> List[str]:

        params = dict()

        endpoint = "/fapi/v1/exchangeInfo" if self.futures else "/api/v3/exchangeInfo"
        data = self._make_request(endpoint, params)

        symbols = [x["symbol"] for x in data["symbols"]]

        return symbols

    def get_historical_data(self, symbol: str, interval: Optional[str] = "1m", start_time: Optional[int] = None, end_time: Optional[int] = None, limit: Optional[int] = 1500):

        params = dict()

        params["symbol"] = symbol
        params["interval"] = interval
        params["limit"] = limit

        if start_time is not None:
            params["startTime"] = start_time
        if end_time is not None:
            params["endTime"] = end_time

        endpoint = "/fapi/v1/klines" if self.futures else "/api/v3/klines"
        raw_candles = self._make_request(endpoint, params)

        candles = []

        if raw_candles is not None:
            for c in raw_candles:
                candles.append((float(c[0]), float(c[1]), float(c[2]), float(c[3]), float(c[4]), float(c[5]),))
            return candles
        else:
            return None

def ms_to_dt_utc(ms: int) -> datetime:
    return datetime.utcfromtimestamp(ms / 1000)

def ms_to_dt_local(ms: int) -> datetime:
    return datetime.fromtimestamp(ms / 1000)

def GetDataFrame(data):
    df = pd.DataFrame(data, columns=['Timestamp', "Open", "High", "Low", "Close", "Volume"])
    df["Timestamp"] = df["Timestamp"].apply(lambda x: ms_to_dt_local(x))
#     df['Date'] = df["Timestamp"].dt.strftime("%d/%m/%Y")
    df['Date'] = df["Timestamp"].dt.strftime("%Y-%m-%d")
    df['Time'] = df["Timestamp"].dt.strftime("%H:%M:%S")
    column_names = ["Date", "Time", "Open", "High", "Low", "Close", "Volume"]
    df = df.set_index('Timestamp')
    df = df.reindex(columns=column_names)

    return df

def GetHistoricalData(client, symbol, start_time, end_time, limit=1500, interval="1m"):
    collection = []

    while start_time < end_time:
        data = client.get_historical_data(symbol, start_time=start_time, end_time=end_time, limit=limit, interval=interval)
        if len(data)!=0:
            print(client.exchange + " " + symbol + " : Collected " + str(len(data)) + " initial data from "+ str(ms_to_dt_local(data[0][0])) +" to " + str(ms_to_dt_local(data[-1][0])))
            start_time = int(data[-1][0] + 1000)
            collection +=data
            time.sleep(1.1)
        else:
            break

    return collection
    
client = BinanceClient(futures=False)
symbol = "BTCUSDT"
interval = "1d"
fromDate = int(datetime.strptime('2018-02-01', '%Y-%m-%d').timestamp() * 1000)
toDate = int(datetime.strptime('2022-06-01', '%Y-%m-%d').timestamp() * 1000)

data = GetHistoricalData(client, symbol, fromDate, toDate, interval=interval)
df = GetDataFrame(data)
# df

# STEP 2. get fear and greed index
import requests 
import pandas as pd 
import numpy as np
r = requests.get("https://api.alternative.me/fng/?limit=0") 
df2 = pd.DataFrame(r.json()["data"])
df2.value = df2.value.astype(int)
df2.timestamp = pd.to_datetime(df2.timestamp, unit="s")
df2.set_index("timestamp", inplace=True)
df2 = df2[::-1]
# plot and show df
# df2.plot(figsize=(15,10))
# df2

# STEP 3. merge df and df2
df3 = df[["Date", "Close"]]
df3.set_index("Date", inplace=True)
df3.index.names = ['Date']
# print(df3)
df4 = df2[["value"]]
# print(df4)
df5 = df4.join(df3)
# print(df5)
df5 = df5.drop(df5.index[0])
# df5

# STEP 4. plot
import matplotlib.pyplot as plt

f, ax = plt.subplots(figsize=(16, 5))

heatmap = ax.scatter(df5.index, df5.Close, c=df5.value)
ax.set_yscale("log")
ax.set_ylabel("USD")
ax.set_title("BTC/USDT colored by Fear&Greed (Binance)")

cbar = plt.colorbar(heatmap)
cbar.ax.get_yaxis().labelpad = 15
cbar.ax.set_ylabel('Fear & Greed', rotation=270)
plt.savefig(fname="FaG.png", format="png", dpi=500)
plt.show()