清算ヒートマップを自炊する方法
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)突っ込むデータは以下。
OHLCは取引所APIで取得してください。OIは、Binanceの場合はヒストリカルデータが5分足しか取れないので、1分ごとに保存するプログラムを別途書いています。
こんな感じになります。
じゃあの。
コメント