開発記録#135(2025/3/20)「新規上場株式およびスピンオフの実現ボラティリティ予測におけるマルチソース転移学習の活用」

2025年3月20日 2025年3月22日

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

Yodaka

今回も論文を元にしてbot開発をしています。

本記事では「新規上場株式およびスピンオフの実現ボラティリティ予測におけるマルチソース転移学習の活用」を元に取引戦略の仮説を立てて、6種類のbotを紹介します。

新規上場株式およびスピンオフの実現ボラティリティ予測におけるマルチソース転移学習の活用https://t.co/VGlBWxLTe8

— よだか(夜鷹/yodaka) (@yodakablog) March 19, 2025

参考

Realized Volatility Forecasting for New Issues and Spin-Offs using Multi-Source Transfer Learning

論文の要約

研究の背景と目的

金融資産のボラティリティ予測は、リスク管理や投資戦略において重要な課題である。
しかし、新規上場株式 (New Issues) やスピンオフ (Spin-Offs) など、過去のデータが極めて限られた資産のボラティリティを正確に予測するのは困難である。
本研究では、このデータ不足問題を克服するために、マルチソース転移学習 (Multi-Source Transfer Learning: MTL) を活用した新たな手法を提案する。

提案手法: マルチソース転移学習 (MTL)

データの豊富な資産 (ソースデータ) から、新規上場株式/スピンオフ (ターゲットデータ) に類似した部分時系列データを選択して活用する。
動的時間ワーピング (Dynamic Time Warping: DTW) を用いて、ターゲットデータと最も類似したソースデータのサブシーケンス (部分データ) を選定する。
この選定したデータをターゲットデータに統合し、以下の3種類のモデルを用いてボラティリティを予測した：
- HARモデル (線形時系列モデル)
- FNNモデル (フィードフォワード型ニューラルネットワーク)
- XGBoostモデル (勾配ブースティング機械学習モデル)

データセット

ターゲットデータ: 2010年〜2020年の間に新規上場またはスピンオフとして登場し、2022年11月1日時点でS&P 500に含まれていた10銘柄。
ソースデータ: 66のS&P 500銘柄の1分足データ (各GICSセクターから幅広く選定)。
ボラティリティ予測には、5分足データを用いて実現分散 (Realized Variance) を算出。

主な結果

MTLアプローチの優位性:
- MTLアプローチは、ターゲットデータのみを使用したモデル (TOモデル) や、ソースデータをすべて組み合わせたナイーブプール法 (NPモデル) に対して、最も高い予測精度を示した。
- 特に、XGBoostモデルにMTLを組み合わせた「MTL-75 XGB-EXT」が、すべての手法の中で最も優れたパフォーマンスを発揮した。
データが限られた初期段階でも精度向上:
- 新規上場株式/スピンオフの初日の取引終了直後からの1日先ボラティリティ予測でも、MTLアプローチが高い精度を発揮した。
選択するデータ量の影響:
- MTLアプローチでは、ソースデータのうち最も類似性の高い25%〜75%のサブシーケンスを選択するのが最も効果的だった。
- データの選定が少なすぎると、パフォーマンスが低下する一方、75%以上のデータを活用しても大きな改善は得られなかった。

結論

本研究は、金融資産のデータが不足する状況において、マルチソース転移学習が非常に有効なアプローチであることを示した。
特に、XGBoostモデルとの組み合わせが優れた成果を示し、今後の新規上場株式/スピンオフ銘柄の予測に役立つ可能性が高い。

仮想通貨の自動取引ボットへの応用アイデア

本論文で提案された 「マルチソース転移学習 (MTL)」 のアプローチは、仮想通貨の自動取引ボットの開発に応用できる可能性があります。特に、新規上場トークンや取引開始直後の資産に対するボラティリティ予測やトレンド分析に有効です。

Yodaka

以下に具体的なアイデアを6つ紹介します。

1. 新規上場トークンの初期価格変動予測

⏩ 応用方法

新規上場トークンは、過去データが極端に少ないため、従来のボラティリティ予測手法がうまく機能しません。
MTLアプローチを用いて、類似のトークン (ソースデータ) から最も相関の高い価格推移パターンを抽出し、新規上場トークン (ターゲットデータ) の初期ボラティリティや短期トレンドを予測する。
これにより、新規上場直後の急騰・急落のリスク管理や早期参入のタイミング判断に役立つ。

🚀 実装例

BybitやBinanceの上場直後のトークンをターゲットとし、ソースデータには同様の特性を持つトークン (例: 同種のDeFiトークン、Metaverseトークン) を利用する。
DTW (Dynamic Time Warping) を用いて類似データを選定し、XGBoostモデルで予測精度を高める。

コード：新規上場トークンの初期価格変動予測＆取引Bot

Yodaka

以下は、論文の「マルチソース転移学習 (MTL)」のアプローチを応用し、Bybitの新規上場トークンの初期価格変動を予測し、自動取引を行うPythonコードです。

機能の概要

✅ 新規上場トークンのデータ収集 (Bybit API)
✅ 過去の類似資産データ (ソースデータ) の収集
✅ DTW (Dynamic Time Warping) を用いた類似データ選定
✅ XGBoostを使用した1分足ベースのボラティリティ予測
✅ 予測に基づく自動取引機能 (エントリー＆イグジット)

必要ライブラリ: ccxt, numpy, pandas, xgboost, dtaidistance (DTW用)

import ccxt
import pandas as pd
import numpy as np
from xgboost import XGBRegressor
from dtaidistance import dtw
import time

# Bybit API接続設定
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"
bybit = ccxt.bybit({
    'apiKey': api_key,
    'secret': api_secret,
    'enableRateLimit': True,
})

# ========== データ取得関数 ==========
def fetch_data(symbol, timeframe='1m', limit=500):
    """Bybitから1分足データを取得"""
    ohlcv = bybit.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df

# ========== DTWによる類似データ選定 ==========
def select_similar_data(target_series, source_data):
    distances = []
    for symbol, df in source_data.items():
        dist = dtw.distance(target_series, df['close'].values[-len(target_series):])
        distances.append((symbol, dist))
    return sorted(distances, key=lambda x: x[1])[:5]  # 最も類似した5つのデータを選定

# ========== 価格予測モデル (XGBoost) ==========
def train_model(X_train, y_train):
    model = XGBRegressor(objective='reg:squarederror', n_estimators=100)
    model.fit(X_train, y_train)
    return model

# ========== 自動取引ロジック ==========
def execute_trade(prediction, current_price, threshold=0.02):
    if prediction > current_price * (1 + threshold):
        print(f"[BUY] 価格上昇予測: {prediction:.2f}")
        # bybit.create_market_buy_order("BTC/USDT", 0.01)
    elif prediction < current_price * (1 - threshold):
        print(f"[SELL] 価格下落予測: {prediction:.2f}")
        # bybit.create_market_sell_order("BTC/USDT", 0.01)
    else:
        print(f"[HOLD] 価格変動なし: {prediction:.2f}")

# ========== メインロジック ==========
new_token = "BTC/USDT"
source_symbols = ["ETH/USDT", "SOL/USDT", "XRP/USDT", "DOGE/USDT"]

# データ取得
new_token_data = fetch_data(new_token)
source_data = {symbol: fetch_data(symbol) for symbol in source_symbols}

# DTWによる類似データ選定
target_series = new_token_data['close'].values[-50:]
similar_data = select_similar_data(target_series, source_data)

# 学習データ作成
X_train = []
y_train = []
for symbol, _ in similar_data:
    df = source_data[symbol]
    X_train.append(df[['close']].values[-50:-1])
    y_train.append(df['close'].values[-49:])

X_train = np.array(X_train).reshape(-1, 1)
y_train = np.array(y_train).reshape(-1)

# モデル学習と予測
model = train_model(X_train, y_train)
latest_price = new_token_data['close'].values[-1]
prediction = model.predict(np.array([[latest_price]]))[0]

# 取引実行
execute_trade(prediction, latest_price)

コードの説明

データ取得：Bybit APIを利用して、指定したトークンの1分足データを取得。
DTWによる類似データ選定：ターゲットデータに最も類似する5つのソースデータを選定。
XGBoostモデル：選定したデータを使い、1分後の価格予測モデルを学習。
自動取引ロジック：予測価格が現価格の2%上昇/下降で「買い」「売り」を判断。

実行手順

BybitのAPIキーとシークレットキーを取得して、コードに入力。
必要ライブラリをインストール。
上記コードを実行し、新規上場トークンのデータ収集と価格予測を行う。
予測が「2%以上の上昇/下降」となった場合に自動売買が実行されます。

#ライブラリのインストール
pip install ccxt xgboost dtaidistance pandas numpy

カスタマイズの提案

🔹 学習データ量の調整: target_series = new_token_data['close'].values[-50:] の部分を -100 や -200 などに変更可能。
🔹 取引条件の調整: threshold を 0.01 (1%) などに調整して、より積極的な取引ロジックに変更可能。
🔹 サーバー運用: このBotはDockerコンテナ化することで、Bybitのリアルタイム環境で安定的に稼働可能。

次のステップ

✅ コードの動作確認
✅ 取引ロジックの最適化 (例: RSIやMACDなどの指標追加)
✅ サーバー環境へのデプロイ (Docker + Kubernetes対応)

2. 低流動性コインの価格予測

⏩ 応用方法

流動性の低い仮想通貨は、価格が不安定で予測が困難です。
MTLアプローチを利用して、他の同様の流動性パターンを持つコインデータから有益なデータを抽出し、短期価格変動や急騰/急落のタイミングを予測する。

🚀 実装例

例えば、マイナーなDeFiトークンの価格予測に対し、同カテゴリの主要トークン (例: Uniswap, SushiSwap) のデータを補完データとして使用する。
ソースデータを「流動性が低く、急騰経験のある資産」に限定して選択すると効果的。

コード: 低流動性コインの価格予測＆自動取引Bot

Yodaka

以下は、論文の「マルチソース転移学習 (MTL)」のアプローチを応用し、低流動性の仮想通貨の価格変動を予測し、Bybitで自動取引を行うPythonコードです。

機能の概要

✅ Bybit APIから低流動性コインのデータ取得
✅ 似たボラティリティパターンを持つ通貨をデータソースとして選定 (DTW活用)
✅ XGBoostを使用した1分足ベースのボラティリティ予測
✅ 予測に基づく自動取引機能 (エントリー＆イグジット)

必要ライブラリ: ccxt, numpy, pandas, xgboost, dtaidistance (DTW用)

import ccxt
import pandas as pd
import numpy as np
from xgboost import XGBRegressor
from dtaidistance import dtw
import time

# Bybit API接続設定
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"
bybit = ccxt.bybit({
    'apiKey': api_key,
    'secret': api_secret,
    'enableRateLimit': True,
})

# ========== データ取得関数 ==========
def fetch_data(symbol, timeframe='1m', limit=500):
    """Bybitから1分足データを取得"""
    ohlcv = bybit.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df

# ========== DTWによる類似データ選定 ==========
def select_similar_data(target_series, source_data):
    distances = []
    for symbol, df in source_data.items():
        dist = dtw.distance(target_series, df['close'].values[-len(target_series):])
        distances.append((symbol, dist))
    return sorted(distances, key=lambda x: x[1])[:5]  # 最も類似した5つのデータを選定

# ========== 価格予測モデル (XGBoost) ==========
def train_model(X_train, y_train):
    model = XGBRegressor(objective='reg:squarederror', n_estimators=100)
    model.fit(X_train, y_train)
    return model

# ========== 自動取引ロジック ==========
def execute_trade(prediction, current_price, threshold=0.03):
    if prediction > current_price * (1 + threshold):
        print(f"[BUY] 価格上昇予測: {prediction:.2f}")
        # bybit.create_market_buy_order("LOW_LIQ_COIN/USDT", 0.01)
    elif prediction < current_price * (1 - threshold):
        print(f"[SELL] 価格下落予測: {prediction:.2f}")
        # bybit.create_market_sell_order("LOW_LIQ_COIN/USDT", 0.01)
    else:
        print(f"[HOLD] 価格変動なし: {prediction:.2f}")

# ========== メインロジック ==========
low_liquidity_token = "LOW_LIQ_COIN/USDT"  # 低流動性コインのシンボル
source_symbols = ["BTC/USDT", "ETH/USDT", "XRP/USDT", "DOGE/USDT"]

# データ取得
low_liquidity_data = fetch_data(low_liquidity_token)
source_data = {symbol: fetch_data(symbol) for symbol in source_symbols}

# DTWによる類似データ選定
target_series = low_liquidity_data['close'].values[-50:]
similar_data = select_similar_data(target_series, source_data)

# 学習データ作成
X_train = []
y_train = []
for symbol, _ in similar_data:
    df = source_data[symbol]
    X_train.append(df[['close']].values[-50:-1])
    y_train.append(df['close'].values[-49:])

X_train = np.array(X_train).reshape(-1, 1)
y_train = np.array(y_train).reshape(-1)

# モデル学習と予測
model = train_model(X_train, y_train)
latest_price = low_liquidity_data['close'].values[-1]
prediction = model.predict(np.array([[latest_price]]))[0]

# 取引実行
execute_trade(prediction, latest_price)

コードの説明

データ取得：Bybit APIを利用して、指定した低流動性コインの1分足データを取得。
DTWによる類似データ選定：ターゲットデータに最も類似する5つのソースデータを選定。
XGBoostモデル：選定したデータを使い、1分後の価格予測モデルを学習。
自動取引ロジック：予測価格が現価格の3%上昇/下降で「買い」「売り」を判断。

実行手順

BybitのAPIキーとシークレットキーを取得して、コード内の YOUR_API_KEY と YOUR_API_SECRET に入力。
必要ライブラリをインストール。
コード内の low_liquidity_token には、ターゲットとなる低流動性の銘柄を指定。
コードを実行し、リアルタイムのデータ収集＆価格予測が開始されます。

#ライブラリのインストール
pip install ccxt xgboost dtaidistance pandas numpy

カスタマイズの提案

🔹 学習データ量の調整: target_series = low_liquidity_data['close'].values[-50:] の部分を -100 や -200 に変更可能。
🔹 取引条件の調整: threshold を 0.02 や 0.01 などに調整して、より積極的な取引に変更可能。
🔹 エラー処理: ネットワーク障害やAPIの制限に対応するリトライ処理を追加して安定性を向上させる。

次のステップ

✅ コードの動作確認
✅ 取引ロジックの最適化 (例: RSIやMACDなどの指標追加)
✅ サーバー環境へのデプロイ (Docker + Kubernetes対応)

3. サーキットブレーカー (損切り) トリガーの最適化

⏩ 応用方法

サーキットブレーカーBotのロジックとして、MTLアプローチで急激な価格変動の直前に発生するパターン検出を導入する。
急落前のボラティリティ上昇パターンを抽出し、損切りトリガーの最適化に活用する。

🚀 実装例

急落前に共通して現れる「価格の上下振れの加速パターン」をDTWを用いて検出し、Botのサーキットブレーカートリガーを調整する。

コード: サーキットブレーカー最適化Bot

Yodaka

以下は、論文の「マルチソース転移学習 (MTL)」のアプローチを応用し、Bybitの自動取引Botにおいて、サーキットブレーカー (損切り) トリガーの最適化を行うPythonコードです。

必要ライブラリ: ccxt, numpy, pandas, xgboost, dtaidistance (DTW用)

import ccxt
import pandas as pd
import numpy as np
from xgboost import XGBClassifier
from dtaidistance import dtw
import time

# Bybit API接続設定
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"
bybit = ccxt.bybit({
    'apiKey': api_key,
    'secret': api_secret,
    'enableRateLimit': True,
})

# ========== データ取得関数 ==========
def fetch_data(symbol, timeframe='1m', limit=500):
    """Bybitから1分足データを取得"""
    ohlcv = bybit.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df

# ========== DTWによる急落直前のパターン選定 ==========
def select_crash_patterns(target_series, source_data):
    distances = []
    for symbol, df in source_data.items():
        dist = dtw.distance(target_series, df['close'].values[-len(target_series):])
        distances.append((symbol, dist))
    return sorted(distances, key=lambda x: x[1])[:5]  # 最も類似した5つのデータを選定

# ========== XGBoostモデル (急落シグナル予測) ==========
def train_model(X_train, y_train):
    model = XGBClassifier(objective='binary:logistic', n_estimators=100)
    model.fit(X_train, y_train)
    return model

# ========== サーキットブレーカー (損切り) ロジック ==========
def circuit_breaker(prediction, current_price, stop_loss=0.03):
    if prediction == 1:
        print(f"[ALERT] 急落の可能性検出 - 損切り実行: {current_price:.2f}")
        # bybit.create_market_sell_order("BTC/USDT", 0.01)
    else:
        print(f"[NORMAL] 異常検出なし: {current_price:.2f}")

# ========== メインロジック ==========
trading_token = "BTC/USDT"
source_symbols = ["ETH/USDT", "SOL/USDT", "XRP/USDT", "DOGE/USDT"]

# データ取得
target_data = fetch_data(trading_token)
source_data = {symbol: fetch_data(symbol) for symbol in source_symbols}

# DTWによる類似データ選定
target_series = target_data['close'].values[-50:]
similar_data = select_crash_patterns(target_series, source_data)

# 学習データ作成
X_train = []
y_train = []
for symbol, _ in similar_data:
    df = source_data[symbol]
    X_train.append(df[['close']].values[-50:-1])
    # 急落直前のデータを「1」、通常時を「0」としてラベル化
    y_train.append([1 if df['close'].pct_change().iloc[-1] < -0.03 else 0])

X_train = np.array(X_train).reshape(-1, 1)
y_train = np.array(y_train).reshape(-1)

# モデル学習と予測
model = train_model(X_train, y_train)
latest_price = target_data['close'].values[-1]
prediction = model.predict(np.array([[latest_price]]))[0]

# サーキットブレーカー作動
circuit_breaker(prediction, latest_price)

機能の概要

✅ Bybit APIから1分足データを取得
✅ DTW (Dynamic Time Warping) を活用した急落直前のパターン検出
✅ XGBoostで急落直前のシグナルを予測し、損切りトリガーを最適化
✅ 急落を検出すると即時ポジションをクローズして損失回避

コードの説明

データ取得：Bybit APIを利用して、指定した銘柄の1分足データを取得。
DTWによる類似データ選定：ターゲットデータに最も類似する急落パターンを持つデータを選定。
XGBoostモデル：選定したデータを使用し、急落の直前パターンを予測。
サーキットブレーカー (損切り) トリガー：
- 予測が 1 なら「急落検出」とみなし、ポジションをクローズ。
- 予測が 0 なら「異常なし」として、ホールドを継続。

実行手順

BybitのAPIキーとシークレットキーを取得し、コード内の YOUR_API_KEY と YOUR_API_SECRET に入力。
必要ライブラリをインストール。
コード内の trading_token にターゲット通貨を指定し、 source_symbols に類似した資産を登録。
コードを実行すると、リアルタイムの急落検出＆損切りが実行されます。

#ライブラリのインストール
pip install ccxt xgboost dtaidistance pandas numpy

カスタマイズの提案

🔹 学習データ量の調整: target_series = target_data['close'].values[-50:] の部分を -100 や -200 に変更可能。
🔹 急落基準の調整: stop_loss=0.03 を 0.02 や 0.01 などに変更して、感度を調整可能。
🔹 リトライ処理の追加: ネットワーク障害やAPIの制限に対応する処理を追加すると安定性が向上します。

次のステップ

✅ コードの動作確認
✅ サーキットブレーカーの最適化 (例: 短期ボラティリティやRSIの組み合わせ)
✅ サーバー環境へのデプロイ (Docker + Kubernetes対応)

4. スキャルピングボットのエントリー/イグジット最適化

⏩ 応用方法

スキャルピングボットは超短期の価格変動に基づいてトレードしますが、ランダムノイズに惑わされるリスクが高い。
MTLアプローチにより、「ノイズではない短期的なトレンド」を識別することで、誤ったエントリーやイグジットを防止できる。

🚀 実装例

BybitのBTC/USDT 1分足データにおいて、過去の「急騰直前のノイズ」と「実際の急騰前シグナル」を比較し、MTLアプローチでパターンマッチングを行う。

コード: スキャルピングBot (MTL + XGBoost)

Yodaka

以下は、論文の「マルチソース転移学習 (MTL)」のアプローチを応用し、Bybitのスキャルピングボットにおいて、エントリー/イグジット最適化を行うPythonコードです。

機能の概要

✅ Bybit APIから1分足データを取得
✅ DTW (Dynamic Time Warping) を活用したノイズと実際のシグナルのパターン識別
✅ XGBoostで「エントリー」「イグジット」の最適タイミングを予測
✅ 予測に基づく自動取引機能

必要ライブラリ: ccxt, numpy, pandas, xgboost, dtaidistance (DTW用)

import ccxt
import pandas as pd
import numpy as np
from xgboost import XGBClassifier
from dtaidistance import dtw
import time

# Bybit API接続設定
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"
bybit = ccxt.bybit({
    'apiKey': api_key,
    'secret': api_secret,
    'enableRateLimit': True,
})

# ========== データ取得関数 ==========
def fetch_data(symbol, timeframe='1m', limit=500):
    """Bybitから1分足データを取得"""
    ohlcv = bybit.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df

# ========== DTWによる類似パターン検出 ==========
def select_patterns(target_series, source_data):
    distances = []
    for symbol, df in source_data.items():
        dist = dtw.distance(target_series, df['close'].values[-len(target_series):])
        distances.append((symbol, dist))
    return sorted(distances, key=lambda x: x[1])[:5]  # 最も類似した5つのデータを選定

# ========== XGBoostモデル (エントリー/イグジット予測) ==========
def train_model(X_train, y_train):
    model = XGBClassifier(objective='binary:logistic', n_estimators=100)
    model.fit(X_train, y_train)
    return model

# ========== 自動取引ロジック ==========
def scalping_trade(prediction, current_price):
    if prediction == 1:
        print(f"[BUY] エントリー: {current_price:.2f}")
        # bybit.create_market_buy_order("BTC/USDT", 0.01)
    elif prediction == -1:
        print(f"[SELL] イグジット: {current_price:.2f}")
        # bybit.create_market_sell_order("BTC/USDT", 0.01)
    else:
        print(f"[HOLD] 様子見: {current_price:.2f}")

# ========== メインロジック ==========
trading_token = "BTC/USDT"
source_symbols = ["ETH/USDT", "SOL/USDT", "XRP/USDT", "DOGE/USDT"]

# データ取得
target_data = fetch_data(trading_token)
source_data = {symbol: fetch_data(symbol) for symbol in source_symbols}

# DTWによる類似データ選定
target_series = target_data['close'].values[-50:]
similar_data = select_patterns(target_series, source_data)

# 学習データ作成
X_train = []
y_train = []
for symbol, _ in similar_data:
    df = source_data[symbol]
    X_train.append(df[['close']].values[-50:-1])
    price_change = df['close'].pct_change().iloc[-1]
    if price_change > 0.005:  # 0.5%以上の急騰 ⇒ 買いシグナル
        y_train.append(1)
    elif price_change < -0.005:  # 0.5%以上の急落 ⇒ 売りシグナル
        y_train.append(-1)
    else:
        y_train.append(0)

X_train = np.array(X_train).reshape(-1, 1)
y_train = np.array(y_train).reshape(-1)

# モデル学習と予測
model = train_model(X_train, y_train)
latest_price = target_data['close'].values[-1]
prediction = model.predict(np.array([[latest_price]]))[0]

# 自動取引
scalping_trade(prediction, latest_price)

コードの説明

データ取得：Bybit APIを利用して、指定した銘柄の1分足データを取得。
DTWによる類似データ選定：ターゲットデータに最も類似する価格変動パターンを選定。
XGBoostモデル：選定したデータを使用し、エントリー (買い) と イグジット (売り) の最適タイミングを予測。
自動取引ロジック：
- 買いシグナル (1)：0.5%以上の急騰を検出した場合。
- 売りシグナル (-1)：0.5%以上の急落を検出した場合。
- 様子見 (0)：小幅な変動の場合。

実行手順

BybitのAPIキーとシークレットキーを取得し、コード内の YOUR_API_KEY と YOUR_API_SECRET に入力。
必要ライブラリをインストール。
コード内の trading_token にターゲット通貨を指定し、 source_symbols に類似した資産を登録。
コードを実行すると、リアルタイムのエントリー/イグジット予測＆自動取引が実行されます。

pip install ccxt xgboost dtaidistance pandas numpy

カスタマイズの提案

🔹 学習データ量の調整: target_series = target_data['close'].values[-50:] の部分を -100 や -200 に変更可能。
🔹 シグナル感度の調整: price_change > 0.005 (0.5%) などの閾値を変更して、エントリー/イグジットの頻度を最適化。
🔹 リトライ処理の追加: ネットワーク障害やAPIの制限に対応する処理を追加すると安定性が向上する。

次のステップ

✅ コードの動作確認
✅ シグナル精度の向上 (例: RSIやMACDの組み合わせ)
✅ サーバー環境へのデプロイ (Docker + Kubernetes対応)

5. 積立Botの「押し目買い」戦略強化

⏩ 応用方法

積立Botの運用では、価格下落時に積立額を増やす「押し目買い (DCA)」戦略が重要です。
MTLアプローチを利用して「価格が底を打つ直前のボラティリティパターン」を識別し、より精度の高い押し目買いタイミングを実現する。

🚀 実装例

BTCの長期価格データと、過去の急落時のボラティリティパターンをDTWで分析し、最適なDCAタイミングをBotに組み込む。

コード：積立Botの「押し目買い」戦略強化＆自動取引Bot

Yodaka

論文の「マルチソース転移学習 (MTL)」のアプローチを応用し、Bybitの積立Botにおいて、押し目買い (DCA: Dollar Cost Averaging) 戦略の強化を行うPythonコードです。

機能の概要

✅ Bybit APIから日足データを取得
✅ DTW (Dynamic Time Warping) を活用した押し目パターン検出
✅ XGBoostを用いて「押し目」の最適タイミングを予測
✅ 予測に基づく積立額の増額を実行

必要ライブラリ: ccxt, numpy, pandas, xgboost, dtaidistance (DTW用)

import ccxt
import pandas as pd
import numpy as np
from xgboost import XGBClassifier
from dtaidistance import dtw
import time

# Bybit API接続設定
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"
bybit = ccxt.bybit({
    'apiKey': api_key,
    'secret': api_secret,
    'enableRateLimit': True,
})

# ========== データ取得関数 ==========
def fetch_data(symbol, timeframe='1d', limit=500):
    """Bybitから日足データを取得"""
    ohlcv = bybit.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df

# ========== DTWによる押し目パターン選定 ==========
def select_dip_patterns(target_series, source_data):
    distances = []
    for symbol, df in source_data.items():
        dist = dtw.distance(target_series, df['close'].values[-len(target_series):])
        distances.append((symbol, dist))
    return sorted(distances, key=lambda x: x[1])[:5]  # 最も類似した5つのデータを選定

# ========== XGBoostモデル (押し目買い予測) ==========
def train_model(X_train, y_train):
    model = XGBClassifier(objective='binary:logistic', n_estimators=100)
    model.fit(X_train, y_train)
    return model

# ========== 押し目買いロジック ==========
def dca_strategy(prediction, current_price, base_amount=10):
    if prediction == 1:
        buy_amount = base_amount * 2  # 押し目買いで積立額を増額
        print(f"[BUY MORE] 押し目検出 - 追加購入: {buy_amount} USDT")
        # bybit.create_market_buy_order("BTC/USDT", buy_amount)
    else:
        print(f"[HOLD] 通常積立: {base_amount} USDT")
        # bybit.create_market_buy_order("BTC/USDT", base_amount)

# ========== メインロジック ==========
trading_token = "BTC/USDT"
source_symbols = ["ETH/USDT", "SOL/USDT", "XRP/USDT", "DOGE/USDT"]

# データ取得
target_data = fetch_data(trading_token)
source_data = {symbol: fetch_data(symbol) for symbol in source_symbols}

# DTWによる類似データ選定
target_series = target_data['close'].values[-50:]
similar_data = select_dip_patterns(target_series, source_data)

# 学習データ作成
X_train = []
y_train = []
for symbol, _ in similar_data:
    df = source_data[symbol]
    X_train.append(df[['close']].values[-50:-1])
    # 押し目パターンが発生した場合は「1」、それ以外は「0」
    y_train.append([1 if df['close'].pct_change().iloc[-1] < -0.03 else 0])

X_train = np.array(X_train).reshape(-1, 1)
y_train = np.array(y_train).reshape(-1)

# モデル学習と予測
model = train_model(X_train, y_train)
latest_price = target_data['close'].values[-1]
prediction = model.predict(np.array([[latest_price]]))[0]

# 押し目買いロジック発動
dca_strategy(prediction, latest_price)

コードの説明

データ取得：Bybit APIを利用して、指定した銘柄の日足データを取得。
DTWによる押し目パターン選定：ターゲットデータに最も類似する押し目パターンを持つデータを選定。
XGBoostモデル：選定したデータを使用し、押し目シグナルを予測。
押し目買いロジック：
- 押し目検出時 (予測 = 1)：通常の2倍の積立額を購入。
- 押し目以外のタイミング (予測 = 0)：通常の積立額を購入。

実行手順

BybitのAPIキーとシークレットキーを取得し、コード内の YOUR_API_KEY と YOUR_API_SECRET に入力。
必要ライブラリをインストール： bashコピーする編集するpip install ccxt xgboost dtaidistance pandas numpy
コード内の trading_token にターゲット通貨を指定し、 source_symbols に類似した資産を登録。
コードを実行すると、リアルタイムの押し目検出＆積立増額が実行されます。

カスタマイズの提案

🔹 学習データ量の調整: target_series = target_data['close'].values[-50:] の部分を -100 や -200 に変更可能。
🔹 押し目基準の調整: y_train.append([1 if df['close'].pct_change().iloc[-1] < -0.03 else 0]) で -0.03 の値を変更し、押し目検出の感度を最適化できます。
🔹 積立額の変更: base_amount を調整し、資金管理を最適化可能。

次のステップ

✅ コードの動作確認
✅ 押し目検出の最適化 (例: RSIやMACDの組み合わせ)
✅ サーバー環境へのデプロイ (Docker + Kubernetes対応)

6. 異常検出と市場の変化察知

⏩ 応用方法

仮想通貨市場は外部要因 (規制発表、ハッキング事件など) で急変することが多い。
MTLアプローチを利用して、過去の異常事象 (例: LUNA暴落、FTX破綻) の直前の市場パターンを学習し、異常検出アラートを生成する。

🚀 実装例

「異常発生直前のトレンドパターン」と類似した状況が発生した場合にアラートを出し、Botが自動でポジションを閉じるように設計。

コード: 異常検出＆市場変化察知Bot

Yodaka

以下は、論文の「マルチソース転移学習 (MTL)」のアプローチを応用し、Bybitの自動取引Botにおいて、異常検出＆市場の変化察知を行い、即時対応するコードです。

機能の概要

✅ Bybit APIから1分足データを取得
✅ DTW (Dynamic Time Warping) を活用した市場急変の類似パターン検出
✅ XGBoostで「異常検出」と「市場急変の予測」を実行
✅ 予測に基づき、ポジションのクローズやリスク回避のエントリーを自動化

必要ライブラリ: ccxt, numpy, pandas, xgboost, dtaidistance (DTW用)

import ccxt
import pandas as pd
import numpy as np
from xgboost import XGBClassifier
from dtaidistance import dtw
import time

# Bybit API接続設定
api_key = "YOUR_API_KEY"
api_secret = "YOUR_API_SECRET"
bybit = ccxt.bybit({
    'apiKey': api_key,
    'secret': api_secret,
    'enableRateLimit': True,
})

# ========== データ取得関数 ==========
def fetch_data(symbol, timeframe='1m', limit=500):
    """Bybitから1分足データを取得"""
    ohlcv = bybit.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    return df

# ========== DTWによる異常パターン選定 ==========
def select_anomaly_patterns(target_series, source_data):
    distances = []
    for symbol, df in source_data.items():
        dist = dtw.distance(target_series, df['close'].values[-len(target_series):])
        distances.append((symbol, dist))
    return sorted(distances, key=lambda x: x[1])[:5]  # 最も類似した5つのデータを選定

# ========== XGBoostモデル (異常検出＆市場急変予測) ==========
def train_model(X_train, y_train):
    model = XGBClassifier(objective='binary:logistic', n_estimators=100)
    model.fit(X_train, y_train)
    return model

# ========== 異常検出と緊急対応ロジック ==========
def emergency_trade(prediction, current_price):
    if prediction == 1:
        print(f"[ALERT] 市場急変検出 - ポジションをクローズ: {current_price:.2f}")
        # bybit.create_market_sell_order("BTC/USDT", 0.01)
    else:
        print(f"[NORMAL] 市場安定 - 継続ホールド: {current_price:.2f}")

# ========== メインロジック ==========
trading_token = "BTC/USDT"
source_symbols = ["ETH/USDT", "SOL/USDT", "XRP/USDT", "DOGE/USDT"]

# データ取得
target_data = fetch_data(trading_token)
source_data = {symbol: fetch_data(symbol) for symbol in source_symbols}

# DTWによる異常パターン選定
target_series = target_data['close'].values[-50:]
similar_data = select_anomaly_patterns(target_series, source_data)

# 学習データ作成
X_train = []
y_train = []
for symbol, _ in similar_data:
    df = source_data[symbol]
    X_train.append(df[['close']].values[-50:-1])
    # 3%以上の急騰 or 急落を「異常 (1)」、通常変動を「通常 (0)」
    price_change = df['close'].pct_change().iloc[-1]
    y_train.append(1 if abs(price_change) > 0.03 else 0)

X_train = np.array(X_train).reshape(-1, 1)
y_train = np.array(y_train).reshape(-1)

# モデル学習と予測
model = train_model(X_train, y_train)
latest_price = target_data['close'].values[-1]
prediction = model.predict(np.array([[latest_price]]))[0]

# 異常検出と緊急対応
emergency_trade(prediction, latest_price)

コードの説明

データ取得：Bybit APIを利用して、指定した銘柄の1分足データを取得。
DTWによる異常パターン選定：ターゲットデータに最も類似する急変パターンを持つデータを選定。
XGBoostモデル：選定したデータを使用し、市場急変のシグナルを予測。
異常検出ロジック：
- 市場急変検出時 (予測 = 1)：ポジションを即時クローズして損失回避。
- 通常の値動き (予測 = 0)：ホールド継続。

実行手順

BybitのAPIキーとシークレットキーを取得し、コード内の YOUR_API_KEY と YOUR_API_SECRET に入力。
必要ライブラリをインストール： bashコピーする編集するpip install ccxt xgboost dtaidistance pandas numpy
コード内の trading_token にターゲット通貨を指定し、 source_symbols に類似した資産を登録。
コードを実行すると、リアルタイムの異常検出＆緊急対応が実行されます。

カスタマイズの提案

🔹 学習データ量の調整: target_series = target_data['close'].values[-50:] の部分を -100 や -200 に変更可能。
🔹 異常検出の感度調整: abs(price_change) > 0.03 の値を 0.02 や 0.05 に変更して、異常検出の感度を最適化できます。
🔹 リトライ処理の追加: ネットワーク障害やAPIの制限に対応する処理を追加すると安定性が向上します。

次のステップ

✅ コードの動作確認
✅ 異常検出の最適化 (例: RSIやボリンジャーバンドの組み合わせ)
✅ サーバー環境へのデプロイ (Docker + Kubernetes対応)

まとめ

今回参考した論文で提案されたMTLアプローチは、データの少ない新規資産やボラティリティの高い仮想通貨に対応する有力な手法であり、特に以下の点で強力な効果を発揮できる可能性があります。

✅ データが不足する新規上場トークンや流動性の低い銘柄のボラティリティ予測
✅ サーキットブレーカーやスキャルピング戦略のトリガー最適化
✅ 異常検出や市場急変への迅速な対応

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

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

comment コメントをキャンセル

Bot

2025/4/23

開発記録#194(2025/4/23)開発計画まとめロードマップ

現在、MMbot開発にコミットしていることの合理性とその根拠及び今後の方針と不確実性・リスクについて詳細をまとめておく。 1. 「MM botから始める」合理性と論理的根拠観点根拠学習効率ROI秒単位で取引→PDCAサイクルが高速。板取得・非同期I/O・約定監視など“全戦略の共通基礎”を一気に経験できる。資本効率小口でもポジション時間が短く在庫リスク限定。手数料リベート込みで日利0.05〜0.15 %程度を狙えるため“学習用実弾”として適量。心理負荷の管理大損が出にくく、損失ストレスが小さい。集中力を開 ...

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

2024/8/29

仮想通貨botの開発記録#71(2024/4/19)「書籍：アルゴリズム取引まとめ」【高頻度bot開発のTips】

前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。今回は「アルゴリズム取引」を読んだので、その内容とBot開発に活かせそうな部分を記録します。参考にしている資料の一つ。Botの開発・運用についてかなり具体的な部分まで踏み込んで解説している良著です。自動売買bot作成を少しでもかじったことがあれば、かなりスラスラ読めると思います。 pic.twitter.com/UMmsCpmSlq— よだか(夜鷹/yodaka) (@yodakablog) April 17, 202 ...

Bot

2025/4/21

開発記録#189(2025/4/21)MM bot開発ログ7「bybitテストネットでのデータ通信トラブルまとめ＆設計構造の考察」

前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。本日取り組んだ「bybitテストネットで動かす際のデータ通信周りのトラブル」と、その失敗→原因分析→修正フロー→最終対応を整理します。最後に、Testnet の設計構造について考察します。 1. 発生した問題と経緯番号現象エラー内容／挙動①約定状況取得で KeyError: 'data'd["result"]["data"][0] へアクセス中に data キーが存在せず落ちる②/v5/order/realtime 呼び出し時に ...

Bot 機械学習・データサイエンス

2024/11/6

開発記録#121(2024/11/1)【MLbot開発】「LSTMとアンサンブル学習(バギングとブースティング)」

前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。今回は「LSTMとアンサンブル学習」というアプローチでMLbotを作ってみました。本記事では、このアプローチについて整理するととともにプログラムのプロトタイプの解説も行います。解決したかったこと・アンサンブル学習について理解する・LSTMとアンサンブル学習を組み合わせることの優位性を知る・LSTMとアンサンブル学習を組み合わせることの問題点を知る・上記の問題点を踏まえて、それらの解決策を調べる・MLbot開発の引き出し ...

Bot 機械学習・データサイエンス

2024/10/27

開発記録#119(2024/10/26)ML bot開発「GAとLSTM」

前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。今回はGA(Genetic Algorithm, 遺伝的アルゴリズム)とLSTMを組み合わせたbotを作ってみました。今日もコツコツやる。触って理解を深める。 pic.twitter.com/emOo5WBdLD— よだか(夜鷹/yodaka) (@yodakablog) October 26, 2024 解決したかったこと・GAの概念を理解する・GAとLSTMを組み合わせたプログラムの雛形を作る・GAとLSTM ...

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

開発記録#136(2025/3/22)「ブロックチェーンデータ分析に基づくbot開発のアイデアpart.1」