บทนำ: ทำไม Mean Reversion ถึงเหมาะกับ Crypto

สกุลเงินดิจิทัลมีความผันผวนสูงกว่าสินทรัพย์ทั่วไปถึง 5-10 เท่า ทำให้กลยุทธ์ Mean Reversion มีประสิทธิภาพเป็นพิเศษในตลาดนี้ เมื่อราคาเบี่ยงเบนจากค่าเฉลี่ยมาก็ มีแนวโน้มว่าจะกลับมาสู่ระดับปกติเร็วกว่าตลาดหุ้น ในบทความนี้ เราจะสำรวจ Tardis ซึ่งเป็นเครื่องมือ backtesting ระดับ production ที่รองรับข้อมูลราคาคริปโตคุณภาพสูงจาก exchange หลายร้อยแห่ง เราจะเจาะลึกถึงสถาปัตยกรรม การปรับแต่งประสิทธิภาพ และโค้ดที่พร้อมใช้งานจริงใน production

Tardis คืออะไร และทำไมถึงเป็นมาตรฐานอุตสาหกรรม

Tardis เป็น time-series database ที่ออกแบบมาเพื่อรองรับข้อมูล market data ปริมาณมหาศาล โดยมีคุณสมบัติเด่นดังนี้

การติดตั้งและ Configuration

# ติดตั้ง Tardis client
pip install tardis-dev


สำหรับ Python backtesting framework

pip install pandas numpy


สร้าง Docker compose สำหรับ Tardis local development

cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
  tardis:
    image: ghcr.io/tardis-dev/tardis:latest
    ports:
      - "27199:27199"
    environment:
      - TARDIS_CONFIG=/app/config.yml
    volumes:
      - ./config.yml:/app/config.yml
      - tardis_data:/data

volumes:
  tardis_data:
EOF
# config.yml - กำหนดค่า exchanges และ data retention
exchanges:
  - name: binance
    channels:
      - trades
      - candles
    symbols:
      - BTCUSDT
      - ETHUSDT
      - SOLUSDT
    intervals:
      - 1m
      - 5m
      - 1h

data_retention:
  trades: 90d
  candles: 730d

performance:
  cache_size_mb: 4096
  max_connections: 100

สถาปัตยกรรม Mean Reversion Strategy

กลยุทธ์ Mean Reversion ที่เราจะ implement อาศัยหลักการว่าราคาจะกลับมาที่ค่าเฉลี่ยเคลื่อนที่ (Moving Average) เมื่อเบี่ยงเบนเกินกว่า threshold ที่กำหนด 
import pandas as pd
import numpy as np
from tardis_client import TardisClient, Playground
from datetime import datetime, timedelta
import asyncio

class MeanReversionStrategy:
    def __init__(
        self,
        symbol: str,
        lookback_period: int = 20,
        std_multiplier: float = 2.0,
        rsi_oversold: int = 30,
        rsi_overbought: int = 70
    ):
        self.symbol = symbol
        self.lookback_period = lookback_period
        self.std_multiplier = std_multiplier
        self.rsi_oversold = rsi_oversold
        self.rsi_overbought = rsi_overbought
        
        self.position = 0
        self.entry_price = 0
        self.trades = []
        
    def calculate_bollinger_bands(self, df: pd.DataFrame) -> pd.DataFrame:
        """คำนวณ Bollinger Bands พร้อม bandwidth และ %B"""
        df['SMA'] = df['close'].rolling(window=self.lookback_period).mean()
        df['STD'] = df['close'].rolling(window=self.lookback_period).std()
        df['Upper'] = df['SMA'] + (self.std_multiplier * df['STD'])
        df['Lower'] = df['SMA'] - (self.std_multiplier * df['STD'])
        df['Bandwidth'] = (df['Upper'] - df['Lower']) / df['SMA']
        df['PercentB'] = (df['close'] - df['Lower']) / (df['Upper'] - df['Lower'])
        return df
    
    def calculate_rsi(self, df: pd.DataFrame, period: int = 14) -> pd.DataFrame:
        """คำนวณ RSI สำหรับ confirmation"""
        delta = df['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
        rs = gain / loss
        df['RSI'] = 100 - (100 / (1 + rs))
        return df
    
    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """สร้างสัญญาณ long/short/hold"""
        df = self.calculate_bollinger_bands(df)
        df = self.calculate_rsi(df)
        
        # Long signal: ราคาต่ำกว่า Lower Band + RSI oversold
        df['LongSignal'] = (
            (df['close'] < df['Lower']) & 
            (df['RSI'] < self.rsi_oversold)
        )
        
        # Short signal: ราคาสูงกว่า Upper Band + RSI overbought
        df['ShortSignal'] = (
            (df['close'] > df['Upper']) & 
            (df['RSI'] > self.rsi_overbought)
        )
        
        # Exit signals
        df['ExitLong'] = df['close'] > df['SMA']
        df['ExitShort'] = df['close'] < df['SMA']
        
        return df
    
    def execute_backtest(self, df: pd.DataFrame, initial_capital: float = 100000) -> dict:
        """รัน backtest และคืนค่าผลลัพธ์"""
        df = self.generate_signals(df)
        capital = initial_capital
        position = 0
        equity_curve = []
        
        for idx, row in df.iterrows():
            # Long entry
            if row['LongSignal'] and position == 0:
                position = capital / row['close']
                entry_price = row['close']
                capital = 0
                
            # Short entry
            elif row['ShortSignal'] and position == 0:
                position = -capital / row['close']
                entry_price = row['close']
                capital = 0
                
            # Exit long
            elif row['ExitLong'] and position > 0:
                capital = position * row['close']
                pnl = capital - initial_capital
                self.trades.append({'type': 'LONG', 'pnl': pnl, 'entry': entry_price, 'exit': row['close']})
                position = 0
                
            # Exit short
            elif row['ExitShort'] and position < 0:
                capital = abs(position) * (2 * entry_price - row['close'])
                pnl = capital - initial_capital
                self.trades.append({'type': 'SHORT', 'pnl': pnl, 'entry': entry_price, 'exit': row['close']})
                position = 0
            
            equity = capital + abs(position) * row['close'] if position != 0 else capital
            equity_curve.append({'timestamp': idx, 'equity': equity})
        
        return {
            'final_capital': capital + abs(position) * df['close'].iloc[-1] if position != 0 else capital,
            'trades': self.trades,
            'equity_curve': pd.DataFrame(equity_curve),
            'total_trades': len(self.trades)
        }

การเชื่อมต่อกับ Tardis API และดึงข้อมูล

import asyncio
from tardis_client import TardisClient, Playground
from datetime import datetime

async def fetch_candles_tardis(
    exchange: str,
    symbol: str,
    start_date: datetime,
    end_date: datetime,
    interval: str = "1h"
):
    """
    ดึงข้อมูล candle จาก Tardis API
    สำหรับ production แนะนำใช้ HolySheep AI 
    สำหรับการ parse และ clean ข้อมูลอัตโนมัติ
    """
    client = TardisClient(playground=False)
    
    # แปลง interval string เป็น ISO format
    interval_map = {
        "1m": "1T", "5m": "5T", "15m": "15T",
        "1h": "1H", "4h": "4H", "1d": "1D"
    }
    
    candles = []
    
    # Replay mode สำหรับ historical data
    async for capsule in client.replay(
        exchange=exchange,
        filters=[{
            "type": "candles",
            "symbol": symbol,
            "interval": interval_map.get(interval, "1H")
        }],
        from_timestamp=start_date,
        to_timestamp=end_date
    ):
        if capsule.type == "candle":
            candles.append({
                'timestamp': capsule.timestamp,
                'open': capsule.candle.open,
                'high': capsule.candle.high,
                'low': capsule.candle.low,
                'close': capsule.candle.close,
                'volume': capsule.candle.volume
            })
    
    return pd.DataFrame(candles)


ตัวอย่างการใช้งาน

async def main():
    start = datetime(2024, 1, 1)
    end = datetime(2024, 12, 31)
    
    df = await fetch_candles_tardis(
        exchange="binance",
        symbol="BTCUSDT",
        start_date=start,
        end_date=end,
        interval="1h"
    )
    
    # Initialize strategy
    strategy = MeanReversionStrategy(
        symbol="BTCUSDT",
        lookback_period=20,
        std_multiplier=2.0,
        rsi_oversold=30,
        rsi_overbought=70
    )
    
    # Run backtest
    results = strategy.execute_backtest(df, initial_capital=100000)
    print(f"Final Capital: ${results['final_capital']:,.2f}")
    print(f"Total Trades: {results['total_trades']}")

if __name__ == "__main__":
    asyncio.run(main())

การวัดผลและ Performance Metrics

import matplotlib.pyplot as plt
from scipy import stats

def calculate_performance_metrics(trades: list, equity_curve: pd.DataFrame, initial_capital: float) -> dict:
    """คำนวณ performance metrics ครบถ้วนตามมาตรฐาน industry"""
    
    equity = equity_curve['equity'].values
    returns = np.diff(equity) / equity[:-1]
    
    # Basic metrics
    total_return = (equity[-1] - initial_capital) / initial_capital * 100
    num_trades = len(trades)
    win_rate = len([t for t in trades if t['pnl'] > 0]) / num_trades * 100 if num_trades > 0 else 0
    
    # Risk metrics
    sharpe_ratio = np.mean(returns) / np.std(returns) * np.sqrt(252 * 24) if np.std(returns) > 0 else 0
    max_drawdown = np.max(np.maximum.accumulate(equity) - equity) / initial_capital * 100
    sortino_ratio = (np.mean(returns) / np.std(returns[returns < 0]) * np.sqrt(252 * 24) 
                     if len(returns[returns < 0]) > 0 else 0)
    
    # Trade analysis
    pnls = [t['pnl'] for t in trades]
    profit_factor = abs(sum([p for p in pnls if p > 0]) / sum([p for p in pnls if p < 0])) if sum([p for p in pnls if p < 0]) != 0 else 0
    avg_win = np.mean([p for p in pnls if p > 0]) if [p for p in pnls if p > 0] else 0
    avg_loss = np.mean([p for p in pnls if p < 0]) if [p for p in pnls if p < 0] else 0
    
    return {
        'total_return': f"{total_return:.2f}%",
        'num_trades': num_trades,
        'win_rate': f"{win_rate:.2f}%",
        'sharpe_ratio': f"{sharpe_ratio:.2f}",
        'max_drawdown': f"{max_drawdown:.2f}%",
        'sortino_ratio': f"{sortino_ratio:.2f}",
        'profit_factor': f"{profit_factor:.2f}",
        'avg_win': f"${avg_win:,.2f}",
        'avg_loss': f"${avg_loss:,.2f}",
        'expectancy': f"${np.mean(pnls):,.2f}" if pnls else "$0.00"
    }

def plot_equity_curve(equity_curve: pd.DataFrame, title: str = "Equity Curve"):
    """Plot equity curve พร้อม drawdown"""
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), sharex=True)
    
    # Equity curve
    ax1.plot(equity_curve['timestamp'], equity_curve['equity'], 'b-', linewidth=1.5)
    ax1.set_ylabel('Equity ($)')
    ax1.set_title(title)
    ax1.grid(True, alpha=0.3)
    ax1.fill_between(equity_curve['timestamp'], equity_curve['equity'], alpha=0.3)
    
    # Drawdown
    rolling_max = equity_curve['equity'].cummax()
    drawdown = (equity_curve['equity'] - rolling_max) / rolling_max * 100
    ax2.fill_between(equity_curve['timestamp'], drawdown, 0, color='red', alpha=0.5)
    ax2.set_ylabel('Drawdown (%)')
    ax2.set_xlabel('Date')
    ax2.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.savefig('equity_curve.png', dpi=150)
    plt.show()

Advanced: Multi-Timeframe Strategy

การใช้หลาย timeframe ช่วยเพิ่มความแม่นยำของสัญญาณ โดย timeframe ที่ใหญ่กว่าใช้สำหรับ trend direction และ timeframe เล็กกว่าใช้สำหรับ entry timing 
class MultiTimeframeStrategy:
    def __init__(self, strategy_config: dict):
        self.config = strategy_config
        
    async def fetch_multi_timeframe_data(
        self,
        symbol: str,
        start: datetime,
        end: datetime
    ) -> dict:
        """ดึงข้อมูลหลาย timeframe พร้อมกัน"""
        tasks = [
            fetch_candles_tardis("binance", symbol, start, end, interval)
            for interval in ['4h', '1h', '15m']
        ]
        results = await asyncio.gather(*tasks)
        return {
            '4h': results[0],
            '1h': results[1],
            '15m': results[2]
        }
    
    def calculate_trend_direction(self, df_4h: pd.DataFrame) -> str:
        """กำหนด trend direction จาก 4H timeframe"""
        sma_20 = df_4h['close'].rolling(20).mean().iloc[-1]
        sma_50 = df_4h['close'].rolling(50).mean().iloc[-1]
        current_price = df_4h['close'].iloc[-1]
        
        if sma_20 > sma_50 and current_price > sma_20:
            return "BULLISH"
        elif sma_20 < sma_50 and current_price < sma_20:
            return "BEARISH"
        return "NEUTRAL"
    
    def generate_entry(self, df_4h: pd.DataFrame, df_15m: pd.DataFrame, trend: str) -> dict:
        """สร้างสัญญาณ entry จาก 15m พร้อม trend confirmation"""
        signal = None
        
        # 15m Mean Reversion signals
        lower_band = df_15m['close'].rolling(20).mean() - 2 * df_15m['close'].rolling(20).std()
        upper_band = df_15m['close'].rolling(20).mean() + 2 * df_15m['close'].rolling(20).std()
        
        if trend == "BULLISH" and df_15m['close'].iloc[-1] < lower_band.iloc[-1]:
            signal = {'action': 'LONG', 'confidence': 'HIGH', 'reason': 'Mean reversion + Uptrend'}
        elif trend == "BEARISH" and df_15m['close'].iloc[-1] > upper_band.iloc[-1]:
            signal = {'action': 'SHORT', 'confidence': 'HIGH', 'reason': 'Mean reversion + Downtrend'}
            
        return signal

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. Look-Ahead Bias ในการคำนวณ Indicator

ปัญหา: การใช้ข้อมูลในอนาคตร่วมคำนวณ (data leakage) ทำให้ผล backtest ดีกว่าความเป็นจริงอย่างมาก วิธีแก้: ใช้ only ข้อมูลที่ available ณ เวลาที่ทำการตัดสินใจ โดย shift indicator กลับไป 1 period 
# วิธีแก้ไข - ใช้ .shift(1) เพื่อป้องกัน look-ahead bias
def calculate_indicators_correct(df: pd.DataFrame) -> pd.DataFrame:
    """
    คำนวณ indicator อย่างถูกต้องโดยไม่ใช้ future data
    """
    # สำหรับ entry signal ใช้ indicator จาก bar ก่อนหน้า
    df['SMA_safe'] = df['close'].rolling(window=20).mean().shift(1)
    df['Upper_safe'] = df['SMA_safe'] + 2 * df['close'].rolling(window=20).std().shift(1)
    df['Lower_safe'] = df['SMA_safe'] - 2 * df['close'].rolling(window=20).std().shift(1)
    
    # ตรวจสอบว่าใช้ index ถูกต้อง
    assert df['SMA_safe'].isna().sum() >= 19, "Still have look-ahead bias!"
    
    return df

2. Survivorship Bias ในข้อมูล Historical

ปัญหา: Backtest กับเฉพาะเหรียญที่ยังอยู่ ทำให้พลาดเหรียญที่ล้มเลิก (delisted) ซึ่งมักมีผลตอบแทนแย่ วิธีแก้: รวมเหรียญที่ delisted ใน dataset หรือใช้ filter วันที่เหมาะสม 
# วิธีแก้ไข - ดึงข้อมูล historical แบบ include delisted
async def fetch_with_survivorship_bias_handling(
    exchange: str,
    symbols: list,
    start: datetime,
    end: datetime
) -> pd.DataFrame:
    """
    ดึงข้อมูลพร้อม handle survivorship bias
    โดยดึง symbols ที่มีอยู่ ณ ช่วงเวลานั้นๆ
    """
    # ดึง list ของ symbols ที่มีอยู่จริงในแต่ละช่วงเวลา
    # จาก exchange API
    
    all_candles = []
    
    # ดึงข้อมูลทุกเหรียญที่มีอยู่ในช่วงเวลานั้น
    historical_symbols = await get_historical_symbols(exchange, start, end)
    
    for symbol in historical_symbols:
        try:
            df = await fetch_candles_tardis(exchange, symbol, start, end)
            if len(df) > 0:
                df['symbol'] = symbol
                all_candles.append(df)
        except Exception as e:
            # เหรียญที่ delisted อาจมี error ให้ skip
            continue
    
    return pd.concat(all_candles, ignore_index=True) if all_candles else pd.DataFrame()

3. Transaction Costs และ Slippage ที่ไม่สมจริง

ปัญหา: Backtest ไม่รวมค่าธรรมเนียม หรือใช้ค่าธรรมเนียมต่ำเกินไป ทำให้ดูว่ากลยุทธ์ทำกำไรได้ทั้งที่จริงไม่ได้ วิธีแก้: รวมค่าธรรมเนียมแบบ conservative และ slippage model 
# วิธีแก้ไข - ใช้ cost model แบบ conservative
class RealisticBacktester:
    def __init__(self, 
                 maker_fee: float = 0.001,  # 0.1%
                 taker_fee: float = 0.002,  # 0.2%
                 slippage_bps: float = 5,   # 5 basis points
                 funding_rate: float = 0.0001):  # สำหรับ futures
        self.maker_fee = maker_fee
        self.taker_fee = taker_fee
        self.slippage_bps = slippage_bps / 10000  # แปลงเป็น percentage
        self.funding_rate = funding_rate
        
    def calculate_trade_cost(self, price: float, size: float, side: str) -> float:
        """คำนวณ cost ทั้งหมดของ trade"""
        # Slippage
        slippage_cost = price * self.slippage_bps * (1 if side == 'BUY' else -1)
        execution_price = price + slippage_cost
        
        # Commission (round trip = 2x)
        commission = execution_price * size * self.taker_fee * 2
        
        # Total cost
        total_cost = abs(execution_price - price) * size + commission
        
        return total_cost
    
    def apply_costs(self, trades: list) -> list:
        """ปรับ PnL ของ trades ทุกรายการหลังหักค่าธรรมเนียม"""
        adjusted_trades = []
        for trade in trades:
            entry_cost = self.calculate_trade_cost(
                trade['entry'], trade['size'], 'BUY'
            )
            exit_cost = self.calculate_trade_cost(
                trade['exit'], trade['size'], 'SELL'
            )
            trade['net_pnl'] = trade['pnl'] - entry_cost - exit_cost
            adjusted_trades.append(trade)
        return adjusted_trades

Best Practices สำหรับ Production Deployment

สรุปและแนวทางต่อไป

การสร้าง mean reversion strategy ที่ทำงานได้จริงในตลาด crypto ต้องอาศัยข้อมูลคุณภาพสูงจาก Tardis หรือแหล่งข้อมูลที่เชื่อถือได้ ร่วมกับการ implement ที่รอบคอบเพื่อหลีกเลี่ยง common pitfalls เช่น look-ahead bias และ survivorship bias สิ่งสำคัญคือต้องใช้ cost model ที่สมจริงและทำ paper trading ก่อน deploy จริง สำหรับการ optimize strategy parameters อัตโนมัติด้วย AI หรือการ parse ข้อมูล market ขนาดใหญ่ ลองพิจารณาใช้ HolySheep AI ซึ่งให้บริการ API คุณภาพสูงในราคาที่ประหยัดกว่า 85% เมื่อเทียบกับ OpenAI โดยรองรับ GPT-4.1, Claude Sonnet 4.5 และ Gemini 2.5 Flash 👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง