🔥
【ハイリキ】Google ColaboratoryからHyperliquid Python SDKを軽く触ってみたときのメモ 【info編】
ChromeブラウザベースからHyperliquidのAPI叩いて結果得るときにやったことのメモ
LLMにendpointのドキュメントをPDFで食わせてからサンプルコード出力させてみました
Google Colabを開く
右上のNew Notebookから新しいNotebookを作る
SDKを導入する
公式ドキュメント
- Github
- Hyperliquid Docs API Info endpoint
code欄でSDKをインストール
pip install hyperliquid-python-sdk
動作テスト
from hyperliquid.info import Info
from hyperliquid.utils import constants
info = Info(constants.MAINNET_API_URL, skip_ws=True)
user_state = info.user_state("0xfae95f601f3a25ace60d19dbb929f2a5c57e3571")
print(user_state)
Hyperliquid APIで遊んでみる
初期化コード
from hyperliquid.info import Info
from hyperliquid.utils import constants
from typing import Union, Dict, Optional
# API URLを指定（デフォルトはMAINNET）
API_URL = constants.MAINNET_API_URL
# Infoクラスのインスタンスを初期化
info = Info(API_URL, skip_ws=True)
トークン別の現在価格の抽出
import pandas as pd
def format_mids(data):
"""
レスポンスデータを整形し、Pandas DataFrameに変換。
ミッドプライスを基準に降順にソートします。
"""
# データを整形してPandas DataFrameに変換
df = pd.DataFrame(list(data.items()), columns=["Coin", "Mid Price"])
df["Mid Price"] = pd.to_numeric(df["Mid Price"], errors="coerce")
# Mid Price の降順でソート
df = df.sort_values(by="Mid Price", ascending=False)
return df
# 実行例
mids_data = get_all_mids()
formatted_data = format_mids(mids_data)
# Jupyter NotebookやGoogle Colabで直接表示
formatted_data # これにより、Notebook環境で綺麗に表示されます
# またはすべてのデータを文字列として表示
print(formatted_data.to_string())
出力結果
Coin Mid Price
158 BTC 97780.500000
176 ETH 3592.950000
235 PANDORA 3026.300000
217 MKR 1580.000000
153 BNB 715.905000
269 TAO 565.490000
147 BCH 472.825000
特定アドレスのオープンオーダーの取得
def get_open_orders(user_address):
"""
ユーザーのオープンオーダーを取得。
"""
payload = {
"type": "openOrders",
"user": user_address
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
user_address = "0x000000000000000000" # サンプルアドレス
open_orders = get_open_orders(user_address)
print(open_orders)
出力結果
[{'coin': 'ZEREBRO', 'side': 'A', 'limitPx': '0.54966', 'sz': '7474.0', 'oid': 60602338066, 'timestamp': 1735986831152, 'origSz': '7474.0'},...
特定アドレスの約定情報
def get_user_fills(user_address):
"""
ユーザーの約定情報を取得。
"""
payload = {
"type": "userFills",
"user": user_address
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
user_address = "0x0000000000000000000000000000000000000000" # サンプルアドレス
user_fills = get_user_fills(user_address)
print(user_fills)
出力結果
{'coin': '@8', 'px': '0.0055895', 'sz': '5334.0', 'side': 'A', 'time': 1717356003754, 'startPosition': '17841691.0', 'dir': 'Sell', 'closedPnl': '0.0', 'hash': '0x65413d4556264878fc7d040a8782770142002e2955fb842c80476da174971115', 'oid': 24748117580, 'crossed': True, 'fee': '0.010435', 'tid': 995089357498554, 'feeToken': 'USDC'}
APIリクエストの制限情報
def get_user_rate_limits(user_address):
"""
ユーザーのリクエスト制限情報を取得。
"""
payload = {
"type": "userRateLimit",
"user": user_address
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
user_address = "0x0000000000000000000000000000000000000000" # サンプルアドレス
rate_limits = get_user_rate_limits(user_address)
print(rate_limits)
出力結果
{'cumVlm': '6000.47489', 'nRequestsUsed': 18, 'nRequestsCap': 15490}
ローソク足の取得
def get_candle_snapshot(coin, interval, start_time, end_time=None):
"""
キャンドルデータを取得。
"""
payload = {
"type": "candleSnapshot",
"req": {
"coin": coin,
"interval": interval,
"startTime": start_time,
"endTime": end_time
}
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
coin = "BTC"
interval = "15m"
start_time = 1681923600000 # サンプルの開始時間 (エポックミリ秒)
candle_data = get_candle_snapshot(coin, interval, start_time)
print(candle_data)
出力結果
{'t': 1731484800000, 'T': 1731485699999, 's': 'BTC', 'i': '15m', 'o': '87424.0', 'c': '87372.0', 'h': '87480.0', 'l': '87220.0', 'v': '40.61307', 'n': 646}
あるアドレスの過去の取引データ
def get_historical_orders(user_address):
"""
ユーザーの過去の注文履歴を取得。
"""
payload = {
"type": "historicalOrders",
"user": user_address
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
user_address = "0x0000000000000000000000000000000000000000" # サンプルアドレス
historical_orders = get_historical_orders(user_address)
print(historical_orders)
出力結果
{'order': {'coin': 'PURR/USDC', 'side': 'A', 'limitPx': '0.15038', 'sz': '15044.0', 'oid': 38697786123, 'timestamp': 1726989155780, 'triggerCondition': 'N/A', 'isTrigger': False, 'triggerPx': '0.0', 'children': [], 'isPositionTpsl': False, 'reduceOnly': False, 'orderType': 'Limit', 'origSz': '56443.0', 'tif': 'Gtc', 'cloid': None}
あるアドレスののVault預金
def get_vault_deposits(user_address):
"""
ユーザーのVault預金状況を取得。
"""
payload = {
"type": "userVaultEquities",
"user": user_address
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
user_address = "0x0000000000000000000000000000000000000000" # サンプルアドレス
vault_deposits = get_vault_deposits(user_address)
print(vault_deposits)
出力結果
[{'vaultAddress': '0xdfc24b077bc1425ad1dea75bcb6f8158e10df303', 'equity': '25109277.276554'}]
指定したトークンのL2オーダーブックスナップショット
def get_l2_book_snapshot(coin):
"""
指定したコインのL2オーダーブックスナップショットを取得。
"""
payload = {
"type": "l2Book",
"coin": coin
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
l2_book_snapshot = get_l2_book_snapshot("BTC")
print(l2_book_snapshot)
出力結果
{'coin': 'BTC', 'time': 1735995288943, 'levels': [[{'px': '97894.0', 'sz': '7.89457', 'n': 3}, {'px': '97893.0', 'sz': '0.15328', 'n': 1}, {'px': '97892.0', 'sz': '0.15328', 'n': 1}
指定した時間範囲でユーザーの約定情報
def get_fills_by_time(user_address, start_time, end_time=None):
"""
指定した時間範囲でユーザーの約定情報を取得。
"""
payload = {
"type": "userFillsByTime",
"user": user_address,
"startTime": start_time,
"endTime": end_time
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
user_address = "0x0000000000000000000000000000000000000000" # ユーザーアドレス
start_time = 1681923600000 # エポックミリ秒形式
end_time = 1735593589085 # エポックミリ秒形式
fills_by_time = get_fills_by_time(user_address, start_time, end_time)
print(fills_by_time)
出力結果
{'coin': '@8', 'px': '0.0055895', 'sz': '5334.0', 'side': 'A', 'time': 1717356003754, 'startPosition': '17841691.0', 'dir': 'Sell', 'closedPnl': '0.0', 'hash': '0x65413d4556264878fc7d040a8782770142002e2955fb842c80476da174971115', 'oid': 24748117580, 'crossed': True, 'fee': '0.010435', 'tid': 995089357498554, 'feeToken': 'USDC'}
あるアドレスのTWAPスライスフィル情報
TWAP（Time-Weighted Average Price）スライスフィルは、一定の時間にわたって平均価格で取引を行うために使用される注文手法（アルゴリズムトレード）の一部
def get_user_twap_slice_fills(user_address):
"""
ユーザーのTWAPスライスフィル情報を取得。
"""
payload = {
"type": "userTwapSliceFills",
"user": user_address
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
user_address = "0x0000000000000000000000000000000000000000" # ユーザーアドレス
twap_slice_fills = get_user_twap_slice_fills(user_address)
print(twap_slice_fills)
出力結果
{'fill': {'coin': 'PURR/USDC', 'px': '0.18204', 'sz': '329.0', 'side': 'A', 'time': 1717621498022, 'startPosition': '16982710.0', 'dir': 'Sell', 'closedPnl': '0.0', 'hash': '0x0000000000000000000000000000000000000000000000000000000000000000', 'oid': 25092136340, 'crossed': True, 'fee': '0.020961', 'tid': 527113877452908, 'feeToken': 'USDC'}, 'twapId': 79230}
Vaultの詳細情報を取得
def get_vault_details(vault_address, user_address=None):
"""
Vaultの詳細情報を取得。
"""
payload = {
"type": "vaultDetails",
"vaultAddress": vault_address
}
if user_address:
payload["user"] = user_address # 必要ならユーザーアドレスを追加
headers = {
"Content-Type": "application/json"
}
response = requests.post(f"{API_URL}/info", json=payload, headers=headers)
response.raise_for_status()
return response.json()
# 実行例
vault_address = "0xdfc24b077bc1425ad1dea75bcb6f8158e10df303" # Vaultアドレス
vault_details = get_vault_details(vault_address)
print(vault_details)
出力結果
{'name': 'Hyperliquidity Provider (HLP)', 'vaultAddress': '0xdfc24b077bc1425ad1dea75bcb6f8158e10df303', 'leader': '0x677d831aef5328190852e24f13c46cac05f984e7' 省略
グラフ作ってみる
ローソク足のグラフ表示
pip install
!pip install plotly
実装コード
#========================================================
# 1) ローソク足データを取得するための関数
#========================================================
import requests
import datetime
from typing import Optional, List, Dict
def get_btc_candle_data(
interval: str = "15m",
start_time_ms: Optional[int] = None,
end_time_ms: Optional[int] = None
) -> List[Dict]:
"""
HyperliquidのAPIからビットコイン（BTC）のローソク足データを取得する関数。
取得可能なローソク足は直近5000本まで。
"""
if end_time_ms is None:
end_time_ms = int(datetime.datetime.now().timestamp() * 1000)
if start_time_ms is None:
start_time_ms = end_time_ms - 24 * 60 * 60 * 1000 # 例: 24時間前
url = "https://api.hyperliquid.xyz/info"
headers = { "Content-Type": "application/json" }
payload = {
"type": "candleSnapshot",
"req": {
"coin": "BTC",
"interval": interval,
"startTime": start_time_ms,
"endTime": end_time_ms
}
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
data = response.json()
# 取得データは配列の可能性が高い
if isinstance(data, list):
return data
else:
print("想定していないレスポンス形式です。レスポンス:", data)
return []
except requests.exceptions.RequestException as e:
print(f"APIリクエストでエラーが発生しました: {e}")
return []
#========================================================
# 2) DataFrame・Plotlyを使ってローソク足を描画する
#========================================================
import pandas as pd
import plotly.graph_objects as go
def plot_candlestick(candle_data: List[Dict]):
"""
取得したローソク足データを用いて、Plotlyでローソク足チャートを描画する関数。
"""
if not candle_data:
print("ローソク足データが空です。")
return
# candle_data中の1レコードは以下のようなフォーマット
# {
# "t": 1735816500000, # 開始時刻 (Unixミリ秒)
# "T": 1735817399999, # 終了時刻 (Unixミリ秒)
# "s": "BTC", # シンボル
# "i": "15m", # 足の区切り (例: 15分足)
# "o": "96824.0", # 始値
# "c": "96622.0", # 終値
# "h": "96824.0", # 高値
# "l": "96497.0", # 安値
# "v": "283.6426", # 出来高
# "n": 1406 # (トレード数など？)
# }
# 必要に応じて float に変換
df = pd.DataFrame(candle_data)
# 数値へ変換
df['o'] = df['o'].astype(float)
df['h'] = df['h'].astype(float)
df['l'] = df['l'].astype(float)
df['c'] = df['c'].astype(float)
# ローソク足の時系列データとして、開始時刻 t を datetime に変換
df['t'] = pd.to_datetime(df['t'], unit='ms') # エポックミリ秒 -> 日時
# Plotlyのローソク足データ
fig = go.Figure(
data=[
go.Candlestick(
x=df['t'],
open=df['o'],
high=df['h'],
low=df['l'],
close=df['c'],
name='BTC 15m'
)
]
)
fig.update_layout(
title="BTC 15m Candlestick (Hyperliquid)",
xaxis_title="Time",
yaxis_title="Price (USD?)",
xaxis_rangeslider_visible=False, # 下のレンジスライダーを消したい場合は False
template="plotly_white"
)
fig.show()
#========================================================
# 3) Google Colabで実行した場合の一連の流れ
#========================================================
if __name__ == "__main__":
# Colabの場合、最初に「!pip install plotly」してから実行
btc_candles = get_btc_candle_data(interval="15m")
if btc_candles:
print(f"取得したBTCのローソク足データ: {len(btc_candles)}本")
print("例: 1本目のデータ:", btc_candles[0])
# 描画
plot_candlestick(btc_candles)
else:
print("BTCのローソク足データが取得できませんでした。")
表示例
移動平均線の描写
pip install
!pip install plotly pandas requests
実装コード
import requests
import datetime
from typing import Optional, List, Dict
import pandas as pd
import plotly.graph_objects as go
#========================================================
# 1) Hyperliquid APIからBTCのローソク足データを取得する関数
#========================================================
def get_btc_candle_data(
interval: str = "15m",
start_time_ms: Optional[int] = None,
end_time_ms: Optional[int] = None
) -> List[Dict]:
"""
HyperliquidのAPIからビットコイン（BTC）のローソク足データを取得する関数。
取得可能なローソク足は直近5000本まで。
"""
if end_time_ms is None:
end_time_ms = int(datetime.datetime.now().timestamp() * 1000)
if start_time_ms is None:
start_time_ms = end_time_ms - 24 * 60 * 60 * 1000 # 例: 24時間前
url = "https://api.hyperliquid.xyz/info"
headers = { "Content-Type": "application/json" }
payload = {
"type": "candleSnapshot",
"req": {
"coin": "BTC",
"interval": interval,
"startTime": start_time_ms,
"endTime": end_time_ms
}
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
data = response.json()
# 取得データは配列形式
if isinstance(data, list):
return data
else:
print("想定していないレスポンス形式です。レスポンス:", data)
return []
except requests.exceptions.RequestException as e:
print(f"APIリクエストでエラーが発生しました: {e}")
return []
#========================================================
# 2) Plotlyでローソク足 + 移動平均を描画する関数
#========================================================
def plot_candlestick_with_ma(
candle_data: List[Dict],
short_window: int = 7,
long_window: int = 30
):
"""
ローソク足データに、2つの移動平均（短期・長期）を重ねたチャートを作成します。
Args:
candle_data (List[Dict]): get_btc_candle_data() 等で取得したローソク足データ
short_window (int): 短期移動平均の期間
long_window (int): 長期移動平均の期間
"""
if not candle_data:
print("ローソク足データが空です。")
return
# DataFrameに変換
df = pd.DataFrame(candle_data)
# string -> float
df['o'] = df['o'].astype(float)
df['h'] = df['h'].astype(float)
df['l'] = df['l'].astype(float)
df['c'] = df['c'].astype(float)
# 時刻 t を datetime へ
df['t'] = pd.to_datetime(df['t'], unit='ms')
#========================================================
# 移動平均の計算
# - 短期移動平均: short_window で指定した期間 (例: 7足)
# - 長期移動平均: long_window で指定した期間 (例: 30足)
# - この例では「終値 c」で移動平均を計算します。
#========================================================
df['MA_short'] = df['c'].rolling(window=short_window).mean()
df['MA_long'] = df['c'].rolling(window=long_window).mean()
#========================================================
# Plotlyでローソク足チャートを作成
#========================================================
fig = go.Figure()
# --- ローソク足 ---
fig.add_trace(
go.Candlestick(
x=df['t'],
open=df['o'],
high=df['h'],
low=df['l'],
close=df['c'],
name='BTC Candle'
)
)
# --- 短期移動平均 ---
fig.add_trace(
go.Scatter(
x=df['t'],
y=df['MA_short'],
mode='lines',
line=dict(color='blue', width=2),
name=f'MA({short_window})'
)
)
# --- 長期移動平均 ---
fig.add_trace(
go.Scatter(
x=df['t'],
y=df['MA_long'],
mode='lines',
line=dict(color='red', width=2),
name=f'MA({long_window})'
)
)
# レイアウト調整
fig.update_layout(
title = "BTC Candlestick with Moving Averages",
xaxis_title = "Time",
yaxis_title = "Price",
xaxis_rangeslider_visible=False,
template="plotly_white"
)
# グラフ表示
fig.show()
#========================================================
# 3) 実行例 (Google Colab で動かす場合)
#========================================================
if __name__ == "__main__":
# 例: 直近24hの15分足を取得
btc_candles = get_btc_candle_data(interval="15m")
if btc_candles:
print(f"取得したBTCのローソク足データ: {len(btc_candles)}本")
print("例: 1本目のデータ:", btc_candles[0])
# 移動平均チャートを表示 (短期7, 長期30)
plot_candlestick_with_ma(btc_candles, short_window=7, long_window=30)
else:
print("BTCのローソク足データが取得できませんでした。")
表示グラフ
クロスオーバーでのバックテストと売買シグナルの表示
実装コード
import requests
import datetime
from typing import Optional, List, Dict
import pandas as pd
import plotly.graph_objects as go
#========================================================
# 1) Hyperliquid APIからBTCのローソク足データを取得する関数
# (すでに提供されていたもの)
#========================================================
def get_btc_candle_data(
interval: str = "15m",
start_time_ms: Optional[int] = None,
end_time_ms: Optional[int] = None
) -> List[Dict]:
"""
HyperliquidのAPIからビットコイン（BTC）のローソク足データを取得する関数。
取得可能なローソク足は直近5000本まで。
Args:
interval (str): ローソク足の時間足 (例: "1m", "5m", "15m", "1h", "4h", "1d", etc.)
start_time_ms (int): 取得開始時間（エポックミリ秒）。指定しない場合は現在時刻付近で取得。
end_time_ms (int): 取得終了時間（エポックミリ秒）。指定しない場合は現在時刻付近で取得。
Returns:
List[Dict]: ローソク足データのリスト。APIのレスポンスに準拠した構造のまま返す。
"""
if end_time_ms is None:
end_time_ms = int(datetime.datetime.now().timestamp() * 1000)
if start_time_ms is None:
# 例: 24時間前をstart_timeに設定
start_time_ms = end_time_ms - 24 * 60 * 60 * 1000
url = "https://api.hyperliquid.xyz/info"
headers = {"Content-Type": "application/json"}
payload = {
"type": "candleSnapshot",
"req": {
"coin": "BTC",
"interval": interval,
"startTime": start_time_ms,
"endTime": end_time_ms
}
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
data = response.json()
if isinstance(data, list):
return data
else:
print("想定していないレスポンス形式です。レスポンス:", data)
return []
except requests.exceptions.RequestException as e:
print(f"APIリクエストでエラーが発生しました: {e}")
return []
#========================================================
# 2) 移動平均クロスオーバー戦略 + 簡易バックテスト
#========================================================
def ma_crossover_backtest(
df: pd.DataFrame,
short_window: int = 7,
long_window: int = 25
) -> pd.DataFrame:
"""
短期移動平均と長期移動平均のクロスオーバーを用いた
バックテストのサンプル実装。
Args:
df (pd.DataFrame): ローソク足DataFrame（列名: 't','o','h','l','c'などを想定）
short_window (int): 短期移動平均の期間
long_window (int): 長期移動平均の期間
Returns:
pd.DataFrame: dfに以下の列を追加したもの
- MA_short: 短期移動平均
- MA_long: 長期移動平均
- Signal: 売買シグナル (1=BUY, -1=SELL, 0=何もしない)
- Position: ポジション (1=ロング, 0=ノーポジ, -1=ショート等)
"""
# 移動平均を計算 (終値ベース)
df['MA_short'] = df['c'].rolling(window=short_window).mean()
df['MA_long'] = df['c'].rolling(window=long_window).mean()
# シグナル列の初期化
df['Signal'] = 0
# ===== ゴールデンクロス (買いシグナル) =====
df.loc[
(df['MA_short'] > df['MA_long']) &
(df['MA_short'].shift(1) <= df['MA_long'].shift(1)),
'Signal'
] = 1
# ===== デッドクロス (売りシグナル) =====
df.loc[
(df['MA_short'] < df['MA_long']) &
(df['MA_short'].shift(1) >= df['MA_long'].shift(1)),
'Signal'
] = -1
# Position 列を作成 (1=ロング, 0=ノーポジ, -1=ショート等)
# ここでは簡単に、BUYシグナルが出たら1、SELLシグナルが出たら0に戻すとします。
df['Position'] = 0
current_pos = 0
for i in range(len(df)):
sig = df.iloc[i]['Signal']
if sig == 1:
# 買いシグナル => ロングに変更
current_pos = 1
elif sig == -1:
# 売りシグナル => ノーポジに変更
current_pos = 0
df.at[i, 'Position'] = current_pos
return df
#========================================================
# 3) Plotlyでバックテスト結果 (ローソク足 + 移動平均 + 売買シグナル) を可視化
#========================================================
def plot_backtest_results(df: pd.DataFrame):
"""
ローソク足チャートに短期・長期移動平均線、買い/売りシグナルを重ねて表示。
"""
# Plotly Figure
fig = go.Figure()
# ------ ローソク足 ------
fig.add_trace(
go.Candlestick(
x=df['t'],
open=df['o'],
high=df['h'],
low=df['l'],
close=df['c'],
name="BTC Candle"
)
)
# ------ 短期移動平均 ------
fig.add_trace(
go.Scatter(
x=df['t'],
y=df['MA_short'],
mode='lines',
line=dict(color='blue', width=2),
name=f"MA_short"
)
)
# ------ 長期移動平均 ------
fig.add_trace(
go.Scatter(
x=df['t'],
y=df['MA_long'],
mode='lines',
line=dict(color='red', width=2),
name=f"MA_long"
)
)
# ------ 買いシグナル (ゴールデンクロス) ------
buy_signals = df[df['Signal'] == 1]
fig.add_trace(
go.Scatter(
x=buy_signals['t'],
y=buy_signals['c'],
mode='markers',
marker=dict(symbol='triangle-up', color='green', size=12),
name="Buy Signal"
)
)
# ------ 売りシグナル (デッドクロス) ------
sell_signals = df[df['Signal'] == -1]
fig.add_trace(
go.Scatter(
x=sell_signals['t'],
y=sell_signals['c'],
mode='markers',
marker=dict(symbol='triangle-down', color='red', size=12),
name="Sell Signal"
)
)
# レイアウト設定
fig.update_layout(
title="BTC MA Crossover Backtest",
xaxis_title="Time",
yaxis_title="Price",
xaxis_rangeslider_visible=False,
template="plotly_white"
)
fig.show()
#========================================================
# 4) メイン処理 (バックテスト + 可視化)
#========================================================
if __name__ == "__main__":
# 例: 直近1日の15分足を取得
btc_candles = get_btc_candle_data(interval="15m")
if btc_candles:
print(f"取得したBTCのローソク足データ: {len(btc_candles)}本")
print("例: 1本目のデータ:", btc_candles[0])
# --- DataFrameへ変換 ---
df = pd.DataFrame(btc_candles)
# 数値に変換
df['o'] = df['o'].astype(float)
df['h'] = df['h'].astype(float)
df['l'] = df['l'].astype(float)
df['c'] = df['c'].astype(float)
# 時刻を datetime へ
df['t'] = pd.to_datetime(df['t'], unit='ms')
# --- バックテスト実行 (例: 短期=7, 長期=25) ---
df_bt = ma_crossover_backtest(df, short_window=7, long_window=25)
# --- 結果を可視化 ---
plot_backtest_results(df_bt)
# --- シグナルの出力例 ---
signals = df_bt[df_bt['Signal'] != 0].copy()
print("\n--- 売買シグナル一覧 ---")
for idx, row in signals.iterrows():
t_str = row['t'].strftime("%Y-%m-%d %H:%M")
if row['Signal'] == 1:
print(f"[{t_str}] BUY Signal : price={row['c']}")
else:
print(f"[{t_str}] SELL Signal: price={row['c']}")
else:
print("BTCのローソク足データが取得できませんでした。")
出力結果
--- 売買シグナル一覧 ---
[2025-01-02 22:00] SELL Signal: price=96989.0
[2025-01-03 02:45] BUY Signal : price=96960.0
[2025-01-03 04:30] SELL Signal: price=96879.0
[2025-01-03 11:00] BUY Signal : price=96502.0
書きながら使ってたライブラリーバージョンメモ
python = "^3.11"
ipykernel = "^6.17.0"
pymongo = "^4.3.2"
hyperliquid-python-sdk = "^0.9.0"
plotly = "^5.24.1"
pandas = "^2.2.3"
requests = "^2.32.3"
参考コード
Discussion