開発記録#113(2024/10/13)「オンチェーンデータを用いた価格予測MLbotのプロトタイプ」

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

今回はオンチェーンデータを用いた価格予測MLbotのプロトタイプを作ってみました。

ML×DeFibot、アリだな。研究のつもりでプロトタイプ作ってみよ。

— よだか(夜鷹/yodaka) (@yodakablog) October 12, 2024

コードのプロトタイプ

from web3 import Web3
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Input
import logging
import time
from tqdm import tqdm  # tqdmをインポート

# ロギングの設定
logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Ethereumノードに接続
infura_url = "https://mainnet.infura.io/v3/YOUR_INFURA_PROJECT_ID"
web3 = Web3(Web3.HTTPProvider(infura_url))

# Uniswap V3のペアコントラクトアドレス（例: ETH/USDCペア）
uniswap_v3_pair_address = "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8"  # 例としてETH/USDCペアのアドレス

# チェックサムアドレスに変換
uniswap_v3_pair_address = Web3.to_checksum_address(uniswap_v3_pair_address)

# Uniswap V3ペアのABI
uniswap_v3_pair_abi = [
    {
        "inputs": [],
        "name": "slot0",
        "outputs": [
            {"internalType": "uint160", "name": "sqrtPriceX96", "type": "uint160"},
            {"internalType": "int24", "name": "tick", "type": "int24"},
            {"internalType": "uint16", "name": "observationIndex", "type": "uint16"},
            {"internalType": "uint16", "name": "observationCardinality", "type": "uint16"},
            {"internalType": "uint16", "name": "observationCardinalityNext", "type": "uint16"},
            {"internalType": "uint8", "name": "feeProtocol", "type": "uint8"},
            {"internalType": "bool", "name": "unlocked", "type": "bool"}
        ],
        "stateMutability": "view",
        "type": "function"
    }
]

# Uniswapペアコントラクトのインスタンスを作成
uniswap_v3_pair_contract = web3.eth.contract(address=uniswap_v3_pair_address, abi=uniswap_v3_pair_abi)

# データ取得モジュール
def fetch_data(num_samples=100, interval=60):
    data = []
    for _ in tqdm(range(num_samples), desc="Fetching Data"):  # tqdmを使用して進行状況を表示
        slot0 = uniswap_v3_pair_contract.functions.slot0().call()
        sqrt_price_x96 = slot0[0]
        eth_price = (sqrt_price_x96 / (2**96))**2  # ETH/USDCの価格を計算
        data.append({'timestamp': pd.Timestamp.now(), 'eth_price': eth_price})
        time.sleep(interval)  # 指定した間隔でデータを取得
    df = pd.DataFrame(data)
    return df

# モデル構築モジュール
def build_model(input_shape):
    model = Sequential([
        Input(shape=input_shape),
        Dense(32, activation='relu'),
        Dense(16, activation='relu'),
        Dense(1)  # 価格予測
    ])
    model.compile(optimizer='adam', loss='mean_squared_error')
    return model

# メインロジック
def main():
    try:
        df = fetch_data()
        
        # 特徴量とラベル
        X = df[['eth_price']].values
        y = df['eth_price'].shift(-1).fillna(df['eth_price']).values
        
        # 訓練・テストデータ分割
        if len(X) > 1:  # データが十分にあるか確認
            X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)
            
            # データの標準化
            scaler = StandardScaler()
            X_train_scaled = scaler.fit_transform(X_train)
            X_test_scaled = scaler.transform(X_test)
            
            # モデルの構築
            model = build_model((X_train_scaled.shape[1],))
            
            # モデルの訓練
            for epoch in tqdm(range(5), desc="Training Progress"):  # tqdmを使用して進行状況を表示
                model.fit(X_train_scaled, y_train, epochs=1, batch_size=64, validation_data=(X_test_scaled, y_test), verbose=0)
            
            # 最新のデータを取得して予測
            df = fetch_data(num_samples=1, interval=0)
            X_live = scaler.transform(df[['eth_price']].values[-1].reshape(1, -1))
            predicted_price = model.predict(X_live)[0][0]
            
            print(f"Predicted ETH Price: {predicted_price}")
        else:
            logger.error("データが不足しています。")

    except Exception as e:
        logger.error(f"メインロジックでエラーが発生しました: {str(e)}")

if __name__ == "__main__":
    main()

このコードは、イーサリアム（ETH）とUSDC（ステーブルコイン）の交換レートをリアルタイムで取得し、そのデータを基にイーサリアムの将来の価格を予測するためのプログラムです。

1. ロギングの設定:
   • システムの動作状況やエラーなどのログを記録するための設定を行います。これにより、プログラムの実行時に何が起こっているかを追跡しやすくなります。
2. Ethereumノードへの接続:
   • Infuraというサービスを利用して、Ethereumのブロックチェーンに接続します。これにより、ブロックチェーン上のデータにアクセスできます。
3. Uniswap V3ペアコントラクトへのアクセス:
   • UniswapはEthereum上の分散型取引所で、特定のコントラクトを介してETHとUSDCのペアの交換レートデータを取得します。
4. データ取得:
   • fetch_data関数を使用して、指定されたサンプル数と時間間隔でETH/USDCの交換レートを取得し、それをデータフレームに保存します。この関数では、進行状況を視覚的に表示するためにtqdmライブラリも使用しています。
5. データの前処理:
   • 取得した価格データを機械学習モデルが扱いやすいように処理します。具体的には、データを標準化して、平均が0、分散が1になるように変換します。
6. モデルの構築と訓練:
   • TensorFlowを用いてニューラルネットワークモデルを構築し、取得したデータを使ってモデルを訓練します。訓練の進行状況もtqdmで表示されます。
7. 価格の予測:
   • 訓練されたモデルを使用して、最新のデータに基づいてETHの将来価格を予測します。
8. エラーハンドリング:
   • エラーが発生した場合には、ログに記録し、何が問題だったかを警告します。

1. ロギングの設定

ロギングはプログラムの動作やエラー情報を記録する重要な機能です。logging ライブラリを使用して設定を行い、異常が発生した際にそれをログファイルやコンソールに出力できます。具体的には、ログのレベル（情報、警告、エラー等）やフォーマット（日時、ログレベル、メッセージ）を設定しています。これにより、プログラムの問題解析や動作の確認が容易になります。

2. Ethereumノードへの接続

Ethereumのノードに接続するために、Infuraというサービスを利用しています。InfuraはEthereumのブロックチェーンデータにアクセスするためのAPIを提供しており、このAPIを通じてプログラムがブロックチェーンのデータにアクセスできるようにします。接続はWeb3.pyライブラリを使用し、HTTPSプロバイダを介して行います。

3. Uniswap V3ペアコントラクトへのアクセス

Uniswap V3はEthereum上で動作する分散型取引所の一つで、各種通貨ペアの交換レート情報を提供するスマートコントラクトを持っています。このプログラムでは、特定の通貨ペア（ETH/USDC）のスマートコントラクトにアクセスし、価格情報を取得するためのABI（アプリケーションバイナリインターフェース）とアドレスを使用しています。

4. データ取得

データ取得は、指定された数と間隔でUniswapのコントラクトから価格データを取得する処理を行います。この関数はプログレスバーを用いて進捗状況を表示し、実行中の状況がわかりやすくなっています。取得されたデータはタイムスタンプと価格情報を含むデータフレームに保存されます。

5. データの前処理

取得したデータはそのままでは機械学習モデルで効率的に処理できないため、標準化を行います。標準化はデータの平均を0、標準偏差を1にする処理で、モデルの学習効率を向上させることができます。これにより、モデルがデータの特徴を捉えやすくなります。

6. モデルの構築と訓練

ニューラルネットワークモデルをTensorFlow/Kerasを用いて構築し、前処理されたデータを用いて訓練を行います。モデルは複数の層を持ち、ETHの価格を予測するために最適化されます。訓練では、データを訓練用とテスト用に分割し、モデルの性能を評価します。

7. 価格の予測

訓練されたモデルを使用して、最新のデータに基づいて将来のETH価格を予測します。この予測は、投資判断や市場分析に利用することができます。

8. エラーハンドリング

プログラムの実行中に発生する可能性のあるエラーを捕捉し、ログに記録します。これにより、プログラムの問題点を迅速に特定し、修正を行うことが可能です。

実行プロセスと各コードの役割

プログラム全体の実行プロセスと各コードセグメントの役割を具体的に説明します。このPythonスクリプトは、分散型取引所UniswapからETH/USDCの価格データを取得し、それを用いて機械学習モデルによる価格予測を行うことを目的としています。

モジュールのインポート

from web3 import Web3
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Input
import logging
import time
from tqdm import tqdm

Web3: Ethereumノードへの接続とインタラクションを可能にするライブラリ。

pandas: データの操作や分析を行うためのライブラリ。

sklearn.model_selection, sklearn.preprocessing: 機械学習データの前処理とデータセットの分割を行う。

tensorflow: ニューラルネットワークのモデルを構築・訓練するためのライブラリ。

logging: プログラムの動作状況やエラーを記録するためのライブラリ。

time: 待機時間の制御に使用。

tqdm: プログレスバーを表示し、長期間の処理の進捗を視覚的に示す。

ロギングの設定

logging.basicConfig(level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

ログの基本設定を行い、エラー発生時などに情報を記録します。これによりデバッグやシステムの監視が容易になります。

Ethereumノードへの接続

infura_url = "https://mainnet.infura.io/v3/YOUR_INFURA_PROJECT_ID"
web3 = Web3(Web3.HTTPProvider(infura_url))

Infuraを通じてEthereumのメインネットに接続。YOUR_INFURA_PROJECT_IDはユーザーごとのプロジェクトIDに置き換える必要があります。

Uniswap V3ペアコントラクトのセットアップ

uniswap_v3_pair_address = "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8"
uniswap_v3_pair_address = Web3.to_checksum_address(uniswap_v3_pair_address)
uniswap_v3_pair_contract = web3.eth.contract(address=uniswap_v3_pair_address, abi=uniswap_v3_pair_abi)

Uniswap V3から特定のペアの価格データを取得するために、適切なアドレスとABIを使用してスマートコントラクトに接続します

データ取得モジュール

def fetch_data(num_samples=100, interval=60):
    data = []
    for _ in tqdm(range(num_samples), desc="Fetching Data"):
        slot0 = uniswap_v3_pair_contract.functions.slot0().call()
        sqrt_price_x96 = slot0[0]
        eth_price = (sqrt_price_x96 / (2**96))**2
        data.append({'timestamp': pd.Timestamp.now(), 'eth_price': eth_price})
        time.sleep(interval)
    df = pd.DataFrame(data)
    return df

UniswapからETH/USDCの交換レートを定期的に取得し、タイムスタンプとともに保存します。tqdmを用いて進行状況を表示します。

モデル構築モジュール

def build_model(input_shape):
    model = Sequential([
        Input(shape=input_shape),
        Dense(32, activation='relu'),
        Dense(16, activation='relu'),
        Dense(1)
    ])
    model.compile(optimizer='adam', loss='mean_squared_error')
    return model

TensorFlowを使用して、ETH価格を予測するためのシンプルなニューラルネットワークを構築します。

メインロジック

def main():
    df = fetch_data()
    X = df[['eth_price']].values
    y = df['eth_price'].shift(-1).fillna(df['eth_price']).values
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    model = build_model((X_train_scaled.shape[1],))
    for epoch in tqdm(range(5), desc="Training Progress"):
        model.fit(X_train_scaled, y_train, epochs=1, batch_size=64, validation_data=(X_test_scaled, y_test), verbose=0)
    df = fetch_data(num_samples=1, interval=0)
    X_live = scaler.transform(df[['eth_price']].values[-1].reshape(1, -1))
    predicted_price = model.predict(X_live)[0][0]
    print(f"Predicted ETH Price: {predicted_price}")

上記の各モジュールを統合して、データ取得、前処理、モデル訓練、および価格予測を行います。プログレスバーで訓練の進行状況を表示し、最終的には未来の価格を予渡して出力します。

まとめ

今回は、オンチェーンデータを用いた価格予測MLbotのプロトタイプを作ってみました。

処理の高速化や実際に使えるデータ群の抽出・精撰などについては、今後の開発の過程で行なっていきます。

オンチェーンデータを参照させたら、結果の出力まで90分程度かかってしまう。
これをタイムリーに更新しながら変数を調整するbotを作る。
課題山積だが、これで良い。やるべきことが見えてきた。 pic.twitter.com/uvsjexKnSc

— よだか(夜鷹/yodaka) (@yodakablog) October 12, 2024

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