บทนำ: ทำไมต้องมี Backtesting Framework ของตัวเอง
การทำ backtest คือหัวใจสำคัญของการพัฒนา quantitative trading strategy การซื้อเครื่องมือสำเร็จรูปอย่าง Backtrader หรือ Zipline มาใช้นั้นสะดวก แต่เมื่อต้องการควบคุมทุก tick ของข้อมูล ทำ custom risk management หรือ integrate กับ AI model ที่ต้องใช้ LLM วิเคราะห์ market sentiment การสร้าง framework เองจะยืดหยุ่นกว่ามาก
บทความนี้ผมจะแชร์ประสบการณ์ตรงจากการสร้าง backtesting system ที่รองรับข้อมูล crypto หลายล้าน records, ทำ parallel execution ของ strategies, และใช้ AI ช่วยวิเคราะห์ผลลัพธ์ผ่าน
HolySheep AI ซึ่งมี latency ต่ำกว่า 50ms และราคาถูกกว่า 85% เมื่อเทียบกับ OpenAI
สถาปัตยกรรมโดยรวมของระบบ
┌─────────────────────────────────────────────────────────────┐
│ Backtesting Framework │
├─────────────┬─────────────┬─────────────┬───────────────────┤
│ Data Layer │ Engine Core │ Risk Engine │ Analysis Layer │
│ │ │ │ │
│ - CSV/Parquet│ - Vectorized│ - Position │ - Performance │
│ - REST API │ - Event-based│ - Portfolio │ - Sharpe/MaxDD │
│ - WebSocket │ - Scheduling │ - Drawdown │ - AI Summarize │
└─────────────┴─────────────┴─────────────┴───────────────────┘
│
┌───────┴───────┐
│ HolySheep AI │
│ (Sentiment/ │
│ Analysis) │
└───────────────┘
การติดตั้ง Dependencies และ Project Structure
pip install pandas numpy pyarrow fastapi uvicorn sqlalchemy aiohttp pydantic
สำหรับ visualization
pip install plotly kaleido
สำหรับ statistical analysis
pip install scipy statsmodels
Project structure
mkdir crypto_backtest/{data,engine,risk,analysis,config}
touch crypto_backtest/__init__.py
touch crypto_backtest/{data,engine,risk,analysis,config}/__init__.py
Layer 1: Data Management - การโหลดและจัดการข้อมูล History
การออกแบบ data layer ที่ดีต้องรองรับทั้ง historical data (CSV/Parquet) และ real-time streaming ผมใช้ PyArrow สำหรับ columnar storage ซึ่งอ่านเร็วกว่า CSV 10-50 เท่า เมื่อจำนวน records มากกว่า 10 ล้าน
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
from dataclasses import dataclass
from typing import Optional, List
from datetime import datetime
import aiohttp
import asyncio
@dataclass
class OHLCV:
"""Standard OHLCV data structure for crypto"""
timestamp: pd.DatetimeIndex
open: pd.Series
high: pd.Series
low: pd.Series
close: pd.Series
volume: pd.Series
@classmethod
def from_dataframe(cls, df: pd.DataFrame) -> "OHLCV":
required = ["timestamp", "open", "high", "low", "close", "volume"]
if not all(col in df.columns for col in required):
missing = [c for c in required if c not in df.columns]
raise ValueError(f"Missing columns: {missing}")
return cls(
timestamp=pd.to_datetime(df["timestamp"]),
open=df["open"].astype("float32"),
high=df["high"].astype("float32"),
low=df["low"].astype("float32"),
close=df["close"].astype("float32"),
volume=df["volume"].astype("float32")
)
def to_dataframe(self) -> pd.DataFrame:
return pd.DataFrame({
"timestamp": self.timestamp,
"open": self.open,
"high": self.high,
"low": self.low,
"close": self.close,
"volume": self.volume
})
class CryptoDataLoader:
"""
Production-grade data loader รองรับหลาย formats และ sources
รวมถึงการดึงข้อมูลจาก exchange APIs
"""
SUPPORTED_FORMATS = {".csv", ".parquet", ".feather"}
def __init__(self, data_dir: str = "./data"):
self.data_dir = Path(data_dir)
self._cache: dict[str, OHLCV] = {}
def load_csv(
self,
symbol: str,
timeframe: str,
start: Optional[datetime] = None,
end: Optional[datetime] = None
) -> OHLCV:
"""
โหลดข้อมูลจาก CSV files
Expected filename format: {symbol}_{timeframe}.csv
"""
filepath = self.data_dir / f"{symbol}_{timeframe}.csv"
if not filepath.exists():
raise FileNotFoundError(f"Data file not found: {filepath}")
df = pd.read_csv(
filepath,
parse_dates=["timestamp"],
dtype={
"open": "float32",
"high": "float32",
"low": "float32",
"close": "float32",
"volume": "float32"
}
)
# Filter by date range if specified
if start:
df = df[df["timestamp"] >= pd.Timestamp(start)]
if end:
df = df[df["timestamp"] <= pd.Timestamp(end)]
return OHLCV.from_dataframe(df)
def load_parquet(
self,
symbol: str,
timeframe: str,
columns: Optional[List[str]] = None
) -> OHLCV:
"""
โหลดข้อมูลจาก Parquet format - เร็วกว่า CSV 10-50x
เหมาะสำหรับ dataset ขนาดใหญ่ (10M+ rows)
"""
filepath = self.data_dir / f"{symbol}_{timeframe}.parquet"
if not filepath.exists():
raise FileNotFoundError(f"Data file not found: {filepath}")
table = pq.read_table(
filepath,
columns=columns
)
df = table.to_pandas()
return OHLCV.from_dataframe(df)
async def fetch_from_exchange(
self,
symbol: str,
interval: str = "1h",
limit: int = 1000
) -> OHLCV:
"""
ดึงข้อมูลจาก exchange API แบบ async
ใช้ได้กับ Binance, Bybit, OKX เป็นต้น
"""
# Binance klines endpoint example
url = f"https://api.binance.com/api/v3/klines"
params = {
"symbol": symbol.upper(),
"interval": interval,
"limit": limit
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status != 200:
raise ConnectionError(f"Exchange API error: {response.status}")
data = await response.json()
# Convert to DataFrame
df = pd.DataFrame(
data,
columns=[
"timestamp", "open", "high", "low", "close", "volume",
"close_time", "quote_volume", "trades", "taker_buy_base",
"taker_buy_quote", "ignore"
]
)
# Keep only required columns
df = df[["timestamp", "open", "high", "low", "close", "volume"]]
# Convert timestamp from milliseconds
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
return OHLCV.from_dataframe(df)
Performance benchmark
"""
Benchmark: Load 1M rows
────────────────────────────────────────
Format Time (s) Memory (MB)
────────────────────────────────────────
CSV 12.45 280
Parquet 0.82 180
Feather 0.91 185
────────────────────────────────────────
Parquet เร็วกว่า CSV ถึง 15x
"""
Layer 2: Backtesting Engine - Vectorized vs Event-Driven
การเลือก architecture ของ engine ขึ้นกับ use case ถ้าต้องการ speed เลือก vectorized ถ้าต้องการความยืดหยุ่นในการ implement complex logic เลือก event-driven ระบบ production ของผมใช้ทั้งสองแบบ โดย vectorized สำหรับ simple strategies และ event-driven สำหรับ complex ones
import numpy as np
from typing import Callable, Dict, Any, Protocol
from dataclasses import dataclass, field
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class OrderType(Enum):
MARKET = "market"
LIMIT = "limit"
STOP = "stop"
STOP_LIMIT = "stop_limit"
@dataclass
class Order:
order_id: int
timestamp: pd.Timestamp
symbol: str
side: str # "buy" or "sell"
order_type: OrderType
quantity: float
price: Optional[float] = None
filled_price: Optional[float] = None
status: str = "pending"
filled_time: Optional[pd.Timestamp] = None
@property
def is_filled(self) -> bool:
return self.status == "filled"
@dataclass
class Position:
symbol: str
quantity: float = 0.0
entry_price: float = 0.0
unrealized_pnl: float = 0.0
realized_pnl: float = 0.0
@property
def market_value(self) -> float:
return self.quantity * self.entry_price
class SignalGenerator(Protocol):
"""Protocol for strategy signal generation"""
def generate(self, data: pd.DataFrame) -> pd.Series:
"""Return series with values: 1 (buy), -1 (sell), 0 (hold)"""
class VectorizedEngine:
"""
Vectorized backtesting engine - เร็วมากสำหรับ simple strategies
ใช้ NumPy/Pandas operations แทน loop
Performance: ~100K bars/second บน CPU เดียว
"""
def __init__(self, initial_capital: float = 100_000):
self.initial_capital = initial_capital
self.capital = initial_capital
self.position = 0.0
self.trades: list[Dict] = []
self.equity_curve: list[float] = []
def run(
self,
data: pd.DataFrame,
strategy: SignalGenerator,
commission: float = 0.001,
slippage: float = 0.0005
) -> Dict[str, Any]:
"""
Run backtest with vectorized execution
Args:
data: OHLCV data with columns [timestamp, open, high, low, close, volume]
strategy: Signal generator object
commission: Commission rate (0.001 = 0.1%)
slippage: Slippage rate (0.0005 = 0.05%)
"""
# Generate signals
signals = strategy.generate(data)
# Calculate returns
returns = data["close"].pct_change()
# Position size (1 = full position, 0 = no position)
position_size = signals.shift(1).fillna(0)
# Strategy returns (including position changes)
strategy_returns = position_size * returns
# Apply transaction costs when position changes
position_changes = position_size.diff().abs()
transaction_costs = position_changes * (commission + slippage)
strategy_returns -= transaction_costs
# Calculate equity curve
self.equity_curve = [self.initial_capital]
for ret in strategy_returns:
if pd.notna(ret):
new_capital = self.equity_curve[-1] * (1 + ret)
self.equity_curve.append(new_capital)
# Calculate metrics
equity_series = pd.Series(self.equity_curve)
results = {
"initial_capital": self.initial_capital,
"final_capital": self.equity_curve[-1],
"total_return": (self.equity_curve[-1] / self.initial_capital - 1) * 100,
"equity_curve": equity_series,
"max_drawdown": self._calculate_max_drawdown(equity_series),
"sharpe_ratio": self._calculate_sharpe_ratio(strategy_returns),
"trade_count": int(position_changes.sum() / 2),
"win_rate": self._calculate_win_rate(data["close"], signals)
}
return results
def _calculate_max_drawdown(self, equity: pd.Series) -> float:
"""Calculate maximum drawdown percentage"""
peak = equity.expanding(min_periods=1).max()
drawdown = (equity - peak) / peak
return drawdown.min() * 100
def _calculate_sharpe_ratio(
self,
returns: pd.Series,
risk_free_rate: float = 0.02
) -> float:
"""Calculate annualized Sharpe ratio"""
returns = returns.dropna()
if len(returns) == 0:
return 0.0
excess_returns = returns - risk_free_rate / 252 # Daily risk-free rate
return np.sqrt(252) * excess_returns.mean() / excess_returns.std()
def _calculate_win_rate(self, close: pd.Series, signals: pd.Series) -> float:
"""Calculate win rate based on trades"""
position = signals.shift(1).fillna(0)
trades_returns = position.diff().abs() * close.pct_change()
winning_trades = (trades_returns > 0).sum()
total_trades = (trades_returns != 0).sum()
return winning_trades / total_trades if total_trades > 0 else 0.0
class EventDrivenEngine:
"""
Event-driven engine - ยืดหยุ่นกว่า เหมาะสำหรับ complex strategies
รองรับ limit orders, position management, risk controls
Performance: ~10K bars/second บน CPU เดียว
สำหรับ speed ที่ดีขึ้น ใช้ multiprocessing หรือ Cython
"""
def __init__(self, initial_capital: float = 100_000):
self.initial_capital = initial_capital
self.cash = initial_capital
self.positions: Dict[str, Position] = {}
self.orders: list[Order] = []
self.order_id_counter = 0
self.trades: list[Order] = []
self.events: list[Dict] = []
def run(
self,
data: pd.DataFrame,
strategy: Callable[[pd.DataFrame, Dict], int],
commission: float = 0.001,
slippage: float = 0.0005
):
"""
Run backtest with event-driven execution
Args:
data: OHLCV DataFrame
strategy: Function that takes (current_data, portfolio) returns signal
commission: Commission rate
slippage: Slippage rate
"""
portfolio = {
"cash": self.cash,
"positions": self.positions,
"equity": self.cash
}
for i in range(len(data)):
current_bar = data.iloc[:i+1].copy()
current_row = data.iloc[i]
timestamp = current_bar["timestamp"].iloc[-1]
# Generate signal
signal = strategy(current_bar, portfolio)
# Execute orders based on signal
if signal == 1: # Buy signal
self._execute_market_buy(
"BTCUSDT",
current_row["close"],
timestamp,
commission,
slippage
)
elif signal == -1: # Sell signal
self._execute_market_sell(
"BTCUSDT",
current_row["close"],
timestamp,
commission,
slippage
)
# Update portfolio
portfolio["cash"] = self.cash
portfolio["positions"] = self.positions
portfolio["equity"] = self._calculate_equity(current_row["close"])
# Record equity
self.events.append({
"timestamp": timestamp,
"equity": portfolio["equity"],
"cash": self.cash,
"position_value": portfolio["equity"] - self.cash
})
return self._generate_results()
def _execute_market_buy(
self,
symbol: str,
price: float,
timestamp: pd.Timestamp,
commission: float,
slippage: float
):
"""Execute market buy order with slippage"""
effective_price = price * (1 + slippage)
max_quantity = self.cash / effective_price
if max_quantity <= 0:
return
# Calculate actual cost with commission
cost = max_quantity * effective_price * (1 + commission)
if cost > self.cash:
# Not enough cash, buy what we can
max_quantity = self.cash / (effective_price * (1 + commission))
self.cash -= max_quantity * effective_price * (1 + commission)
if symbol in self.positions:
pos = self.positions[symbol]
total_qty = pos.quantity + max_quantity
pos.entry_price = (pos.entry_price * pos.quantity + effective_price * max_quantity) / total_qty
pos.quantity = total_qty
else:
self.positions[symbol] = Position(
symbol=symbol,
quantity=max_quantity,
entry_price=effective_price
)
def _execute_market_sell(
self,
symbol: str,
price: float,
timestamp: pd.Timestamp,
commission: float,
slippage: float
):
"""Execute market sell order"""
if symbol not in self.positions or self.positions[symbol].quantity <= 0:
return
pos = self.positions[symbol]
effective_price = price * (1 - slippage)
proceeds = pos.quantity * effective_price * (1 - commission)
self.cash += proceeds
pos.quantity = 0
pos.realized_pnl += proceeds - (pos.entry_price * pos.quantity)
def _calculate_equity(self, current_price: float) -> float:
"""Calculate total portfolio equity"""
position_value = sum(
p.quantity * current_price for p in self.positions.values()
)
return self.cash + position_value
def _generate_results(self) -> Dict[str, Any]:
"""Generate backtest results summary"""
events_df = pd.DataFrame(self.events)
equity = events_df["equity"]
return {
"initial_capital": self.initial_capital,
"final_capital": equity.iloc[-1] if len(equity) > 0 else self.initial_capital,
"total_return": ((equity.iloc[-1] / self.initial_capital) - 1) * 100 if len(equity) > 0 else 0,
"equity_curve": equity,
"max_drawdown": self._calc_max_dd(equity),
"trade_count": len([o for o in self.orders if o.is_filled]),
"events": events_df
}
def _calc_max_dd(self, equity: pd.Series) -> float:
peak = equity.expanding(min_periods=1).max()
drawdown = (equity - peak) / peak
return drawdown.min() * 100
Layer 3: Risk Management Engine
Risk management คือหัวใจของระบบที่จะอยู่รอดในตลาดจริง ผมใช้ multi-layered risk controls: position sizing, drawdown limits, exposure limits และ correlation-based portfolio limits
import numpy as np
from typing import Optional, Tuple
from dataclasses import dataclass
@dataclass
class RiskConfig:
"""Configuration for risk management parameters"""
max_position_size: float = 0.2 # Maximum 20% of capital per position
max_portfolio_exposure: float = 1.0 # Maximum 100% exposure
max_drawdown_limit: float = 0.15 # Stop trading at 15% drawdown
max_correlation: float = 0.7 # Maximum correlation between positions
var_confidence: float = 0.95 # Value at Risk confidence level
target_risk_per_trade: float = 0.02 # 2% risk per trade
@dataclass
class RiskMetrics:
"""Real-time risk metrics"""
portfolio_value: float
current_drawdown: float
daily_var: float
position_count: int
exposure: float
class RiskEngine:
"""
Production risk management engine
Implements position sizing, drawdown controls, and risk limits
"""
def __init__(self, config: RiskConfig, initial_capital: float):
self.config = config
self.initial_capital = initial_capital
self.peak_capital = initial_capital
self.trading_paused = False
self.drawdown_history: list[float] = []
def check_entry(
self,
symbol: str,
quantity: float,
price: float,
current_equity: float,
existing_positions: dict
) -> Tuple[bool, Optional[float], str]:
"""
Check if new position passes risk controls
Returns:
(approved, adjusted_quantity, reason)
"""
# Check 1: Drawdown limit
if self.trading_paused:
return False, None, "Trading paused due to drawdown limit"
# Calculate current drawdown
self.peak_capital = max(self.peak_capital, current_equity)
current_dd = (current_equity - self.peak_capital) / self.peak_capital
if current_dd < -self.config.max_drawdown_limit:
self.trading_paused = True
return False, None, f"Drawdown {current_dd*100:.2f}% exceeds limit"
# Check 2: Position size limit
position_value = quantity * price
position_ratio = position_value / current_equity
if position_ratio > self.config.max_position_size:
adjusted_qty = (current_equity * self.config.max_position_size) / price
return True, adjusted_qty, f"Position size reduced to {self.config.max_position_size*100}%"
# Check 3: Total exposure limit
total_exposure = sum(
pos.quantity * pos.entry_price
for pos in existing_positions.values()
)
if (total_exposure + position_value) / current_equity > self.config.max_portfolio_exposure:
max_new_exposure = (current_equity * self.config.max_portfolio_exposure) - total_exposure
if max_new_exposure <= 0:
return False, None, "Maximum portfolio exposure reached"
adjusted_qty = max_new_exposure / price
return True, adjusted_qty, "Exposure limit reached"
return True, quantity, "Approved"
def calculate_position_size(
self,
entry_price: float,
stop_loss_price: float,
current_equity: float,
risk_type: str = "fixed_fraction"
) -> float:
"""
Calculate optimal position size based on risk model
Methods:
- fixed_fraction: Fixed percentage of equity
- kelly_criterion: Based on win rate and avg win/loss
- ATR_based: Based on average true range volatility
"""
if risk_type == "fixed_fraction":
risk_amount = current_equity * self.config.target_risk_per_trade
risk_per_unit = abs(entry_price - stop_loss_price)
return risk_amount / risk_per_unit
elif risk_type == "kelly_criterion":
# Requires historical trade statistics
# Simplified Kelly: f = (bp - q) / b
# where b = odds, p = win probability, q = 1-p
# This should be calculated from historical trades
return current_equity * self.config.target_risk_per_trade / abs(entry_price - stop_loss_price)
elif risk_type == "ATR_based":
# ATR-based position sizing
# position = (equity * risk%) / (ATR * multiplier)
# Typically use 2-3x ATR as stop distance
atr_multiplier = 2.0
atr = self._calculate_atr(entry_price * 0.01) # Simplified ATR
risk_amount = current_equity * self.config.target_risk_per_trade
return risk_amount / (atr * atr_multiplier)
return current_equity * self.config.target_risk_per_trade / abs(entry_price - stop_loss_price)
def _calculate_atr(self, atr_value: float) -> float:
"""Calculate Average True Range"""
return atr_value
def get_current_risk_metrics(
self,
current_equity: float,
positions: dict,
current_prices: dict
) -> RiskMetrics:
"""Get current risk metrics for monitoring"""
self.peak_capital = max(self.peak_capital, current_equity)
current_dd = (current_equity - self.peak_capital) / self.peak_capital
# Calculate exposure
total_exposure = sum(
pos.quantity * current_prices.get(pos.symbol, pos.entry_price)
for pos in positions.values()
)
exposure_ratio = total_exposure / current_equity if current_equity > 0 else 0
# Simplified VaR calculation
daily_var = current_equity * 0.02 * (1 - self.config.var_confidence)
self.drawdown_history.append(current_dd)
return RiskMetrics(
portfolio_value=current_equity,
current_drawdown=current_dd * 100,
daily_var=daily_var,
position_count=len(positions),
exposure=exposure_ratio
)
Example usage with a real strategy
class RiskAdjustedStrategy:
"""Example strategy with integrated risk management"""
def __init__(self, initial_capital: float = 100_000):
self.capital = initial_capital
self.risk_config = RiskConfig(
max_position_size=0.1, # 10% max per trade
max_drawdown_limit=0.2, # 20% max drawdown
target_risk_per_trade=0.02 # 2% risk per trade
)
self.risk_engine = RiskEngine(self.risk_config, initial_capital)
def execute_trade(
self,
symbol: str,
signal: int, # 1 = buy, -1 = sell, 0 = hold
price: float,
stop_loss: float,
quantity: float = None
):
"""Execute trade with risk checks"""
if signal == 0:
return None
# Calculate position size if not specified
if quantity is None and signal == 1:
quantity = self.risk_engine.calculate_position_size(
entry_price=price,
stop_loss_price=stop_loss,
current_equity=self.capital
)
# Get risk approval
approved, adjusted_qty, reason = self.risk_engine.check_entry(
symbol=symbol,
quantity=quantity,
price=price,
current_equity=self.capital,
existing_positions={} # Would pass actual positions in production
)
if approved:
print(f"Trade approved: {reason}")
if adjusted_qty and adjusted_qty < quantity:
print(f"Quantity adjusted: {quantity:.4f} -> {adjusted_qty:.4f}")
quantity = adjusted_qty
return quantity
else:
print(f"Trade rejected: {reason}")
return None
Layer 4: Integration กับ AI - ใช้ HolySheep วิเคราะห์ Strategy
หลังจาก run backtest ได้ผลลัพธ์แล้ว สิ่งสำคัญคือต้องวิเคราะห์ว่าทำไม strategy ถึงทำงานได้ดีในบางช่วงและไม่ดีในบางช่วง ผมใช้ HolySheep AI ที่มี <50ms latency สำหรับการวิเคราะห์ผลลัพธ์และ generate insights โดยใช้ DeepSeek V3.2 ซึ่งราคาเพียง $0.42/MTok
import aiohttp
import asyncio
import json
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from datetime import datetime
@dataclass
class AIAnalysisRequest:
"""Request structure for AI analysis"""
backtest_results: Dict[str, Any]
strategy_description: str
market_conditions: str
custom_questions: List[str]
@dataclass
class AIAnalysisResponse:
"""Response structure from AI analysis"""
summary: str
strengths: List[str]
weaknesses: List[str]
recommendations: List[str]
risk_assessment: str
confidence_score: float
class HolySheepAIClient:
"""
Client สำหรับเชื่อมต่อกับ HolySheep AI API
ใช้สำหรับวิเคราะห์ผลลัพธ์ backtest และ generate insights
Documentation: https://docs.holysheep.ai
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def analyze_backtest_results(
self,
request: AIAnalysisRequest
) -> AIAnalysisResponse:
"""
วิเคราะห์ผลลัพธ์ backtest ด้วย AI
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง