Bot

開発記録#131(2025/3/9)「リミットオーダーブック（LOB）データを基にしたホークスモデルの活用」

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

Yodaka

興味深い論文を読んだので、それを基にしたbot開発のアイデアをまとめておきます。

参考

Hawkes-based cryptocurrency forecasting via Limit Order Book data

ドキュメントの要約

この研究では、暗号通貨のリターン予測のための新しい予測アルゴリズムを提示している。このアルゴリズムは、リミットオーダーブック（LOB）データを基にしたホークスモデルを使用する。ホークスモデルとは、点過程の一種で、過去のイベントの時間に基づいて未来のイベントの発生確率をモデル化する。このモデルを利用し、連続的な出力誤差（COE）モデルと組み合わせることで、リターンの符号をより正確に予測する。

モンテカルロシミュレーションを通じて50のシナリオでアプローチの効果を検証。その結果、この戦略はベンチマークモデルよりも予測精度と累積利益の点で優れていることが確認された。

また、実際の高頻度取引環境においても性能を検証し、予測戦略が理想的な基準であるオラクルに近い性能を示すことが分かった。この研究は、特に暗号通貨のLOBデータに適用可能であり、LOBの株式市場データで開発されたベースインバランス回帰器(?)が、暗号通貨データにおいても有効。

仮説

Yodaka

ホークスプロセスを活用した予測モデルと連続的な出力誤差（COE）モデルを組み合わせた戦略をbot開発に応用できるのでは？

【実行手順】
1,データの取得と前処理：LOBデータをリアルタイムまたは履歴データから取得し、必要な前処理を行う。
2,ホークスプロセスモデルの定義と訓練：過去のデータを基にモデルを訓練する。
3,COEモデルの定義と訓練：ホークスプロセスモデルからの予測を利用して、リターンの符号を予測する。
4,取引ロジックの実装：予測されたリターンの符号に基づいて、売買の判断を行う。

参考

hawkeslib(ライブラリのドキュメント)

hawkeslibとは？

hawkeslibは、ホークスプロセス（Hawkes process）と呼ばれる数学的モデルを実装したPythonライブラリです。ホークスプロセスは、過去のイベントが将来のイベントの発生率に影響を与えるという特性を持つ、自己励起型の点過程（self-exciting point process）です。このモデルは、金融市場の取引データ分析、社会ネットワークの行動分析、地震の発生予測など、多様な分野で応用されています。

hawkeslibの主な特徴と機能

  1. モデリングの柔軟性：
    • hawkeslibは、ホークスプロセスを用いて時間的なデータのパターンをモデル化する際に、パラメータの推定やシミュレーションを簡単に行うことができます。
  2. イベントの自己励起性：
    • ホークスプロセスは、あるイベントが他のイベントの発生確率を時間経過とともに増加させる効果をモデル化します。例えば、ある株の取引が活発になると、その後も引き続き取引が活発になると予測できます。
  3. 統計的推定：
    • 最尤推定（Maximum Likelihood Estimation, MLE）やその他の統計的手法を使用して、モデルのパラメータをデータから推定する機能が含まれています。
  4. 応用の広さ：
    • ホークスプロセスは、単に連続するイベント間の時間だけでなく、イベントの強度やその他の特性を考慮に入れることができるため、非常に柔軟なモデリングが可能です。

hawkeslibを用いる利点

  • データドリブンな洞察の提供： ホークスプロセスを通じて、イベントデータ内の潜在的な自己励起パターンや相互依存性を発見し、よりデータに基づいた意思決定を行うことができます。
  • 多様なデータソースへの適用： ソーシャルメディアの分析、金融市場のモニタリング、自然災害の予測など、さまざまなタイプの時系列データに対して適用可能です。

使用時の考慮点

  • ホークスプロセスの理解が必要：hawkeslibを効果的に使用するためには、ホークスプロセスの理論的背景についての理解が必要です。
  • データの質と前処理が重要：入力データの質と適切な前処理が、モデルの性能に直接影響を与えます。

hawkeslibはその特殊な用途と数学的背景により、専門的な知識を必要とするツールですが、適切に使用すれば非常に強力な洞察を提供することができます。

プログラムの雛形

仮想通貨の自動取引botを作る場合、ホークスプロセスを活用した予測モデルと連続的な出力誤差（COE）モデルを組み合わせることが提案されています。以下は、このアプローチに基づいた取引botの基本的なコードの概念をPythonで記述する一例です。

Yodaka

この例では、ポイントプロセスのイベント時間予測と取引判断を行います。

  1. データの取得と前処理：LOBデータをリアルタイムまたは履歴データから取得し、必要な前処理を行います。
  2. ホークスプロセスモデルの定義と訓練：過去のデータを基にモデルを訓練します。
  3. COEモデルの定義と訓練：ホークスプロセスモデルからの予測を利用して、リターンの符号を予測します。
  4. 取引ロジックの実装：予測されたリターンの符号に基づいて、売買の判断を行います。
Yodaka

まずは簡単なコードの例を示します。

import numpy as np
import pandas as pd
from hawkeslib import HawkesProcess
from sklearn.linear_model import LogisticRegression

class CryptoTradingBot:
    def __init__(self):
        self.hawkes = HawkesProcess(mu=0.1, alpha=0.5, beta=0.7)  # パラメータは適宜調整
        self.coe_model = LogisticRegression()

    def train_hawkes(self, data):
        # データからホークスプロセスを訓練
        self.hawkes.fit(data['timestamps'])

    def train_coe(self, features, labels):
        # COEモデルを訓練
        self.coe_model.fit(features, labels)

    def predict_event_time(self, current_time):
        # 次のイベント時間を予測
        return self.hawkes.predict_next(current_time)

    def predict_return_sign(self, features):
        # リターンの符号を予測
        return self.coe_model.predict(features)

    def trade(self, prediction):
        # 予測に基づいて取引
        if prediction > 0:
            print("Buying...")
        else:
            print("Selling...")

# 使用例
data = pd.read_csv('lob_data.csv')
timestamps = pd.to_datetime(data['timestamp'])

bot = CryptoTradingBot()
bot.train_hawkes(timestamps)
features = data[['feature1', 'feature2']]  # 予測に使用する特徴量
labels = np.sign(data['returns'])  # リターンの符号

bot.train_coe(features, labels)

current_time = pd.Timestamp.now()
next_event_time = bot.predict_event_time(current_time)
next_event_features = np.array([[data.iloc[-1]['feature1'], data.iloc[-1]['feature2']]])
prediction = bot.predict_return_sign(next_event_features)

bot.trade(prediction)

このコードは非常に基本的な形であるため、実際の取引ではさらに多くの要素を考慮に入れる必要があります。具体的には、市場のノイズや異常値への対策、取引手数料やスリッページの管理、リスク管理戦略の適用などが挙げられます。

改良型1

実践的な自動取引botを作るためには、より詳細なデータの取得、効果的なモデルの訓練、および取引戦略の実装が必要です。ここでは、リミットオーダーブック（LOB）データを用いた取引botの実装例を示します。

Yodaka

この例では、実際にデータを取得し、ホークスプロセスモデルを訓練し、取引判断を行うまでの一連のステップをPythonで実装します。

import numpy as np
import pandas as pd
from datetime import datetime
from hawkeslib import ContinuousTimeHawkesModel
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler

class CryptoTradingBot:
    def __init__(self):
        self.hawkes_model = ContinuousTimeHawkesModel(uv_prior=1.0)  # ホークスプロセスモデルの初期化
        self.coe_model = LogisticRegression()  # COEモデルの初期化
        self.scaler = StandardScaler()  # データのスケーリング用

    def fetch_data(self, filepath):
        # データの読み込み
        return pd.read_csv(filepath, parse_dates=['timestamp'])

    def preprocess_data(self, data):
        # 必要な前処理を行う
        data['time_diff'] = (data['timestamp'] - data['timestamp'].shift()).dt.total_seconds().fillna(0)
        return data

    def train_hawkes(self, timestamps):
        # ホークスプロセスモデルの訓練
        self.hawkes_model.fit(timestamps)

    def train_coe(self, features, labels):
        # 特徴量のスケーリング
        features_scaled = self.scaler.fit_transform(features)
        # COEモデルを訓練
        self.coe_model.fit(features_scaled, labels)

    def predict_event_time(self, last_event_time):
        # 次のイベント時間を予測
        intensity = self.hawkes_model.intensity(last_event_time)
        return np.random.exponential(1 / intensity)

    def predict_return_sign(self, features):
        # リターンの符号を予測
        features_scaled = self.scaler.transform([features])
        return self.coe_model.predict(features_scaled)[0]

    def trade(self, prediction):
        # 予測に基づいて取引
        if prediction > 0:
            print("Buying...")
        elif prediction < 0:
            print("Selling...")
        else:
            print("Holding...")

# 使用例
data_filepath = 'lob_data.csv'  # データファイルのパス
bot = CryptoTradingBot()
data = bot.fetch_data(data_filepath)
data = bot.preprocess_data(data)

# ホークスモデルの訓練
bot.train_hawkes(data['timestamp'].astype(int) / 10**9)  # Unix時間に変換

# 特徴量とラベルの準備
features = data[['time_diff', 'order_imbalance']]  # 時間差とオーダー不均衡を特徴量とする
labels = np.sign(data['return'])  # リターンの符号をラベルとする

# COEモデルの訓練
bot.train_coe(features, labels)

# 現在時刻の取得と次のイベント時間の予測
current_time = datetime.now()
next_event_time = bot.predict_event_time(current_time.timestamp())

# 次のイベントの特徴量の取得（ここでは例として最新のデータを使用）
next_event_features = features.iloc[-1].values

# リターンの符号の予測
prediction = bot.predict_return_sign(next_event_features)

# 取引の実行
bot.trade(prediction)

このコードは、LOBデータから特定の特徴量を抽出し、それを用いてホークスプロセスに基づく次のイベント時間を予測し、COEモデルを用いてリターンの符号を予測し取引を行う一連の流れを示しています。実際の運用には、さらに詳細な例外処理、リアルタイムのデータフィードへの接続、パフォーマンスの監視とログの取り扱いなどが必要になります。また、市場の状況に応じてモデルのパラメータ調整や再訓練が必要です。

改良型2

このコードは、データの取得、前処理、モデルの訓練、および取引決定までを自動化し、より実際の取引環境に近い形で動作します。

Yodaka

以下は、リアルタイム取引を考慮した実用的な自動取引botの改良版のコード例です。

import numpy as np
import pandas as pd
from datetime import datetime
from hawkeslib import ContinuousTimeHawkesModel
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler

class CryptoTradingBot:
    def __init__(self):
        # ホークスプロセスモデルの初期化
        self.hawkes_model = ContinuousTimeHawkesModel(uv_prior=1.0)
        # ロジスティック回帰モデルの初期化
        self.coe_model = LogisticRegression()
        # 特徴量スケーラーの初期化
        self.scaler = StandardScaler()

    def fetch_data(self, filepath):
        # CSVファイルからデータを読み込む
        return pd.read_csv(filepath, parse_dates=['timestamp'])

    def preprocess_data(self, data):
        # 時間差分の計算
        data['time_diff'] = (data['timestamp'] - data['timestamp'].shift()).dt.total_seconds().fillna(0)
        return data

    def train_hawkes(self, timestamps):
        # ホークスプロセスモデルの訓練
        self.hawkes_model.fit(timestamps)

    def train_coe(self, features, labels):
        # 特徴量のスケーリングとCOEモデルの訓練
        features_scaled = self.scaler.fit_transform(features)
        self.coe_model.fit(features_scaled, labels)

    def predict_event_time(self, last_event_time):
        # 次のイベント時間を予測
        intensity = self.hawkes_model.intensity(last_event_time)
        return np.random.exponential(1 / intensity)

    def predict_return_sign(self, features):
        # リターンの符号を予測
        features_scaled = self.scaler.transform([features])
        return self.coe_model.predict(features_scaled)[0]

    def trade(self, prediction):
        # 取引判断の実施
        if prediction > 0:
            print("Buying...")
        elif prediction < 0:
            print("Selling...")
        else:
            print("Holding...")

# メインプロセス
if __name__ == "__main__":
    data_filepath = 'lob_data.csv'  # データファイルのパス
    bot = CryptoTradingBot()
    data = bot.fetch_data(data_filepath)
    data = bot.preprocess_data(data)

    # ホークスプロセスの訓練
    bot.train_hawkes(data['timestamp'].astype(int) / 10**9)

    # 特徴量とラベルの準備
    features = data[['time_diff', 'order_imbalance']]
    labels = np.sign(data['return'])

    # COEモデルの訓練
    bot.train_coe(features, labels)

    # 最新のイベント時間を取得し次のイベント時間を予測
    last_event_time = data['timestamp'].iloc[-1].timestamp()
    next_event_time = bot.predict_event_time(last_event_time)

    # 次のイベントの特徴量を取得
    next_event_features = features.iloc[-1].values

    # リターンの符号を予測
    prediction = bot.predict_return_sign(next_event_features)

    # 取引の実行
    bot.trade(prediction)

こちらのPythonコードは、仮想通貨の自動取引botを構築するためのものです。主にホークスプロセスとロジスティック回帰モデルを活用して、取引タイミングと方向性を決定します。データ駆動型のアプローチでリアルタイム取引判断を自動化するための実用的なフレームワークです。

Yodaka

それぞれの部分について詳しく解説します。

クラスとメソッドの概要

  • CryptoTradingBot: このクラスは取引botの全機能を包含しています。モデルの初期化、データの取得と前処理、モデルの訓練、イベント時間の予測、取引決定の実行までを行います。

初期化 (__init__)

  • ContinuousTimeHawkesModel: ホークスプロセスモデルを用いて、イベント間の時間依存性をモデル化します。このモデルは、イベントの自己励起性をキャプチャするのに適しています。
  • LogisticRegression: ロジスティック回帰モデルは、特定の特徴からリターンの符号（上昇または下降）を予測します。
  • StandardScaler: 特徴量のスケーリングを行い、モデルの性能を向上させます。

データ取得 (fetch_data)

  • pd.read_csv: 指定されたパスからCSVファイルを読み込みます。parse_dates引数により、日付データを適切に解析します。

データ前処理 (preprocess_data)

  • 時間差分を計算し、イベント間の秒数を新たな列として追加します。これにより、イベントの時間的な連続性が特徴量として利用されます。

モデル訓練

  • train_hawkes: ホークスプロセスモデルをLOBデータのタイムスタンプに基づいて訓練します。
  • train_coe: リターンの符号を予測するためのロジスティック回帰モデルを訓練します。ここで、StandardScalerを使用して特徴量を正規化します。

イベント時間予測 (predict_event_time)

  • hawkes_model.intensity: 最後のイベント時間から次のイベント発生の強度を計算し、その強度を基に次のイベントの時間を指数分布からサンプリングします。

リターン符号予測 (predict_return_sign)

  • 予測された特徴量を用いてリターンの符号を予測します。このステップでは、特徴量のスケーリングが必要です。

取引実行 (trade)

  • 予測されたリターンの符号に基づいて、「買い」、「売り」、または「保持」の取引指示を出します。

メインプロセス

  • __main__: スクリプトが直接実行された場合に、上記のプロセスを順に実行します。データの取得、前処理、モデルの訓練、最新のイベント時間の取得、リターン符号の予測、そして取引指示の出力までを行います。

上記のコードを実行するために

上記のコードを実行するために必要な環境構築を行う際には、以下の手順に従って準備を進めることができます。このセットアップはPythonがインストールされたシステムで行います。

1. Python のインストール

Pythonがまだインストールされていない場合は、公式Pythonウェブサイトからインストーラーをダウンロードしてインストールします。Python 3.6 以上を推奨します。

2. 仮想環境のセットアップ

Pythonプロジェクトでは、プロジェクトごとに依存関係を隔離するために仮想環境を使用することが一般的です。以下のコマンドを使って仮想環境を作成し、アクティベートします。

# 仮想環境の作成
python -m venv trading-bot-env

# 仮想環境のアクティベーション
# Windows
trading-bot-env\Scripts\activate

# macOS / Linux
source trading-bot-env/bin/activate

3. 必要なライブラリのインストール

次に、必要なPythonライブラリをインストールします。これにはpandas, numpy, scikit-learn、そして特定のライブラリであるhawkeslibが含まれます。これらのライブラリはpipを使用してインストールできます。

pip install pandas numpy scikit-learn
pip install hawkeslib

hawkeslibは特殊なライブラリであり、すべての環境で簡単にインストールできるとは限りません。インストールに問題がある場合は、そのライブラリのドキュメントやGitHubのリポジトリを参照してください。

4. コードの準備

コードをテキストエディタにコピーして.py拡張子で保存します。例えば、trading_bot.pyという名前で保存することができます。

5. データファイルの準備

コードが参照するLOBデータが含まれるCSVファイルを準備します。このファイル名はスクリプト内で参照されるので、適切なファイルパスを指定してください。データファイルはスクリプトが置かれているディレクトリ内、またはスクリプトからアクセス可能なパスに置かれている必要があります。

6. スクリプトの実行

環境が整ったら、以下のコマンドを使ってPythonスクリプトを実行します。

python trading_bot.py

これでスクリプトが実行され、定義された取引ロジックに従って出力が表示されます。

7. 環境のデアクティベーション

作業が完了したら、仮想環境をデアクティベートします。

deactivate
Yodaka

これらのステップに従って、Pythonで書かれた自動取引botを実行するための環境を構築できます。

まとめ

今回は「ホークスプロセスを活用した予測モデル」についてまとめました。

論文の熟読を通して新戦略を考案していくのは楽しいです。

本記事の戦略実行において進展があれば、また更新します。

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

-Bot