Bot トレードロジック 書籍・論文・web記事

開発記録#134(2025/3/19)「バミューダンオプション（Bermudan option）の価格に対するモデル非依存の上限」

2025年3月19日

前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。

Yodaka

今回も論文を元にbot作成をする内容の記事です。

本記事では「バミューダンオプション（Bermudan option）の価格に対するモデル非依存の上限についての研究」を元に仮説を立てて、botのプロトタイプを４種類紹介します。

参考

Model-independent upper bounds for the price of Bermudan option with convex payoffs

論文の要旨

本論文では、バミューダンオプション（Bermudan option）の価格に対するモデル非依存の上限について研究しています。著者らは、確率測度 μ\muμ と ν\nuν が凸順序の関係（convex order）を満たす場合に、バミューダンオプションの価格の上限を求める問題を扱っています。具体的には、2つのエクササイズ日（行使日）を持つオプションの価格の上限を、ヨーロピアンオプションの価格情報をもとに算出するモデルを構築しました。

Yodaka

本研究の主要な貢献は以下の2点です。

  1. 双対問題（Dual Problem）の特性と簡素化
    • 問題の性質を整理し、双対問題の形に簡素化する手法を提示しました。
  2. 対称的なケースでの完全解決
    • 特に対称的な支払い関数と対称的な分布 μ\muμ と ν\nuν の場合に、最適なモデルと最適なヘッジ戦略を特定しました。

重要な発見として、フィルター生成された標準的な設定では、最適なモデルを定義するには十分ではなく、追加のランダム化が必要であることを示しました。これは、分布 μおよび ν が原子なし（atom-free）であっても成り立つ結果です。

この研究は、バミューダンオプションの堅牢なノーアービトラージ価格の上限を求める問題において、モデルフリーなアプローチを確立する重要な一歩となっています。

キーワード

ロバストプライシング, バミューダンオプション, マルチンゲール最適輸送, 双対性, スーパー・ヘッジング

Yodaka

本論文の「モデル非依存のバミューダンオプション価格の上限評価」の手法や概念を応用して、仮想通貨の自動取引Botに活用するための具体的なアイデアを以下に提案します。

🔍 基本的なアイデアの発展の方向性

本論文のアプローチは、以下の3つの特徴に焦点を当てています。

  1. モデル非依存性 (Model Independence)
    • 市場の確率モデルを厳密に仮定せず、観測データと価格関係から上限・下限を推定するアプローチ。
  2. マルチンゲール最適輸送 (MOT)
    • 価格の時間推移を確率過程として扱い、堅牢な最適化手法を用いる。
  3. スーパー・ヘッジ戦略 (Superhedging Strategy)
    • 最悪の市場変動でも利益を守るヘッジポートフォリオの構築。
Yodaka

これらを活用することで、堅牢で損失リスクの抑制に優れた仮想通貨取引Botを設計できると予測しました。

💡 具体的な取引Bot設計のアイデア

1. モデル非依存の「レンジブレークアウト」戦略

概要

  • 仮想通貨市場では、特定の価格帯（レンジ）を突破すると急騰・急落が起こる傾向があります。
  • MOTの「Left-Curtain Coupling」や「Right-Curtain Coupling」の概念を活用し、相場の転換点を確率的に予測する戦略が可能です。

実装のアイデア

  • バックテストデータを活用し、価格帯の上限・下限を動的に計算。
  • MOT理論に基づき、**現在の価格が「右側の価格帯境界」に近づいた場合は「上昇のシグナル」、「左側の価格帯境界」**に近づいた場合は「下落のシグナル」として判断。
  • ストップロス (損切り) 水準をスーパー・ヘッジの理論を基に設計し、最大損失額を抑制。

期待される効果

✅ レンジ相場でも効率的なエントリーが可能
✅ 損失リスクを最小化するストップロス設計

2. デリバティブ市場での「ノーアービトラージ検出Bot」

概要

  • 本論文では「モデル非依存のアービトラージ価格帯」の評価が提示されています。
  • 仮想通貨市場では、先物 (Futures) や オプション市場 の価格に不整合が発生することがあり、これを検出するBotが有効です。

実装のアイデア

  • BybitやBinanceのデリバティブ価格から、以下の関係をチェック
    • スポット価格 + デリバティブ価格の関係が「マルチンゲール制約」を満たしているか確認
  • 異常が発生した場合、次のアクションを行う
    • 「スポット市場で買い」＋「先物市場で売り」
    • あるいはその逆のポジションをとる
  • ディープヘッジング戦略 (Deep Hedging Strategy) を組み込み、損失リスクを制御する。

期待される効果

✅ デリバティブ市場の歪みを利用したアービトラージ収益が可能
✅ 「価格が理論価格に収束する」原理を活用した堅実な利益獲得

3. スーパー・ヘッジ戦略を活用した「安全重視型Bot」

概要

  • 本論文のスーパー・ヘッジの概念を活用し、最大損失を事前に制限した安全なBotの開発が可能です。

実装のアイデア

  • **「最大損失」と「目標利益」**を指定し、それに基づく取引ルールを設計。
  • MOTの「凸関数」を用いて、以下の2つの関数を設計：
    • a(x)a(x)a(x)：利益関数
    • b(y)b(y)b(y)：損失関数
  • 価格が a(x)a(x)a(x) を上回るならポジションを決済し利益確定
  • 価格が b(y)b(y)b(y) を下回るならポジションを自動損切り

期待される効果

✅ 急激な価格変動に対応可能
✅ 大きな損失を回避しつつ、安定的に利益を狙う戦略

4. 「確率的価格予測」型Bot

概要

  • MOT理論に基づき、将来の価格が「ある価格帯にとどまる確率」を評価し、それに基づくエントリーやイグジットの戦略を設計します。

実装のアイデア

  • BybitのAPIからリアルタイムデータを取得し、価格の移動平均やボラティリティの情報を利用
  • LSTMやDQN、DDPGの強化学習モデルと組み合わせ、将来の価格帯を確率的に予測
  • 価格帯が「ターゲットエリア」に入った場合にのみエントリー

期待される効果

✅ ダマシ相場を回避し、信頼性の高い取引が可能
✅ AIによる最適化で、変動の激しい市場に対応

⚙️ 実装の技術要素

Yodaka

上記のアイデアを具体的に実装するには、以下の技術要素を活用できると考えました。

✅ Python (戦略の設計、API連携、データ処理)
✅ Rust (データストリーミングやパフォーマンス最適化)
✅ Docker + Kubernetes (Botの並列稼働および監視)
✅ Bybit API (リアルタイムデータの取得および発注)
✅ Poetry (仮想環境と依存関係管理)

プロトタイプ１「Bybit Range Bot」とコードの説明

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import requests
import time
import hmac
import hashlib
import json
from datetime import datetime, timedelta
# Bybit API情報
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://api.bybit.com"
# 範囲ブレークアウトのパラメータ
RANGE_PERIOD = 20 # 過去20期間の最高値と最安値を基準
STOP_LOSS_PERCENT = 0.02 # 2%のストップロス
TAKE_PROFIT_PERCENT = 0.04 # 4%の利益確定
# Bybit APIの署名関数
def sign_request(params):
sorted_params = sorted(params.items())
query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
signature = hmac.new(
API_SECRET.encode(), query_string.encode(), hashlib.sha256
).hexdigest()
return signature
# BybitのK線データ取得
def get_klines(symbol="BTCUSDT", interval="15m", limit=RANGE_PERIOD):
url = f"{BASE_URL}/v2/public/kline/list"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
response = requests.get(url, params=params)
return response.json()["result"]
# 注文関数
def place_order(symbol, side, qty, price=None):
url = f"{BASE_URL}/v2/private/order/create"
params = {
"api_key": API_KEY,
"symbol": symbol,
"side": side,
"order_type": "Market" if not price else "Limit",
"qty": qty,
"time_in_force": "GoodTillCancel",
"timestamp": int(time.time() * 1000)
}
params["sign"] = sign_request(params)
response = requests.post(url, data=params)
print(response.json())
# 範囲ブレークアウト戦略
def range_breakout_strategy(symbol="BTCUSDT", qty=0.01):
candles = get_klines(symbol)
highs = [float(candle[3]) for candle in candles] # 高値データ
lows = [float(candle[4]) for candle in candles] # 安値データ
range_high = max(highs)
range_low = min(lows)
current_price = float(candles[-1][2])
if current_price >= range_high:
print(f"[BUY SIGNAL] Current Price: {current_price} | Breakout Above: {range_high}")
place_order(symbol, "Buy", qty)
elif current_price <= range_low:
print(f"[SELL SIGNAL] Current Price: {current_price} | Breakout Below: {range_low}")
place_order(symbol, "Sell", qty)
# メインループ
def main():
symbol = "BTCUSDT"
qty = 0.01
while True:
try:
range_breakout_strategy(symbol, qty)
except Exception as e:
print(f"Error: {e}")
time.sleep(300) # 5分ごとにチェック
if __name__ == "__main__":
main()
import requests import time import hmac import hashlib import json from datetime import datetime, timedelta # Bybit API情報 API_KEY = "YOUR_API_KEY" API_SECRET = "YOUR_API_SECRET" BASE_URL = "https://api.bybit.com" # 範囲ブレークアウトのパラメータ RANGE_PERIOD = 20 # 過去20期間の最高値と最安値を基準 STOP_LOSS_PERCENT = 0.02 # 2%のストップロス TAKE_PROFIT_PERCENT = 0.04 # 4%の利益確定 # Bybit APIの署名関数 def sign_request(params): sorted_params = sorted(params.items()) query_string = "&".join(f"{key}={value}" for key, value in sorted_params) signature = hmac.new( API_SECRET.encode(), query_string.encode(), hashlib.sha256 ).hexdigest() return signature # BybitのK線データ取得 def get_klines(symbol="BTCUSDT", interval="15m", limit=RANGE_PERIOD): url = f"{BASE_URL}/v2/public/kline/list" params = { "symbol": symbol, "interval": interval, "limit": limit } response = requests.get(url, params=params) return response.json()["result"] # 注文関数 def place_order(symbol, side, qty, price=None): url = f"{BASE_URL}/v2/private/order/create" params = { "api_key": API_KEY, "symbol": symbol, "side": side, "order_type": "Market" if not price else "Limit", "qty": qty, "time_in_force": "GoodTillCancel", "timestamp": int(time.time() * 1000) } params["sign"] = sign_request(params) response = requests.post(url, data=params) print(response.json()) # 範囲ブレークアウト戦略 def range_breakout_strategy(symbol="BTCUSDT", qty=0.01): candles = get_klines(symbol) highs = [float(candle[3]) for candle in candles] # 高値データ lows = [float(candle[4]) for candle in candles] # 安値データ range_high = max(highs) range_low = min(lows) current_price = float(candles[-1][2]) if current_price >= range_high: print(f"[BUY SIGNAL] Current Price: {current_price} | Breakout Above: {range_high}") place_order(symbol, "Buy", qty) elif current_price <= range_low: print(f"[SELL SIGNAL] Current Price: {current_price} | Breakout Below: {range_low}") place_order(symbol, "Sell", qty) # メインループ def main(): symbol = "BTCUSDT" qty = 0.01 while True: try: range_breakout_strategy(symbol, qty) except Exception as e: print(f"Error: {e}") time.sleep(300) # 5分ごとにチェック if __name__ == "__main__": main()
import requests
import time
import hmac
import hashlib
import json
from datetime import datetime, timedelta

# Bybit API情報
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://api.bybit.com"

# 範囲ブレークアウトのパラメータ
RANGE_PERIOD = 20  # 過去20期間の最高値と最安値を基準
STOP_LOSS_PERCENT = 0.02  # 2%のストップロス
TAKE_PROFIT_PERCENT = 0.04  # 4%の利益確定

# Bybit APIの署名関数
def sign_request(params):
    sorted_params = sorted(params.items())
    query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
    signature = hmac.new(
        API_SECRET.encode(), query_string.encode(), hashlib.sha256
    ).hexdigest()
    return signature

# BybitのK線データ取得
def get_klines(symbol="BTCUSDT", interval="15m", limit=RANGE_PERIOD):
    url = f"{BASE_URL}/v2/public/kline/list"
    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }
    response = requests.get(url, params=params)
    return response.json()["result"]

# 注文関数
def place_order(symbol, side, qty, price=None):
    url = f"{BASE_URL}/v2/private/order/create"
    params = {
        "api_key": API_KEY,
        "symbol": symbol,
        "side": side,
        "order_type": "Market" if not price else "Limit",
        "qty": qty,
        "time_in_force": "GoodTillCancel",
        "timestamp": int(time.time() * 1000)
    }
    params["sign"] = sign_request(params)
    response = requests.post(url, data=params)
    print(response.json())

# 範囲ブレークアウト戦略
def range_breakout_strategy(symbol="BTCUSDT", qty=0.01):
    candles = get_klines(symbol)
    highs = [float(candle[3]) for candle in candles]  # 高値データ
    lows = [float(candle[4]) for candle in candles]  # 安値データ

    range_high = max(highs)
    range_low = min(lows)

    current_price = float(candles[-1][2])

    if current_price >= range_high:
        print(f"[BUY SIGNAL] Current Price: {current_price} | Breakout Above: {range_high}")
        place_order(symbol, "Buy", qty)
    elif current_price <= range_low:
        print(f"[SELL SIGNAL] Current Price: {current_price} | Breakout Below: {range_low}")
        place_order(symbol, "Sell", qty)

# メインループ
def main():
    symbol = "BTCUSDT"
    qty = 0.01
    while True:
        try:
            range_breakout_strategy(symbol, qty)
        except Exception as e:
            print(f"Error: {e}")
        time.sleep(300)  # 5分ごとにチェック

if __name__ == "__main__":
    main()

🚀 主要な機能

✅ Bybit API を使用してリアルタイムのK線データ (ローソク足データ) を取得
✅ 過去20期間の最高値と最安値を範囲のブレークアウトラインとして設定
✅ 現在価格が「範囲の上限を超えた場合 → 買い (Buy)」
✅ 現在価格が「範囲の下限を割った場合 → 売り (Sell)」
✅ 定期的にデータを更新し、5分ごとにシグナルを確認

🔧 カスタマイズ可能なパラメータ

  • RANGE_PERIOD：価格の範囲を決定する期間 (例: 20 → 過去20期間)
  • STOP_LOSS_PERCENT：損失を抑制するためのストップロス (例: 2% 損失制限)
  • TAKE_PROFIT_PERCENT：利益確定のための目標価格設定 (例: 4% 利益確定)

⚠️ 注意点

  1. Bybit APIのキーとシークレットキーは適切に管理する。
  2. 実際の取引を行う前に、Bybitの「テストネット」を使用してテストする。
  3. ストップロスと利益確定のアルゴリズムは、追加でカスタマイズする。

📈 実用化へ向けた次のステップ

  • ストップロスやテイクプロフィットのロジックを組み込む
  • MOT理論に基づく「価格の上限と下限の動的計算」の導入
  • バックテスト機能の実装で戦略のパフォーマンスを検証する

プロトタイプ２「Bybit Arbitrage Bot」とコードの説明

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import requests
import time
import hmac
import hashlib
import json
# Bybit API情報
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://api.bybit.com"
# パラメータ設定
SPREAD_THRESHOLD = 0.5 # アービトラージの利幅 (0.5%差異で実行)
TRADE_AMOUNT = 0.01 # トレード量
# 署名作成関数
def sign_request(params):
sorted_params = sorted(params.items())
query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
signature = hmac.new(
API_SECRET.encode(), query_string.encode(), hashlib.sha256
).hexdigest()
return signature
# スポット価格取得
def get_spot_price(symbol="BTCUSDT"):
url = f"{BASE_URL}/v2/public/tickers"
response = requests.get(url).json()
for ticker in response["result"]:
if ticker["symbol"] == symbol:
return float(ticker["last_price"])
return None
# 先物価格取得
def get_futures_price(symbol="BTCUSDT"):
url = f"{BASE_URL}/v2/public/funding/prev-funding-rate"
params = {"symbol": symbol}
response = requests.get(url, params=params).json()
if "result" in response:
return float(response["result"]["lastFundingRate"])
return None
# 注文実行
def place_order(symbol, side, qty, price=None):
url = f"{BASE_URL}/v2/private/order/create"
params = {
"api_key": API_KEY,
"symbol": symbol,
"side": side,
"order_type": "Market" if not price else "Limit",
"qty": qty,
"time_in_force": "GoodTillCancel",
"timestamp": int(time.time() * 1000)
}
params["sign"] = sign_request(params)
response = requests.post(url, data=params)
print(response.json())
# アービトラージ戦略
def arbitrage_strategy(symbol="BTCUSDT"):
spot_price = get_spot_price(symbol)
futures_price = get_futures_price(symbol)
if spot_price is None or futures_price is None:
print("価格データの取得に失敗しました。")
return
spread = ((futures_price - spot_price) / spot_price) * 100
if spread >= SPREAD_THRESHOLD:
print(f"[Arbitrage Opportunity] Spread: {spread:.2f}%")
print("[ACTION] - Spot: Buy | Futures: Sell")
place_order(symbol, "Buy", TRADE_AMOUNT) # 現物市場で買い
place_order(symbol, "Sell", TRADE_AMOUNT) # 先物市場で売り
elif spread <= -SPREAD_THRESHOLD:
print(f"[Arbitrage Opportunity] Spread: {spread:.2f}%")
print("[ACTION] - Spot: Sell | Futures: Buy")
place_order(symbol, "Sell", TRADE_AMOUNT) # 現物市場で売り
place_order(symbol, "Buy", TRADE_AMOUNT) # 先物市場で買い
else:
print(f"[INFO] Spread: {spread:.2f}% (No action)")
# メインループ
def main():
symbol = "BTCUSDT"
while True:
try:
arbitrage_strategy(symbol)
except Exception as e:
print(f"Error: {e}")
time.sleep(300) # 5分ごとにチェック
if __name__ == "__main__":
main()
import requests import time import hmac import hashlib import json # Bybit API情報 API_KEY = "YOUR_API_KEY" API_SECRET = "YOUR_API_SECRET" BASE_URL = "https://api.bybit.com" # パラメータ設定 SPREAD_THRESHOLD = 0.5 # アービトラージの利幅 (0.5%差異で実行) TRADE_AMOUNT = 0.01 # トレード量 # 署名作成関数 def sign_request(params): sorted_params = sorted(params.items()) query_string = "&".join(f"{key}={value}" for key, value in sorted_params) signature = hmac.new( API_SECRET.encode(), query_string.encode(), hashlib.sha256 ).hexdigest() return signature # スポット価格取得 def get_spot_price(symbol="BTCUSDT"): url = f"{BASE_URL}/v2/public/tickers" response = requests.get(url).json() for ticker in response["result"]: if ticker["symbol"] == symbol: return float(ticker["last_price"]) return None # 先物価格取得 def get_futures_price(symbol="BTCUSDT"): url = f"{BASE_URL}/v2/public/funding/prev-funding-rate" params = {"symbol": symbol} response = requests.get(url, params=params).json() if "result" in response: return float(response["result"]["lastFundingRate"]) return None # 注文実行 def place_order(symbol, side, qty, price=None): url = f"{BASE_URL}/v2/private/order/create" params = { "api_key": API_KEY, "symbol": symbol, "side": side, "order_type": "Market" if not price else "Limit", "qty": qty, "time_in_force": "GoodTillCancel", "timestamp": int(time.time() * 1000) } params["sign"] = sign_request(params) response = requests.post(url, data=params) print(response.json()) # アービトラージ戦略 def arbitrage_strategy(symbol="BTCUSDT"): spot_price = get_spot_price(symbol) futures_price = get_futures_price(symbol) if spot_price is None or futures_price is None: print("価格データの取得に失敗しました。") return spread = ((futures_price - spot_price) / spot_price) * 100 if spread >= SPREAD_THRESHOLD: print(f"[Arbitrage Opportunity] Spread: {spread:.2f}%") print("[ACTION] - Spot: Buy | Futures: Sell") place_order(symbol, "Buy", TRADE_AMOUNT) # 現物市場で買い place_order(symbol, "Sell", TRADE_AMOUNT) # 先物市場で売り elif spread <= -SPREAD_THRESHOLD: print(f"[Arbitrage Opportunity] Spread: {spread:.2f}%") print("[ACTION] - Spot: Sell | Futures: Buy") place_order(symbol, "Sell", TRADE_AMOUNT) # 現物市場で売り place_order(symbol, "Buy", TRADE_AMOUNT) # 先物市場で買い else: print(f"[INFO] Spread: {spread:.2f}% (No action)") # メインループ def main(): symbol = "BTCUSDT" while True: try: arbitrage_strategy(symbol) except Exception as e: print(f"Error: {e}") time.sleep(300) # 5分ごとにチェック if __name__ == "__main__": main()
import requests
import time
import hmac
import hashlib
import json

# Bybit API情報
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://api.bybit.com"

# パラメータ設定
SPREAD_THRESHOLD = 0.5  # アービトラージの利幅 (0.5%差異で実行)
TRADE_AMOUNT = 0.01     # トレード量

# 署名作成関数
def sign_request(params):
    sorted_params = sorted(params.items())
    query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
    signature = hmac.new(
        API_SECRET.encode(), query_string.encode(), hashlib.sha256
    ).hexdigest()
    return signature

# スポット価格取得
def get_spot_price(symbol="BTCUSDT"):
    url = f"{BASE_URL}/v2/public/tickers"
    response = requests.get(url).json()
    for ticker in response["result"]:
        if ticker["symbol"] == symbol:
            return float(ticker["last_price"])
    return None

# 先物価格取得
def get_futures_price(symbol="BTCUSDT"):
    url = f"{BASE_URL}/v2/public/funding/prev-funding-rate"
    params = {"symbol": symbol}
    response = requests.get(url, params=params).json()
    if "result" in response:
        return float(response["result"]["lastFundingRate"])
    return None

# 注文実行
def place_order(symbol, side, qty, price=None):
    url = f"{BASE_URL}/v2/private/order/create"
    params = {
        "api_key": API_KEY,
        "symbol": symbol,
        "side": side,
        "order_type": "Market" if not price else "Limit",
        "qty": qty,
        "time_in_force": "GoodTillCancel",
        "timestamp": int(time.time() * 1000)
    }
    params["sign"] = sign_request(params)
    response = requests.post(url, data=params)
    print(response.json())

# アービトラージ戦略
def arbitrage_strategy(symbol="BTCUSDT"):
    spot_price = get_spot_price(symbol)
    futures_price = get_futures_price(symbol)

    if spot_price is None or futures_price is None:
        print("価格データの取得に失敗しました。")
        return

    spread = ((futures_price - spot_price) / spot_price) * 100

    if spread >= SPREAD_THRESHOLD:
        print(f"[Arbitrage Opportunity] Spread: {spread:.2f}%")
        print("[ACTION] - Spot: Buy | Futures: Sell")
        place_order(symbol, "Buy", TRADE_AMOUNT)  # 現物市場で買い
        place_order(symbol, "Sell", TRADE_AMOUNT)  # 先物市場で売り

    elif spread <= -SPREAD_THRESHOLD:
        print(f"[Arbitrage Opportunity] Spread: {spread:.2f}%")
        print("[ACTION] - Spot: Sell | Futures: Buy")
        place_order(symbol, "Sell", TRADE_AMOUNT)  # 現物市場で売り
        place_order(symbol, "Buy", TRADE_AMOUNT)   # 先物市場で買い
    else:
        print(f"[INFO] Spread: {spread:.2f}% (No action)")

# メインループ
def main():
    symbol = "BTCUSDT"
    while True:
        try:
            arbitrage_strategy(symbol)
        except Exception as e:
            print(f"Error: {e}")
        time.sleep(300)  # 5分ごとにチェック

if __name__ == "__main__":
    main()

🚀 主要な機能

✅ Bybit API を使用してリアルタイムの価格データを取得
✅ スポット価格と先物価格のスプレッドを計算し、指定した閾値 (例：0.5%) を超えたらアービトラージを実行
✅ シグナル発生時に次の行動を取る

  • スプレッドがプラス (先物が割高) → 「スポット市場で買い」＋「先物市場で売り」
  • スプレッドがマイナス (先物が割安) → 「スポット市場で売り」＋「先物市場で買い」
    ✅ 定期的に価格データを更新し、5分ごとにシグナルを確認

🔧 カスタマイズ可能なパラメータ

  • SPREAD_THRESHOLD：スプレッドの閾値 (例: 0.5 → 0.5%の差異が発生時にエントリー)
  • TRADE_AMOUNT：取引量 (例: 0.01 BTC)
  • API_KEY と API_SECRET：BybitのAPIキー情報

⚠️ 注意点

  1. Bybit APIのキーとシークレットキーは適切に管理する。
  2. 実際の取引を行う前に、Bybitの「テストネット」を使用してテストする。
  3. スプレッド閾値は、手数料やスリッページを考慮して適切に設定する。

📈 実用化に向けた次のステップ

  • スプレッドの変動を動的に計算するロジックの追加
  • 手数料の影響を考慮した利益計算機能の実装
  • アービトラージ機会の統計データを取得して、最適な取引条件の分析

プロトタイプ３「Bybit Superhedge Bot」とコードの説明

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import requests
import time
import hmac
import hashlib
import json
from datetime import datetime
# Bybit API情報
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://api.bybit.com"
# スーパー・ヘッジのパラメータ
MAX_LOSS_PERCENT = 0.02 # 2%の最大損失制限
TARGET_PROFIT_PERCENT = 0.04 # 4%の目標利益
TRADE_AMOUNT = 0.01 # トレード量
# 署名作成関数
def sign_request(params):
sorted_params = sorted(params.items())
query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
signature = hmac.new(
API_SECRET.encode(), query_string.encode(), hashlib.sha256
).hexdigest()
return signature
# Bybit価格取得
def get_price(symbol="BTCUSDT"):
url = f"{BASE_URL}/v2/public/tickers"
response = requests.get(url).json()
for ticker in response["result"]:
if ticker["symbol"] == symbol:
return float(ticker["last_price"])
return None
# 注文実行
def place_order(symbol, side, qty, price=None):
url = f"{BASE_URL}/v2/private/order/create"
params = {
"api_key": API_KEY,
"symbol": symbol,
"side": side,
"order_type": "Market" if not price else "Limit",
"qty": qty,
"time_in_force": "GoodTillCancel",
"timestamp": int(time.time() * 1000)
}
params["sign"] = sign_request(params)
response = requests.post(url, data=params)
print(response.json())
# スーパー・ヘッジ戦略
def superhedge_strategy(symbol="BTCUSDT"):
entry_price = get_price(symbol)
if entry_price is None:
print("価格データの取得に失敗しました。")
return
print(f"[ENTRY] Entry Price: {entry_price}")
place_order(symbol, "Buy", TRADE_AMOUNT)
target_price = entry_price * (1 + TARGET_PROFIT_PERCENT)
stop_loss_price = entry_price * (1 - MAX_LOSS_PERCENT)
print(f"[INFO] Target Price: {target_price}, Stop Loss Price: {stop_loss_price}")
while True:
current_price = get_price(symbol)
if current_price is None:
continue
if current_price >= target_price:
print(f"[PROFIT] Target Achieved! Closing Position at {current_price}")
place_order(symbol, "Sell", TRADE_AMOUNT)
break
if current_price <= stop_loss_price:
print(f"[LOSS] Stop Loss Triggered. Closing Position at {current_price}")
place_order(symbol, "Sell", TRADE_AMOUNT)
break
time.sleep(60) # 1分ごとに価格チェック
# メインループ
def main():
symbol = "BTCUSDT"
while True:
try:
superhedge_strategy(symbol)
except Exception as e:
print(f"Error: {e}")
time.sleep(300) # 5分ごとに新しいポジションをチェック
if __name__ == "__main__":
main()
import requests import time import hmac import hashlib import json from datetime import datetime # Bybit API情報 API_KEY = "YOUR_API_KEY" API_SECRET = "YOUR_API_SECRET" BASE_URL = "https://api.bybit.com" # スーパー・ヘッジのパラメータ MAX_LOSS_PERCENT = 0.02 # 2%の最大損失制限 TARGET_PROFIT_PERCENT = 0.04 # 4%の目標利益 TRADE_AMOUNT = 0.01 # トレード量 # 署名作成関数 def sign_request(params): sorted_params = sorted(params.items()) query_string = "&".join(f"{key}={value}" for key, value in sorted_params) signature = hmac.new( API_SECRET.encode(), query_string.encode(), hashlib.sha256 ).hexdigest() return signature # Bybit価格取得 def get_price(symbol="BTCUSDT"): url = f"{BASE_URL}/v2/public/tickers" response = requests.get(url).json() for ticker in response["result"]: if ticker["symbol"] == symbol: return float(ticker["last_price"]) return None # 注文実行 def place_order(symbol, side, qty, price=None): url = f"{BASE_URL}/v2/private/order/create" params = { "api_key": API_KEY, "symbol": symbol, "side": side, "order_type": "Market" if not price else "Limit", "qty": qty, "time_in_force": "GoodTillCancel", "timestamp": int(time.time() * 1000) } params["sign"] = sign_request(params) response = requests.post(url, data=params) print(response.json()) # スーパー・ヘッジ戦略 def superhedge_strategy(symbol="BTCUSDT"): entry_price = get_price(symbol) if entry_price is None: print("価格データの取得に失敗しました。") return print(f"[ENTRY] Entry Price: {entry_price}") place_order(symbol, "Buy", TRADE_AMOUNT) target_price = entry_price * (1 + TARGET_PROFIT_PERCENT) stop_loss_price = entry_price * (1 - MAX_LOSS_PERCENT) print(f"[INFO] Target Price: {target_price}, Stop Loss Price: {stop_loss_price}") while True: current_price = get_price(symbol) if current_price is None: continue if current_price >= target_price: print(f"[PROFIT] Target Achieved! Closing Position at {current_price}") place_order(symbol, "Sell", TRADE_AMOUNT) break if current_price <= stop_loss_price: print(f"[LOSS] Stop Loss Triggered. Closing Position at {current_price}") place_order(symbol, "Sell", TRADE_AMOUNT) break time.sleep(60) # 1分ごとに価格チェック # メインループ def main(): symbol = "BTCUSDT" while True: try: superhedge_strategy(symbol) except Exception as e: print(f"Error: {e}") time.sleep(300) # 5分ごとに新しいポジションをチェック if __name__ == "__main__": main()
import requests
import time
import hmac
import hashlib
import json
from datetime import datetime

# Bybit API情報
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://api.bybit.com"

# スーパー・ヘッジのパラメータ
MAX_LOSS_PERCENT = 0.02    # 2%の最大損失制限
TARGET_PROFIT_PERCENT = 0.04  # 4%の目標利益
TRADE_AMOUNT = 0.01       # トレード量

# 署名作成関数
def sign_request(params):
    sorted_params = sorted(params.items())
    query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
    signature = hmac.new(
        API_SECRET.encode(), query_string.encode(), hashlib.sha256
    ).hexdigest()
    return signature

# Bybit価格取得
def get_price(symbol="BTCUSDT"):
    url = f"{BASE_URL}/v2/public/tickers"
    response = requests.get(url).json()
    for ticker in response["result"]:
        if ticker["symbol"] == symbol:
            return float(ticker["last_price"])
    return None

# 注文実行
def place_order(symbol, side, qty, price=None):
    url = f"{BASE_URL}/v2/private/order/create"
    params = {
        "api_key": API_KEY,
        "symbol": symbol,
        "side": side,
        "order_type": "Market" if not price else "Limit",
        "qty": qty,
        "time_in_force": "GoodTillCancel",
        "timestamp": int(time.time() * 1000)
    }
    params["sign"] = sign_request(params)
    response = requests.post(url, data=params)
    print(response.json())

# スーパー・ヘッジ戦略
def superhedge_strategy(symbol="BTCUSDT"):
    entry_price = get_price(symbol)
    if entry_price is None:
        print("価格データの取得に失敗しました。")
        return

    print(f"[ENTRY] Entry Price: {entry_price}")
    place_order(symbol, "Buy", TRADE_AMOUNT)

    target_price = entry_price * (1 + TARGET_PROFIT_PERCENT)
    stop_loss_price = entry_price * (1 - MAX_LOSS_PERCENT)

    print(f"[INFO] Target Price: {target_price}, Stop Loss Price: {stop_loss_price}")

    while True:
        current_price = get_price(symbol)
        if current_price is None:
            continue

        if current_price >= target_price:
            print(f"[PROFIT] Target Achieved! Closing Position at {current_price}")
            place_order(symbol, "Sell", TRADE_AMOUNT)
            break

        if current_price <= stop_loss_price:
            print(f"[LOSS] Stop Loss Triggered. Closing Position at {current_price}")
            place_order(symbol, "Sell", TRADE_AMOUNT)
            break

        time.sleep(60)  # 1分ごとに価格チェック

# メインループ
def main():
    symbol = "BTCUSDT"
    while True:
        try:
            superhedge_strategy(symbol)
        except Exception as e:
            print(f"Error: {e}")
        time.sleep(300)  # 5分ごとに新しいポジションをチェック

if __name__ == "__main__":
    main()

🚀 主要な機能

✅ Bybit API を使用してリアルタイムの価格データを取得
✅ エントリー後、「利益確定」 と 「損失制限」 の水準を自動で計算
✅ シグナル発生時に次の行動を取る

  • 価格が目標利益ラインを超えたら → 利益確定のためにポジションを決済
  • 価格が最大損失ラインを割ったら → 損切りのためにポジションを決済
    ✅ 定期的に価格データを更新し、リアルタイムで監視

🔧 カスタマイズ可能なパラメータ

  • MAX_LOSS_PERCENT：最大許容損失 (例: 0.02 → 2%の損失)
  • TARGET_PROFIT_PERCENT：目標利益 (例: 0.04 → 4%の利益)
  • TRADE_AMOUNT：取引量 (例: 0.01 BTC)
  • API_KEY と API_SECRET：BybitのAPIキー情報

⚠️ 注意点

  1. Bybit APIのキーとシークレットキーは適切に管理する。
  2. 実際の取引を行う前に、Bybitの「テストネット」を使用してテストする。
  3. 損切りや利益確定の水準は、ボラティリティや市場状況に応じて柔軟に調整する。

📈 実用化に向けた次のステップ

  • トレーリングストップの導入で「利益を最大化しつつ、損失を最小化」する戦略を強化
  • マルチポジション対応で複数のエントリー・イグジット管理を実装
  • リアルタイムアラート機能の追加で、Botの行動をすぐに把握可能にする

プロトタイプ４「Bybit Price Predictor」とコードの説明

Plain text
Copy to clipboard
Open code in new window
EnlighterJS 3 Syntax Highlighter
import numpy as np
import pandas as pd
import requests
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import time
# Bybit API情報
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://api.bybit.com"
# データ取得関数
def get_klines(symbol="BTCUSDT", interval="15m", limit=200):
url = f"{BASE_URL}/v2/public/kline/list"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
response = requests.get(url, params=params)
data = response.json()["result"]
df = pd.DataFrame(data)
df["close"] = pd.to_numeric(df["close"])
return df["close"].values
# データの前処理
def create_dataset(data, look_back=20):
X, y = [], []
for i in range(len(data) - look_back - 1):
X.append(data[i:(i + look_back)])
y.append(data[i + look_back])
return np.array(X), np.array(y)
# LSTMモデルの構築
def build_model(look_back):
model = keras.Sequential([
layers.LSTM(64, return_sequences=True, input_shape=(look_back, 1)),
layers.LSTM(32),
layers.Dense(16, activation='relu'),
layers.Dense(1)
])
model.compile(optimizer='adam', loss='mse')
return model
# 予測とエントリー/エグジットの判定
def predict_and_trade(model, symbol, look_back, threshold=0.01):
data = get_klines(symbol)
current_data = np.array(data[-look_back:]).reshape(1, look_back, 1)
predicted_price = model.predict(current_data)[0][0]
current_price = data[-1]
if predicted_price >= current_price * (1 + threshold):
print(f"[BUY SIGNAL] Current: {current_price} | Predicted: {predicted_price}")
place_order(symbol, "Buy", TRADE_AMOUNT)
elif predicted_price <= current_price * (1 - threshold):
print(f"[SELL SIGNAL] Current: {current_price} | Predicted: {predicted_price}")
place_order(symbol, "Sell", TRADE_AMOUNT)
# 注文実行関数
def place_order(symbol, side, qty):
url = f"{BASE_URL}/v2/private/order/create"
params = {
"api_key": API_KEY,
"symbol": symbol,
"side": side,
"order_type": "Market",
"qty": qty,
"time_in_force": "GoodTillCancel",
"timestamp": int(time.time() * 1000)
}
params["sign"] = sign_request(params)
response = requests.post(url, data=params)
print(response.json())
# API署名生成関数
def sign_request(params):
sorted_params = sorted(params.items())
query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
signature = hmac.new(
API_SECRET.encode(), query_string.encode(), hashlib.sha256
).hexdigest()
return signature
# メイン関数
def main():
symbol = "BTCUSDT"
look_back = 20
data = get_klines(symbol)
X, y = create_dataset(data, look_back)
X = X.reshape(X.shape[0], X.shape[1], 1)
model = build_model(look_back)
model.fit(X, y, epochs=20, batch_size=32, verbose=1)
while True:
try:
predict_and_trade(model, symbol, look_back)
except Exception as e:
print(f"Error: {e}")
time.sleep(300) # 5分ごとにチェック
if __name__ == "__main__":
main()
import numpy as np import pandas as pd import requests import tensorflow as tf from tensorflow import keras from tensorflow.keras import layers import time # Bybit API情報 API_KEY = "YOUR_API_KEY" API_SECRET = "YOUR_API_SECRET" BASE_URL = "https://api.bybit.com" # データ取得関数 def get_klines(symbol="BTCUSDT", interval="15m", limit=200): url = f"{BASE_URL}/v2/public/kline/list" params = { "symbol": symbol, "interval": interval, "limit": limit } response = requests.get(url, params=params) data = response.json()["result"] df = pd.DataFrame(data) df["close"] = pd.to_numeric(df["close"]) return df["close"].values # データの前処理 def create_dataset(data, look_back=20): X, y = [], [] for i in range(len(data) - look_back - 1): X.append(data[i:(i + look_back)]) y.append(data[i + look_back]) return np.array(X), np.array(y) # LSTMモデルの構築 def build_model(look_back): model = keras.Sequential([ layers.LSTM(64, return_sequences=True, input_shape=(look_back, 1)), layers.LSTM(32), layers.Dense(16, activation='relu'), layers.Dense(1) ]) model.compile(optimizer='adam', loss='mse') return model # 予測とエントリー/エグジットの判定 def predict_and_trade(model, symbol, look_back, threshold=0.01): data = get_klines(symbol) current_data = np.array(data[-look_back:]).reshape(1, look_back, 1) predicted_price = model.predict(current_data)[0][0] current_price = data[-1] if predicted_price >= current_price * (1 + threshold): print(f"[BUY SIGNAL] Current: {current_price} | Predicted: {predicted_price}") place_order(symbol, "Buy", TRADE_AMOUNT) elif predicted_price <= current_price * (1 - threshold): print(f"[SELL SIGNAL] Current: {current_price} | Predicted: {predicted_price}") place_order(symbol, "Sell", TRADE_AMOUNT) # 注文実行関数 def place_order(symbol, side, qty): url = f"{BASE_URL}/v2/private/order/create" params = { "api_key": API_KEY, "symbol": symbol, "side": side, "order_type": "Market", "qty": qty, "time_in_force": "GoodTillCancel", "timestamp": int(time.time() * 1000) } params["sign"] = sign_request(params) response = requests.post(url, data=params) print(response.json()) # API署名生成関数 def sign_request(params): sorted_params = sorted(params.items()) query_string = "&".join(f"{key}={value}" for key, value in sorted_params) signature = hmac.new( API_SECRET.encode(), query_string.encode(), hashlib.sha256 ).hexdigest() return signature # メイン関数 def main(): symbol = "BTCUSDT" look_back = 20 data = get_klines(symbol) X, y = create_dataset(data, look_back) X = X.reshape(X.shape[0], X.shape[1], 1) model = build_model(look_back) model.fit(X, y, epochs=20, batch_size=32, verbose=1) while True: try: predict_and_trade(model, symbol, look_back) except Exception as e: print(f"Error: {e}") time.sleep(300) # 5分ごとにチェック if __name__ == "__main__": main()
import numpy as np
import pandas as pd
import requests
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import time

# Bybit API情報
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
BASE_URL = "https://api.bybit.com"

# データ取得関数
def get_klines(symbol="BTCUSDT", interval="15m", limit=200):
    url = f"{BASE_URL}/v2/public/kline/list"
    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }
    response = requests.get(url, params=params)
    data = response.json()["result"]
    df = pd.DataFrame(data)
    df["close"] = pd.to_numeric(df["close"])
    return df["close"].values

# データの前処理
def create_dataset(data, look_back=20):
    X, y = [], []
    for i in range(len(data) - look_back - 1):
        X.append(data[i:(i + look_back)])
        y.append(data[i + look_back])
    return np.array(X), np.array(y)

# LSTMモデルの構築
def build_model(look_back):
    model = keras.Sequential([
        layers.LSTM(64, return_sequences=True, input_shape=(look_back, 1)),
        layers.LSTM(32),
        layers.Dense(16, activation='relu'),
        layers.Dense(1)
    ])
    model.compile(optimizer='adam', loss='mse')
    return model

# 予測とエントリー/エグジットの判定
def predict_and_trade(model, symbol, look_back, threshold=0.01):
    data = get_klines(symbol)
    current_data = np.array(data[-look_back:]).reshape(1, look_back, 1)
    predicted_price = model.predict(current_data)[0][0]
    current_price = data[-1]

    if predicted_price >= current_price * (1 + threshold):
        print(f"[BUY SIGNAL] Current: {current_price} | Predicted: {predicted_price}")
        place_order(symbol, "Buy", TRADE_AMOUNT)

    elif predicted_price <= current_price * (1 - threshold):
        print(f"[SELL SIGNAL] Current: {current_price} | Predicted: {predicted_price}")
        place_order(symbol, "Sell", TRADE_AMOUNT)

# 注文実行関数
def place_order(symbol, side, qty):
    url = f"{BASE_URL}/v2/private/order/create"
    params = {
        "api_key": API_KEY,
        "symbol": symbol,
        "side": side,
        "order_type": "Market",
        "qty": qty,
        "time_in_force": "GoodTillCancel",
        "timestamp": int(time.time() * 1000)
    }
    params["sign"] = sign_request(params)
    response = requests.post(url, data=params)
    print(response.json())

# API署名生成関数
def sign_request(params):
    sorted_params = sorted(params.items())
    query_string = "&".join(f"{key}={value}" for key, value in sorted_params)
    signature = hmac.new(
        API_SECRET.encode(), query_string.encode(), hashlib.sha256
    ).hexdigest()
    return signature

# メイン関数
def main():
    symbol = "BTCUSDT"
    look_back = 20
    data = get_klines(symbol)
    X, y = create_dataset(data, look_back)
    X = X.reshape(X.shape[0], X.shape[1], 1)

    model = build_model(look_back)
    model.fit(X, y, epochs=20, batch_size=32, verbose=1)

    while True:
        try:
            predict_and_trade(model, symbol, look_back)
        except Exception as e:
            print(f"Error: {e}")
        time.sleep(300)  # 5分ごとにチェック

if __name__ == "__main__":
    main()

🚀 主要な機能

✅ Bybit API を使用してリアルタイムのK線データを取得
✅ LSTM (Long Short-Term Memory) モデルを用いて将来の価格を予測
✅ 予測価格と現在価格の比較により、以下のアクションを実行

  • 予測価格が「現在価格 + 1%以上」なら 買い (Buy)
  • 予測価格が「現在価格 - 1%以下」なら 売り (Sell)
    ✅ 定期的にデータを更新し、5分ごとにシグナルを確認

🔧 カスタマイズ可能なパラメータ

  • look_back：LSTMが過去のデータを学習する期間 (例: 20 → 過去20期間のデータ)
  • TRADE_AMOUNT：取引量 (例: 0.01 BTC)
  • threshold：エントリー/エグジットの判定閾値 (例: 0.01 → 1%)
  • API_KEY と API_SECRET：BybitのAPIキー情報

⚠️ 注意点

  1. Bybit APIのキーとシークレットキーは適切に管理する。
  2. LSTMモデルの学習に十分なデータが必要なため、limit=200 以上のK線データを取得するのが望ましい。
  3. 実際の取引を行う前に、Bybitの「テストネット」を使用してテストする。
  4. ボラティリティが激しい相場では、threshold の値を調整することで、より安定したエントリー/エグジットが可能。

📈 実用化に向けた次のステップ

  • トレーリングストップの追加で、利益確定の最適化
  • 強化学習 (DQN / DDPG) と組み合わせ、より高度な判断モデルを構築
  • リスク管理モジュールを追加して、最大損失をコントロール

まとめ

今回は、「バミューダンオプション（Bermudan option）の価格に対するモデル非依存の上限についての研究」を元に仮説を立てて、４種類のbotを作成してみました。

Yodaka

論文ベースのbot作りは、開発ペースを早めることができるのでとても良いなと感じます。

今後もこの調子で開発の状況を発信していきます。

-Bot, トレードロジック, 書籍・論文・web記事

S