本記事は、仮想通貨botter Advent Calendar 2024の16日目の記事となります。
こんにちは、takeと申します。私は普段はサラリーマンとして働きつつ、トレーディング bot の開発や仮想通貨のトレーディング活動に没頭しています。前回の仮想通貨バブル時に勢いに乗ってsolanaのアビトラに参入したものの、年末の利確~資産の引き上げやバブル崩壊もあって仮想通貨からはしばらく離れていました。しかし、今年に入って、ふたたびバブルの兆しが見え始めたことや、yoshisoさんの記事の影響もあって「ML/DL」に興味を持ち始めました。
今年の頭までは、アクティブ・ポートフォリオ・マネジメントやファイナンス機械学習も読みながら真面目に試行錯誤していたもののなかなか結果が出ず、加えて今年はAIへの関心が高まりすぎてしまい、実際にMLで利益が出るところまでいきつけていません。
ということで、お役に立てるTipsなどが一切ないのですが、「AIを活用することで、トレーディング戦略を始めとした開発がどれほど楽になるのか」の実践レポくらいは書けそうということで今回の記事をまとめました。
ゴール
Hyperliquidでヒゲを感知して拾うbotを作成し、実際に動かせるようになる
縛り
- ChatGPTとCursorをフル活用
- 自分ではコードは一切書かない
- デバッグは全てAIに行わせる
- 制限時間は3時間
事前準備
トレードに必要となるbybitとHyperliquidのAPIkeyの取得やアカウントの設定、資金のDepositなどのお膳立ては自分で行いました。
自分のアカウント情報の取得
まずはじめに、Hyperliquidにアクセスしてアカウント情報を取得する部分からです。
ChatGPT4oで検索を駆使してコードを書いてもらいました。
1回目は
user_state = exchange.get_user_state()
などと存在しない関数を教えてくれましたが、何度かやり取りして下記のコードを書かせることに成功。きちんとアカウント情報を取得することに成功です。
最初からBybitとのサヤ取りを依頼しているのでbybitのapiも読み込んでいますが御愛嬌
import os
from dotenv import load_dotenv
import eth_account
from hyperliquid.exchange import Exchange
from hyperliquid.utils import constants
from hyperliquid.info import Info
import os
from dotenv import load_dotenv
import eth_account
from hyperliquid.exchange import Exchange
from hyperliquid.utils import constants
from hyperliquid.info import Info
# .envファイルの読み込み
load_dotenv()
# 環境変数の読み込みとバリデーション
def load_env_variables():
required_vars = {
'HYPERLIQUID_SECRET_KEY': None,
'BYBIT_API_KEY': None,
'BYBIT_SECRET': None
}
for var in required_vars:
value = os.getenv(var)
if value is None:
raise ValueError(f"環境変数 '{var}' が設定されていません。.envファイルを確認してください。")
required_vars[var] = value
return required_vars
# 環境変数の読み込み
env_vars = load_env_variables()
# アカウントの作成
account = eth_account.Account.from_key(env_vars['HYPERLIQUID_SECRET_KEY'])
# Exchangeオブジェクトの作成
exchange = Exchange(account, constants.MAINNET_API_URL)
# Infoオブジェクトの作成
info = Info(constants.MAINNET_API_URL, skip_ws=True)
print(info.user_state(account.address))
価格を取得
続いて、価格を取得するコードです。
これは1発で動くコードを書いてくれました。
from hyperliquid.info import Info
from hyperliquid.utils import constants
from typing import Union, Dict, Optional
def get_crypto_price(symbol: str = 'DOGE', api_url: str = constants.MAINNET_API_URL) -> Optional[float]:
"""
Hyperliquidから指定された暗号資産の現在の価格を取得します。
Args:
symbol (str): 価格を取得したい暗号資産のシンボル（例：'DOGE', 'BTC', 'ETH'）。
api_url (str): HyperliquidのAPI URL。デフォルトはメインネットのURL。
Returns:
float or None: 指定された暗号資産の現在の価格。取得できない場合はNone。
Raises:
Exception: API接続やデータ取得に失敗した場合。
Examples:
>>> doge_price = get_crypto_price('DOGE')
>>> if doge_price:
... print(f"DOGEの価格: {doge_price}")
>>> btc_price = get_crypto_price('BTC')
>>> if btc_price:
... print(f"BTCの価格: {btc_price}")
"""
try:
info = Info(api_url, skip_ws=True)
all_mids = info.all_mids()
return all_mids.get(symbol)
except Exception as e:
print(f"エラーが発生しました: {e}")
return None
def get_all_prices(api_url: str = constants.MAINNET_API_URL) -> Dict[str, float]:
"""
Hyperliquidから全ての暗号資産の現在の価格を取得します。
Args:
api_url (str): HyperliquidのAPI URL。デフォルトはメインネットのURL。
Returns:
Dict[str, float]: 全ての暗号資産のシンボルと価格のディクショナリ。
Raises:
Exception: API接続やデータ取得に失敗した場合。
Examples:
>>> prices = get_all_prices()
>>> for symbol, price in prices.items():
... print(f"{symbol}の価格: {price}")
"""
try:
info = Info(api_url, skip_ws=True)
return info.all_mids()
except Exception as e:
print(f"エラーが発生しました: {e}")
return {}
if __name__ == "__main__":
# 特定の暗号資産の価格を取得
symbol = 'kPEPE'
price = get_crypto_price(symbol)
print(f"{symbol}の現在の価格: {price}")
symbols = ['DOGE', 'BTC', 'ETH']
for symbol in symbols:
price = get_crypto_price(symbol)
if price:
print(f"{symbol}の現在の価格: {price}")
else:
print(f"{symbol}の価格情報が取得できませんでした。")
print("\n全ての暗号資産の価格:")
# 全ての価格を取得
all_prices = get_all_prices()
for symbol, price in all_prices.items():
print(f"{symbol}: {price}")
指値で注文を出す
続いて、注文です。
注文を出す部分が一番時間がかかりました。
Hyperliquidでは、サイズの桁数が多すぎるとエラーになってしまうので、その調整を行う必要があります。
通貨ごとの桁数の取得を行う必要がありますが、ChatGPTは存在しない関数を教えてくれるので、最終的には自分で関数を見つけてそれをAIに教えて実装する形としました。コードは書いていないのでチャレンジとしてはギリセーフですね。
ドキュメントをうまく読み込ませられればこのあたりも解決できるのかもしれません。
from hyperliquid.info import Info
from hyperliquid.exchange import Exchange
from hyperliquid.utils import constants
from typing import Optional
import eth_account
from eth_account.signers.local import LocalAccount
import json
import math
import os
from dotenv import load_dotenv
import sys
import requests
def adjust_order_size(size: float, price: float, symbol: str, sz_decimals: int, min_order_value: float = 10.0) -> float:
"""
注文サイズを最小注文金額に基づいて調整します。
Args:
size (float): 元の注文サイズ
price (float): 注文価格
symbol (str): 取引対象の暗号資産のシンボル
sz_decimals (int): サイズの小数点以下の桁数
min_order_value (float): 最小注文金額（デフォルト: $10.0）
Returns:
float: 調整後の注文サイズ
Raises:
ValueError: 調整後も最小注文金額を満たさない場合
"""
# サイズをszDecimalsの桁数で丸める
size = round(size, sz_decimals)
# 注文の合計金額を計算
total_value = size * price
if total_value < min_order_value:
required_size = math.ceil((min_order_value / price) * (10 ** sz_decimals)) / (10 ** sz_decimals)
print(f"警告: 注文金額が最小金額(${min_order_value})を下回っています。")
print(f"現在の注文金額: ${total_value:.2f}")
print(f"サイズを{required_size} {symbol}に調整して注文を実行します。")
size = required_size
total_value = size * price
if total_value < min_order_value:
raise ValueError(f"エラー: 調整後も最小注文金額(${min_order_value})を満たしていません。")
return size
def place_limit_order(exchange: Exchange, symbol: str, size: float, is_buy: bool = True, limit_price: Optional[float] = None, discount_rate: Optional[float] = 0.98):
"""
指定された暗号資産の指値注文を出します。
Args:
account (LocalAccount): eth_accountで生成したアカウントインスタンス
symbol (str): 取引対象の暗号資産のシンボル
size (float): 注文サイズ(例：kPEPEを550枚など)
is_buy (bool): 買い注文の場合True、売り注文の場合False
limit_price (Optional[float]): 指値価格。Noneの場合はdiscount_rateを使用して計算
discount_rate (Optional[float]): 現在価格に対する割引率（買い注文時）または上乗せ率（売り注文時）（デフォルト: 0.98）
Returns:
dict: 注文結果の詳細情報
Raises:
ValueError: 入力値が不正な場合
Exception: API通信エラーなど
"""
# market_info.jsonを読み込む
try:
with open('market_info.json', 'r') as f:
market_info = json.load(f)
# シンボルに対応するszDecimalsを取得
sz_decimals = None
for asset in market_info['universe']:
if asset['name'] == symbol:
sz_decimals = asset['szDecimals']
break
if sz_decimals is None:
print(f"{symbol}の情報がmarket_info.jsonに見つかりませんでした。")
return
except Exception as e:
print(f"market_info.jsonの読み込み中にエラーが発生しました: {e}")
return
# サイズを調整
size = adjust_order_size(size, limit_price, symbol, sz_decimals)
total_value = size * limit_price
print(f"注文情報:")
print(f"- 注文タイプ: {'買い' if is_buy else '売り'}")
print(f"- 指値価格: {limit_price}")
print(f"- 注文サイズ: {size} {symbol}")
print(f"- 注文金額: ${total_value:.2f}")
# 指値注文を送信
try:
order_result = exchange.order(
name=symbol,
is_buy=is_buy,
sz=size,
limit_px=limit_price,
order_type={"limit": {"tif": "Gtc"}}
)
# レスポンスの詳細な検証
if not isinstance(order_result, dict):
raise ValueError(f"予期しない応答形式です: {order_result}")
if order_result.get("status") != "ok":
raise ValueError(f"注文が失敗しました: {order_result}")
response_data = order_result.get('response', {}).get('data', {})
statuses = response_data.get('statuses', [])
if not statuses:
raise ValueError("注文ステータスが空です")
order_status = statuses[0]
# 注文の詳細情報を取得
order_info = {
'order_id': order_status.get('oid', 'Unknown'),
'status': 'resting' if 'resting' in order_status else 'unknown',
'symbol': symbol,
'size': size,
'price': limit_price,
'total_value': total_value,
'raw_response': order_status
}
print(f"\n注文実行結果:")
print(f"注文ID: {order_info['order_id']}")
print(f"ステータス: {order_info['status']}")
print(f"取引ペア: {order_info['symbol']}")
print(f"サイズ: {order_info['size']}")
print(f"価格: {order_info['price']}")
print(f"注文金額: ${order_info['total_value']:.2f}")
# 注文が本当に約定待ちになっているか確認
if order_info['status'] != 'resting':
print(f"警告: 注文が約定待ち状態になっていない可能性があります")
print(f"詳細なレスポンス: {order_info['raw_response']}")
return order_info
except Exception as e:
error_msg = f"注文処理中にエラーが発生しました: {str(e)}"
print(error_msg)
print(f"完全なレスポンス: {order_result}") # デバッグ用
raise Exception(error_msg)
if __name__ == "__main__":
# 環境変数から秘密鍵を読み込む
load_dotenv()
secret_key = os.getenv('HYPERLIQUID_SECRET_KEY')
if not secret_key:
raise ValueError("環境変数 'HYPERLIQUID_SECRET_KEY' が設定されていません")
# アカウントの初期化
account = eth_account.Account.from_key(secret_key)
# Exchangeオブジェクトの作成
exchange = Exchange(account, constants.MAINNET_API_URL)
symbol = 'kPEPE'
# 価格を直接指定する場合
place_limit_order(exchange, "kPEPE", 1, is_buy=True, limit_price=0.017)
通貨情報の取得
さきほどのHypwerliquidで注文する際に、価格やサイズで桁数が多すぎるとエラーになるため、そのエラーを回避するためにszDecimalsを取得するためのコードです。
一度に全てのコードを書かせるとあっているところまで修正されるので、このようなかたちで細かいパーツごとに完成させていったほうが結果的には近道でした。
from hyperliquid.info import Info
from hyperliquid.utils import constants
import json
def fetch_and_save_market_info():
info = Info(constants.MAINNET_API_URL, skip_ws=True)
market_info = info.meta()
with open('market_info.json', 'w') as f:
json.dump(market_info, f, indent=2)
print("Market information saved to 'market_info.json'.")
if __name__ == "__main__":
fetch_and_save_market_info()
import pandas as pd
import json
# JSONデータをファイルから読み込み
def load_market_data():
with open('market_info.json', 'r') as f:
data = json.load(f)
# universeキーの中身をデータフレームに変換
df = pd.DataFrame(data['universe'])
print("Market data loaded into DataFrame.")
print(df.head()) # データ表示確認
return df
if __name__ == "__main__":
df = load_market_data()
Market data loaded into DataFrame.
szDecimals name maxLeverage onlyIsolated
0 5 BTC 50 NaN
1 4 ETH 50 NaN
2 2 ATOM 10 NaN
3 1 MATIC 20 NaN
レバレッジの設定
アカウントに残高がたりません、というエラーが発生したので、レバレッジ設定用のコードも書いてもらいました。一発で書いてくれました。
import os
from dotenv import load_dotenv
import eth_account
from hyperliquid.exchange import Exchange
from hyperliquid.utils import constants
from eth_account.signers.local import LocalAccount
import requests
import time
def set_leverage(exchange: Exchange, symbol: str, leverage: int = 1) -> bool:
"""
指定された取引ペアのレバレッジを設定します。
Args:
exchange (Exchange): Hyperliquid Exchangeインスタンス
symbol (str): 取引ペアのシンボル
leverage (int): 設定するレバレッジ
Returns:
bool: 設定が成功した場合True、失敗した場合False
"""
try:
result = exchange.update_leverage(leverage, symbol)
if result and result.get('status') == 'ok':
print(result)
print(f"{symbol}のレバレッジを{leverage}倍に設定しました。")
return True
elif result and result.get('message'): # より詳細なエラーメッセージの出力
print(f"レバレッジの設定に失敗しました: {result['message']}")
return False
else:
print(f"レバレッジの設定に失敗しました: 不明なエラー")
return False
except requests.exceptions.RequestException as e:
print(f"ネットワークエラーが発生しました: {e}")
return False
except Exception as e:
print(f"レバレッジ設定中に予期せぬエラーが発生しました: {e}")
return False
if __name__ == "__main__":
load_dotenv()
secret_key = os.getenv('HYPERLIQUID_SECRET_KEY')
if not secret_key:
raise ValueError("環境変数 'HYPERLIQUID_SECRET_KEY' が設定されていません")
try:
account: LocalAccount = eth_account.Account.from_key(secret_key)
except Exception as e:
raise ValueError(f"アカウントの初期化中にエラーが発生しました: {e}")
exchange = Exchange(account, constants.MAINNET_API_URL) # または constants.TESTNET_API_URL
symbols = ["kPEPE"]
leverage = 20
for symbol in symbols:
if not set_leverage(exchange, symbol, leverage):
print(f"{symbol}のレバレッジ設定に失敗しました。")
time.sleep(1) # レート制限対策として1秒待機
print("プログラムを終了します。")
ヒゲ取りbotの実装
ここまでで、Hyperliquid側で必要な関数は使えるようになったので、いよいよbotの実装です。
botの実装については、Cursorで今まで作成したコードのファイルをすべて選択して読み込ませたうえで、bybitとHyperliquidの価格差があった場合に注文を出すようなbotを作りました。
細かいパラメーターはAI側で適当に決めてもらっています。
Hyperliquid側の注文の実装などと比べると、ここの実装は非常にスムーズで、最初から動くものが出てきました
エラーは、価格の型が異なっていたこと、シンボルが存在しないものになっていたことくらいでした
Hyperliquidとbybitでシンボルの指定が違うのですが、そこもうまくハンドリングしてくれていますね
import ccxt
import os
from dotenv import load_dotenv
import eth_account
from hyperliquid.exchange import Exchange
from hyperliquid.utils import constants
from hyperliquid.info import Info
import time
import logging
import pandas as pd
from datetime import datetime
import schedule
# ログ設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(f'arbitrage_bot_{datetime.now().strftime("%Y%m%d")}.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# 設定パラメータ
DIVERGENCE_THRESHOLD = 0.02 # 2%以上の乖離で取引
POSITION_SIZE = 100 # 1回あたりの取引サイズ（USDT）
MAX_POSITION = 1000 # 最大ポジションサイズ（USDT）
TAKE_PROFIT_RATIO = 0.01 # 1%の利益で決済
STOP_LOSS_RATIO = 0.01 # 1%の損失で決済
LEVERAGE = 3 # レバレッジ
class ArbitrageBot:
"""
HyperliquidとBybit間の価格乖離を利用した裁定取引ボット
Attributes:
hl_exchange (Exchange): Hyperliquidのexchangeインスタンス
bybit_exchange (ccxt.bybit): Bybitのexchangeインスタンス
info (Info): Hyperliquidの情報取得用インスタンス
current_positions (dict): 現在のポジション状況
"""
def __init__(self):
"""
ボットの初期化を行います。
取引所の接続設定とアカウント認証を行います。
"""
# 環境変数の読み込み
load_dotenv()
# Hyperliquid設定
secret_key = os.getenv('HYPERLIQUID_SECRET_KEY')
if not secret_key:
raise ValueError("HYPERLIQUID_SECRET_KEYが設定されていません")
self.account = eth_account.Account.from_key(secret_key)
self.hl_exchange = Exchange(self.account, constants.MAINNET_API_URL)
self.info = Info(constants.MAINNET_API_URL, skip_ws=True)
# Bybit設定
bybit_api_key = os.getenv('BYBIT_API_KEY')
bybit_secret = os.getenv('BYBIT_SECRET')
if not bybit_api_key or not bybit_secret:
raise ValueError("Bybit APIキーが設定されていません")
self.bybit_exchange = ccxt.bybit({
'apiKey': bybit_api_key,
'secret': bybit_secret,
'options': {'defaultType': 'future'}
})
self.current_positions = {}
logger.info("ArbitrageBot初期化完了")
def get_prices(self, symbol: str) -> tuple:
"""
両取引所の価格を取得します。
Args:
symbol (str): 取引ペアのシンボル
Returns:
tuple: (Hyperliquid価格, Bybit価格)
"""
try:
# Hyperliquid価格の取得
hl_price = self.info.all_mids().get(symbol.replace('USDT', ''))
if hl_price is not None:
hl_price = float(hl_price) # 文字列を数値に変換
# Bybit価格の取得
bybit_ticker = self.bybit_exchange.fetch_ticker(symbol)
bybit_price = float(bybit_ticker['last']) # 確実に数値に変換
logger.debug(f"{symbol} - HL: {hl_price}, Bybit: {bybit_price}")
return hl_price, bybit_price
except Exception as e:
logger.error(f"価格取得エラー: {e}")
return None, None
def calculate_divergence(self, hl_price: float, bybit_price: float) -> float:
"""
価格乖離率を計算します。
Args:
hl_price (float): Hyperliquid価格
bybit_price (float): Bybit価格
Returns:
float: 乖離率
"""
try:
if hl_price is None or bybit_price is None:
return 0
# 両方の価格が数値型であることを確認
hl_price = float(hl_price)
bybit_price = float(bybit_price)
return (hl_price - bybit_price) / bybit_price
except (TypeError, ValueError) as e:
logger.error(f"価格乖離率の計算エラー: {e}")
return 0
def execute_trade(self, symbol: str, is_buy: bool, divergence: float):
"""
取引を実行します。
Args:
symbol (str): 取引ペアのシンボル
is_buy (bool): 買いポジションの場合True
divergence (float): 価格乖離率
"""
try:
# レバレッジの設定
self.hl_exchange.update_leverage(LEVERAGE, symbol.replace('USDT', ''))
# 注文サイズの計算
current_price = self.info.all_mids().get(symbol.replace('USDT', ''))
order_size = POSITION_SIZE / current_price
# 利確・損切り価格の計算
take_profit = current_price * (1 + TAKE_PROFIT_RATIO) if is_buy else current_price * (1 - TAKE_PROFIT_RATIO)
stop_loss = current_price * (1 - STOP_LOSS_RATIO) if is_buy else current_price * (1 + STOP_LOSS_RATIO)
# 注文パラメータの設定
params = {
'stopLoss': str(stop_loss),
'takeProfit': str(take_profit),
'slTriggerBy': 'MarkPrice',
'tpTriggerBy': 'MarkPrice'
}
# 注文実行
order = self.hl_exchange.order(
name=symbol.replace('USDT', ''),
is_buy=is_buy,
sz=order_size,
limit_px=str(current_price),
order_type={"limit": {"tif": "Gtc"}},
params=params
)
logger.info(f"取引実行: {symbol}, {'買い' if is_buy else '売り'}, 乖離率: {divergence:.2%}")
logger.info(f"注文詳細: {order}")
# ポジション管理の更新
self.current_positions[symbol] = {
'side': 'buy' if is_buy else 'sell',
'size': order_size,
'entry_price': current_price,
'take_profit': take_profit,
'stop_loss': stop_loss
}
except Exception as e:
logger.error(f"取引実行エラー: {e}")
def monitor_prices(self):
"""
価格監視と取引実行の主要ロジック
"""
try:
# 監視対象のシンボルリスト
symbols = ['GMTUSDT'] # 必要に応じて追加
print(symbols)
for symbol in symbols:
# 価格取得
hl_price, bybit_price = self.get_prices(symbol)
if not hl_price or not bybit_price:
continue
# 乖離率計算
divergence = self.calculate_divergence(hl_price, bybit_price)
# 乖離率が閾値を超えた場合の取引判断
if abs(divergence) >= DIVERGENCE_THRESHOLD:
# ポジション確認
if symbol not in self.current_positions:
# Hyperliquidの価格が高い場合は売り
if divergence > 0:
self.execute_trade(symbol, False, divergence)
# Hyperliquidの価格が低い場合は買い
else:
self.execute_trade(symbol, True, divergence)
logger.info(f"{symbol} - Hyperliquid: ${hl_price:.6f}, Bybit: ${bybit_price:.6f}, 乖離率: {divergence:.2%}")
except Exception as e:
logger.error(f"価格監視エラー: {e}")
def main():
"""
メイン実行関数
"""
bot = ArbitrageBot()
logger.info("価格監視を開始します")
while True:
try:
# 価格監視の実行
bot.monitor_prices()
# 1秒待機
time.sleep(1)
except KeyboardInterrupt:
logger.info("ボットを停止します")
break
except Exception as e:
logger.error(f"予期せぬエラー: {e}")
time.sleep(60) # エラー時は1分待機
if __name__ == "__main__":
main()
3. チャレンジ結果
達成したこと
- 3時間でHyperliquidでのデータ取得から乖離計算、トレードの実装まで完了。
- 少額であれば十分実戦投入できるレベルのものができた
改善点と今後の展望
- 想定しているロジックで動くのか動作検証
- 複数銘柄への対応
- アカウント残高のモニタリング
- 想定以上の乖離があった場合の対応
- 乖離幅設定の最適化
- wsでの価格取得
など
おわりに
今回の3時間チャレンジでは、AIを活用することで短時間でも動作するトレードボットのプロトタイプを構築できることが確認できました。特にエラーのデバッグやコード生成のスピード感は2020年であれば想像すらできないものです。
AIによって可能性が爆発的に増えたものの、一つ一つを確実に実現していかなければ結局今までと何も変わらないということを実感した年でした
来年はAIを活用しながらbotにも真面目に取り組む所存です