การทำ Backtest สำหรับกลยุทธ์เทรดคริปโตเป็นขั้นตอนสำคัญที่นักลงทุนและนักพัฒนาโมเดล AI ทุกคนต้องทำ บทความนี้จะพาคุณเรียนรู้วิธีดึงข้อมูลประวัติราคาจาก OKX Exchange API อย่างละเอียด พร้อมโค้ด Python ที่พร้อมใช้งานจริง ครอบคลุมตั้งแต่การตั้งค่า API Key ไปจนถึงการนำข้อมูลไปวิเคราะห์และเทรดจริง
ทำไมต้องใช้ OKX API สำหรับการดึงข้อมูลคริปโต
OKX (เดิมชื่อ OKEx) เป็นหนึ่งใน Exchange ที่ใหญ่ที่สุดของโลก มีข้อได้เปรียบหลายประการสำหรับการทำ Backtest:
- ปริมาณการซื้อขายสูง — ข้อมูลมีความน่าเชื่อถือและสะท้อนราคาตลาดจริง
- ค่าธรรมเนียมต่ำ — Maker fee เพียง 0.08% สำหรับผู้ใช้ทั่วไป
- API มีความเสถียร — Uptime สูงกว่า 99.9%
- รองรับคู่เทรดมากกว่า 300 คู่ — ครอบคลุม BTC, ETH และ Altcoins ยอดนิยม
การตั้งค่า OKX API Key
ก่อนเริ่มเขียนโค้ด คุณต้องสร้าง API Key จาก OKX ก่อน:
- เข้าไปที่ OKX官网 → ล็อกอินบัญชี
- ไปที่ Account Settings → API
- คลิก Create API Key
- เลือก API Key type เป็น Trade (สำหรับดึงข้อมูลเท่านั้น ไม่ต้องเปิด Trading Permission)
- บันทึก API Key, Secret Key และ Passphrase ไว้อย่างปลอดภัย
# ติดตั้งไลบรารีที่จำเป็น
pip install requests pandas numpy okx
ไฟล์ config สำหรับเก็บ API credentials
ห้าม commit ไฟล์นี้ขึ้น GitHub เด็ดขาด
OKX API Configuration
OKX_API_KEY = "your_okx_api_key_here"
OKX_SECRET_KEY = "your_okx_secret_key_here"
OKX_PASSPHRASE = "your_okx_passphrase_here"
ตั้งค่า environment variables (แนะนำวิธีนี้มากกว่า)
import os
os.environ['OKX_API_KEY'] = "your_okx_api_key_here"
os.environ['OKX_SECRET_KEY'] = "your_okx_secret_key_here"
os.environ['OKX_PASSPHRASE'] = "your_okx_passphrase_here"โค้ด Python สำหรับดึงข้อมูลประวัติราคาจาก OKX
ต่อไปนี้คือโค้ดที่สมบูรณ์สำหรับดึงข้อมูล OHLCV (Open, High, Low, Close, Volume) จาก OKX API:
import requests
import pandas as pd
import time
from datetime import datetime, timedelta
class OKXDataFetcher:
"""Class สำหรับดึงข้อมูลประวัติราคาจาก OKX Exchange"""
BASE_URL = "https://www.okx.com"
def __init__(self, api_key, secret_key, passphrase, passphrase_type="plain"):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
self.passphrase_type = passphrase_type
def get_klines(self, inst_id, bar="1H", start=None, end=None, limit=100):
"""
ดึงข้อมูล OHLCV
Parameters:
- inst_id: คู่เทรด เช่น "BTC-USDT", "ETH-USDT"
- bar: Timeframe เช่น "1m", "5m", "1H", "1D"
- start: วันที่เริ่มต้น (ISO format)
- end: วันที่สิ้นสุด (ISO format)
- limit: จำนวนข้อมูลสูงสุด (100-3000)
Returns:
- DataFrame ที่มีคอลัมน์: timestamp, open, high, low, close, volume
"""
endpoint = f"{self.BASE_URL}/api/v5/market/history-candles"
params = {
"instId": inst_id,
"bar": bar,
"limit": limit
}
if start:
params["after"] = start # OKX ใช้ 'after' สำหรับข้อมูลก่อนหน้า
if end:
params["before"] = end # และ 'before' สำหรับข้อมูลหลังจาก
try:
response = requests.get(endpoint, params=params, timeout=30)
response.raise_for_status()
data = response.json()
if data.get("code") != "0":
print(f"❌ API Error: {data.get('msg')}")
return None
candles = data.get("data", [])
if not candles:
print("⚠️ ไม่พบข้อมูลในช่วงเวลาที่กำหนด")
return None
# แปลงข้อมูลเป็น DataFrame
df = pd.DataFrame(candles, columns=[
"timestamp", "open", "high", "low", "close", "volume",
"close_vol", "quote_vol", "num_trades", "buy_vol", "buy_quote_vol", "flag"
])
# แปลงประเภทข้อมูล
numeric_cols = ["open", "high", "low", "close", "volume"]
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors="coerce")
# แปลง timestamp เป็น datetime
df["datetime"] = pd.to_datetime(df["timestamp"].astype(float), unit="ms")
df = df.sort_values("datetime").reset_index(drop=True)
return df
except requests.exceptions.RequestException as e:
print(f"❌ Connection Error: {e}")
return None
def get_multi_year_data(self, inst_id, bar="1H", start_date=None, end_date=None):
"""
ดึงข้อมูลย้อนหลังหลายปี โดยใช้การเรียก API หลายครั้ง
เนื่องจาก OKX จำกัดการดึงข้อมูลครั้งละไม่เกิน 3000 candles
"""
if not start_date:
start_date = (datetime.now() - timedelta(days=365)).isoformat()
if not end_date:
end_date = datetime.now().isoformat()
all_data = []
current_start = start_date
print(f"📥 กำลังดึงข้อมูล {inst_id} จาก {start_date[:10]} ถึง {end_date[:10]}...")
max_iterations = 50 # ป้องกัน infinite loop
iteration = 0
while iteration < max_iterations:
df = self.get_klines(inst_id, bar, start=current_start, limit=3000)
if df is None or df.empty:
break
all_data.append(df)
# ดึงข้อมูลช่วงก่อนหน้า
current_start = str(int(df["timestamp"].min()) - 1)
print(f" ดึงได้แล้ว {len(df)} records (รวม {sum(len(d) for d in all_data)} records)")
# หยุดพักเพื่อไม่ให้ถูก rate limit
time.sleep(0.2)
iteration += 1
# ตรวจสอบว่าถึงวันที่ต้องการหรือยัง
if int(current_start) < int(datetime.fromisoformat(end_date.replace("Z", "+00:00")).timestamp() * 1000):
break
if all_data:
combined_df = pd.concat(all_data, ignore_index=True)
combined_df = combined_df.drop_duplicates(subset=["timestamp"]).sort_values("datetime")
# กรองตามช่วงวันที่ที่ต้องการ
end_dt = datetime.fromisoformat(end_date.replace("Z", "+00:00"))
combined_df = combined_df[combined_df["datetime"] <= end_dt]
print(f"✅ ดึงข้อมูลสำเร็จ: {len(combined_df)} records")
return combined_df
return None
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# สร้าง instance
fetcher = OKXDataFetcher(
api_key=os.environ.get('OKX_API_KEY'),
secret_key=os.environ.get('OKX_SECRET_KEY'),
passphrase=os.environ.get('OKX_PASSPHRASE')
)
# ดึงข้อมูล BTC/USDT รายชั่วโมงย้อนหลัง 2 ปี
btc_data = fetcher.get_multi_year_data(
inst_id="BTC-USDT",
bar="1H",
start_date=(datetime.now() - timedelta(days=730)).isoformat()
)
if btc_data is not None:
# บันทึกเป็น CSV
btc_data.to_csv("btc_usdt_2years.csv", index=False)
print(f"💾 บันทึกไฟล์: btc_usdt_2years.csv")
# แสดงตัวอย่างข้อมูล
print("\n📊 ตัวอย่างข้อมูล 5 แถวแรก:")
print(btc_data[["datetime", "open", "high", "low", "close", "volume"]].head())การทำ Backtest ด้วยข้อมูลที่ได้
เมื่อได้ข้อมูลประวัติราคาแล้ว ต่อไปจะเป็นการนำข้อมูลไปทำ Backtest ด้วยกลยุทธ์ Moving Average Crossover:
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
def backtest_ma_crossover(df, short_window=20, long_window=50, initial_balance=10000):
"""
ทำ Backtest กลยุทธ์ Moving Average Crossover
Parameters:
- df: DataFrame ที่มีข้อมูล OHLCV
- short_window: ค่าเฉลี่ยเคลื่อนที่ระยะสั้น
- long_window: ค่าเฉลี่ยเคลื่อนที่ระยะยาว
- initial_balance: จำนวนเงินเริ่มต้น (USDT)
Returns:
- ผลลัพธ์การ Backtest พร้อม Metrics
"""
df = df.copy()
# คำนวณ Moving Averages
df["SMA_Short"] = df["close"].rolling(window=short_window).mean()
df["SMA_Long"] = df["close"].rolling(window=long_window).mean()
# สร้างสัญญาณซื้อ/ขาย
# Signal = 1 เมื่อ SMA Short ตัด SMA Long ขึ้น (Golden Cross)
# Signal = -1 เมื่อ SMA Short ตัด SMA Long ลง (Death Cross)
df["Signal"] = 0
df.loc[df["SMA_Short"] > df["SMA_Long"], "Signal"] = 1
df.loc[df["SMA_Short"] <= df["SMA_Long"], "Signal"] = -1
# คำนวณสัญญาณการซื้อขาย
df["Position"] = df["Signal"].diff()
# จำลองการซื้อขาย
balance = initial_balance
btc_holding = 0
trades = []
portfolio_value = []
for i, row in df.iterrows():
current_price = row["close"]
# ซื้อเมื่อมีสัญญาณ Golden Cross และยังไม่มี BTC
if row["Position"] == 2 and balance > 0: # Signal changed from -1 to 1
btc_holding = balance / current_price
trades.append({
"datetime": row["datetime"],
"type": "BUY",
"price": current_price,
"amount_btc": btc_holding,
"balance_before": balance
})
balance = 0
# ขายเมื่อมีสัญญาณ Death Cross และมี BTC
elif row["Position"] == -2 and btc_holding > 0: # Signal changed from 1 to -1
balance = btc_holding * current_price
trades.append({
"datetime": row["datetime"],
"type": "SELL",
"price": current_price,
"amount_btc": btc_holding,
"balance_after": balance
})
btc_holding = 0
# คำนวณมูลค่าพอร์ตรวม
total_value = balance + (btc_holding * current_price)
portfolio_value.append({
"datetime": row["datetime"],
"value": total_value
})
# สร้าง DataFrame สำหรับผลลัพธ์
result_df = pd.DataFrame(portfolio_value)
# คำนวณ Metrics
total_return = (result_df["value"].iloc[-1] - initial_balance) / initial_balance * 100
max_value = result_df["value"].max()
max_drawdown = ((result_df["value"] - result_df["value"].cummax()) / result_df["value"].cummax()).min() * 100
# คำนวณ Buy & Hold Return
buy_hold_return = (df["close"].iloc[-1] - df["close"].iloc[0]) / df["close"].iloc[0] * 100
# จำนวนการซื้อขาย
num_trades = len(trades)
metrics = {
"Initial Balance": initial_balance,
"Final Value": result_df["value"].iloc[-1],
"Total Return": total_return,
"Buy & Hold Return": buy_hold_return,
"Excess Return": total_return - buy_hold_return,
"Max Drawdown": max_drawdown,
"Number of Trades": num_trades,
"Win Rate": len([t for t in trades if t["type"] == "SELL" and
t["price"] > next((x["price"] for x in trades if x["type"] == "BUY"), t["price"])]) / max(num_trades // 2, 1) * 100 if num_trades > 0 else 0
}
return metrics, trades, result_df
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# โหลดข้อมูลที่ดึงไว้ก่อนหน้า
df = pd.read_csv("btc_usdt_2years.csv")
df["datetime"] = pd.to_datetime(df["datetime"])
# ทำ Backtest ด้วยกลยุทธ์ SMA 20/50
metrics, trades, result_df = backtest_ma_crossover(
df,
short_window=20,
long_window=50,
initial_balance=10000
)
# แสดงผลลัพธ์
print("=" * 50)
print("📊 BACKTEST RESULTS - MA Crossover Strategy")
print("=" * 50)
for key, value in metrics.items():
if isinstance(value, float):
print(f"{key}: {value:.2f}%")
else:
print(f"{key}: {value}")
print("=" * 50)
# วาดกราฟ
fig, axes = plt.subplots(3, 1, figsize=(14, 10), sharex=True)
# กราฟราคาและ Moving Averages
axes[0].plot(df["datetime"], df["close"], label="BTC Price", alpha=0.7)
axes[0].plot(df["datetime"], df["SMA_Short"], label=f"SMA 20", alpha=0.8)
axes[0].plot(df["datetime"], df["SMA_Long"], label=f"SMA 50", alpha=0.8)
# กำหนดจุดซื้อขาย
buy_signals = df[df["Position"] == 2]
sell_signals = df[df["Position"] == -2]
axes[0].scatter(buy_signals["datetime"], buy_signals["close"],
marker="^", color="green", s=100, label="Buy Signal", zorder=5)
axes[0].scatter(sell_signals["datetime"], sell_signals["close"],
marker="v", color="red", s=100, label="Sell Signal", zorder=5)
axes[0].set_ylabel("Price (USDT)")
axes[0].set_title("BTC/USDT - MA Crossover Strategy Backtest")
axes[0].legend(loc="upper left")
axes[0].grid(True, alpha=0.3)
# กราฟมูลค่าพอร์ต
axes[1].plot(result_df["datetime"], result_df["value"], color="blue", linewidth=2)
axes[1].axhline(y=10000, color="gray", linestyle="--", alpha=0.5, label="Initial Balance")
axes[1].fill_between(result_df["datetime"], 10000, result_df["value"],
where=result_df["value"] >= 10000, alpha=0.3, color="green")
axes[1].fill_between(result_df["datetime"], 10000, result_df["value"],
where=result_df["value"] < 10000, alpha=0.3, color="red")
axes[1].set_ylabel("Portfolio Value (USDT)")
axes[1].legend(loc="upper left")
axes[1].grid(True, alpha=0.3)
# กราฟ Drawdown
drawdown = (result_df["value"] - result_df["value"].cummax()) / result_df["value"].cummax() * 100
axes[2].fill_between(result_df["datetime"], 0, drawdown, color="red", alpha=0.5)
axes[2].set_ylabel("Drawdown (%)")
axes[2].set_xlabel("Date")
axes[2].grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("backtest_result.png", dpi=150)
print("💾 บันทึกกราฟ: backtest_result.png")การใช้ AI วิเคราะห์ผลลัพธ์ Backtest
เมื่อได้ผลลัพธ์การ Backtest แล้ว คุณสามารถนำข้อมูลไปวิเคราะห์ด้วย AI เพื่อหากลยุทธ์ที่เหมาะสมที่สุด โดยใช้ HolySheep AI ซึ่งมีความได้เปรียบด้านราคาและความเร็ว:
import requests
import json
def analyze_backtest_with_ai(backtest_results, trading_pairs=["BTC-USDT", "ETH-USDT"]):
"""
ใช้ AI วิเคราะห์ผลลัพธ์ Backtest และเสนอกลยุทธ์ที่ดีที่สุด
ใช้ HolySheep AI API สำหรับความเร็วและประหยัดค่าใช้จ่าย
"""
# เตรียมข้อมูลสำหรับ AI
analysis_prompt = f"""
ผลลัพธ์การ Backtest สำหรับคู่เทรด: {', '.join(trading_pairs)}
📊 Metrics สรุป:
- Total Return: {backtest_results.get('Total Return', 0):.2f}%
- Buy & Hold Return: {backtest_results.get('Buy & Hold Return', 0):.2f}%
- Max Drawdown: {backtest_results.get('Max Drawdown', 0):.2f}%
- Number of Trades: {backtest_results.get('Number of Trades', 0)}
กรุณาวิเคราะห์และแนะนำ:
1. กลยุทธ์ที่เหมาะสมที่สุดสำหรับตลาดปัจจุบัน
2. ค่า Parameters ที่ควรปรับ (เช่น MA windows, Stop Loss %, Take Profit %)
3. ความเสี่ยงและคำแนะนำในการจัดการความเสี่ยง
4. ข้อเสนอแนะในการปรับปรุงกลยุทธ์
"""
# เรียกใช้ HolySheep AI API
base_url = "https://api.holysheep.ai/v1"
api_key = "YOUR_HOLYSHEEP_API_KEY" # แทนที่ด้วย API Key จริงของคุณ
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1", # ใช้โมเดลที่เหมาะสม
"messages": [
{
"role": "system",
"content": "คุณเป็นผู้เชี่ยวชาญด้าน Cryptocurrency Trading และ Quantitative Analysis ที่มีประสบการณ์มากกว่า 10 ปี"
},
{
"role": "user",
"content": analysis_prompt
}
],
"temperature": 0.7,
"max_tokens": 2000
}
try:
print("🤖 กำลังวิเคราะห์ด้วย AI...")
response = requests.post(
f"{base_url}/chat/completions",
headers=headers,
json=payload,
timeout=60
)
if response.status_code == 200:
result = response.json()
ai_analysis = result["choices"][0]["message"]["content"]
print("\n" + "=" * 60)
print("📝 AI ANALYSIS RESULT")
print("=" * 60)
print(ai_analysis)
print("=" * 60)
return ai_analysis
else:
print(f"❌ API Error: {response.status_code}")
print(response.text)
return None
except requests.exceptions.RequestException as e:
print(f"❌ Connection Error: {e}")
return None
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# ผลลัพธ์จากการ Backtest
sample_results = {
"Total Return": 45.5,
"Buy & Hold Return": 62.3,
"Max Drawdown": -18.5,
"Number of Trades": 12
}
analysis = analyze_backtest_with_ai(
backtest_results=sample_results,
trading_pairs=["BTC-USDT", "ETH-USDT", "SOL-USDT"]
)ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. ได้รับข้อผิดพลาด 1001 "System busy" จาก OKX API
สาเหตุ: Server ของ OKX มีปริมาณการใช้งานสูง หรือถูก Rate Limit
# วิธีแก้ไข: ใช้ Retry Logic พร้อม Exponential Backoff
import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry(max_retries=5):
"""สร้าง Requests Session ที่มี Retry Logic ในตัว"""
session = requests.Session()
retry_strategy = Retry(
total=max_retries,
backoff_factor=1, # 1, 2, 4, 8, 16 วินาที
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
การใช้งาน
session = create_session_with_retry()
หากยังได้รับ Error ให้ลองใช้ Alternative Endpoint
def fetch_klines_alternative(inst_id, bar="1H", limit=100):
"""ดึงข้อมูลจาก Public API ของ OKX (ไม่ต้องมี API Key)"""
# ลอง endpoint หลัก
try:
url = f"https://www.okx.com/api/v5/market/history-candles"
params = {"instId": inst_id, "bar": bar, "limit": limit}
response = session.get(url, params=params, timeout=30)
if response.status_code == 200:
return response.json()
except Exception as e:
print(f"Primary endpoint failed: {