清算ヒートマップを自炊する方法

2024年1月12日 14:09

coinglassとかでよくある清算ヒートマップを自炊する方法。

注意点として、清算マップは「確実に清算対象となるオーダーが溜まっている場所」ではないことです。
ある時間帯の価格とOI変化とを照らし合わせて、あくまで目安として可視化を行っているだけです。
なので今回紹介するものはまったく厳密なものではなく、1分足のmid_priceと1分ごとのOI変化を利用してざっくり作ったものです。

コードは以下。
完璧なものではないので、あくまで参考にしてください。
クラス設計もきちんとできていないチンパンコードなので、変なところに流用しないでほしいです。
また、JupyterLabのコードを貼り付けただけなので多分そのまま動きません。importとか抜け漏れあるかも。使いたい人はがんばって動かしてください。

ロジックは以下です。

1分足のhighとlowからmid_priceを計算。
そこから+5%と-5%の価格帯（レバ20倍想定）に、OI増減分を加える。
これを24時間分とか繰り返す。
で、ヒートマップで可視化。

あとは自分好みにカスタマイズしてください。いろんな改良点があると思います。で、裁量に使うなりボットに使うなりしてください。

import numba
import pandas
import numpy as np
import matplotlib.pyplot as plt


@numba.njit
def id_to_price(id, tick, min_price, digit):
    return round(id * tick + min_price, digit)

@numba.njit
def price_to_id(price, min_price, tick):
    return round((price - min_price) / tick)

@numba.njit
def round_njit(a, b):
    return round(a, b)


class LiquidationMap:

    def __init__(self, data, tick, digit, liquidate_percentage=[0.05], with_mid_price=False):
        self.data = data  # DataFrame. 必要カラムは以下.
        self.idx_open_price = list(self.data.columns).index('open_price')
        self.idx_high_price = list(self.data.columns).index('high_price')
        self.idx_low_price = list(self.data.columns).index('low_price')
        self.idx_close_price = list(self.data.columns).index('close_price')
        self.idx_oi_usdt_diff = list(self.data.columns).index('open_interest_usdt_diff')

        self.tick = tick  # OI溜まりを計算する各価格帯の幅.
        self.digit = digit  # その小数点桁数.
        self.liquidate_percentage = liquidate_percentage  # 清算が発生するとされるポイント. 0.05ならレバ20倍の想定で、OI増減箇所*1.05と*0.95のポイントに
        self.with_mid_price = with_mid_price  # 現在価格を描画するかのフラグ
        self.min_price = round(data['low_price'].min() * 0.90, digit)  # 描画する最小価格帯
        self.max_price = round(data['high_price'].max() * 1.10, digit)  # 描画する最大価格帯
        
        self.liq_map = np.zeros((len(self.data), self.price_to_id(self.max_price) + 1))  # 各価格帯のOI溜まりを保持する配列 = 清算MAP

    def id_to_price(self, id):
        return id_to_price(id, self.tick, self.min_price, self.digit)

    def price_to_id(self, price):
        return price_to_id(price, self.min_price, self.tick)

    def calc(self):
        values = self.data.values
        self.last_mid_price = None
        for idx in range(len(values)):
            # 今回の中央価格を計算、そこから清算価格を計算
            mid_price = (values[idx][self.idx_high_price] + values[idx][self.idx_low_price]) / 2
            short_liq_prices = [mid_price * (1+liquidate_percentage) for liquidate_percentage in self.liquidate_percentage]
            long_liq_prices = [mid_price * (1-liquidate_percentage) for liquidate_percentage in self.liquidate_percentage]
            if self.last_mid_price is None:
                self.last_mid_price = mid_price
            # 前回からの清算マップを引き継ぎ
            if idx > 0:
                self.liq_map[idx] = self.liq_map[idx-1][:]
            # 価格が横切った場所を0に戻す
            self.last_mid_price_idx = self.price_to_id(self.last_mid_price)
            high_price_idx = self.price_to_id(values[idx][self.idx_high_price])
            low_price_idx = self.price_to_id(values[idx][self.idx_low_price])
            from_id = min(self.last_mid_price_idx, low_price_idx)
            to_id = max(self.last_mid_price_idx, high_price_idx)
            for idx2 in range(from_id, to_id + 1):
                self.liq_map[idx][idx2] = 0
            # 5%下に精算追加
            for long_liq_price in long_liq_prices:
                self.liq_map[idx][self.price_to_id(long_liq_price)] += values[idx][self.idx_oi_usdt_diff] / 1000 / len(long_liq_prices)
                self.liq_map[idx][self.price_to_id(long_liq_price)] = max(self.liq_map[idx][self.price_to_id(long_liq_price)], 0)
            # 5%上に精算追加
            for short_liq_price in short_liq_prices:
                self.liq_map[idx][self.price_to_id(short_liq_price)] += values[idx][self.idx_oi_usdt_diff] / 1000 / len(short_liq_prices)
                self.liq_map[idx][self.price_to_id(short_liq_price)] = max(self.liq_map[idx][self.price_to_id(short_liq_price)], 0)
            # 前回mid_priceにコピー
            self.last_mid_price = mid_price
        if self.with_mid_price:
            for idx in range(len(values)):
                # 今回の中央価格を計算、そこから清算価格を計算
                mid_price = (values[idx][self.idx_high_price] + values[idx][self.idx_low_price]) / 2
                mid_price_idx = self.price_to_id(mid_price)
                self.liq_map[idx][mid_price_idx] = 10000

    def plot(self, from_id, to_id):
        # ヒートマップの作成
        plt.figure(figsize=(25, 5))
        plt.imshow(self.liq_map[from_id:to_id, :].T, interpolation='nearest', origin='lower')
        plt.colorbar()
        plt.show()

    def print_liq_map_at(self, idx):
        print(get_datestr(self.data.iloc[idx]['open_timestamp']))
        for i in range(len(self.liq_map[idx])):
            print(self.id_to_price(i), self.liq_map[idx][i])


map = LiquidationMap(df, 0.1, 1, with_mid_price=True)
map.calc()
map.plot(0, -1)