AIトレーディングボットの作り方:開発者向け完全ガイド(2026年版)

最終更新日: 2025年12月10日8分で読めます

過去数年で、AIトレーディングツールが急速に普及しました。

小口投資家の約58%がポートフォリオ構築に何らかの形でAIを利用しており、ベンチャーキャピタルはWeb3のAIトレーディング技術に2億1,300万ドルを投じ、自律型のAIエージェントの中にはわずか1ヶ月で4億ドル以上の取引高を処理するものも現れ、市場分析、実行速度、他のエージェントとの連携がますます高度化しています。

しかし、この盛り上がりの裏には、より複雑な現実がある。成功は高度なアルゴリズムだけでなく、信頼できるインフラ、堅実なリスク管理、そしてAIにできることとできないことについての現実的な理解にかかっている。

この包括的なガイドでは、独自のAIトレーディングボットを構築するために必要な知識をすべて解説する。自律型エージェント経済の台頭を含むAIトレーディングの現状を概観し、利用可能なツールやプラットフォームを調査し、ボットの各戦略を分解し、Alchemyのブロックチェーンインフラを使って自分自身のトレーディングボットを構築するためのチュートリアルを提供する。

AIトレーディングボットとは何か

AIトレーディングボットとは、機械学習アルゴリズムを使って市場データを分析し、取引機会を特定し、自動的に買い注文や売り注文を実行するソフトウェアである。静的な「if-then」ロジックに従う従来のルールベースのボットとは異なり、AIを活用したボットは変化する市場状況に適応し、過去のパターンから学習し、膨大なデータセットに基づいて確率的な判断を下すことができる。これらは24時間365日稼働し、取引判断から感情的なバイアスを取り除き、人間よりも速く市場の動きに反応する。

2025年のAIトレーディングボットの状況は、対照的な側面を示している。AI支援トレーディングツールの採用は大きく伸び、自動化を戦略に取り入れる小口投資家はこれまで以上に増えている。しかし、実際の成功率を見ると、より厳しい現実が見えてくる。ボット利用者のうち、継続的な収益を上げているのは10〜30%に過ぎない。

この採用率と成功率のギャップは驚くにはあたらない。暗号資産市場はボラティリティが高いことで知られ、予測モデルには継続的な改良が必要であり、多くの新規参入者はリスク管理の重要性を過小評価しているからだ。成功するボットは必ずしも最も高度なAIを搭載したものではなく、信頼できるデータ、明確な戦略、規律あるエクゼキューションといった堅実な基盤の上に構築されたものである。

教訓は何か。AIトレーディングボットは強力なツールだが、魔法ではない。成功は、その能力と限界の両方を理解しているかどうかにかかっている。

ブロックチェーンインフラの役割

AIトレーディングボットは強力なツールだが、成功はその能力と限界の両方を理解しているかどうかにかかっている。ただし見落とされがちなのは、モデルの精巧さは方程式の一部分にすぎないということだ。ほとんどのボットが失敗するのは、AIの予測が悪かったからではなく、その下にあるインフラが追いつかなかったからだ。データが古かった。ボラティリティの高い瞬間にAPIがダウンしていた。より速いパイプラインを持つ競合に先を越された。これらは例外的なケースではなく、アルゴリズムトレーディングの日常的な現実である。

継続的に利益を上げるボットと破綻するボットを分ける、3つの重要なインフラ要件がある。

データの正確性: AIモデルの精度は、学習に使用し反応するデータの質に左右される。古くなったトークン価格、更新されていない流動性プールの状態、見逃されたクジラの動きは、そのまま質の悪い取引判断につながる。ボットが悪いデータに基づいて動けば損失を出す。根本的に欠陥のある入力を補えるようなアルゴリズム上の巧妙さは存在しない。
レイテンシー: アルゴリズムトレーディングでは、速度が誰が利益を得るかを決める。ボットが裁定取引の機会を特定したとき、他の多くの参加者も同じシグナルを見ている可能性が高い。最も速いデータパイプラインを持つボットが最初に約定し、スプレッドを獲得する。それ以外はより悪い約定条件を得るか、取引そのものを逃す。低レイテンシーのインフラは、パフォーマンスの最適化ではなく、競争上の必須要件である。
信頼性: ボラティリティの高い市場でのインフラのダウンタイムは致命的になりかねない。数分間のオフラインが、暴落時の損切りの見逃し、防御できなかったポジションの強制決済、ボットの再接続前に過ぎ去ってしまった機会を意味することもある。実際の資本を運用するトレーディングシステムにとって、稼働率はサービスレベルの指標ではなく、生存のための必須要件である。

ここで、Alchemyのようなデータプロバイダが不可欠になる。Alchemyは、AIトレーディングボットが効果的に機能するために必要な、信頼性の高いリアルタイムデータアクセスを提供する。トークン価格、取引量、クジラの動き、流動性プールの状態など、複数のチェーンにわたるデータだ。Alchemyの拡張APIとインフラは、これらの要件を大規模に処理し、100以上のブロックチェーンに対応しながら、競争の激しいトレーディングが求める低レイテンシーのパフォーマンスをサポートしている。

安全性と可観測性を備えたトレーディングボットの構築

堅実なインフラがあっても、AIトレーディングボットの挙動は予測不能になり得る。市場はモデルが想定していない動きをし、エッジケースが予期しない挙動を引き起こし、エラーの積み重ねが小さなミスを壊滅的な損失に変えることがある。ボットのアーキテクチャに安全機構を組み込むことは選択肢ではなく、それがあってこそ生き残り、改良を重ねることができる。

可観測なシステムを構築する: すべての取引は追跡可能であるべきだ。どのデータが判断の根拠となったのか。ボットがその特定のアクションを選んだ理由は何か。何か問題が起きたとき(必ず起きる)、それが悪いデータによるものか、ロジックの欠陥によるものか、モデルが想定していなかったエッジケースによるものかを診断する必要がある。
サーキットブレーカーを追加する: 自動的にトレーディングを一時停止する最大損失のしきい値を設定する。暴走的な実行を防ぐレートリミッターを実装する。大きな損失後のクールダウン期間を組み込む。これらの機構は、問題に気づく前に不具合のあるボットが口座を空にしてしまうことを防ぐ。
人間をループに残す: 完全自律は印象的に聞こえるが、最も堅牢なシステムは、リスクの高い判断については人間を関与させ続けている。一定のしきい値を超える取引には手動承認を求める、あるいはボットの挙動が想定パターンから逸脱した際に通知するアラートシステムを構築することを検討するとよい。

長期的に成功するボットは、単に最良のモデルを持つものではなく、問題を早期に察知し、優雅に回復できるだけの可観測性と安全対策を備えたものである。

ツールと技術:AIトレーディングボットのスタック

AIトレーディングボットのインフラ層(データの正確性、レイテンシー、信頼性)を押さえたところで、次は実際のツールキットを組み立てる段階だ。以下は、本番運用に耐えるAIトレーディングボットのスタックに必要な要素である。

コアプログラミングスタック

Pythonは、その豊富な機械学習エコシステムとシンプルさから、AIトレーディングボットにおいて依然として主流の言語である。コアとなるPythonライブラリには以下を含めるべきだ。

Pandas: データ操作と時系列分析に使用
NumPy: 価格・出来高データの数値計算に使用
scikit-learn: 従来型の機械学習モデル(回帰、分類、クラスタリング)に使用
PyTorchまたはTensorFlow: 系列予測のためのLSTMやトランスフォーマーなどのディープラーニングモデルに使用

これらのライブラリを組み合わせることで、単純な移動平均クロスオーバーボット(一日の午後で構築できる)から、テクニカル分析、センチメントスコアリング、オンチェーン指標を融合させた高度なマルチモデルボットまで、幅広い選択肢が手に入る。これがPythonエコシステムの強みだ。シンプルに始めて、必要に応じて複雑さを加えていける。線形回帰で十分な場合に、初日からディープラーニング用のアーキテクチャを設計する必要はない。

補足:マイクロ秒単位が重要な高頻度取引ボットを構築する場合、C++は生の速度と低レベルの制御を提供する。Rustも、パフォーマンスとメモリ安全性から暗号資産分野で支持を集めており、特にセキュリティが最優先されるオンチェーンコンポーネントやカスタムDEX連携を構築する場合に有用だ。Goは、複数の市場を同時に監視する必要がある並行処理システムにとって、もう一つの堅実な選択肢である。

とはいえ、ほとんどの暗号資産トレーディング戦略、たとえ高度なものであっても、Pythonのパフォーマンスで十分であり、それによって得られる開発速度は、ほとんどの小口トレーダーにとってわずかな速度改善を大きく上回る。まずPythonで始め、プロファイリングによって実際に必要だと判明した場合に限り、重要なコンポーネントをより速い言語に移行すればよい。

ブロックチェーンインフラ:Alchemyの層

ここでAlchemyが不可欠になる。Alchemyは、信頼性のあるAIトレーディングボットを大規模に稼働させるためのインフラ層を提供する。主要なツールは以下の通りだ。

リアルタイムブロックチェーンデータ: 50以上のチェーンにわたって、トークン価格、残高、取引履歴、NFTメタデータをクエリできる。サンドボックスで実際に試してみてほしい。
WebSocket接続: ペンディング中のトランザクション、新規ブロック、スマートコントラクトの状態変化といったリアルタイムイベントをサブスクライブできる。
拡張API: ガス価格予測、メタデータ付きトークン残高、過去の取引データにアクセスできる。

暗号資産トレーディングボットには、以下も必要になるだろう。

Web3.py: Ethereumとやり取りするためのインターフェース。トランザクションの送信、スマートコントラクトの呼び出し、データのエンコーディングを行う。DEXで取引したり、DeFiプロトコルと直接やり取りする場合には必須だ。
CCXT: 一貫した構文で100以上の中央集権型取引所(Binance、Coinbase、Krakenなど)で取引できる統一APIラッパー。各取引所のAPIの癖を個別に覚える代わりに、CCXTがすべてを正規化してくれる。
AlchemyのMCP(Model Context Protocol)サーバー: ブロックチェーンとAIが交わる場所だ。AlchemyのMCPサーバーにより、AIエージェントは自然言語でオンチェーンデータをクエリできる。複雑なクエリを書く代わりに、AIに「ETHの現在の価格は?」と尋ねれば、構造化されたデータが返ってくる。

Alchemy Skillsをインストールし、MCPサーバーに接続し、CLIを使用する。

bashCopied
npx skills add alchemyplatform/skills --yes

# Claude Code
claude mcp add alchemy --transport http https://mcp.alchemy.com/mcp

# Codex
codex mcp add alchemy --url https://mcp.alchemy.com/mcp

npm i -g @alchemy/cli@latest
alchemy auth
alchemy --json --no-interactive prices symbol ETH,BTC
alchemy --json --no-interactive balance vitalik.eth -n eth-mainnet

自律的な取引実行については、CLIのエージェントウォレットを参照してほしい。

AI強化ツール

トレーディングボットの次の進化は、単なるデータ分析にとどまらず、AIを使ってますます高度な判断を行うことにある。この1年で、開発者たちがそれぞれのプロセスを洗練させるために使ってきた、2つの主要な手法が登場した。

大規模言語モデルのAPI(Claude、GPT-4など)は、従来の機械学習を超えた能力を解放する。

センチメント分析: 数千件のツイート、Redditの投稿、ニュース記事を処理して市場のムードを測る
戦略の生成: 平易な英語でトレーディングのアイデアを記述すれば、AIがそれをコードとして形式化する手助けをしてくれる
異常検知: LLMを使って、機会を示唆する可能性のある市場データの異常なパターンを見つける

2025年には、これをさらに推し進め、自律型システムを構築する新しいエージェントフレームワークが登場した。

ElizaOS(ai16zによる): 分散型AIファンドの背後にあるフレームワークで、取引を提案するだけでなく実行するエージェントを構築できる。ポートフォリオを管理し、他のエージェントと連携することさえ可能だ。
Virtuals GAME SDK: より大きなエージェント経済に参加するエージェントを作成できる。手数料を稼いだり、他のエージェントにサービスを提供したり、複雑な戦略で協調したりする。他者と「うまく協調できる」ボットを構築するようなものと考えればよい。
LangChain: 複雑なAIワークフローをオーケストレーションするためのスイスアーミーナイフだ。複数のAI呼び出しを組み合わせ、異なるソースからのデータを統合し、ハードコーディングすると面倒な判断木を構築できる。戦略が複数段階の推論を必要とする場合に特に有用だ。

これらのツールは、「ルールに従うボット」から「適応し学習するエージェント」へと役割をシフトさせる。開発における学習曲線は急になるが、得られる成果の可能性は大幅に高い。

重要な補助ツール

市場データ用API: CoinGecko、CoinMarketCap、あるいは取引所固有のAPI(Binance、Coinbase、Kraken)といったサービスは、価格データ、取引量、時価総額ランキング、過去のOHLCV(始値、高値、安値、終値、出来高)データを提供する。Alchemyがオンチェーンデータを扱う一方で、これらのAPIは中央集権型取引所の価格や、モデルが判断の根拠として必要とする、より広い市場状況のギャップを埋めてくれる。
バックテストフレームワーク: 実際の資本をリスクにさらす前に、過去のデータに対して戦略を検証する必要がある。BacktraderやZiplineのようなフレームワークを使えば、強気相場、暴落、レンジ相場など過去の市場状況において、ボットがどのようなパフォーマンスを発揮したかをシミュレートできる。優れたバックテストインフラは、戦略の弱点を特定し、パラメータを最適化し、実運用前に確信を持つ助けになる。
バージョン管理: 本格的なトレーディングシステムにおいて、Gitは選択肢ではない。戦略を進化させていく中で、パラメータの調整、新しいシグナルの追加、バグの修正など、すべての変更を追跡し、失敗した実験をロールバックし、特定の取引が実行された時点でどのコードが動いていたかの明確な履歴を維持する必要がある。これは、ボットの挙動を時間とともにデバッグし監査する上で極めて重要になる。

AIトレーディングボットの種類

AIトレーディングボットは、単純な自動化から実験的なマルチエージェントシステムまで、幅広い戦略と複雑さのレベルにまたがる。以下は、よく目にする主なカテゴリだ。

テクニカル分析・センチメント分析ボット: テクニカルボットは、価格チャート、移動平均、各種指標を分析して取引機会を特定する。AIによる強化により、シグナルを動的に重み付けし、静的なルールでは見逃してしまう複雑なパターンを認識できる。センチメントボットは、Twitter/Xの投稿、Redditの議論、ニュース記事、オンチェーンシグナル(クジラの動き、取引所への入金、ガス価格の急騰)を解析し、市場のムードに基づいて短期的な価格変動を予測する。ハイプサイクルがボラティリティを引き起こすアルトコインやミームトークンで特に人気が高い。
裁定取引・マーケットメイキングボット: 裁定取引ボットは、取引所や流動性プール間の価格差を利用する。ある取引所でETHを2,000ドルで購入し、別の取引所で2,010ドルで売却するといった具合だ。AIは、こうした価格差が発生する前にそれを予測し、DEX間の複数ホップのルートを最適化するのに役立つ。マーケットメイキングボットは、買いと売りを同時に提示することで流動性を提供し、スプレッドから利益を得る。DeFiにおいては、UniswapやCurveのプールに預け入れ、AIを使ってインパーマネントロスを管理し、ボラティリティに応じてスプレッドを調整することを意味する。
トレンドフォロー・ポートフォリオ管理ボット: モメンタムボットは、勢いを増している資産を特定し、シグナルが反転するまでトレンドに乗る。LSTM(Long Short-Term Memory network)のようなモデルは、トレンドが継続するか崩れるかを予測する助けになる。ポートフォリオボットは、AIの予測に基づいて保有資産を自動的にリバランスし、全体のリスクエクスポージャーを管理しながら、期待リターンがより高い資産への配分をシフトする。
自律型AIエージェントファンド: 実験的な最前線では、ElizaOSのようなプロジェクトが、AIエージェントがトレーディング戦略を調整し、シグナルを共有し、集団パフォーマンスを最適化する、DAOガバナンスによるファンドを運用している。aixbtと同様の仕組みだ。これらのマルチエージェントシステムは、トレーディングボットと自律的な金融主体との境界を曖昧にしており、まだ初期段階だが、AI駆動型金融の向かう先を垣間見せている。

最初のAIトレーディングボットを構築する:ステップバイステップのチュートリアル

このセクションでは、オンチェーンデータと機械学習を組み合わせて取引判断を行う、Ethereum向けのセンチメント強化型トレンド予測ボットを実際に作ってみる。

構築するもの:

Alchemy経由でリアルタイムのETH価格とブロックチェーンデータを取得するボット
価格指標とオンチェーンセンチメントシグナルを組み合わせた特徴量エンジニアリング
短期的な価格変動を予測する機械学習モデル
信頼度のしきい値とリスク管理を備えたトレーディングロジック
実運用前に戦略を検証するためのバックテスト機能

これは原始的な例であり、本番運用に耐えるシステムではなく、これから発展させていける基盤と考えてほしい。目的は、核となる概念とワークフローを理解し、自分自身の戦略に基づいて拡張していくことだ。

前提条件

コーディングを始める前に、以下を確認しておこう。

Python 3.8以上がマシンにインストールされていること
APIキーを持つAlchemyアカウント(こちらから無料登録)
Pythonの基礎知識: 関数、ループ、データ構造に抵抗がないこと
機械学習の基礎知識: 学習/テスト分割やモデル評価といった概念に馴染みがあること

まず、必要なパッケージをインストールする。

bash

Copied

pip install pandas numpy scikit-learn requests web3 python-dotenv

各パッケージが何をするか:

pandas: データ操作と時系列を扱う
numpy: 数値計算を担う
scikit-learn: 機械学習アルゴリズムを提供する
requests: APIへのHTTPリクエストを行う
web3: AlchemyのEthereumエンドポイントに接続し、ブロックチェーンとやり取りする
python-dotenv: 環境変数を安全に管理する

ステップ1: Alchemyを設定し、過去のブロックチェーンデータを取得する

まず、ブロックチェーンへの接続を確立し、モデルの学習に使う過去データを収集しよう。

APIキーを安全に保存するため、プロジェクトディレクトリに.envファイルを作成する。

markdownCopied
ALCHEMY_API_KEY=your_api_key_here

次に、blockchain\_data\_fetcher.pyというファイルを作成し、ブロックチェーンデータを取得するコードを書こう。

pythonCopied
import os
from dotenv import load_dotenv
from web3 import Web3
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import time

# Load environment variables from .env file

load_dotenv()

# Connect to Alchemy's Ethereum mainnet endpoint

w3 = Web3(Web3.HTTPProvider(
f"https://eth-mainnet.g.alchemy.com/v2/{os.getenv('ALCHEMY_API_KEY')}"
))

def fetch_historical_data(hours=168): # Default: last week (168 hours)
  """
Fetch historical blockchain data that we'll use as features.

  We're collecting:
  - Block numbers and timestamps (for temporal ordering)
  - Gas used per block (proxy for network activity)
  - Transaction counts (proxy for market interest)

  These on-chain metrics can signal market conditions before they
  appear in price data.
  """
  current_block = w3.eth.block_number

  # Ethereum produces ~1 block every 12 seconds# So roughly 300 blocks per hour
  blocks_per_hour = 300
  blocks_to_fetch = hours * blocks_per_hour

  # We'll sample every N blocks to avoid excessive API calls# Sampling hourly (every 300 blocks) gives us manageable data
  data = []
  for i in range(current_block - blocks_to_fetch, current_block, blocks_per_hour):
      try:
          block = w3.eth.get_block(i)
          data.append({
              'block_number': i,
              'timestamp': block['timestamp'],
              'gas_used': block['gasUsed'],
              'transaction_count': len(block['transactions'])
          })
      except Exception as e:
          print(f"Error fetching block {i}: {e}")
          continue

  return pd.DataFrame(data)

# Fetch a week of data

print("Fetching historical blockchain data...")
df = fetch_historical_data(hours=168)
print(f"Retrieved {len(df)} data points")
print(df.head())

このスクリプトは、あなたのAlchemy APIキーを使ってweb3.pyでEthereumメインネットに接続する。次に、希望する期間に基づいて取得すべきブロック数を計算する。Ethereumの約12秒というブロックタイムから、1時間あたり約300ブロックとなる。データセットを扱いやすい規模に保ち、過剰なAPI呼び出しを避けるため、すべてのブロックを取得するのではなく、1時間ごとに1ブロックをサンプリングする。サンプリングした各ブロックについて、ネットワーク活動の代理指標としてガス使用量とトランザクション数を抽出し、後のステップで扱いやすいようにすべてをpandasのDataFrameに保存する。

これらのオンチェーン指標は、価格データに現れる前に市場活動の早期シグナルを提供することが多い。ガス使用量が急上昇するのは、通常、取引活動の増加、DeFiとのやり取り、あるいは重要なオンチェーンイベントを示しており、それらはいずれも価格のボラティリティに先行し得る。同様に、トランザクション数の増加は、ETHへの関心の高まりやDeFi活動の活発化を示すことがある。これらブロックチェーン固有のシグナルを従来の価格データと合わせてモデルに組み込むことで、純粋に価格ベースの戦略では見逃してしまう情報をボットに与えることになる。

ステップ2: Alchemyを使ってETH価格データを取得する

ブロックチェーンデータだけでは不十分だ。モデルを学習させるには実際の価格情報が必要になる。fetch\_eth\_price.pyというファイルを作成し、複数のチェーンにわたって低レイテンシーでリアルタイム価格を提供する、AlchemyのPrices APIを組み込もう。

なぜAlchemyのPrices APIなのか。

統一されたプロバイダ: 同じAPIキー、同じレート制限、一貫した課金
取引所の集約: 価格は複数の取引所にわたる実際の市場状況を反映する
低レイテンシー: トレーディングアプリケーション向けに最適化されている(典型的には100ミリ秒未満のレスポンス)
マルチチェーン対応: 対応する任意のチェーン上の任意のトークンで、同じAPI構造が使える

pythonCopied
import requests
from datetime import datetime, timedelta

def fetch_eth_prices_alchemy(hours=168):
  """
Fetch historical ETH prices using Alchemy's Token Prices API.

  This gives us reliable, exchange-aggregated pricing that reflects
  actual market conditions across major venues.
  """
  # Alchemy's prices endpoint
  url = f"https://eth-mainnet.g.alchemy.com/prices/v1/{os.getenv('ALCHEMY_API_KEY')}/tokens/by-symbol"

  # Calculate our time window
  end_time = datetime.now()
  start_time = end_time - timedelta(hours=hours)

  params = {
      'symbols': 'ETH',
      'startTime': int(start_time.timestamp()),
      'endTime': int(end_time.timestamp()),
      'interval': '1h'  # Hourly price points
  }

  headers = {
      'Accept': 'application/json'
  }

  response = requests.get(url, params=params, headers=headers)

  if response.status_code == 200:
      data = response.json()
      prices = data['data'][0]['prices']

      # Convert to DataFrame format
      price_data = []
      for price_point in prices:
          price_data.append({
              'timestamp': datetime.fromtimestamp(price_point['timestamp']),
              'price': price_point['value']
          })

      return pd.DataFrame(price_data)
  else:
      raise Exception(f"Failed to fetch prices: {response.status_code} - {response.text}")

# Fetch prices

print("Fetching ETH price data from Alchemy...")
price_df = fetch_eth_prices_alchemy(hours=168)

# Ensure timestamps are datetime objects for merging

df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')

# Merge blockchain data with price data# merge_asof handles slight timestamp mismatches by finding nearest match

df = pd.merge_asof(
df.sort_values('timestamp'),
price_df.sort_values('timestamp'),
on='timestamp',
direction='nearest'
)

print(f"Merged data shape: {df.shape}")
print(df.head())

このスクリプトは、ブロックチェーンデータと同じ期間(デフォルトで168時間)にわたって、AlchemyのToken Prices APIから過去のETH価格を取得する。開始時刻と終了時刻を定義し、1時間ごとの価格ポイントをリクエストし、レスポンスをタイムスタンプと価格値を持つpandasのDataFrameに解析する。

最後の重要なステップは、この価格データをステップ1のブロックチェーンデータとマージすることだ。ブロックデータと価格データのタイムスタンプは完全には一致しないため、標準的なマージの代わりにmerge\_asofを使う。merge\_asofは各行に最も近いタイムスタンプを見つけてくれるため、わずかなタイミングのずれによるデータの欠落を防げる。このステップの後、オンチェーン指標(ガス使用量、トランザクション数)と価格データの両方を含む統一されたDataFrameが手に入り、次のステップの特徴量エンジニアリングに進む準備が整う。

ステップ3: 機械学習のための特徴量を設計する

生のデータが手に入ったので、次はそれをモデルが学習できるパターンを捉える特徴量に変換する必要がある。ここでドメイン知識とデータサイエンスが交わる。

pythonCopied
def engineer_features(df):
  """
  Transform raw data into predictive features.
  
  We're creating two types of features:
  1. Price-based technical indicators (moving averages, volatility)
  2. On-chain sentiment proxies (gas and transaction trends)
  
  The goal is to give the model multiple perspectives on market conditions.
  """
  
  # ============================================# PRICE-BASED FEATURES# ============================================
  
  # Percentage change from previous hour# Captures momentum - is price accelerating up or down?
  df['price_change'] = df['price'].pct_change()
  
  # Moving averages - smooth out noise, identify trends
  df['price_ma_12'] = df['price'].rolling(window=12).mean()  # 12-hour
  df['price_ma_24'] = df['price'].rolling(window=24).mean()  # 24-hour
  
  # When short-term MA crosses above long-term MA = bullish signal# When it crosses below = bearish signal
  
  # Volatility - standard deviation of recent price changes# High volatility = risky/unstable market
  df['volatility'] = df['price_change'].rolling(window=12).std()
  
  # ============================================# ON-CHAIN SENTIMENT FEATURES# ============================================
  
  # Gas usage trend - are people paying more to transact?# Spikes often precede price movements
  df['gas_trend'] = df['gas_used'].pct_change()
  
  # Transaction count trend - is activity increasing?# More transactions = more interest = potential price catalyst
  df['tx_trend'] = df['transaction_count'].pct_change()
  
  # ============================================# MOMENTUM INDICATORS# ============================================
  
  # Price change over last 6 hours# Positive momentum = upward trajectory
  df['momentum'] = df['price'] - df['price'].shift(6)
  
  # ============================================# TARGET VARIABLE# ============================================
  
  # What we're trying to predict: will price go up in the next hour?# 1 = yes (buy signal), 0 = no (sell/hold signal)
  df['target'] = (df['price'].shift(-1) > df['price']).astype(int)
  
  # Drop rows with NaN values (from rolling windows and shifts)
  df = df.dropna()
  
  return df

df = engineer_features(df)
print(f"\nEngineered features: {df.columns.tolist()}")
print(f"Data shape: {df.shape}")
print(f"\nSample of engineered data:")
print(df[['price', 'price_change', 'volatility', 'momentum', 'target']].head())

この関数は、生のブロックチェーンデータと価格データを、機械学習モデルが予測に使える特徴量に変換する。各特徴量は市場状況に関する異なるシグナルを捉えており、価格の動きから導かれるものもあれば、オンチェーン活動から導かれるものもある。以下は、各特徴量が何を表し、価格方向の予測になぜ重要なのかを説明する。

価格変化(pct\_change): 絶対的な価格変動をパーセンテージに変換する。2,000ドルから2,020ドルへの変化は1%の動きであり、単に「20ドル上昇」と言うよりも意味があり比較可能だ。パーセンテージは、異なる価格水準にまたがる変動を正規化する。
移動平均(12時間と24時間): これらは短期的な価格ノイズを平滑化し、根底にあるトレンドを浮かび上がらせる。短期MAが長期MAを上抜けするとき、伝統的には強気シグナルと解釈され、下抜けは弱気モメンタムを示唆する。モデルはこうしたクロスオーバーパターンを学習できる。
ボラティリティ: 直近の価格変化の標準偏差として測定される。高いボラティリティは、より大きな価格変動を伴う不安定な市場を示す。リスクは高いが、潜在的なリターンも高い。一部の戦略はボラティリティの高い局面を避け、逆にそれを特に狙う戦略もある。
ガスとトランザクションのトレンド: 価格の動きに先行し得るオンチェーン指標。ガス使用量が突如急増した場合、大規模なNFTドロップ、DeFiのエクスプロイト、クジラの活動といった何か重要なことが起きている可能性があり、それが取引データに現れる前にETH価格に影響を与えることがある。
モメンタム: 過去6時間にわたって価格が上昇トレンドにあるかどうかを示す、シンプルだが効果的な指標。正のモメンタムは上昇の継続を示唆し、負のモメンタムは下落を示唆する。モメンタムトレーダーは、これらのシグナルを使って既存のトレンドに乗る。
目的変数: これは、モデルに予測させたい対象だ。次の1時間で価格は現在より上がるか。値が1なら上がる(潜在的な買いシグナル)、0なら上がらない(潜在的な売りまたはホールドのシグナル)を意味する。この二値分類アプローチにより、予測問題がシンプルになる。

ステップ4: 機械学習モデルを学習させる

次に、特徴量からパターンを学習するモデルを訓練する。ここではRandom Forestを使う。ノイズの多いデータに頑健で、広範なハイパーパラメータチューニングを必要としないアンサンブル手法だ。

pythonCopied
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import accuracy_score, classification_report

# Select the features we engineered

feature_columns = [
'price_change', 'price_ma_12', 'price_ma_24', 'volatility',
'gas_trend', 'tx_trend', 'momentum'
]

X = df[feature_columns]
y = df['target']

# Split data: 80% for training, 20% for testing# CRITICAL: shuffle=false to avoid look-ahead bias# in time series, future data can't inform past predictions

X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, shuffle=False
)

print(f"Training samples: {len(X_train)}")
print(f"Testing samples: {len(X_test)}")

# Initialize random forest classifier

model = RandomForestClassifier(
n_estimators=100, # 100 decision trees in the forest
max_depth=10, # Prevent trees from growing too deep (overfitting)
random_state=42, # For reproducibility
class_weight='balanced' # Handle imbalanced classes (more ups than downs or vice versa)
)

print("\nTraining model...")
model.fit(X_train, y_train)
print("✓ Model trained")

# ============================================# EVALUATE ON TEST set# ============================================

y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)

print(f"\n{'='*50}")
print(f"TEST SET PERFORMANCE")
print(f"{'='*50}")
print(f"Accuracy: {accuracy:.2%}")
print(f"\nDetailed Classification Report:")
print(classification_report(y_test, y_pred, target_names=['Down', 'Up']))

# ============================================# CROSS-validation# ============================================# cross-validation gives us more robust performance estimates# by training/testing on multiple data splits

cv_scores = cross_val_score(model, X_train, y_train, cv=5)
print(f"\nCross-validation scores: {cv_scores}")
print(f"Average CV score: {cv_scores.mean():.2%} (+/- {cv_scores.std():.2%})")

# ============================================# FEATURE importance# ============================================# random forests tell us which features mattered most

feature_importance = pd.DataFrame({
  'feature': feature_columns,
  'importance': model.feature_importances_
}).sort_values('importance', ascending=False)

print(f"\n{'='*50}")
print("FEATURE IMPORTANCE")
print(f"{'='*50}")
print(feature_importance)

このスクリプトは、前のステップで設計した特徴量を使ってRandom Forest分類器を学習させる。データを学習用(80%)とテスト用(20%)に分割するが、時系列の順序を保つためにshuffle=Falseを使う。これは時系列データにとって重要で、未来の情報が過去の予測に漏れ込むのを防ぐ必要があるからだ。モデルは学習データからパターンを学び、その後、それまで見たことのないテストセットに対して予測を行う。また、複数の異なるデータ分割で学習とテストを行い、結果が一貫しているのか、それとも単なる幸運なのかを確認するクロスバリデーションも実行する。

結果の解釈

機械学習モデルは大量の指標を出力する。ボットが有効な戦略を持っているかどうかを理解するには、これらを正しく解釈することが不可欠だ。以下は、それぞれの出力が何を意味し、何に注目すべきかをまとめたものだ。

精度(Accuracy): 正しい予測の割合。トレーディングにおいては、55〜60%の精度でもかなり良い方だ。市場は本質的にノイズが多く、予測が難しい。90%を超える精度は期待しないこと。もしそれほど高い数値が出た場合、モデルが過学習しているか、データリークの問題がある可能性が高い。
分類レポート: 両方の予測クラス(上昇と下降)についてパフォーマンスを分解する。Precisionは、モデルが「上昇」と予測したときにどれだけ正しいかを示す。誤検知は悪い取引につながるため重要だ。Recallは、実際の「上昇」の動きのうちモデルが捉えた割合を示す。機会を逃さないために重要だ。F1スコアは両者の調和平均であり、パフォーマンスのバランスの取れた見方を与えてくれる。
クロスバリデーションスコア: データの異なるサブセットで学習することで、モデルが一貫したパフォーマンスを発揮するか、それとも使用する特定のデータポイントによって結果が大きくぶれるかが分かる。各フォールドで一貫したスコアが出れば、それは頑健な戦略を示唆する。分散が大きい場合は、モデルが実際のパターンではなくノイズを学習している可能性を示唆する。
特徴量の重要度: Random Forestは、どの特徴量が予測に最も寄与したかを教えてくれる。momentumの重要度が最も高ければ、あなたの戦略は主にモメンタム主導だ。volatilityが支配的であれば、主に市場の不安定性に基づいて取引していることになる。この洞察は、モデルが実際に何をしているのか、それが自分のトレーディングの仮説と一致しているのかを理解するのに役立つ。

何が「良い」結果とされるのか。

トレーディングモデルとしては、55%を超える精度(50%のランダムな推測より良い)、学習セットとテストセットで同程度のパフォーマンス(過学習していないことを示す)、そして一貫したクロスバリデーションスコア(異なる市場状況にわたって戦略が頑健であることを示唆する)を目指すべきだ。テストの精度が学習の精度より著しく低い場合、モデルは汎化可能なパターンを学習するのではなく、学習データを丸暗記してしまっている。

ステップ5: リアルタイムのトレーディングロジックを実装する

次に、ボットのメインループ、つまり現在のデータを取得し、予測を行い、取引を実行するロジックを構築しよう。

pythonCopied
def fetch_current_data():
  """
  Fetch the most recent data point for making a prediction.
  
  In production, this runs every N minutes to stay current.
  """
  # Get latest block from blockchain
  current_block = w3.eth.block_number
  block = w3.eth.get_block(current_block)
  
  # Get current ETH price (using Alchemy Prices API)
  price_url = f"https://eth-mainnet.g.alchemy.com/prices/v1/{os.getenv('ALCHEMY_API_KEY')}/tokens/by-symbol"
  price_response = requests.get(
      price_url,
      params={'symbols': 'ETH'},
      headers={'Accept': 'application/json'}
  )
  current_price = price_response.json()['data'][0]['prices'][0]['value']
  
  return {
      'block_number': current_block,
      'timestamp': datetime.now(),
      'gas_used': block.gas_used,
      'transaction_count': len(block.transactions),
      'price': current_price
  }

def prepare_features_for_prediction(current_data, historical_df):
  """
Engineer features from current data point using historical context.

  Why we need historical context: Features like moving averages and
  momentum require previous data points to calculate.
  """
  # Append current data to our historical DataFrame
  temp_df = pd.concat([
      historical_df,
      pd.DataFrame([current_data])
  ], ignore_index=True)

  # Re-run feature engineering to get current features
  temp_df = engineer_features(temp_df)

  # Return only the latest row (current features)
  return temp_df[feature_columns].iloc[-1:].values

def execute_trade(action, amount, current_price):
  """
Execute a trade. Currently simulated—prints trade details.

  In production, replace with:
  - Exchange API calls (via CCXT)
  - Signed onchain transactions sent through Alchemy's Node API
  - Order execution with slippage controls
  """
  print(f"\n{'='*50}")
  print(f"TRADE SIGNAL: {action.upper()}")
  print(f"Amount: {amount} ETH")
  print(f"Price: ${current_price:,.2f}")
  print(f"Est. Value: ${amount * current_price:,.2f}")
  print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
  print(f"{'='*50}\n")

  # TODO: In production, add real execution:# if action == 'BUY':#     exchange.create_market_buy_order('ETH/USD', amount)# elif action == 'SELL':#     exchange.create_market_sell_order('ETH/USD', amount)

def run_trading_bot(model, historical_df, interval_seconds=300):
  """
Main bot loop: fetch data, predict, execute trades.

  This runs continuously, checking market conditions every N seconds.
  """
  print("="*60)
  print("AI TRADING BOT STARTED")
  print("="*60)
  print(f"Checking market every {interval_seconds} seconds")
  print(f"Model accuracy: {accuracy:.2%}")
  print(f"Press Ctrl+C to stop\n")

  # Track bot state
  position = None  # Current position: 'long', 'short', or None
  entry_price = 0
  trades_executed = 0

  while True:
      try:
          # ============================================# 1. FETCH CURRENT MARKET DATA# ============================================
          current_data = fetch_current_data()
          current_price = current_data['price']

          # ============================================# 2. PREPARE FEATURES & MAKE PREDICTION# ============================================
          features = prepare_features_for_prediction(current_data, historical_df)
          prediction = model.predict(features)[0]
          prediction_proba = model.predict_proba(features)[0]

          # Display current state
          print(f"[{datetime.now().strftime('%H:%M:%S')}] Price: ${current_price:,.2f} | "
                f"Prediction: {'UP' if prediction == 1 else 'DOWN'} "
                f"(Confidence: {prediction_proba[prediction]:.1%}) | "
                f"Position: {position or 'None'}")

          # ============================================# 3. TRADING LOGIC WITH CONFIDENCE THRESHOLD# ============================================

          # Only trade when model is confident (>60%)
          confidence_threshold = 0.6

          # BUY SIGNAL: Predict UP with high confidence, no position
          if prediction == 1 and prediction_proba[1] > confidence_threshold:
              if position != 'long':
                  execute_trade('BUY', 0.1, current_price)
                  position = 'long'
                  entry_price = current_price
                  trades_executed += 1

          # SELL SIGNAL: Predict DOWN with high confidence, have long position
          elif prediction == 0 and prediction_proba[0] > confidence_threshold:
              if position == 'long':
                  profit_pct = ((current_price - entry_price) / entry_price) * 100
                  print(f"Closing long position. Profit: {profit_pct:+.2f}%")
                  execute_trade('SELL', 0.1, current_price)
                  position = None
                  trades_executed += 1

          # ============================================# 4. UPDATE HISTORICAL DATA (ROLLING WINDOW)# ============================================

          historical_df = pd.concat([
              historical_df,
              pd.DataFrame([current_data])
          ], ignore_index=True).tail(168)  # Keep last week

          # ============================================# 5. WAIT BEFORE NEXT CHECK# ============================================

          time.sleep(interval_seconds)

      except KeyboardInterrupt:
          print(f"\n{'='*60}")
          print("BOT STOPPED BY USER")
          print(f"Total trades executed: {trades_executed}")
          print(f"{'='*60}")
          break

      except Exception as e:
          print(f"ERROR: {e}")
          print("Retrying in 60 seconds...")
          time.sleep(60)

# Run the bot (check every 5 minutes)

run_trading_bot(model, df, interval_seconds=300)

このステップで、すべてを統合して機能するトレーディングボットに仕上げる。コードは、順番に動作する4つの主要な関数を定義している。fetch_current_dataは最新のブロックチェーン指標とETH価格をAlchemyから取得し、prepare_features_for_predictionはその生データをモデルが学習したのと同じ特徴量に変換し、execute_tradeは実際の買い/売りの実行を扱い(ここではシミュレーションだが、本番連携の準備はできている)、run_trading_botは5分ごとに市場状況をチェックする継続的なループですべてを統括する。

メインのトレーディングループは、単純なサイクルに従う。現在のデータを取得し、特徴量を生成し、予測を行い、信頼度のしきい値に基づいて取引するかどうかを判断し、過去データを更新し、繰り返す前に待機する。このループに組み込まれた主要な設計上の判断は以下の通りだ。

信頼度のしきい値: ボットは、モデルの予測に対する信頼度が60%を超える場合にのみ取引を実行する。これにより、本物の機会というよりノイズである可能性が高い弱いシグナルを除外できる。このしきい値はリスク許容度に応じて調整できる。しきい値を高くすれば、取引数は減るがより選別された取引になる。
ポジション追跡: ボットは現在の状態(ロング、ショート、フラット)を常に把握している。これにより、すでにETHを保有しているのに繰り返し買い注文を出す、あるいは売るものがないのに売り注文を出すといった、非論理的な挙動を防ぐ。
ローリングウィンドウ: 過去データは直近168時間(1週間)に制限される。これにより、長時間実行中のメモリ問題を防ぎ、ボットの特徴量計算が古いデータに歪められることなく現在の市場状況と関連性を保つことを保証する。
エラーハンドリング: try/exceptブロックにより、一時的な障害(APIタイムアウト、ネットワークの問題、レート制限)がボット全体をクラッシュさせないようにする。代わりに、エラーはログに記録され、ボットは短い休止の後に再試行する。
優雅な終了: Ctrl+Cを押すとボットはきれいに停止し、サマリー統計を表示する。状態情報を失うことなくパフォーマンスをレビューできる。

なぜ5分間隔なのか。このデフォルト値は、応答性とAPIのレート制限、計算オーバーヘッドのバランスを取ったものだ。これは戦略に応じて調整できる。高頻度アプローチでは数秒ごとにチェックすることもあれば、スイングトレーディング戦略では1時間ごとの更新で十分な場合もある。より短い間隔はより多くの機会を捉えるが、より多くのAPI呼び出しを消費し、より高速な実行インフラを必要とする。

ステップ6: Alchemyでオンチェーンセンチメント分析を追加する

価格とテクニカル指標だけでは、物語の一部しか語れない。暗号資産において、特に「クジラ」による大口送金といったオンチェーン活動は、価格データに現れる前に、差し迫った価格の動きを示唆することがある。クジラがETHを取引所に移動させることは、しばしば売り圧力を示し、逆に取引所からETHを引き出すことは蓄積を示唆する。これらの送金をリアルタイムで監視することで、ボットは価格がその活動を反映するより数秒から数分早く、事前の警告を得ることができる。

このステップでは、ボットの追加の入力層としてクジラの監視を加える。

以下の2つのコードブロックでは、まず各ブロックで新しいWETH送金をAlchemyにポーリングし、定義したしきい値(100 ETH)を超える送金をフィルタリングし、資金が既知の取引所アドレスとの間で移動しているかどうかに基づいて分類する。純粋なETH送金はイベントを発行しないため、WETHのTransferイベントが大口ETH移動の実用的な代理指標となる。

2つ目のコードブロックでは、そのクジラ活動データを既存のトレーディングロジックに取り込み、クジラが買っているように見えるか売っているように見えるかに基づいて信頼度のしきい値を調整する方法を示す。

クジラの送金を監視するモニターの設定:

pythonCopied
def monitor_whale_activity():
  """
  Monitor large WETH transfers as sentiment signals using Alchemy's Node API.
  
  Why this matters:
  - Whales moving ETH to exchanges often signals selling pressure
  - Whales moving ETH off exchanges signals accumulation (bullish)
  - Large transfers between wallets can indicate OTC deals or fund movements
  
  This real-time data can give you seconds or minutes of advance warning
  before price reflects the activity.
  """
  # Define what constitutes a "whale" transaction
  WHALE_THRESHOLD = 100  # 100 ETH (~$200K-400K depending on price)
  
  # Known exchange addresses (for context)
  EXCHANGE_ADDRESSES = {
      '0x3f5ce5fbfe3e9af3971dd833d26ba9b5c936f0be': 'Binance',
      '0x28c6c06298d514db089934071355e5743bf21d60': 'Binance 2',
      '0x21a31ee1afc51d94c2efccaa2092ad1028285549': 'Binance 3',
      '0xdfd5293d8e347dfe59e90efd55b2956a1343963d': 'Binance 4',
      '0x56eddb7aa87536c09ccc2793473599fd21a8b17f': 'Binance 5',
      '0x9696f59e4d72e237be84ffd425dcad154bf96976': 'Binance 6',
      '0x4e9ce36e442e55ecd9025b9a6e0d88485d628a67': 'Binance 7',
      '0xbe0eb53f46cd790cd13851d5eff43d12404d33e8': 'Binance 8',
      '0xf977814e90da44bfa03b6295a0616a897441acec': 'Binance 9',
      '0x001866ae5b3de6caa5a51543fd9fb64f524f5478': 'Coinbase 1',
      '0x71660c4005ba85c37ccec55d0c4493e66fe775d3': 'Coinbase 2',
      '0x503828976d22510aad0201ac7ec88293211d23da': 'Coinbase 3',
      '0xddfabcdc4d8ffc6d5beaf154f18b778f892a0740': 'Coinbase 4',
      '0x3cd751e6b0078be393132286c442345e5dc49699': 'Coinbase 5',
      '0xb5d85cbf7cb3ee0d56b3bb207d5fc4b82f43f511': 'Coinbase 6',
      '0xeb2629a2734e272bcc07bda959863f316f4bd4cf': 'Coinbase 7',
  }
  
  def handle_transfer(event):
      """
      Process each transfer event as it happens.
      """
      try:
          # Extract transfer details
          value_wei = int.from_bytes(event['data'], 'big')
          value_eth = value_wei / 1e18
          from_addr = '0x' + event['topics'][1].hex()[-40:]
          to_addr = '0x' + event['topics'][2].hex()[-40:]
          tx_hash = event['transactionHash'].to_0x_hex()
          
          # Only process whale-sized transfers
          if value_eth >= WHALE_THRESHOLD:
              # Determine transfer context
              from_exchange = EXCHANGE_ADDRESSES.get(from_addr, None)
              to_exchange = EXCHANGE_ADDRESSES.get(to_addr, None)
              
              # Build alert message
              alert = f"\n🐋 WHALE ALERT: {value_eth:.2f} ETH"
              
              if from_exchange and not to_exchange:
                  # Moving OFF exchange = potential buying/holding
                  alert += f"\n📤 From {from_exchange} to private wallet"
                  alert += "\n💡 Signal: BULLISH (accumulation)"
                  
              elif not from_exchange and to_exchange:
                  # Moving TO exchange = potential selling
                  alert += f"\n📥 From private wallet to {to_exchange}"
                  alert += "\n💡 Signal: BEARISH (potential sell pressure)"
                  
              elif from_exchange and to_exchange:
                  # Exchange to exchange = arbitrage or OTC
                  alert += f"\n🔄 From {from_exchange} to {to_exchange}"
                  alert += "\n💡 Signal: NEUTRAL (arbitrage or OTC)"
                  
              else:
                  # Wallet to wallet = unknown intent
                  alert += f"\n↔️  Between private wallets"
                  alert += "\n💡 Signal: UNCLEAR (monitor for pattern)"
              
              alert += f"\nTx: https://etherscan.io/tx/{tx_hash}"
              print(alert)
              
              # TODO: In production, you might:# - Adjust trading thresholds based on whale activity# - Increase position size if whales are accumulating# - Exit positions early if whales are dumping# - Log whale activity for later analysis
              
      except Exception as e:
          print(f"Error processing whale transfer: {e}")
  
  # WETH contract and the Transfer(address,address,uint256) event signature
  WETH_ADDRESS = Web3.to_checksum_address('0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2')
  TRANSFER_TOPIC = Web3.keccak(text="Transfer(address,address,uint256)").to_0x_hex()
  
  print("\n" + "="*60)
  print("WHALE ACTIVITY MONITOR STARTED")
  print("="*60)
  print(f"Tracking transfers ≥ {WHALE_THRESHOLD} ETH")
  print("Press Ctrl+C to stop\n")
  
  last_block = w3.eth.block_number
  while True:
      try:
          latest_block = w3.eth.block_number
          if latest_block > last_block:
              logs = w3.eth.get_logs({
                  'address': WETH_ADDRESS,
                  'topics': [TRANSFER_TOPIC],
                  'fromBlock': last_block + 1,
                  'toBlock': latest_block,
              })
              for log in logs:
                  handle_transfer(log)
              last_block = latest_block
      except Exception as e:
          print(f"Whale monitor error: {e}")
      
      time.sleep(12)  # Ethereum produces a block roughly every 12 seconds

# Run whale monitor in a separate thread so it doesn't block the main bot

import threading

whale_thread = threading.Thread(target=monitor_whale_activity, daemon=True)
whale_thread.start()

# Now your main trading bot can run alongside the whale monitor# the whale alerts will print in real-time while your bot trades

上記のクジラモニターはデータを収集するが、そのデータは実際に判断に影響を与えるためにトレーディングボットに取り込まれる必要がある。以下のコードは、クジラのセンチメントを累積スコアとして追跡し、それに応じて信頼度のしきい値を調整する方法を示している。クジラが蓄積している場合は買いのハードルを下げ、売っているように見える場合はハードルを上げる。

pythonCopied
# Add this to your trading bot
whale_sentiment = 0  # Global variable tracking recent whale activity

def update_whale_sentiment(signal_type):
  """
  Update sentiment based on whale activity.
  +1 for bullish moves, -1 for bearish, 0 for neutral.
  """
  global whale_sentiment
  if signal_type == 'BULLISH':
      whale_sentiment += 1
  elif signal_type == 'BEARISH':
      whale_sentiment -= 1
  # Decay over time so old signals matter less
  whale_sentiment *= 0.9

# Modify your trading logic to incorporate whale sentiment

def should_trade(prediction, confidence, whale_sentiment):
  """
  Enhanced trading logic that considers both ML prediction and whale activity.
  """
  base_threshold = 0.6

  # If whales are bullish and model predicts up, lower threshold
  if prediction == 1 and whale_sentiment > 2:
      threshold = base_threshold - 0.1  # More aggressive buying# If whales are bearish, raise threshold (be more cautious)
  elif prediction == 1 and whale_sentiment < -2:
      threshold = base_threshold + 0.1  # Less aggressive buying
  else:
      threshold = base_threshold

  return confidence > threshold

これら2つのブロックを合わせることで、トレーディングボットに追加のシグナル層が加わる。1つ目のブロックはバックグラウンドスレッドで実行され、WebSocket経由で大口のETH送金を継続的に監視し、各送金を強気(取引所から出ていく)、弱気(取引所に入っていく)、または中立に分類する。2つ目のブロックは、そのクジラ活動に基づいて累積センチメントスコアを維持し、それに応じてトレーディングのしきい値を修正する。

機械学習モデルの予測がクジラの行動と一致する場合、例えばクジラが蓄積している中で価格上昇を予測する場合、組み合わされたシグナルはより強くなり、ボットはより積極的に取引する。シグナルが矛盾する場合、ボットはより慎重になる。

これが重要なのは、機関投資家や大口保有者がしばしば、小口トレーダーよりも優れた情報やより長い時間軸を持っているからだ。取引所への、あるいは取引所からの大きな移動は、利用可能な供給量に影響を与え、数分から数時間先の価格の動きに先行することがある。オンチェーン活動を機械学習モデルの予測と合わせて取り込むことで、価格データだけでは得られない、より完全な市場状況の全体像をボットに与えることになる。

ステップ7: リスク管理を実装する

これはおそらく、このチュートリアル全体の中で最も重要なステップだ。適切なリスク管理がなければ、どれほど優れた予測モデルでも損失を出す。60%の確率で正しいボットであっても、負け取引である残りの40%が制御されていなければ破綻しかねない。一つの悪いポジションが、多数の勝ち取引を帳消しにすることもある。リスク管理はいくつかの重要な機能を果たす。単一の取引でどれだけ損失を出せるかを制限し(損切り)、市場が反転する前に利益を確定し(利益確定)、破滅的なポートフォリオの下落を防ぎ(ドローダウン制限)、感情的な過剰取引を止める(一日の取引回数制限)。以下のコードは、これらすべてを、トレーディングボットと統合可能な再利用可能なRiskManagerクラスとして実装している。

RiskManagerクラスの構築

以下のクラスは、すべてのリスク管理ロジックを一箇所にカプセル化している。ポートフォリオの価値を追跡し、危険な状況を監視し、信頼度に基づいて適切なポジションサイズを計算し、後の分析のためにすべての取引を記録する。

pythonCopied
class RiskManager:
  """
  Comprehensive risk management for trading bots.
  
  Risk management is what separates profitable bots from bankrupt ones.
  Key principles:
  1. Never risk more than you can afford to lose
  2. Cut losses quickly, let winners run
  3. Position size matters more than win rate
  4. Track everything for post-mortem analysis
  """
  
  def __init__(self, 
               initial_capital=10000,
               max_position_size=1.0,
               max_drawdown=0.15,
               stop_loss=0.05,
               take_profit=0.10,
               max_daily_trades=20):
      """
      Initialize risk parameters.
      
      Args:
          initial_capital: Starting portfolio value in USD
          max_position_size: Maximum ETH per trade
          max_drawdown: Max portfolio decline before pausing (15% = 0.15)
          stop_loss: Per-trade loss limit (5% = 0.05)
          take_profit: Per-trade profit target (10% = 0.10)
          max_daily_trades: Prevent overtrading
      """
      self.initial_capital = initial_capital
      self.portfolio_value = initial_capital
      self.peak_value = initial_capital
      
      self.max_position_size = max_position_size
      self.max_drawdown = max_drawdown
      self.stop_loss = stop_loss
      self.take_profit = take_profit
      self.max_daily_trades = max_daily_trades
      
      self.trades = []
      self.daily_trades = 0
      self.last_trade_date = datetime.now().date()
      self.trading_paused = False
  
  def check_drawdown(self):
      """
      Pause trading if drawdown exceeds threshold.
      
      Drawdown is measured from peak portfolio value.
      This prevents catastrophic losses during bad market conditions.
      """
      current_drawdown = (self.peak_value - self.portfolio_value) / self.peak_value
      
      if current_drawdown >= self.max_drawdown:
          self.trading_paused = True
          print(f"\n{'='*60}")
          print(f"⚠️  MAX DRAWDOWN REACHED: {current_drawdown:.2%}")
          print(f"Portfolio: ${self.portfolio_value:,.2f} (down from ${self.peak_value:,.2f})")
          print(f"Trading PAUSED. Manual review required.")
          print(f"{'='*60}\n")
          return False
      
      return True
  
  def calculate_position_size(self, confidence, current_price):
      """
      Adjust position size based on prediction confidence and portfolio size.
      
      This implements a simplified Kelly Criterion approach:
      - Higher confidence = larger position
      - Never exceed max position size
      - Scale down if portfolio has shrunk
      
      Args:
          confidence: Model's prediction probability (0-1)
          current_price: Current ETH price for value calculation
          
      Returns:
          Position size in ETH
      """
      # Base size scaled by confidence# At 60% confidence: 0.6 * max_size# At 90% confidence: 0.9 * max_size
      base_size = self.max_position_size * confidence
      
      # Scale down if portfolio has lost value
      portfolio_scale = self.portfolio_value / self.initial_capital
      adjusted_size = base_size * portfolio_scale
      
      # Never exceed max position size
      final_size = min(adjusted_size, self.max_position_size)
      
      # Verify we can afford this trade
      trade_value = final_size * current_price
      if trade_value > self.portfolio_value * 0.95:  # Never use >95% of portfolio
          final_size = (self.portfolio_value * 0.95) / current_price
      
      return round(final_size, 4)  # Round to 4 decimals
  
  def check_stop_loss(self, entry_price, current_price, position_type):
      """
      Check if stop loss is triggered for current position.
      
      Stop losses protect against large individual trade losses.
      Even if your model is right 60% of the time, the 40% losers
      can wipe you out without stops.
      """
      if position_type == 'long':
          loss = (entry_price - current_price) / entry_price
          
          if loss >= self.stop_loss:
              print(f"\n🛑 STOP LOSS TRIGGERED")
              print(f"Entry: ${entry_price:,.2f} → Current: ${current_price:,.2f}")
              print(f"Loss: {loss:.2%}")
              return True
      
      return False
  
  def check_take_profit(self, entry_price, current_price, position_type):
      """
      Check if take profit target is reached.
      
      Taking profits locks in gains and prevents giving back winnings
      if the market reverses.
      """
      if position_type == 'long':
          profit = (current_price - entry_price) / entry_price
          
          if profit >= self.take_profit:
              print(f"\n🎯 TAKE PROFIT TARGET REACHED")
              print(f"Entry: ${entry_price:,.2f} → Current: ${current_price:,.2f}")
              print(f"Profit: {profit:.2%}")
              return True
      
      return False
  
  def check_daily_limit(self):
      """
      Prevent overtrading by limiting daily trades.
      
      Overtrading leads to:
      - Death by a thousand fees
      - Emotional decision-making
      - Curve-fitting to noise rather than signal
      """
      today = datetime.now().date()
      
      # Reset counter at start of new day
      if today != self.last_trade_date:
          self.daily_trades = 0
          self.last_trade_date = today
      
      if self.daily_trades >= self.max_daily_trades:
          print(f"⚠️  Daily trade limit reached ({self.max_daily_trades})")
          return False
      
      return True
  
  def can_trade(self):
      """
      Master check: Can we trade right now?
      """
      if self.trading_paused:
          return False
      if not self.check_drawdown():
          return False
      if not self.check_daily_limit():
          return False
      return True
  
  def record_trade(self, trade_type, price, amount, profit_loss=0):
      """
      Track all trades for analysis and performance monitoring.
      
      This data is gold for:
      - Understanding what's working/not working
      - Tax reporting
      - Strategy refinement
      - Performance attribution
      """
      self.trades.append({
          'timestamp': datetime.now(),
          'type': trade_type,
          'price': price,
          'amount': amount,
          'value': price * amount,
          'profit_loss': profit_loss,
          'portfolio_value': self.portfolio_value
      })
      
      # Update portfolio value
      self.portfolio_value += profit_loss
      self.peak_value = max(self.peak_value, self.portfolio_value)
      
      # Increment daily trade counter
      self.daily_trades += 1
  
  def get_performance_stats(self):
      """
      Calculate comprehensive performance metrics.
      """
      if not self.trades:
          return "No trades executed yet"
      
      df = pd.DataFrame(self.trades)
      
      # Separate buys and sells
      buys = df[df['type'] == 'BUY']
      sells = df[df['type'] == 'SELL']
      
      # Calculate returns
      total_return = ((self.portfolio_value - self.initial_capital) / self.initial_capital) * 100
      
      # Win rate (profitable trades / total trades)
      profitable_trades = len(sells[sells['profit_loss'] > 0])
      total_closed_trades = len(sells)
      win_rate = (profitable_trades / total_closed_trades * 100) if total_closed_trades > 0 else 0
      
      # Average profit per winning trade
      avg_win = sells[sells['profit_loss'] > 0]['profit_loss'].mean() if profitable_trades > 0 else 0
      
      # Average loss per losing trade
      losing_trades = sells[sells['profit_loss'] < 0]
      avg_loss = losing_trades['profit_loss'].mean() if len(losing_trades) > 0 else 0
      
      # Profit factor (total wins / total losses)
      total_wins = sells[sells['profit_loss'] > 0]['profit_loss'].sum()
      total_losses = abs(sells[sells['profit_loss'] < 0]['profit_loss'].sum())
      profit_factor = total_wins / total_losses if total_losses > 0 else float('inf')
      
      # Current drawdown
      current_drawdown = ((self.peak_value - self.portfolio_value) / self.peak_value) * 100
      
      stats = f"""
{'='*60}
PERFORMANCE STATISTICS
{'='*60}
Portfolio Value: ${self.portfolio_value:,.2f}
Initial Capital: ${self.initial_capital:,.2f}
Total Return: {total_return:+.2f}%
Peak Value: ${self.peak_value:,.2f}
Current Drawdown: {current_drawdown:.2f}%

Trading Activity:

- Total Trades: {len(self.trades)}
- Closed Trades: {total_closed_trades}
- Win Rate: {win_rate:.1f}%
- Profitable Trades: {profitable_trades}
- Losing Trades: {len(losing_trades)}

Trade Quality:

- Average Win: ${avg_win:,.2f}
- Average Loss: ${avg_loss:,.2f}
- Profit Factor: {profit_factor:.2f}x
- Total Wins: ${total_wins:,.2f}
- Total Losses: ${total_losses:,.2f}

Risk Status:

- Trading Paused: {self.trading_paused}
- Daily Trades: {self.daily_trades}/{self.max_daily_trades}
{'='*60}
"""
      return stats

# Initialize risk manager

risk_manager = RiskManager(
initial_capital=10000,
max_position_size=0.5, # 0.5 ETH max per trade
max_drawdown=0.15, # Stop at 15% portfolio loss
stop_loss=0.03, # 3% stop loss per trade
take_profit=0.06, # 6% take profit target
max_daily_trades=15 # Max 15 trades per day
)

RiskManagerクラスは、すべてのリスク制御を単一の再利用可能なコンポーネントに集約する。初期化時には、リスクパラメータを定義する。初期資本、1取引あたりの最大ポジションサイズ、取引の一時停止をトリガーするポートフォリオのドローダウン率、個々の取引に対する損切りと利益確定のしきい値、そして過剰取引を防ぐための一日の取引回数制限だ。

このクラスはいくつかの重要なメソッドを提供する。check\_drawdownは、ポートフォリオがピーク値から下落しすぎていないかを監視し、しきい値を超えた場合は自動的に取引を一時停止する。これにより、連敗が口座全体を枯渇させることを防ぐ。calculate\_position\_sizeは、簡略化されたケリー基準のアプローチを実装しており、モデルの信頼度に基づいてポジションサイズを決める(信頼度が高いほどポジションは大きくなる)一方、ポートフォリオがすでに損失を出している場合はサイズを縮小する。

check\_stop\_lossとcheck\_take\_profitは、オープンポジションを監視し、価格が定義したしきい値を超えて動いたときに決済をトリガーする。record\_tradeは後の分析のためにすべての取引を記録し、get\_performance\_statsは勝率、プロフィットファクター、現在のドローダウンといった包括的な指標を計算する。

リスク管理をボットに統合する

RiskManagerクラスを定義したら、次のステップはそれをメインのトレーディングループに組み込むことだ。以下のコードは、ステップ5のボットを修正し、各取引の前にリスク制限をチェックし、信頼度に基づく動的なポジションサイジングを使用し、損切りまたは利益確定のレベルに達したときに自動的にポジションを決済する。

pythonCopied
# Modified trading loop with risk management
def run_trading_bot_with_risk_management(model, historical_df, risk_manager, interval_seconds=300):
  """
  Enhanced trading bot with comprehensive risk management.
  """
  print("="*60)
  print("AI TRADING BOT WITH RISK MANAGEMENT")
  print("="*60)
  print(f"Initial Capital: ${risk_manager.initial_capital:,.2f}")
  print(f"Max Position: {risk_manager.max_position_size} ETH")
  print(f"Stop Loss: {risk_manager.stop_loss:.1%}")
  print(f"Max Drawdown: {risk_manager.max_drawdown:.1%}")
  print("="*60 + "\n")
  
  position = None
  entry_price = 0
  position_size = 0
  
  while True:
      try:
          # Check if we can trade
          if not risk_manager.can_trade():
              print("Trading paused due to risk limits")
              time.sleep(interval_seconds)
              continue
          
          # Fetch data and predict
          current_data = fetch_current_data()
          current_price = current_data['price']
          features = prepare_features_for_prediction(current_data, historical_df)
          prediction = model.predict(features)[0]
          prediction_proba = model.predict_proba(features)[0]
          confidence = prediction_proba[prediction]
          
          # Display status
          print(f"[{datetime.now().strftime('%H:%M:%S')}] "
                f"Price: ${current_price:,.2f} | "
                f"Prediction: {'UP' if prediction == 1 else 'DOWN'} ({confidence:.1%}) | "
                f"Portfolio: ${risk_manager.portfolio_value:,.2f} | "
                f"Position: {position or 'None'}")
          
          # Check stop loss and take profit for open positions
          if position == 'long':
              if risk_manager.check_stop_loss(entry_price, current_price, 'long'):
                  # STOP LOSS HIT
                  loss = (entry_price - current_price) * position_size
                  execute_trade('SELL (Stop Loss)', position_size, current_price)
                  risk_manager.record_trade('SELL', current_price, position_size, loss)
                  position = None
                  continue
                  
              elif risk_manager.check_take_profit(entry_price, current_price, 'long'):
                  # TAKE PROFIT HIT
                  profit = (current_price - entry_price) * position_size
                  execute_trade('SELL (Take Profit)', position_size, current_price)
                  risk_manager.record_trade('SELL', current_price, position_size, profit)
                  position = None
                  continue
          
          # Regular trading logic
          if prediction == 1 and confidence > 0.6:
              if position != 'long':
                  # Calculate position size based on confidence
                  position_size = risk_manager.calculate_position_size(confidence, current_price)
                  execute_trade('BUY', position_size, current_price)
                  risk_manager.record_trade('BUY', current_price, position_size)
                  position = 'long'
                  entry_price = current_price
          
          elif prediction == 0 and confidence > 0.6:
              if position == 'long':
                  # Close position
                  profit = (current_price - entry_price) * position_size
                  execute_trade('SELL', position_size, current_price)
                  risk_manager.record_trade('SELL', current_price, position_size, profit)
                  position = None
          
          # Update historical data
          historical_df = pd.concat([
              historical_df,
              pd.DataFrame([current_data])
          ], ignore_index=True).tail(168)
          
          time.sleep(interval_seconds)
          
      except KeyboardInterrupt:
          print("\n" + risk_manager.get_performance_stats())
          break
          
      except Exception as e:
          print(f"ERROR: {e}")
          time.sleep(60)

# Run bot with risk management

run_trading_bot_with_risk_management(model, df, risk_manager, interval_seconds=300)

この強化されたトレーディングループは、ステップ5の基本バージョンに比べて複数の防御層を追加している。各イテレーションの前に、ボットはrisk_manager.can_trade\(\)をチェックして、リスク制限が破られていないことを確認する。ポートフォリオが最大ドローダウンに達した場合や、一日の取引回数制限を使い切った場合、取引は自動的に一時停止する。

オープンポジションについては、ボットは損切りまたは利益確定のしきい値に達したかどうかを継続的に監視し、モデルが新しいシグナルを生成するのを待つことなく、トリガーされ次第すぐに決済する。ポジションサイズは固定量を使う代わりに、リスクマネージャーがモデルの信頼度と現在のポートフォリオ価値に基づいて計算する、動的なものになっている。

結果として、連敗中に資本を守り、自動的に利益を確定し、パフォーマンスの良し悪しに応じてリスクエクスポージャーをスケールするボットが出来上がる。ボットを停止すると、総リターン、勝率、平均勝ち/負けの大きさ、プロフィットファクターを含む包括的なパフォーマンス統計が表示され、戦略を評価し改良するために必要なデータが得られる。

ステップ8: 戦略をバックテストする

実際のお金をリスクにさらす前に、過去のデータで戦略をバックテストしよう。これにより、さまざまな市場状況でボットがどのようなパフォーマンスを発揮したかが分かる。

以下の関数は、現実的な手数料とスリッページの見積もり、損切りと利益確定の実行、詳細なパフォーマンス追跡を含め、過去のデータに対してトレーディング戦略全体をシミュレートする。主要な指標を出力し、時間経過に伴うポートフォリオ価値の推移を示すエクイティカーブの可視化を生成する。

pythonCopied
def backtest_strategy(model, df, risk_manager, initial_capital=10000):
  """
  Comprehensive backtest with risk management.
  
  Backtesting answers critical questions:
  - Would this strategy have been profitable?
  - What's the maximum drawdown I need to stomach?
  - How many trades does it generate?
  - Does it work in different market conditions?
  
  CRITICAL: Backtests are optimistic. Real trading includes:
  - Slippage (getting worse prices than expected)
  - Fees (exchange commissions, gas costs)
  - Execution delays (orders take time to fill)
  - Market impact (your trades affect prices)
  
  Assume real performance will be 20-30% worse than backtest.
  """
  print("\n" + "="*60)
  print("BACKTESTING STRATEGY")
  print("="*60)
  print(f"Data points: {len(df)}")
  print(f"Time period: {df['timestamp'].min()} to {df['timestamp'].max()}")
  print(f"Initial capital: ${initial_capital:,.2f}\n")
  
  # Reset risk manager for clean backtest
  capital = initial_capital
  position = None
  entry_price = 0
  position_size = 0
  trades = []
  portfolio_values = []
  
  # Track statistics
  wins = 0
  losses = 0
  total_fees = 0
  max_drawdown = 0
  peak_value = initial_capital
  
  # Simulation parameters
  FEE_RATE = 0.001  # 0.1% per trade (typical exchange fee)
  SLIPPAGE = 0.0005  # 0.05% slippage (market orders get worse prices)
  
  for i in range(len(df)):
      if i < 24:  # Need history for features
          continue
      
      current_price = df['price'].iloc[i]
      timestamp = df['timestamp'].iloc[i]
      
      # Get features and prediction
      features = df[feature_columns].iloc[i:i+1].values
      prediction = model.predict(features)[0]
      prediction_proba = model.predict_proba(features)[0]
      confidence = prediction_proba[prediction]
      
      # Check stop loss and take profit for open positions
      if position == 'long':
          current_pnl_pct = (current_price - entry_price) / entry_price
          
          # Stop loss hit
          if current_pnl_pct <= -risk_manager.stop_loss:
              sell_price = current_price * (1 - SLIPPAGE)  # Slippage on market sell
              pnl = (sell_price - entry_price) * position_size
              fees = sell_price * position_size * FEE_RATE
              net_pnl = pnl - fees
              
              capital += (position_size * sell_price) - fees
              total_fees += fees
              losses += 1
              
              trades.append({
                  'timestamp': timestamp,
                  'type': 'SELL (Stop)',
                  'price': sell_price,
                  'amount': position_size,
                  'pnl': net_pnl,
                  'capital': capital
              })
              
              position = None
              continue
          
          # Take profit hit
          elif current_pnl_pct >= risk_manager.take_profit:
              sell_price = current_price * (1 - SLIPPAGE)
              pnl = (sell_price - entry_price) * position_size
              fees = sell_price * position_size * FEE_RATE
              net_pnl = pnl - fees
              
              capital += (position_size * sell_price) - fees
              total_fees += fees
              wins += 1
              
              trades.append({
                  'timestamp': timestamp,
                  'type': 'SELL (Profit)',
                  'price': sell_price,
                  'amount': position_size,
                  'pnl': net_pnl,
                  'capital': capital
              })
              
              position = None
              continue
      
      # Regular trading logic
      if prediction == 1 and confidence > 0.6:
          if position != 'long' and capital > 0:
              # BUY
              buy_price = current_price * (1 + SLIPPAGE)  # Slippage on market buy
              position_size = min(
                  risk_manager.max_position_size,
                  (capital * 0.95) / buy_price  # Use max 95% of capital
              )
              cost = position_size * buy_price
              fees = cost * FEE_RATE
              
              capital -= (cost + fees)
              total_fees += fees
              entry_price = buy_price
              position = 'long'
              
              trades.append({
                  'timestamp': timestamp,
                  'type': 'BUY',
                  'price': buy_price,
                  'amount': position_size,
                  'pnl': -fees,  # Fees are a loss
                  'capital': capital
              })
      
      elif prediction == 0 and confidence > 0.6:
          if position == 'long':
              # SELL
              sell_price = current_price * (1 - SLIPPAGE)
              pnl = (sell_price - entry_price) * position_size
              fees = sell_price * position_size * FEE_RATE
              net_pnl = pnl - fees
              
              capital += (position_size * sell_price) - fees
              total_fees += fees
              
              if net_pnl > 0:
                  wins += 1
              else:
                  losses += 1
              
              trades.append({
                  'timestamp': timestamp,
                  'type': 'SELL',
                  'price': sell_price,
                  'amount': position_size,
                  'pnl': net_pnl,
                  'capital': capital
              })
              
              position = None
      
      # Track portfolio value (including open position)
      if position == 'long':
          current_value = capital + (position_size * current_price)
      else:
          current_value = capital
      
      portfolio_values.append({
          'timestamp': timestamp,
          'value': current_value
      })
      
      # Track drawdown
      peak_value = max(peak_value, current_value)
      current_drawdown = (peak_value - current_value) / peak_value
      max_drawdown = max(max_drawdown, current_drawdown)
  
  # Close any open position at end
  if position == 'long':
      final_price = df['price'].iloc[-1] * (1 - SLIPPAGE)
      pnl = (final_price - entry_price) * position_size
      fees = final_price * position_size * FEE_RATE
      capital += (position_size * final_price) - fees
      total_fees += fees
  
  # Calculate final statistics
  total_return = ((capital - initial_capital) / initial_capital) * 100
  total_trades = len([t for t in trades if t['type'] == 'BUY'])
  win_rate = (wins / (wins + losses) * 100) if (wins + losses) > 0 else 0
  
  # Print results
  print("\n" + "="*60)
  print("BACKTEST RESULTS")
  print("="*60)
  print(f"\nCapital:")
  print(f"  Initial: ${initial_capital:,.2f}")
  print(f"  Final: ${capital:,.2f}")
  print(f"  Total Return: {total_return:+.2f}%")
  print(f"  Max Drawdown: {max_drawdown:.2%}")
  
  print(f"\nTrading Activity:")
  print(f"  Total Trades: {total_trades}")
  print(f"  Winning Trades: {wins}")
  print(f"  Losing Trades: {losses}")
  print(f"  Win Rate: {win_rate:.1f}%")
  print(f"  Total Fees Paid: ${total_fees:,.2f}")
  
  if wins > 0 and losses > 0:
      avg_win = sum(t['pnl'] for t in trades if t['pnl'] > 0) / wins
      avg_loss = sum(t['pnl'] for t in trades if t['pnl'] < 0) / losses
      profit_factor = abs(sum(t['pnl'] for t in trades if t['pnl'] > 0) / 
                         sum(t['pnl'] for t in trades if t['pnl'] < 0))
      
      print(f"\nTrade Quality:")
      print(f"  Avg Win: ${avg_win:,.2f}")
      print(f"  Avg Loss: ${avg_loss:,.2f}")
      print(f"  Profit Factor: {profit_factor:.2f}x")
  
  print("="*60 + "\n")
  
  # Plot equity curve
  import matplotlib.pyplot as plt
  
  pv_df = pd.DataFrame(portfolio_values)
  plt.figure(figsize=(12, 6))
  plt.plot(pv_df['timestamp'], pv_df['value'])
  plt.axhline(y=initial_capital, color='r', linestyle='--', label='Initial Capital')
  plt.title('Portfolio Value Over Time (Backtest)')
  plt.xlabel('Date')
  plt.ylabel('Portfolio Value ($)')
  plt.legend()
  plt.grid(True, alpha=0.3)
  plt.tight_layout()
  plt.savefig('backtest_equity_curve.png')
  print("📊 Equity curve saved to 'backtest_equity_curve.png'\n")
  
  return trades, capital

# Run comprehensive backtest

trades, final_capital = backtest_strategy(model, df, risk_manager, initial_capital=10000)

この関数は、収集した過去データに対してトレーディング戦略全体を再現し、ボットが下したであろうすべての買い/売りの判断をシミュレートする。データセット内の各時間帯を繰り返し処理し、学習済みモデルを使って予測を生成し、オープンポジションに対する損切りまたは利益確定のしきい値が達成されたかをチェックし、信頼度がしきい値を超えたときに取引を実行し、全体を通じてポートフォリオ価値を追跡する。

このシミュレーションには現実的な摩擦が含まれている。すべての取引に対する0.1%の手数料(主要取引所で典型的な水準)と、0.05%のスリッページ(成行注文を使用する際の予想約定価格と実際の約定価格の差)だ。これらの小さなパーセンテージは多数の取引を通じて急速に積み重なるため、この関数は支払った総手数料を追跡する。それはしばしば、それ以外は収益性のある戦略を静かに殺してしまう要因となる。

最後に、この関数は包括的な統計を出力し、テスト期間にわたってポートフォリオ価値がどのように推移したかを示すエクイティカーブの可視化を保存する。

バックテスト結果の解釈

バックテストは、戦略が実行可能かどうかを評価するのに役立ういくつかの指標を出力する。以下は、それぞれの意味と注目すべきポイントだ。

総リターン: 戦略の収益性。0%を超えていれば収益性があると言えるが、その結果はETHの買い持ち(buy-and-hold)と比較すべきだ。
最大ドローダウン: ピークからボトムへの最悪の下落幅。15%の損失に耐えられないなら、この戦略は運用できない。
勝率: 収益を上げた取引の割合。勝ちが負けより大きければ、勝率50%未満でも収益性を確保できる。
プロフィットファクター: 総利益÷総損失。1.5倍を超えていれば概ね良好で、1.0倍を下回れば損失を出していることを意味する。
総手数料: しばしば静かな殺し屋となる。高頻度戦略は手数料を差し引く前は収益性があっても、差し引いた後は損失になり得る。

現実チェック:バックテストが100%を超えるリターンを示している場合、おそらくバグがある(先読みバイアス、過学習など)。実際のトレーディングは、スリッページ、手数料、実行の遅延、予期しない出来事のために、バックテストよりも悪い結果になる。

おめでとう。これを実行できたなら、機能するAIトレーディングボットを構築できたことになる。

構築したもの

あなたは今、以下を行う稼働中のトレーディングボットを手にしている。

Alchemyのインフラを使ってリアルタイムのブロックチェーンデータを取得する
複数のデータソース(価格、オンチェーン指標、クジラの活動)を組み合わせる
機械学習を使って短期的な価格変動を予測する
損切り、ポジションサイジング、ドローダウン制限でリスクを管理する
実運用前に戦略を検証するためにバックテストできる

これは原始的だが完結したシステムだ。データ収集から実行に至る、あらゆるAIトレーディングボットの核となるワークフローを示している。

重要な注意事項と警告

AIトレーディングボットを始動させる前に、資本を守るため以下に留意してほしい。

ペーパートレーディングから始める: シミュレーションによる注文で、最低でも1ヶ月はボットを運用する。実際の資本をリスクにさらすことなく、想定パフォーマンスを追跡してバグを特定し、戦略を洗練させる。
小さなポジションを使う: 実運用に移行する際は、意図する資本の10〜20%から始める。十分にテストされたボットでも、本番環境では予期しない挙動をすることがある。
常に監視する: 特に最初の数週間は、ボットを注意深く見守る。異常な挙動(取引回数が多すぎる、大きな損失、エラー)に対するアラートを設定する。
自動化されたキルスイッチを持つ: 重要なしきい値(一日の最大損失、異常なボラティリティの急上昇、システムエラー)が破られたときに自動的に取引を停止する仕組みを組み込む。市場は、手動で介入できるよりも速くあなたに不利に動くことがあるため、手動での停止に頼るだけでは不十分だ。ボットは、あなたが眠っていても、デスクを離れていても、自らを守れるようにしておくべきだ。
手数料に注意する: DEXでの取引にはガス代がかかり、CEXは取引手数料を課す。高頻度戦略は手数料を差し引く前は収益性があっても、差し引いた後は損失になり得る。常に現実的な手数料の見積もりを含めること。
規制遵守: あなたの管轄区域の法律を理解すること。地域によっては、アルゴリズムトレーディングにライセンスを要求する場合がある。他人の資金を運用する場合、証券規制が適用される可能性が高い。
税務上の影響: 多くの管轄区域では、すべての取引が課税対象イベントとなる。詳細な記録を残すこと。1日100回のボットの取引が、税務申告上の悪夢を生み出しかねない。
セキュリティは最優先事項だ:

APIキーを絶対にハードコーディングしない
環境変数を使用する
すべてのアカウントで2FAを有効にする
使用限度額を設定できるスマートコントラクトウォレットを検討する
可能な限り秘密鍵をオフラインに保管する
コードの脆弱性を定期的に監査する

まとめ

このガイドで取り上げたツールと技術、すなわちリアルタイムのブロックチェーンデータの取得、予測特徴量の設計、機械学習モデルの学習、リスク管理の実装、戦略のバックテストは、AIトレーディングシステムを構築するための基盤を提供する。しかしこれは出発点であって、完成品ではない。現実のトレーディングには継続的な反復が求められる。市場状況の変化に応じてモデルを洗練させ、エッジケースに対してストレステストを行い、24時間365日確実に稼働できる堅牢なインフラを構築することだ。

構築を始める準備はできただろうか。Alchemyに登録してAPIキーを取得し、上記のコード例を使って実験を始めよう。ブロックチェーン開発を体系的に学びたい場合は、Alchemy Universityをチェックしてほしい。

よくある質問

自分でAIトレーディングボットを構築できますか?

はい、Python、scikit-learnやPyTorchといった機械学習ライブラリ、そしてリアルタイムデータアクセスのためのブロックチェーンインフラを使って、独自のAIトレーディングボットを構築できます。このガイドでは、始めるための完全なチュートリアルを提供しています。

AIトレーディングボットの構築にはどのプログラミング言語が最適ですか?

Pandas、NumPy、scikit-learn、PyTorchやTensorFlowなど、機械学習とデータ分析のための豊富なライブラリがあるため、Pythonが最も推奨される言語です。

AIトレーディングボットに不可欠な構成要素は何ですか?

完全なAIトレーディングボットには、信頼できるブロックチェーンデータインフラ、予測のための機械学習モデル、価格とオンチェーン指標を組み合わせた特徴量エンジニアリング、損切りとポジションサイジングによるリスク管理、そして実運用前に戦略を検証するためのバックテスト機能が必要です。

AIトレーディングボットにとってブロックチェーンインフラはどれほど重要ですか?

インフラは極めて重要です。ほとんどのボットが失敗するのは、予測が悪いからではなく、古いデータ、APIのダウンタイム、高いレイテンシーが原因です。複数のブロックチェーンにわたって、競争の激しいトレーディングが求めるリアルタイムかつ低レイテンシーなデータアクセスと信頼性が必要です。

AIトレーディングボットの現実的な成功率はどのくらいですか?

ボット利用者のうち、継続的な収益を上げているのはわずか10〜30%です。成功は、高度なアルゴリズムだけでなく、信頼できるインフラ、堅実なリスク管理、現実的な期待にかかっています。ボラティリティの高い暗号資産市場では、55〜60%の予測精度でもかなり良い方です。

AIトレーディングボットでリスクをどう管理すればよいですか?

損切りと利益確定のレベルを設定し、信頼度に基づくポジションサイジング(通常は1取引あたりポートフォリオの最大2%)を実装し、取引を一時停止する最大ドローダウンのしきい値を設け、過剰取引を防ぐ一日の取引回数制限を設定し、予期しない損失に対するサーキットブレーカーを組み込んでください。

AIトレーディングボットはどのようなデータソースを使うべきですか?

最も包括的な市場の全体像を得るには、リアルタイムの価格データ、ガス使用量やトランザクション数といったオンチェーン指標、移動平均やボラティリティといったテクニカル指標、そしてクジラのウォレットの動きから得られるセンチメントシグナルを組み合わせてください。

実際のお金を使う前にトレーディングボットをバックテストすべきですか?

はい、まず必ず過去のデータでバックテストし、その後シミュレーション注文で最低1ヶ月はペーパートレーディングを行ってください。実際のパフォーマンスは、手数料、スリッページ、実行の遅延によりバックテストより20〜30%悪くなるため、実運用は意図する資本のわずか10〜20%から始めてください。

AIトレーディングボットの主な種類は何ですか?

一般的な種類には、価格パターンと指標を使うテクニカル分析ボット、ソーシャルメディアとオンチェーンシグナルを解析するセンチメント分析ボット、取引所間の価格差を利用する裁定取引ボット、複数の戦略を協調させる自律型AIエージェントファンドなどがあります。

トレーディングボットにどのような安全機構を組み込むべきですか?

すべての判断を追跡できる可観測なシステムを構築し、最大損失のしきい値とクールダウン期間を備えたサーキットブレーカーを追加し、暴走的な実行を防ぐレートリミッターを実装し、一定のしきい値を超えるリスクの高い判断については人間をループに残してください。

Alchemy Newsletter

リリース情報をいち早く受け取る

ニュースレターに登録する

Alchemyの最新のプロダクト情報とリソースをお届けします

80,000人以上が購読中

メールアドレスを入力すると、当社のマーケティング情報およびプロダクト最新情報の受信に同意したことになります。Alchemyが受け取った情報をプライバシー通知に従って取り扱うことに同意するものとします。購読はいつでも解除できます。

お知らせ

Glamsterdam: Ethereum開発者のための完全ガイド

GlamsterdamはSepoliaで稼働中です。開発者にとって何を意味するか、Alchemyがネットワークアップグレードにどう備えたかをまとめた完全ガイドです。

技術

東京で構築するなら、どのRPCプロバイダーを選ぶべきか

東京を提供するRPCプロバイダー5社を、5つのチェーンでベンチマークした。比較結果と、レイテンシに敏感なチームにとって差が出るポイントを示す。

製品

業界最速クラスの速度と信頼性を実現するSolana RPC読み取りの設計

AlchemyはSolanaの読み取りレイテンシが最も低く、9.21ミリ秒で、次点のプロバイダーより約26%高速です。