如何打造 AI 交易機器人：完整開發者指南（2026）

最後更新：2025年12月10日10 分鐘閱讀

過去幾年，我們見證了 AI 交易工具的大量湧現。

約有58%的散戶投資人現在會使用某種形式的 AI 來協助建構投資組合，創投資金已向 Web3 AI 交易技術投入2.13 億美元，部分自主AI agents在單月內的交易量已超過4 億美元，在市場分析、執行速度以及與其他 agents 的協調上也愈趨成熟。

但在這股熱潮底下，其實是更為複雜的現實：成功與否不只取決於演算法的精密程度，還取決於可靠的基礎設施、健全的風險管理，以及對 AI 能做什麼、不能做什麼的務實認知。

這份完整指南將帶你了解打造自己的 AI 交易 bot 所需的一切。我們會探討目前 AI 交易的現況，包括新興的自主 agent 經濟體，檢視現有的工具和平台，拆解各種 bot 策略，並提供教學，帶你利用 Alchemy 的區塊鏈基礎設施著手打造自己的交易 bot。

什麼是 AI 交易 bot？

AI 交易 bot 是使用機器學習演算法來分析市場數據、找出交易機會，並自動執行買賣單的軟體。與遵循靜態「if-then」邏輯的傳統規則型 bot 不同，AI 驅動的 bot 能適應變動中的市場條件，從歷史模式中學習，並根據龐大的數據集做出機率性決策。它們全年無休運作，排除交易決策中的情緒偏誤，反應速度也比任何人都快。

2025 年的 AI 交易 bot 生態呈現出強烈對比。AI 輔助交易工具的採用率大幅成長，比以往有更多散戶投資人把自動化納入自己的策略中。然而，實際的成功率卻透露出更嚴峻的現實：只有 10-30% 的 bot 使用者能達到穩定獲利。

這種採用率與成功率之間的落差並不令人意外，因為加密貨幣市場向來波動劇烈，預測模型需要持續調整，而許多新手也低估了風險管理的重要性。能成功的 bot 未必擁有最精密的 AI，而是建立在扎實基礎之上：可靠的數據、清楚的策略，以及有紀律的執行。

這帶來的教訓是：AI 交易 bot 是強大的工具，但並非萬靈丹，成功與否取決於同時理解其能力與限制。

區塊鏈基礎設施的角色

AI 交易 bot 是強大的工具，但成功與否取決於同時理解其能力與限制。然而，經常被忽略的是，模型的精密程度只是方程式的一部分。大多數 bot 失敗，並非因為 AI 做出了錯誤的預測，而是因為底層基礎設施跟不上——數據過時、API 在市場劇烈波動時當機、速度更快的競爭對手搶先成交。這些都不是特例，而是演算法交易的日常現實。

有三項關鍵基礎設施要求，區分出能穩定獲利的 bot 與爆倉的 bot：

數據準確性：AI 模型的好壞取決於訓練及回應所依據的數據。過時的 token 價格、失準的流動性池狀態，或錯過的巨鯨動向，都會直接轉化為糟糕的交易決策。當你的 bot 根據錯誤數據行動時，就會虧錢——沒有任何演算法上的精巧設計能彌補根本上有缺陷的輸入。
延遲：在演算法交易中，速度決定誰能獲利。當你的 bot 找到一個套利機會時，很可能有數十個其他人也看到同樣的訊號。擁有最快數據管線的 bot 會率先執行並吃下價差；其他人則只能拿到較差的成交價，甚至完全錯過交易。低延遲的基礎設施是競爭上的必要條件，而非效能優化的選項。
可靠性：市場劇烈波動期間的基礎設施停機可能造成災難性後果。短短幾分鐘離線，可能意味著在崩盤時錯過停損、無法防守的部位遭到清算，或是在 bot 重新連線前機會已經消失。對於管理實際資金的交易系統來說，正常運行時間不只是服務等級指標，而是生存的必要條件。

這正是像 Alchemy 這樣的數據提供商變得不可或缺之處。Alchemy 提供 AI 交易 bot 有效運作所需的可靠、即時數據存取——包括 token 價格、交易量、巨鯨動向、流動性池狀態等，並涵蓋多條鏈。Alchemy 的增強型 API 與基礎設施能大規模處理這些需求，支援超過 100 條區塊鏈，並具備競爭型交易所要求的低延遲效能。

打造安全且可觀察的交易 bot

即使擁有扎實的基礎設施，AI 交易 bot 仍可能出現無法預測的行為。市場的走勢常常超出模型的預期，邊界情況會觸發意外行為，而複合性的錯誤也可能把小失誤放大成災難性的損失。在 bot 架構中建立安全機制並非可有可無，而是你能撐得夠久、持續迭代改進的關鍵。

建立可觀察的系統：每一筆交易都應該可以追蹤。是什麼數據促成這個決策？bot 為什麼選擇那個特定動作？當出狀況時（一定會發生），你需要判斷問題出在數據不良、邏輯有缺陷，還是模型沒有預料到的邊界情況。
加入斷路機制：設定最大虧損門檻，自動暫停交易。實作速率限制器以防止失控執行。在重大虧損後建立冷卻期。這些機制能防止故障的 bot 在你察覺問題前把帳戶掏空。
讓人類保持在決策迴路中：完全自主聽起來很厲害，但最具韌性的系統仍會讓人類參與高風險決策。可以考慮要求超過特定門檻的交易需要人工核准，或建立在 bot 行為偏離預期模式時通知你的警示系統。

能長期成功的 bot 不只是擁有最好模型的那些，而是建立了足夠的可觀察性與安全防護、能及早發現問題並妥善復原的那些。

工具與技術：你的 AI 交易 bot 技術棧

在涵蓋了 AI 交易 bot 的基礎設施層——數據準確性、延遲、可靠性——之後，下一步就是組裝實際的工具組。以下是打造可上線的 AI 交易 bot 技術棧所需的內容：

核心程式語言技術棧

Python 因其龐大的機器學習生態系與簡潔性，至今仍是 AI 交易 bot 的主流語言。你的核心 Python 函式庫應包括：

Pandas：用於數據操作與時間序列分析
NumPy：用於價格與交易量數據的數值運算
scikit-learn：用於傳統機器學習模型（迴歸、分類、分群）
PyTorch 或 TensorFlow：用於深度學習模型，例如用於序列預測的 LSTM 或 transformer

這些函式庫合起來能涵蓋完整的範圍：從一個下午就能做出來的簡單移動平均交叉 bot，到融合技術分析、情緒評分與鏈上指標的精密多模型 bot。這正是 Python 生態系的優點：你可以從簡單開始，需要時再逐步增加複雜度，如果線性迴歸就能達成目標，第一天就不需要為深度學習搭建架構。

補充說明：如果你在打造微秒級速度攸關的高頻交易 bot，C++ 能提供原生速度與底層控制。Rust 在加密貨幣領域也逐漸受到青睞，因其效能與記憶體安全性，特別是在打造鏈上元件或客製化 DEX 整合、安全性至關重要時。Go 則是另一個適合同時監控多個市場的並行系統的可靠選擇。

話雖如此，對於大多數加密貨幣交易策略，即使是複雜的策略，Python 的效能已經綽綽有餘，你在開發速度上的收穫，遠遠超過對大多數散戶交易者而言邊際效益不大的速度提升。先從 Python 開始，之後如果效能分析顯示你確實需要，再考慮把關鍵元件遷移到更快的語言。

區塊鏈基礎設施：Alchemy 層

這正是 Alchemy 變得不可或缺之處。Alchemy 提供支撐可靠、大規模 AI 交易 bot 的基礎設施層。關鍵工具包括：

即時區塊鏈數據：查詢橫跨超過 50 條鏈的 token 價格、餘額、交易歷史與 NFT 元數據。透過我們的沙盒親自試用看看。
WebSocket 連線：訂閱即時事件，例如待處理交易、新區塊，或智慧合約狀態變更。
增強型 API：存取 gas 價格預測、附帶元數據的 token 餘額，以及歷史交易數據。

對於加密貨幣交易 bot，你還會需要：

Web3.py：進行 Ethereum 互動的介面——發送交易、呼叫智慧合約、編碼數據。如果你在DEXs上交易或直接與 DeFi 協議互動，這是必備工具。
CCXT：一個統一的 API 包裝器，讓你能用一致的語法在超過 100 個中心化交易所（Binance、Coinbase、Kraken 等）上交易。不必學習每個交易所各自的 API 怪癖，CCXT 會把一切標準化。
Alchemy 的 MCP（Model Context Protocol）server：這正是區塊鏈與 AI 交會之處：我們的 MCP server 讓 AI agents 能以自然語言查詢鏈上數據。你的 AI 不需要撰寫複雜的查詢語句，只要問「ETH 目前的價格是多少？」就能取得結構化的數據回覆。

安裝 Alchemy Skills、連接 MCP server，並使用 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 中的 agent wallets。

AI 增強工具

我們已經看到交易 bot 的下一個演進階段：不只是分析數據，而是運用 AI 做出愈趨複雜的決策。過去一年出現了兩種主要方法，不同的開發者都在用它們來精進自己的流程：

大型語言模型 API（Claude、GPT-4 等）開啟了超越傳統機器學習的能力：

情緒分析：處理數千則推文、Reddit 貼文或新聞文章，藉此衡量市場情緒
策略生成：用白話英文描述一個交易想法，讓 AI 協助把它形式化成程式碼
異常偵測：利用 LLM 找出市場數據中可能預示機會的異常模式

到了 2025 年，我們見識到了新的 Agent 框架，透過建構自主系統把這一切推得更遠：

ElizaOS（來自 ai16z）：支撐去中心化 AI 基金背後的框架，讓你能打造不只是建議交易、而是實際執行交易的 agents。它們能管理投資組合，甚至與其他 agents 協調合作。
Virtuals GAME SDK：打造能參與更龐大 agent 經濟體的 agents——賺取手續費、為其他 agents 提供服務，或協作執行複雜策略。你可以把它想成打造一個能「與他人良好協作」的 bot。
LangChain：用來編排複雜 AI 工作流程的瑞士刀，讓你能把多次 AI 呼叫組合在一起、整合來自不同來源的數據，並建構如果要硬寫程式碼會很繁瑣的決策樹。當你的策略需要多步驟推理時特別有用。

這些工具能讓你從「遵循規則的 bot」轉變為「能適應並學習的 agent」。開發的學習曲線更陡峭，但潛在的回報也顯著更高。

必要的輔助工具

市場數據 API：像 CoinGecko、CoinMarketCap，或交易所專屬的 API（Binance、Coinbase、Kraken）等服務，提供價格數據、交易量、市值排名，以及歷史 OHLCV（開盤、最高、最低、收盤、成交量）數據。Alchemy 處理鏈上數據，而這些 API 則補足了中心化交易所價格以及模型做出明智決策所需的更廣泛市場脈絡。
回測框架：在冒險投入真實資金之前，你需要根據歷史數據驗證你的策略。有像 Backtrader 和 Zipline 這樣的框架，能讓你模擬 bot 在過去牛市、崩盤與盤整期間的表現。良好的回測基礎設施能協助你找出策略弱點、優化參數，並在正式上線前建立信心。
版本控制：對認真的交易系統來說，Git 並非可有可無。隨著你的策略演進——調整參數、加入新訊號、修復錯誤——你需要追蹤每一次變更、回退失敗的實驗，並清楚記錄在特定交易執行時運行的是哪份程式碼。這對日後除錯與稽核 bot 的行為至關重要。

AI 交易 bot 的種類

AI 交易 bot 涵蓋了廣泛的策略與複雜程度，從單純的自動化到實驗性的多 agent 系統都有。以下是你會遇到的最常見類別：

技術分析與情緒分析 Bot：技術型 bot 分析價格圖表、移動平均線與各種指標來找出交易機會——AI 增強讓它們能動態調整訊號權重，並辨識出靜態規則會錯過的複雜模式。情緒型 bot 則解析 Twitter/X 貼文、Reddit 討論、新聞文章，以及鏈上訊號（巨鯨動向、交易所流入、gas 費用飆升），根據市場情緒預測短期價格走勢。在炒作循環驅動波動的山寨幣與迷因幣領域特別受歡迎。
套利與造市 Bot：套利型 bot 利用不同交易所或流動性池之間的價差獲利——在某平台以 2,000 美元買入 ETH，在另一平台以 2,010 美元賣出。AI 能在價差出現之前預測，並優化跨 DEX 的多跳路徑。造市型 bot 同時提供買賣報價來提供流動性，靠價差獲利。在 DeFi 中，這意味著存入 Uniswap 或 Curve 資金池，並利用 AI 管理無常損失，根據波動性調整價差。
趨勢跟蹤與投資組合管理 Bot：動量型 bot 找出正在增強力道的資產，並持有直到訊號反轉，像 LSTM（長短期記憶網路）這類模型能協助預測趨勢是會延續還是崩解。投資組合型 bot 根據 AI 預測自動重新平衡持倉，把配置比重轉移到預期報酬更強的資產，同時管理整體風險暴露。
自主 AI Agent 基金：在實驗性的最前沿，像 ElizaOS 這類專案管理由 DAO 治理的基金，其中 AI agents 協調交易策略、共享訊號，並優化整體績效，類似 aixbt。這些多 agent 系統模糊了交易 bot 與自主金融實體之間的界線，雖然仍處於早期階段，但已能一窺 AI 驅動金融的未來走向。

打造你的第一個 AI 交易 bot：逐步教學

在本節中，我們將嘗試為 Ethereum 打造一個結合情緒分析的趨勢預測 bot，融合鏈上數據與機器學習來做出交易決策。

我們要打造的內容：

一個透過 Alchemy 擷取即時 ETH 價格與區塊鏈數據的 bot
結合價格指標與鏈上情緒訊號的特徵工程
用來預測短期價格走勢的機器學習模型
附帶信心門檻與風險管理的交易邏輯
在正式上線前用來驗證策略的回測能力

這是一個原始的範例：把它想成你可以在其上繼續建構的基礎，而非可直接上線的生產系統。目標是理解核心概念與工作流程，之後再根據你自己的策略進行擴充。

前置準備

在開始寫程式之前，請確認你已經具備：

機器上已安裝 Python 3.8 或更高版本
一個附帶 API key 的 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：安全地管理環境變數

第一步：設定 Alchemy 並擷取歷史區塊鏈數據

首先，我們要建立與區塊鏈的連線，並蒐集之後用來訓練模型的歷史數據。

在你的專案目錄中建立一個 .env 檔案，以安全地儲存你的 API key：

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())

這段程式碼使用 web3.py 搭配你的 Alchemy API key 連接到Ethereum mainnet。接著根據你想要的時間範圍計算要擷取多少區塊，Ethereum 大約每 12 秒產生一個區塊，代表每小時約有 300 個區塊。為了讓數據集規模可控並避免過多的 API 呼叫，我們每小時只取樣一個區塊，而非擷取每一個區塊。對於每個取樣的區塊，我們擷取 gas 使用量與交易數量作為網路活躍度的代理指標，然後把所有內容存入 pandas DataFrame，方便後續步驟操作。

這些鏈上指標經常能在市場活動反映在價格數據之前，提供早期訊號。當 gas 使用量飆升時，通常代表交易活動增加、DeFi 互動增多，或發生了重大鏈上事件，這些都可能在價格波動之前發生。同樣地，交易數量上升可能顯示市場對 ETH 的興趣升溫，或 DeFi 活動升溫。透過把這些原生於區塊鏈的訊號與傳統價格數據一併納入模型，我們讓 bot 取得了純粹以價格為基礎的策略會錯過的資訊。

第二步：使用 Alchemy 擷取 ETH 價格數據

單靠區塊鏈數據並不足夠：我們需要實際的價格資訊來訓練模型。讓我們建立一個名為 fetch\_eth\_price.py 的檔案，並整合 Alchemy 的 Prices API，它能以低延遲提供橫跨多條鏈的即時價格。

為什麼選擇 Alchemy 的 Prices API？

統一的提供者：相同的 API key、相同的速率限制、一致的計費方式
交易所聚合：價格反映橫跨多個交易所的實際市場狀況
低延遲：針對交易應用優化（典型回應時間低於 100ms）
多鏈就緒：相同的 API 結構適用於任何受支援鏈上的任何 token

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 價格。我們定義起訖時間、請求每小時的價格點，並把回應解析成含時間戳記與價格數值的 pandas DataFrame。

最後的關鍵步驟，是把這份價格數據與第一步的區塊鏈數據合併。我們使用 merge\_asof 而非標準的合併方式，因為區塊數據與價格數據的時間戳記不會完全對齊，merge\_asof 會為每一列找出最接近的時間戳記，確保不會因為微小的時間差而遺失數據。完成這個步驟後，我們就得到一個統一的 DataFrame，同時包含鏈上指標（gas 使用量、交易數量）與價格數據，準備好進行下一步的特徵工程。

第三步：為機器學習設計特徵

現在我們已經有了原始數據，接下來需要把它轉換成模型能從中學習模式的特徵。這正是領域知識與數據科學交會之處。

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 小時）：這些能平滑掉短期的價格雜訊，顯露出底層趨勢。當短期移動平均線向上穿越長期移動平均線時，傳統上被解讀為看漲訊號；向下穿越則暗示看跌動能。模型可以學習這些交叉模式。
波動性：以近期價格變化的標準差來衡量。高波動性代表市場不穩定、價格波動較大——風險較高，但潛在報酬也可能較高。有些策略會避開波動劇烈的時期，有些則專門鎖定這些時期。
Gas 與交易趨勢：可能領先於價格走勢的鏈上指標。如果 gas 使用量突然飆升，可能代表正在發生一些重要事件（大型 NFT 發行、DeFi 漏洞攻擊、巨鯨活動），這些事件可能在反映到交易數據之前，就已經影響 ETH 價格。
動量：一個簡單但有效的指標，用來衡量過去 6 小時內價格是否呈上升趨勢。正動量代表可能延續上升走勢；負動量則暗示可能下跌。動量交易者利用這些訊號來順勢操作。
目標變數：這是我們要訓練模型去預測的內容——下一個小時的價格是否會高於現在？數值為 1 代表是（潛在買進訊號），0 代表否（潛在賣出或持有訊號）。這種二元分類方式簡化了預測問題。

第四步：訓練機器學習模型

現在我們要訓練一個模型，從我們設計的特徵中學習模式。我們使用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 來保留時間先後順序。這對時間序列數據來說至關重要，因為我們不能讓未來的資訊洩漏到過去的預測中。模型從訓練數據中學習模式，然後對它未曾見過的測試集做出預測。我們也會執行交叉驗證，也就是在多組不同的數據切分上進行訓練與測試，藉此檢查結果是否穩定，還是純粹運氣好。

解讀結果

機器學習模型會輸出許多指標，正確解讀這些指標對於判斷你的 bot 是否具備可行策略至關重要。以下說明每項輸出的意義，以及該注意的重點：

準確率：正確預測的百分比。在交易中，55-60% 的準確率其實已經相當不錯——市場本質上充滿雜訊、難以預測。不要期待 90% 以上的準確率。如果你看到這麼高的數字，你的模型很可能過度擬合，或存在數據洩漏的問題。
分類報告：這會拆解兩種預測類別（上漲與下跌）的表現。Precision（精確率）告訴你，模型預測「上漲」時的正確比例——這很重要，因為誤判會導致虧損的交易。Recall（召回率）告訴你，模型抓到了多少比例的實際「上漲」走勢——這對於不錯過機會很重要。F1-score 是兩者的調和平均，能讓你看到平衡後的整體表現。
交叉驗證分數：在數據的不同子集上訓練，能揭露你的模型表現是否穩定一致，還是會因為使用的特定數據點而劇烈波動。各折之間分數一致，代表策略穩健；變異度高則暗示模型可能學到的是雜訊，而非真正的模式。
特徵重要性：Random Forest 會告訴你哪些特徵對預測貢獻最大。如果 momentum 的重要性最高，代表你的策略主要是由動量驅動。如果 volatility 佔主導地位，代表你主要是在根據市場不穩定性交易。這項洞察能幫助你理解模型實際在做什麼，以及是否符合你的交易論點。

什麼算是「好」的結果？

對於交易模型，你要找的是高於 55% 的準確率（優於隨機猜測的 50%）、訓練集與測試集上表現相近（代表模型沒有過度擬合），以及一致的交叉驗證分數（暗示策略在不同市場條件下都具有穩健性）。如果你的測試準確率明顯低於訓練準確率，代表模型是在記憶訓練數據，而非學習可泛化的模式。

第五步：實作即時交易邏輯

現在讓我們打造 bot 的主迴圈：也就是擷取當前數據、做出預測並執行交易的邏輯。

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)

這一步把所有內容整合成一個可運作的交易 bot。程式碼定義了依序運作的四個主要函式：fetch_current_data 從 Alchemy 拉取最新的區塊鏈指標與 ETH 價格，prepare_features_for_prediction 把這些原始數據轉換成與訓練模型時相同的特徵，execute_trade 處理實際的買賣執行（此處為模擬，但已為正式環境整合做好準備），而 run_trading_bot 則在一個持續運作的迴圈中協調整個流程，每 5 分鐘檢查一次市場狀況。

主要的交易迴圈遵循一個直截了當的循環：擷取當前數據、產生特徵、做出預測、根據信心門檻決定是否交易、更新歷史數據，然後等待後再重複。以下是內建在這個迴圈中的幾項關鍵設計決策：

信心門檻：只有當模型對其預測的信心超過 60% 時，bot 才會執行交易。這能篩掉那些較可能只是雜訊、而非真正機會的弱訊號。你可以根據自己的風險承受度調整這個門檻——門檻越高，交易次數越少，但篩選也更嚴格。
持倉追蹤：bot 會維持對自身當前狀態（多倉、空倉或無持倉）的認知。這能防止不合邏輯的行為，例如在已經持有 ETH 時反覆買進，或在沒有東西可賣時賣出。
滾動窗口：歷史數據上限為過去 168 小時（一週）。這能避免長時間運行過程中出現記憶體問題，並確保 bot 的特徵計算能保持貼近當前市場狀況，而不會被過舊的數據拖累。
錯誤處理：try/except 區塊能確保暫時性的故障（API 逾時、網路問題、速率限制）不會導致整個 bot 當機。錯誤會被記錄下來，bot 會在短暫暫停後重試。
優雅關閉：按下 Ctrl+C 能乾淨地停止 bot，並印出摘要統計數據，讓你能在不遺失狀態資訊的情況下檢視績效。

為什麼選擇 5 分鐘的間隔？這個預設值在反應速度、API 速率限制與運算負擔之間取得平衡。你可以根據自己的策略進行調整：高頻策略可能每幾秒就要檢查一次，而波段交易策略可能只需要每小時更新一次。較短的間隔能捕捉更多機會，但也會消耗更多 API 呼叫，並需要更快的執行基礎設施。

第六步：透過 Alchemy 加入鏈上情緒分析

價格與技術指標只能說明部分的故事。在加密貨幣中，鏈上活動——尤其是「巨鯨」的大額轉帳——可能在反映到價格數據之前，就預示著即將到來的價格走勢。巨鯨把 ETH 轉入交易所，通常代表賣壓；把 ETH 轉出交易所，則暗示囤積累積。透過即時監控這些轉帳，你的 bot 能在價格反映相應活動之前，取得數秒到數分鐘的提前預警。

這一步為你的交易 bot 加入巨鯨監控，作為額外的輸入層。

在接下來的兩段程式碼中，我們首先會輪詢 Alchemy 取得每個區塊上新的 WETH 轉帳，篩選出超過定義門檻（100 ETH）的轉帳，並根據資金是流向還是流出已知的交易所地址進行分類。純粹的 ETH 轉帳不會發出事件，因此 WETH 的 Transfer 事件是大額 ETH 移動的實用替代指標。

在第二段程式碼中，我們會示範如何把巨鯨活動數據餵進你既有的交易邏輯中，根據巨鯨看起來是在買進還是賣出來調整信心門檻。

設定巨鯨轉帳監控：

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

上面的巨鯨監控器蒐集了數據，但這些數據需要被餵進你的交易 bot 中，才能真正影響決策。以下程式碼展示了如何把巨鯨情緒追蹤成一個持續累計的分數，並據此調整你的信心門檻——巨鯨在累積時降低買進門檻，看起來在賣出時則提高門檻。

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

這兩段程式碼加在一起，為你的交易 bot 增添了額外的訊號層。第一段程式碼在背景執行緒中運行，透過 WebSocket 持續監聽大額 ETH 轉帳，並把每一筆歸類為看漲（轉出交易所）、看跌（轉入交易所）或中性。第二段程式碼根據這些巨鯨活動維持一個持續累計的情緒分數，並據此調整你的交易門檻。

當你的機器學習模型預測與巨鯨行為一致時——例如，預測價格上漲的同時巨鯨也在累積——合併後的訊號會更強，bot 也會更積極地交易。當訊號互相衝突時，bot 則會變得更加謹慎。

這一點很重要，因為機構與大戶往往比散戶交易者擁有更好的資訊或更長的時間視野。大額轉入或轉出交易所會影響可用供給，並可能在數分鐘到數小時前預示著價格走勢。透過把鏈上活動與機器學習模型的預測結合起來，你能讓 bot 對市場狀況擁有比單靠價格數據更完整的認知。

第七步：實作風險管理

這可以說是整份教學中最重要的一步。即使是最好的預測模型，若缺乏適當的風險管理，仍然會虧錢。一個勝率 60% 的 bot，如果那 40% 虧損的交易沒有受到控制，仍然可能破產——一筆糟糕的部位就足以抵銷數十筆獲利的交易。風險管理發揮了幾項關鍵功能：限制單一交易可能虧損的幅度（停損）、在市場反轉前鎖定獲利（停利）、防止投資組合出現災難性的下跌（回撤限制），並防止情緒化的過度交易（每日交易次數限制）。以下程式碼把這一切實作成一個可重複使用、能整合進你交易 bot 的 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 類別把所有風險控制集中成單一、可重複使用的元件。初始化時，你要定義風險參數：起始資金、每筆交易的最大部位大小、觸發交易暫停的投資組合回撤百分比、個別交易的停損與停利門檻，以及防止過度交易的每日交易次數限制。

這個類別提供了幾個關鍵方法。check\_drawdown 會監控你的投資組合是否從高峰值下跌過多，一旦超過門檻就會自動暫停交易——這能防止連續虧損把整個帳戶耗盡。calculate\_position\_size 實作了簡化版的 Kelly Criterion 方法，根據模型信心度來決定部位大小（信心度越高，部位越大），同時在投資組合已經出現虧損時縮小部位。

check\_stop\_loss 與 check\_take\_profit 監控未平倉部位，並在價格超出你定義的門檻時觸發出場。record\_trade 記錄每一筆交易供日後分析，get\_performance\_stats 則計算勝率、獲利因子與目前回撤等綜合指標。

把風險管理整合進你的 bot

定義好 RiskManager 類別後，下一步就是把它接入你的主要交易迴圈中。以下程式碼修改了第五步中的 bot，讓它在每次交易前檢查風險限制、根據信心度使用動態部位大小，並在觸及停損或停利水準時自動出場。

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)

與第五步的基礎版本相比，這個增強版交易迴圈加入了數層保護。在每次迭代之前，bot 會檢查 risk_manager.can_trade\(\) 以確認沒有觸犯任何風險限制——如果投資組合已達到最大回撤，或每日交易次數已用完，交易會自動暫停。

對於未平倉部位，bot 會持續監控是否觸及停損或停利門檻，一旦觸發便立即出場，不會等待模型產生新訊號。部位大小現在是動態的，由風險管理器根據模型信心度與當前投資組合價值計算，而非使用固定金額。

結果就是一個能在連續虧損期間保護資金、自動鎖定獲利，並根據自身表現調整風險暴露的 bot。當你停止 bot 時，它會印出完整的績效統計，包括總報酬、勝率、平均勝/負規模，以及獲利因子——提供你評估與精進策略所需的數據。

第八步：回測你的策略

在冒險投入真實資金之前，先在歷史數據上回測你的策略。這能揭露你的 bot 在不同市場條件下的表現。

以下函式會把你整個交易策略拿去對照歷史數據進行模擬，包括逼真的手續費與滑價估算、停損與停利執行，以及詳細的績效追蹤。它會輸出關鍵指標，並產生一個權益曲線視覺化圖表，顯示你的投資組合價值隨時間的演變情況。

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)

這個函式會把你整個交易策略拿去回放對照你所蒐集的歷史數據，模擬 bot 原本會做出的每一個買賣決策。它會逐一走過數據集中的每個時間段、用你訓練好的模型產生預測、檢查未平倉部位是否觸及停損或停利門檻、在信心超過門檻時執行交易，並全程追蹤投資組合價值。

這個模擬包含了逼真的摩擦成本：每筆交易 0.1% 的手續費（主要交易所的典型水準），以及 0.05% 的滑價（使用市價單時，預期與實際成交價格之間的差異）。這些看似微小的百分比在大量交易中會迅速累積，這也是為什麼這個函式會追蹤總手續費支出——這往往是原本能獲利的策略中，默默拖垮績效的隱形殺手。

最後，這個函式會輸出完整的統計數據，並儲存一張權益曲線視覺化圖表，顯示你的投資組合價值在測試期間的演變情況。

解讀回測結果

回測會輸出幾項指標，協助你評估策略是否可行。以下說明每項指標的意義，以及該注意的重點：

總報酬：你策略的獲利能力。任何高於 0% 都算獲利，但你應該把這些結果拿去和買入並持有 ETH 做比較。
最大回撤：從高峰到谷底最嚴重的跌幅。如果你無法承受 15% 的虧損，就無法運行這個策略。
勝率：獲利交易的百分比。只要你的獲利規模大於虧損規模，即使勝率低於 50% 也能獲利。
獲利因子：總獲利 / 總虧損。高於 1.5 倍通常算不錯；低於 1.0 倍則代表你在虧錢。
總手續費：往往是默默的隱形殺手。高頻策略在計入手續費前可能是獲利的，但計入之後反而虧損。

現實檢視：如果你的回測顯示超過 100% 的報酬，你的程式碼很可能有 bug（例如未來函數偏誤、過度擬合等）。實際交易的表現會比回測更差，因為存在滑價、手續費、執行延遲，以及不可預見的事件。

恭喜，如果你成功運行了這一切——代表你已經打造出一個能運作的 AI 交易 bot！

你已經打造出的成果

你現在擁有一個能運作的交易 bot，它能：

利用 Alchemy 的基礎設施擷取即時區塊鏈數據
結合多種數據來源：價格、鏈上指標、巨鯨活動
運用機器學習預測短期價格走勢
透過停損、部位大小與回撤限制來管理風險
在正式上線前可進行回測以驗證策略

這是一個原始但完整的系統。它示範了任何 AI 交易 bot 的核心工作流程，從數據蒐集到執行。

重要提醒與警告

在啟動你的 AI 交易 bot 之前，請留意以下事項以保護你的資金：

先從模擬交易開始：先用模擬訂單運行你的 bot 至少一個月。追蹤假設性的績效，在不冒風險的情況下找出錯誤並精進策略。
使用小額部位：正式上線時，先從你預定資金的 10-20% 開始。即使經過充分測試的 bot，在正式環境中仍可能出現意外行為。
持續監控：尤其是最初幾週，密切關注你的 bot。針對異常行為（交易過於頻繁、大額虧損、錯誤）設定警示。
設置自動終止機制：內建能在觸及關鍵門檻時自動停止交易的機制，例如每日最大虧損、異常的波動飆升，或系統錯誤。市場可能比你手動介入的速度更快地反轉，因此僅依賴手動關閉並不足夠。即使你在睡覺或不在電腦前，你的 bot 也應該能夠自我保護。
注意手續費：在 DEX 上交易會產生 gas 成本；CEX 則會收取交易手續費。高頻策略在計入手續費前可能是獲利的，但計入之後反而虧損。務必納入逼真的手續費估算。
法規遵循：了解你所在司法管轄區的相關法律。有些地區對演算法交易要求執照。如果代管他人的資金，很可能會適用證券法規。
稅務影響：在許多司法管轄區，每一筆交易都是應稅事件。務必詳實記錄。你的 bot 一天執行 100 筆交易，可能造成稅務申報上的惡夢。
安全性至關重要：

絕不要把 API key 寫死在程式碼中
使用環境變數
在所有帳號上啟用雙重驗證（2FA）
考慮使用具有支出限制的智慧合約錢包
盡可能讓私鑰保持離線
定期稽核你的程式碼是否存在漏洞

結論

本指南涵蓋的工具與技巧——擷取即時區塊鏈數據、設計預測性特徵、訓練機器學習模型、實作風險管理，以及回測策略——為打造 AI 交易系統提供了基礎。但這只是起點，而非完成品。真實世界的交易需要持續迭代：隨著市場條件改變而精進模型、針對邊界情況進行壓力測試，並建構能全年無休穩定運行的堅固基礎設施。

準備好開始打造了嗎？註冊 Alchemy，取得你的 API key，並開始用上述程式碼範例進行實驗。如果想要有系統地學習區塊鏈開發，可以參考 Alchemy University。

常見問題

我可以自己打造 AI 交易 bot 嗎？

可以，你可以使用 Python、scikit-learn 或 PyTorch 等機器學習函式庫，以及提供即時數據存取的區塊鏈基礎設施，來打造自己的 AI 交易 bot。本指南提供了完整的教學供你入門。

打造 AI 交易 bot 最適合的程式語言是什麼？

Python 是最推薦的語言，因為它擁有豐富的機器學習與數據分析函式庫，包括 Pandas、NumPy、scikit-learn，以及 PyTorch 或 TensorFlow。

AI 交易 bot 的必要組成元件有哪些？

一個完整的 AI 交易 bot 需要可靠的區塊鏈數據基礎設施、用於預測的機器學習模型、結合價格與鏈上指標的特徵工程、附帶停損與部位大小控管的風險管理，以及在正式上線前用來驗證策略的回測能力。

區塊鏈基礎設施對 AI 交易 bot 有多重要？

基礎設施至關重要，大多數 bot 失敗並非因為預測不佳，而是因為數據過時、API 停機，或延遲過高。你需要即時、低延遲的數據存取，以及跨多條區塊鏈進行競爭型交易所需的可靠性。

AI 交易 bot 的實際成功率大概是多少？

只有 10-30% 的 bot 使用者能達到穩定獲利。成功與否取決於可靠的基礎設施、健全的風險管理，以及務實的期待，而不僅僅是精密的演算法。在波動劇烈的加密貨幣市場中，55-60% 的預測準確率已經相當不錯。

我該如何用 AI 交易 bot 管理風險？

實作停損與停利水準、根據信心度決定部位大小（通常每筆交易最多佔投資組合的 2%）、觸發交易暫停的最大回撤門檻、防止過度交易的每日交易次數限制，以及應對意外虧損的斷路機制。

我的 AI 交易 bot 應該使用哪些數據來源？

結合即時價格數據、gas 使用量與交易數量等鏈上指標、移動平均線與波動性等技術指標，以及來自巨鯨錢包動向的情緒訊號，以取得最完整的市場視野。

我該在使用真實資金前先回測交易 bot 嗎？

是的，一定要先在歷史數據上回測，接著用模擬訂單進行至少一個月的模擬交易。正式交易時只使用你預定資金的 10-20%，因為由於手續費、滑價與執行延遲，實際表現會比回測結果差 20-30%。

AI 交易 bot 的主要種類有哪些？

常見種類包括利用價格模式與指標的技術分析 bot、解析社群媒體與鏈上訊號的情緒分析 bot、利用交易所間價差獲利的套利 bot，以及協調多種策略的自主 AI agent 基金。

我應該在交易 bot 中內建哪些安全機制？

建構能追蹤每個決策的可觀察系統、加入具有最大虧損門檻與冷卻期的斷路機制、實作速率限制器以防止失控執行，並讓人類參與超過特定門檻的高風險決策。

Alchemy 電子報

搶先掌握最新發布消息

訂閱我們的電子報

取得 Alchemy 最新產品更新與資源

超過 80,000 名訂閱者

輸入您的電子郵件地址，即表示您同意接收我們的行銷通訊與產品更新。您了解 Alchemy 會依照我們的隱私權聲明處理所收到的資訊。您可以隨時取消訂閱。

公告

Glamsterdam：Ethereum 開發者完整指南

Glamsterdam 已在 Sepolia 上線。這份完整指南說明升級對開發者意味著什麼，以及 Alchemy 如何為這次網路升級做準備。

技術

在東京建構，應該選擇哪家 RPC 提供商？

我們在五條鏈上對服務東京的五家 RPC 提供商做了基準測試。以下是比較結果，以及差異對延遲敏感團隊的意義。

產品

打造業界領先速度與可靠性的 Solana RPC 讀取架構

Alchemy 擁有最低的 Solana 讀取延遲：9.21 毫秒，比次佳供應商快約 26%。