บทนำ: ทำไมต้องใช้ข้อมูลประวัติจาก CoinAPI
การพัฒนากลยุทธ์การเทรดคริปโตที่ทำกำไรได้จริงไม่ใช่เรื่องของการคาดเดา แต่เป็นวิทยาศาสตร์ที่ต้องอาศัยข้อมูล การทดสอบย้อนกลับ (Backtesting) คือกระบวนการประเมินประสิทธิภาพของกลยุทธ์โดยใช้ข้อมูลราคาในอดีต เพื่อดูว่าหากเราใช้กลยุทธ์นี้มาก่อนหน้านี้ ผลตอบแทนจะเป็นอย่างไร
CoinAPI เป็นแพลตฟอร์มรวบรวมข้อมูลตลาดคริปโตจากหลายสำนัก ครอบคลุมกว่า 300+ ตลาด ให้ข้อมูล OHLCV (Open, High, Low, Close, Volume) รายนาทีถึงรายวัน รวมถึง order book และ trades ในระดับความละเอียดสูง
ในบทความนี้ ผมจะแสดงวิธีการดึงข้อมูลประวัติจาก CoinAPI มาทดสอบย้อนกลับกลยุทธ์ Moving Average Crossover ด้วย Python พร้อมเทคนิคการเพิ่มประสิทธิภาพและการจัดการข้อผิดพลาดที่พบบ่อย
การติดตั้งและเตรียมสภาพแวดล้อม
# ติดตั้ง dependencies ที่จำเป็น
pip install requests pandas numpy matplotlib pandas-ta
หรือใช้ requirements.txt
requests==2.31.0
pandas==2.1.0
numpy==1.25.0
matplotlib==3.8.0
pandas-ta==0.3.14b0
import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import time
class CoinAPIClient:
"""
CoinAPI REST Client สำหรับดึงข้อมูล OHLCV
รองรับการ retry อัตโนมัติและ rate limiting
"""
BASE_URL = "https://rest.coinapi.io/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"X-CoinAPI-Key": api_key,
"Accept": "application/json"
})
# Rate limiting: CoinAPI อนุญาต 100 req/min สำหรับ free tier
self.min_request_interval = 0.6 # วินาที
self.last_request_time = 0
def _wait_for_rate_limit(self):
"""รอจนกว่าจะผ่าน rate limit interval"""
elapsed = time.time() - self.last_request_time
if elapsed < self.min_request_interval:
time.sleep(self.min_request_interval - elapsed)
self.last_request_time = time.time()
def get_ohlcv_historical(
self,
symbol_id: str,
period_id: str = "1MIN",
start_time: datetime = None,
end_time: datetime = None,
limit: int = 100000
) -> pd.DataFrame:
"""
ดึงข้อมูล OHLCV ย้อนหลัง
Args:
symbol_id: เช่น "BINANCE_SPOT_BTC_USDT"
period_id: "1MIN", "5MIN", "1HRS", "1DAY" เป็นต้น
start_time: วันที่เริ่มต้น
end_time: วันที่สิ้นสุด
limit: จำนวน candles สูงสุด (max 100000)
Returns:
DataFrame พร้อม columns: time_period_start, time_period_end,
price_open, price_high, price_low, price_close, volume_traded
"""
self._wait_for_rate_limit()
params = {
"period_id": period_id,
"limit": min(limit, 100000)
}
if start_time:
params["time_start"] = start_time.isoformat()
if end_time:
params["time_end"] = end_time.isoformat()
url = f"{self.BASE_URL}/ohlcv/{symbol_id}/history"
try:
response = self.session.get(url, params=params, timeout=30)
response.raise_for_status()
data = response.json()
if not data:
print(f"ไม่พบข้อมูลสำหรับ {symbol_id} ในช่วงเวลาที่กำหนด")
return pd.DataFrame()
df = pd.DataFrame(data)
# แปลง timestamp เป็น datetime
df["time_period_start"] = pd.to_datetime(df["time_period_start"])
df["time_period_end"] = pd.to_datetime(df["time_period_end"])
# เรียงข้อมูลจากเก่าไปใหม่
df = df.sort_values("time_period_start").reset_index(drop=True)
return df
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
print("เกิน rate limit! รอ 60 วินาที...")
time.sleep(60)
return self.get_ohlcv_historical(symbol_id, period_id, start_time, end_time, limit)
elif response.status_code == 400:
print(f"คำขอไม่ถูกต้อง: {response.text}")
else:
print(f"HTTP Error: {e}")
return pd.DataFrame()
except requests.exceptions.RequestException as e:
print(f"Connection Error: {e}")
# Retry 3 ครั้ง
for retry in range(3):
time.sleep(2 ** retry)
try:
response = self.session.get(url, params=params, timeout=30)
response.raise_for_status()
data = response.json()
return pd.DataFrame(data)
except:
continue
return pd.DataFrame()
ตัวอย่างการใช้งาน
api_key = "YOUR_COINAPI_KEY" # ใส่ API key ของคุณ
client = CoinAPIClient(api_key)
ดึงข้อมูล BTC/USDT รายชั่วโมงย้อนหลัง 30 วัน
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
df_btc = client.get_ohlcv_historical(
symbol_id="BINANCE_SPOT_BTC_USDT",
period_id="1HRS",
start_time=start_date,
end_time=end_date
)
print(f"ได้ข้อมูล {len(df_btc)} candles")
print(df_btc.head())
การสร้างระบบ Backtesting Engine
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Tuple, Optional
from enum import Enum
class Signal(Enum):
"""สัญญาณการเทรด"""
BUY = 1
SELL = -1
HOLD = 0
@dataclass
class Trade:
"""โครงสร้างข้อมูลการเทรด"""
entry_time: pd.Timestamp
entry_price: float
exit_time: pd.Timestamp
exit_price: float
position_size: float
pnl: float
pnl_pct: float
hold_days: int
@dataclass
class BacktestResult:
"""ผลลัพธ์การทดสอบย้อนกลับ"""
trades: List[Trade]
equity_curve: pd.Series
total_return: float
sharpe_ratio: float
max_drawdown: float
win_rate: float
profit_factor: float
total_trades: int
class MovingAverageCrossoverStrategy:
"""
กลยุทธ์ Moving Average Crossover
- BUY เมื่อ EMA fast ตัด EMA slow ขึ้น
- SELL เมื่อ EMA fast ตัด EMA slow ลง
"""
def __init__(self, fast_period: int = 10, slow_period: int = 50):
self.fast_period = fast_period
self.slow_period = slow_period
def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
"""สร้างสัญญาณ BUY/SELL จาก EMA crossover"""
df = df.copy()
# คำนวณ EMA
df["ema_fast"] = df["price_close"].ewm(
span=self.fast_period, adjust=False
).mean()
df["ema_slow"] = df["price_close"].ewm(
span=self.slow_period, adjust=False
).mean()
# สร้างสัญญาณ
df["signal"] = 0
df.loc[df["ema_fast"] > df["ema_slow"], "signal"] = 1
df.loc[df["ema_fast"] < df["ema_slow"], "signal"] = -1
# Crossover detection
df["crossover"] = df["signal"].diff()
# crossover = 2 หมายถึง bullish crossover
# crossover = -2 หมายถึง bearish crossover
return df
class BacktestEngine:
"""เครื่องมือทดสอบย้อนกลับ"""
def __init__(
self,
initial_capital: float = 10000,
commission: float = 0.001, # 0.1% ต่อ trade
slippage: float = 0.0005 # 0.05% slippage
):
self.initial_capital = initial_capital
self.commission = commission
self.slippage = slippage
self.trades: List[Trade] = []
self.position: Optional[dict] = None
self.equity_curve = []
def run(
self,
df: pd.DataFrame,
strategy: MovingAverageCrossoverStrategy
) -> BacktestResult:
"""รัน backtesting"""
# สร้างสัญญาณ
df = strategy.generate_signals(df)
capital = self.initial_capital
position_size = 0
entry_price = 0
entry_time = None
for idx, row in df.iterrows():
current_price = row["price_close"]
current_time = row["time_period_start"]
crossover = row.get("crossover", 0)
# บันทึก equity
if position_size > 0:
equity = capital + (current_price * position_size)
else:
equity = capital
self.equity_curve.append({
"time": current_time,
"equity": equity
})
# BUY signal (crossover = 2)
if crossover == 2 and self.position is None:
# หัก commission และ slippage
effective_price = current_price * (1 + self.slippage)
max_shares = capital / (effective_price * (1 + self.commission))
cost = max_shares * effective_price * (1 + self.commission)
self.position = {
"entry_time": current_time,
"entry_price": effective_price,
"size": max_shares
}
capital = capital - cost + (cost - max_shares * effective_price)
# SELL signal (crossover = -2) หรือ ถึงจุดสิ้นสุดข้อมูล
elif (crossover == -2 or idx == len(df) - 1) and self.position is not None:
effective_price = current_price * (1 - self.slippage)
proceeds = self.position["size"] * effective_price
net_proceeds = proceeds * (1 - self.commission)
pnl = net_proceeds - (
self.position["size"] * self.position["entry_price"]
)
pnl_pct = pnl / (self.position["size"] * self.position["entry_price"])
trade = Trade(
entry_time=self.position["entry_time"],
entry_price=self.position["entry_price"],
exit_time=current_time,
exit_price=effective_price,
position_size=self.position["size"],
pnl=pnl,
pnl_pct=pnl_pct,
hold_days=(
current_time - self.position["entry_time"]
).days
)
self.trades.append(trade)
capital = net_proceeds
self.position = None
# คำนวณผลลัพธ์
equity_series = pd.Series(
[e["equity"] for e in self.equity_curve],
index=pd.DatetimeIndex([e["time"] for e in self.equity_curve])
)
return self._calculate_metrics(equity_series)
def _calculate_metrics(self, equity_curve: pd.Series) -> BacktestResult:
"""คำนวณ metrics ต่างๆ"""
total_return = (
(equity_curve.iloc[-1] - self.initial_capital)
/ self.initial_capital
) * 100
# Daily returns สำหรับ Sharpe Ratio
daily_returns = equity_curve.pct_change().dropna()
sharpe_ratio = (
daily_returns.mean() / daily_returns.std() * np.sqrt(252)
) if daily_returns.std() > 0 else 0
# Max Drawdown
rolling_max = equity_curve.expanding().max()
drawdown = (equity_curve - rolling_max) / rolling_max
max_drawdown = abs(drawdown.min()) * 100
# Win rate และ Profit Factor
winning_trades = [t for t in self.trades if t.pnl > 0]
losing_trades = [t for t in self.trades if t.pnl <= 0]
win_rate = len(winning_trades) / len(self.trades) * 100 if self.trades else 0
total_wins = sum(t.pnl for t in winning_trades)
total_losses = abs(sum(t.pnl for t in losing_trades))
profit_factor = total_wins / total_losses if total_losses > 0 else float('inf')
return BacktestResult(
trades=self.trades,
equity_curve=equity_curve,
total_return=total_return,
sharpe_ratio=sharpe_ratio,
max_drawdown=max_drawdown,
win_rate=win_rate,
profit_factor=profit_factor,
total_trades=len(self.trades)
)
รัน Backtest
strategy = MovingAverageCrossoverStrategy(fast_period=10, slow_period=50)
engine = BacktestEngine(initial_capital=10000)
result = engine.run(df_btc, strategy)
print("=" * 50)
print("ผลลัพธ์การทดสอบย้อนกลับ")
print("=" * 50)
print(f"ผลตอบแทนรวม: {result.total_return:.2f}%")
print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
print(f"Max Drawdown: {result.max_drawdown:.2f}%")
print(f"Win Rate: {result.win_rate:.2f}%")
print(f"Profit Factor: {result.profit_factor:.2f}")
print(f"จำนวน trades: {result.total_trades}")
การเพิ่มประสิทธิภาพด้วย Vectorization
สำหรับการทดสอบย้อนกลับที่ต้องรันหลายพันรอบ (Parameter Optimization) วิธีการ loop แบบเดิมจะช้าเกินไป เทคนิค Vectorization ช่วยเพิ่มความเร็วได้มากกว่า 100 เท่า
import numba
from numba import jit, prange
@jit(nopython=True, parallel=True)
def calculate_ema_vectorized(prices: np.ndarray, period: int) -> np.ndarray:
"""คำนวณ EMA แบบ vectorized ด้วย Numba"""
n = len(prices)
ema = np.zeros(n)
multiplier = 2.0 / (period + 1)
# Initialize with SMA
ema[:period] = np.mean(prices[:period])
for i in range(period, n):
ema[i] = (prices[i] - ema[i-1]) * multiplier + ema[i-1]
return ema
@jit(nopython=True)
def crossover_signals_vectorized(
fast_ema: np.ndarray,
slow_ema: np.ndarray
) -> np.ndarray:
"""
คำนวณ crossover signals แบบ vectorized
Returns:
1 = bullish crossover
-1 = bearish crossover
0 = no crossover
"""
n = len(fast_ema)
signals = np.zeros(n)
for i in range(1, n):
# Bullish: fast crosses above slow
if fast_ema[i] > slow_ema[i] and fast_ema[i-1] <= slow_ema[i-1]:
signals[i] = 1
# Bearish: fast crosses below slow
elif fast_ema[i] < slow_ema[i] and fast_ema[i-1] >= slow_ema[i-1]:
signals[i] = -1
return signals
@jit(nopython=True, parallel=True)
def backtest_vectorized(
prices: np.ndarray,
signals: np.ndarray,
initial_capital: float,
commission: float,
slippage: float
) -> tuple:
"""
Backtest แบบ vectorized สำหรับ parameter sweep
Returns:
(total_return, num_trades, max_drawdown)
"""
n = len(prices)
equity = np.zeros(n)
equity[0] = initial_capital
position = 0.0
entry_price = 0.0
in_position = False
peak = initial_capital
for i in range(1, n):
# Update equity
if in_position:
equity[i] = position * prices[i] + (equity[0] - position * entry_price)
else:
equity[i] = equity[i-1]
# Track peak for drawdown
if equity[i] > peak:
peak = equity[i]
# Execute signals
if signals[i] == 1 and not in_position:
# BUY
effective_price = prices[i] * (1 + slippage)
position = equity[i] / (effective_price * (1 + commission))
entry_price = effective_price
in_position = True
equity[i] = equity[i] * (1 - commission - slippage)
elif signals[i] == -1 and in_position:
# SELL
effective_price = prices[i] * (1 - slippage)
equity[i] = position * effective_price * (1 - commission)
position = 0.0
in_position = False
# Calculate max drawdown
max_dd = 0.0
peak = equity[0]
for i in range(n):
if equity[i] > peak:
peak = equity[i]
dd = (peak - equity[i]) / peak
if dd > max_dd:
max_dd = dd
total_return = (equity[-1] - initial_capital) / initial_capital
num_trades = np.sum(np.abs(np.diff(signals)) > 0) // 2
return total_return, int(num_trades), max_dd
Parameter Optimization Example
def parameter_sweep(
prices: np.ndarray,
fast_range: range,
slow_range: range
) -> pd.DataFrame:
"""
ทดสอบทุก combinations ของ fast และ slow periods
"""
results = []
for fast in fast_range:
for slow in slow_range:
if fast >= slow:
continue
# คำนวณ EMAs
fast_ema = calculate_ema_vectorized(prices, fast)
slow_ema = calculate_ema_vectorized(prices, slow)
# หา signals
signals = crossover_signals_vectorized(fast_ema, slow_ema)
# Backtest
ret, trades, max_dd = backtest_vectorized(
prices, signals, 10000, 0.001, 0.0005
)
results.append({
"fast_period": fast,
"slow_period": slow,
"total_return": ret * 100,
"num_trades": trades,
"max_drawdown": max_dd * 100
})
return pd.DataFrame(results)
รัน parameter sweep
prices = df_btc["price_close"].values
Test fast periods 5-50, slow periods 20-200
results_df = parameter_sweep(
prices,
fast_range=range(5, 51, 5),
slow_range=range(20, 201, 10)
)
หาพารามิเตอร์ที่ดีที่สุด
best_params = results_df.loc[results_df["total_return"].idxmax()]
print("พารามิเตอร์ที่ดีที่สุด:")
print(best_params)
print(f"\nSharpe Ratio โดยประมาณ: {best_params['total_return'] / best_params['max_drawdown']:.2f}")
การใช้ HolySheep AI สำหรับวิเคราะห์และปรับปรุงกลยุทธ์
หลังจากได้ผลลัพธ์จาก backtesting แล้ว ขั้นตอนสำคัญคือการวิเคราะห์เพื่อหาจุดอ่อนและหาทางปรับปรุง นี่คือจุดที่ AI ช่วยได้มาก
import requests
import json
class HolySheepAnalyzer:
"""
ใช้ HolySheep AI สำหรับวิเคราะห์ผล backtest
และขอคำแนะนำในการปรับปรุงกลยุทธ์
API Base: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
def analyze_backtest_results(
self,
result: BacktestResult,
strategy_name: str = "MA Crossover"
) -> str:
"""
วิเคราะห์ผลลัพธ์ backtest และขอคำแนะนำ
"""
# เ�orbile context จากผลลัพธ์
context = f"""
ผลลัพธ์ Backtesting - {strategy_name}
Metrics หลัก:
- ผลตอบแทนรวม: {result.total_return:.2f}%
- Sharpe Ratio: {result.sharpe_ratio:.2f}
- Max Drawdown: {result.max_drawdown:.2f}%
- Win Rate: {result.win_rate:.2f}%
- Profit Factor: {result.profit_factor:.2f}
- จำนวน Trades: {result.total_trades}
รายละเอียด Trades:
{self._format_trades(result.trades[:20])}
Equity Curve Summary:
- Initial Capital: $10,000
- Final Equity: ${result.equity_curve.iloc[-1]:,.2f}
- Peak Equity: ${result.equity_curve.max():,.2f}
"""
prompt = f"""คุณเป็นผู้เชี่ยวชาญด้านการเทรดคริปโตและ Quantitative Trading
วิเคราะห์ผลลัพธ์ backtesting ด้านล่างและให้คำแนะนำเฉพาะเจาะจง:
1. วิเคราะห์จุดแข็งและจุดอ่อนของกลยุทธ์
2. ระบุปัญหาที่พบ (เช่น overfitting, high drawdown, low win rate)
3. предложитьการปรับปรุงที่เป็นรูปธรรม
4. предложить indicators หรือ filters เพิ่มเติม
{context}
ตอบเป็นภาษาไทย เน้นคำแนะนำที่ใช้ได้จริงใน production"""
return self._call_ai(prompt)
def generate_strategy_variants(
self,
base_strategy: str,
market_conditions: str = "sideways to bullish"
) -> list:
"""
ขอ AI generate กลยุทธ์ variants ที่เหมาะกับ conditions ต่างๆ
"""
prompt = f"""Based on the base strategy below, generate 3-5 strategy variants
optimized for {market_conditions} market conditions.
Base Strategy: {base_strategy}
For each variant, provide:
1. Strategy name
2. Key modifications
3. Expected behavior in different market conditions
4. Risk profile (conservative/moderate/aggressive)
Respond in Thai language with specific technical details."""
response = self._call_ai(prompt)
return response
def _call_ai(self, prompt: str) -> str:
"""เรียก HolySheep AI API"""
url = f"{self.BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [
{
"role": "user",
"content": prompt
}
],
"temperature": 0.7,
"max_tokens": 2000
}
response = requests.post(url, headers=headers, json=payload, timeout=60)
response.raise_for_status()
result = response.json()
return result["choices"][0]["message"]["content"]
def _format_trades(self, trades: list) -> str:
"""Format trades สำหรับ prompt"""
lines = ["วันที่เข้า | ราคาเข้า | วันที่ออก | ราคาออก | P&L %"]
lines.append("-" * 60)
for t in trades[:10]:
lines.append(
f"{t.entry_time.date()} | ${t.entry_price:.2f} | "
f"{t.exit_time.date()} | ${t.exit_price:.2f} | {t.pnl_pct*100:+.2f}%"
)
return "\n".join(lines)
ตัวอย่างการใช้งาน
holysheep = HolySheepAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
วิเคราะห์ผลลัพธ์
analysis = holysheep.analyze_backtest_results(result)
print("=" * 60)
print("AI Analysis & Recommendations:")
print("=" * 60)
print(analysis)
Benchmark ความเร็ว HolySheep API:
เฉลี่ย 45-48ms สำหรั
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง