BybitのAPIを使ってFR（資金調達率）のグラフを自作する

Parameter	設定値
category	linear
symbol	BTCUSDT
startTime	設定しない
endTime	初回は未設定（最新が取得できる）。2回目以降は取得済みデータの最も古いタイムスタンプ。
limit	設定しない（デフォルトは200個）

FRヒストリカルデータ取得APIのパラメータ設定値

フローチャート

処理内容は、FR取得APIを取得期間を変えて複数回呼び、必要な期間のFRデータを揃えます。
フローチャートは以下のようになります。

FRデータ取得のサンプルコード

Pythonを使って、実際にFRのヒストリカルデータを取得してみます。

取得するデータの期間は３ヶ月とします。
API呼び出し１回で取得できるデータ数が200個です。
FRは１日に３回更新されるので、１ヶ月だと3✕31=93個なので、APIコールが１回で済んでしまいます。
サンプルコードのループ処理が正しく動くことを示したいので、期間は３ヶ月とします。

サンプルコードでは、必要な関数を作成して、メイン処理から呼び出すようにします。
作成する関数の説明とメイン処理の説明をしていきます。

FRヒストリカルデータ取得関数

まずは、データを取得する関数のコードです。

関数名はget_funding_rate_history()としました。
引数は、「仮想通貨ペアの名前」と「取得期間の終端のタイムスタンプ」です。

def get_funding_rate_history(symbol, end_time):
    endPoint = 'https://api.bybit.com'
    path     = '/v5/market/funding/history'
    url = endPoint + path

    params = {
        'category': 'linear',
        'symbol': symbol,
        'endTime': end_time,
    }

    response = requests.get(url, params=params)
    res = response.json()
    return res

FRヒストリカルデータをDataFrameにする関数

BybitのFRヒストリカルデータ取得APIのレスポンスパラメータは以下です。

実際には、このようなレスポンスが返ります。

{
    "retCode": 0,
    "retMsg": "OK",
    "result": {
        "category": "linear",
        "list": [
            {
                "symbol": "BTCUSDT",
                "fundingRate": "0.0001",
                "fundingRateTimestamp": "1729152000000"
            },
            {
                "symbol": "BTCUSDT",
                "fundingRate": "0.0001",
                "fundingRateTimestamp": "1729123200000"
            },
            {
                "symbol": "BTCUSDT",
                "fundingRate": "0.0001",
                "fundingRateTimestamp": "1729094400000"
            },
            {
                "symbol": "BTCUSDT",
                "fundingRate": "0.0001",
                "fundingRateTimestamp": "1729065600000"
            },
            {
                "symbol": "BTCUSDT",
                "fundingRate": "0.0001",
                "fundingRateTimestamp": "1729036800000"
            },
            {
                "symbol": "BTCUSDT",
                "fundingRate": "0.0001",
                "fundingRateTimestamp": "1729008000000"
            }
        ]
    },
    "retExtInfo": {},
    "time": 1729165018157
}

このレスポンスの情報を扱いやすいようにDataFrameに入れます。
具体的には、”list”部分のFR情報のみ抜き出します。
その関数がconvert_data()です。

def convert_data(data):
    # FRデータを入れるDataFrame。
    df = pd.DataFrame()

    # FRデータが入っているリストから情報を取り出して、DataFrameに入れる。
    for d in data['result']['list']:
        df_add = pd.DataFrame(d, index=[0])
        df = pd.concat([df, df_add], axis='index')
    df = df.reset_index(drop=True)

    # FRはfloat型に変換する。
    df['fundingRate'] = df['fundingRate'].astype('float64')

    # タイムスタンプはint型に変換する。
    df['fundingRateTimestamp'] = df['fundingRateTimestamp'].astype('int64')
    return df

メイン処理

上のフローチャートで示したメインのループ処理のコードです。

３ヶ月分のBTCUSDTのFRデータを取得したら、ループ処理を抜けてdf_fr.csvに保存しています。
ループ処理から、上で説明した関数の get_funding_rate_history() と convert_data() を呼んでいます。

import os
import sys
import requests
import json
import time
from datetime import datetime
import pandas as pd

from dateutil import tz
JST = tz.gettz('Asia/Tokyo')
UTC = tz.gettz('UTC')

# 現在のUTC時刻を取得しておく。
now_utc = datetime.now(UTC)

# 取得するFRの通貨ペアを設定する。
# 今回はBTCUSDTとする。
symbol = 'BTCUSDT'

# エンドタイムスタンプを初期化する。
end_time = None

# FRを入れるDataFrameを初期化する。
df_fr = pd.DataFrame()

# ここから、ループ文で3ヶ月分のFRを取得する。
while True:

    # FRを取得する
    # end_timeはwhile文の中で毎回更新される。
    data = get_funding_rate_history(symbol, end_time)

    # 取得したFRを日付をIndexとするDataFrameに変換する。
    df_add = convert_data(data)

    # FRを保存するDataFrameに追加する。
    df_fr = pd.concat([df_fr, df_add], axis='index')

    # タイムスタンプのUNIX時間でソートする。未来の時刻が下側になるようにソートする。
    df_fr = df_fr.sort_values(by=['fundingRateTimestamp'], ascending=True)

    # 重複行は先頭を残す。
    # タイムスタンプで重複を判断する。
    df_fr = df_fr.drop_duplicates(subset='fundingRateTimestamp', keep='first')

    # Indexの通し番号を再度付与する。
    df_fr = df_fr.reset_index(drop=True)

    # 取得済みのFRの最も古いタイムスタンプを取得し、次に取得するFRの最後のタイムスタンプとする。
    # 先頭が最も古いタイムスタンプとなる。
    end_time = df_fr.iat[0, df_fr.columns.get_loc('fundingRateTimestamp')]

    # 最も古いタイムスタンプをUTC時刻に変換する。
    end_time_utc = datetime.fromtimestamp(end_time/1000, UTC)

    # 3ヶ月分のFRを取得できたら、抜ける。
    if (now_utc - end_time_utc).days > 31 * 3:
        break

    # 短時間にAPIを何度も呼ばないように1秒間寝る。
    time.sleep(1)

# FR情報をファイル出力する。
filename = 'df_fr.csv'
file_path = os.path.join(work_path, filename)
df_fr.to_csv(file_path)

変数work_pathには、ファイルを保存するディレクトリを設定してください。

これで、FRデータを取得できました。

FRヒストリカルデータのグラフ作成方法

次に、取得したFRヒストリカルデータをグラフ化します。

サンプルコード

処理を見やすくするために、グラフ作成処理は関数化します。

def draw_bar(df):
    fig, ax = plt.subplots(1, 1, figsize=(9.6, 3.6))

    x = []
    for timestamp in df['fundingRateTimestamp']:
        # X軸はdatetime型に変換する。
        x.append(timestamp.to_pydatetime())
    print(type(x[0]))

    # Y軸の単位は％にするために、100倍する。
    y = list(df['fundingRate'] * 100)

    ax.bar(
        x, y,
        width=0.2,
        alpha=0.6,
        label='FR[%]',
        )

    ax.legend() # 凡例の表示
    ax.grid(axis='y', linestyle='--', color='red') # グリッドラインの表示
    fig.autofmt_xdate()
    plt.show()
    return

メインの処理です。
FRデータをファイルから読み込んで、グラフ化関数を呼びます。

# FR情報をファイルから読み込む。
filename = 'df_fr.csv'
file_path = os.path.join(work_path, filename)
df_fr = pd.read_csv(file_path, index_col=0)

# タイムスタンプがUNIX時間になっているので、UTC時刻に変換する。
df_fr['fundingRateTimestamp'] = df_fr['fundingRateTimestamp'].apply(lambda x:datetime.fromtimestamp(x/1000, UTC))

# グラフを見やすくするため、データ数を減らす。FR情報を1ヶ月分にする。
df_fr = df_fr[df_fr['fundingRateTimestamp'] > datetime.now(UTC) - timedelta(days=31)]

# グラフ化関数を呼ぶ。
draw_bar(df_fr)

・import部分は省略させていただきます。
・３ヶ月分のデータだとグラフが細かくて見づらかったので、１ヶ月分に減らしています。
・変数work_pathには、FRデータのファイルを保存したディレクトリを設定してください。