บทนำ: ทำไมต้อง Clean Data ก่อน Backtest

การพัฒนาระบบเทรดแบบ Quantitative ต้องอาศัยข้อมูลที่สะอาดและถูกต้อง หาก Data Quality ไม่ดี ผล Backtest จะไม่น่าเชื่อถือ และอาจทำให้สูญเสียเงินจริงในการเทรด จากประสบการณ์ของผู้เขียนในการสร้าง Pipeline สำหรับ Crypto Quant มากว่า 3 ปี พบว่า ข้อมูล Raw จาก Exchange มักมีปัญหาหลายอย่าง เช่น: บทความนี้จะสอนขั้นตอนการ Clean Data จาก Tardis.dev (แพลตฟอร์มรวบรวม Historical Data ของ Exchange) ให้พร้อมใช้งานสำหรับ Backtesting

ขั้นตอนที่ 1: ดาวน์โหลดข้อมูลจาก Tardis.dev

Tardis.dev ให้บริการ Historical Market Data ของ Exchange หลายตัว รวมถึง Bybit โดยมีราคาเริ่มต้นที่ $9/เดือนสำหรับแพลนเริ่มต้น ข้อมูลที่ได้จะอยู่ในรูปแบบ CSV หรือ JSON สำหรับการดาวน์โหลด ให้ไปที่ tardis.dev แล้วเลือก Bybit → BTCUSDT → Trades → กำหนดช่วงเวลาที่ต้องการ หลังจากได้ไฟล์ CSV มาแล้ว โครงสร้างข้อมูลจะมีลักษณะดังนี้: 
timestamp,side,price,amount,trade_id
2024-03-15T08:30:00.123456Z,buy,67845.50,0.01234,123456789
2024-03-15T08:30:00.234567Z,sell,67846.00,0.00567,123456790
2024-03-15T08:30:00.345678Z,buy,67844.75,0.00123,123456791

ขั้นตอนที่ 2: สร้าง Python Script สำหรับ Data Cleaning

import pandas as pd
import numpy as np
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

class BybitDataCleaner:
    """
    Data Cleaning Pipeline สำหรับ Bybit BTCUSDT Trades
    ออกแบบมาเพื่อเตรียมข้อมูลสำหรับ Quantitative Backtesting
    """
    
    def __init__(self, filepath: str):
        self.filepath = filepath
        self.df = None
        self.cleaning_report = {}
    
    def load_data(self) -> pd.DataFrame:
        """โหลดข้อมูลจาก CSV file"""
        print(f"📂 กำลังโหลดข้อมูลจาก: {self.filepath}")
        
        # อ่านไฟล์ CSV
        self.df = pd.read_csv(
            self.filepath,
            parse_dates=['timestamp'],
            dtype={
                'side': str,
                'price': float,
                'amount': float,
                'trade_id': str
            }
        )
        
        original_rows = len(self.df)
        print(f"✅ โหลดสำเร็จ: {original_rows:,} rows")
        print(f"📅 ช่วงเวลา: {self.df['timestamp'].min()} ถึง {self.df['timestamp'].max()}")
        
        return self.df
    
    def clean_timestamp(self, timezone: str = 'UTC') -> pd.DataFrame:
        """จัดการ Timestamp ให้ถูกต้อง"""
        print("\n🔧 กำลัง Clean Timestamp...")
        
        # แปลง timezone
        self.df['timestamp'] = pd.to_datetime(self.df['timestamp'], utc=True)
        if timezone != 'UTC':
            self.df['timestamp'] = self.df['timestamp'].dt.tz_convert(timezone)
        
        # จัดเรียงข้อมูลตาม timestamp
        self.df = self.df.sort_values('timestamp').reset_index(drop=True)
        
        # ตรวจสอบ timestamp ที่ผิดปกติ
        time_diffs = self.df['timestamp'].diff()
        abnormal_gaps = time_diffs[time_diffs > pd.Timedelta(hours=1)]
        
        self.cleaning_report['abnormal_time_gaps'] = len(abnormal_gaps)
        print(f"   - พบช่วงเวลาที่ห่างผิดปกติ: {len(abnormal_gaps)} จุด")
        
        return self.df
    
    def remove_duplicates(self) -> pd.DataFrame:
        """ลบข้อมูลซ้ำซ้อน"""
        print("\n🔧 กำลังลบข้อมูลซ้ำซ้อน...")
        
        before_count = len(self.df)
        
        # ลบตาม trade_id
        self.df = self.df.drop_duplicates(subset=['trade_id'], keep='first')
        
        # ลบ rows ที่ซ้ำกันทุก column
        self.df = self.df.drop_duplicates(keep='first')
        
        removed = before_count - len(self.df)
        self.cleaning_report['duplicates_removed'] = removed
        print(f"   - ลบข้อมูลซ้ำ: {removed} rows")
        
        return self.df
    
    def clean_prices(self, max_deviation: float = 0.05) -> pd.DataFrame:
        """
        ลบราคาที่ผิดปกติ (Outliers)
        max_deviation: % สูงสุดที่ยอมรับได้จาก Median
        """
        print("\n🔧 กำลัง Clean ราคา...")
        
        before_count = len(self.df)
        
        # คำนวณ Median และ Standard Deviation
        median_price = self.df['price'].median()
        std_price = self.df['price'].std()
        
        # กำหนดขอบเขตที่ยอมรับได้
        lower_bound = median_price * (1 - max_deviation)
        upper_bound = median_price * (1 + max_deviation)
        
        # Filter outliers
        mask = (self.df['price'] >= lower_bound) & (self.df['price'] <= upper_bound)
        outliers_count = (~mask).sum()
        
        self.df = self.df[mask].copy()
        
        self.cleaning_report['price_outliers_removed'] = outliers_count
        print(f"   - ราคา Median: ${median_price:,.2f}")
        print(f"   - ลบ Outliers: {outliers_count} rows (deviation > {max_deviation*100}%)")
        
        return self.df
    
    def clean_volume(self) -> pd.DataFrame:
        """ลบ Volume ที่ผิดปกติ"""
        print("\n🔧 กำลัง Clean Volume...")
        
        before_count = len(self.df)
        
        # ลบ rows ที่มี Volume <= 0 หรือ NaN
        self.df = self.df[
            (self.df['amount'] > 0) & 
            (self.df['amount'].notna())
        ].copy()
        
        removed = before_count - len(self.df)
        self.cleaning_report['invalid_volume_removed'] = removed
        print(f"   - ลบ Volume ที่ไม่ถูกต้อง: {removed} rows")
        
        return self.df
    
    def clean_side(self) -> pd.DataFrame:
        """มาตรฐานรูปแบบ Side (buy/sell)"""
        print("\n🔧 กำลัง Clean Side field...")
        
        # แปลงให้เป็น lowercase
        self.df['side'] = self.df['side'].str.lower().str.strip()
        
        # กรองเฉพาะ buy และ sell
        valid_sides = ['buy', 'sell']
        invalid_count = (~self.df['side'].isin(valid_sides)).sum()
        
        self.df = self.df[self.df['side'].isin(valid_sides)].copy()
        
        self.cleaning_report['invalid_side_removed'] = invalid_count
        print(f"   - ลบ Side ที่ไม่ถูกต้อง: {invalid_count} rows")
        
        return self.df
    
    def handle_missing_data(self, strategy: str = 'drop') -> pd.DataFrame:
        """
        จัดการ Missing Data
        strategy: 'drop' หรือ 'forward_fill'
        """
        print(f"\n🔧 กำลัง Handle Missing Data (strategy: {strategy})...")
        
        missing_before = self.df.isnull().sum().sum()
        
        if strategy == 'drop':
            self.df = self.df.dropna()
        elif strategy == 'forward_fill':
            self.df = self.df.fillna(method='ffill')
        
        self.cleaning_report['missing_handled'] = missing_before
        print(f"   - จัดการ Missing Data: {missing_before} cells")
        
        return self.df
    
    def add_features(self) -> pd.DataFrame:
        """เพิ่ม Features ที่มีประโยชน์สำหรับ Backtesting"""
        print("\n🔧 กำลังเพิ่ม Features...")
        
        # คำนวณ Volume ในรูปแบบ USDT
        self.df['volume_usdt'] = self.df['price'] * self.df['amount']
        
        # คำนวณ Timestamp เป็น Unix
        self.df['timestamp_unix'] = self.df['timestamp'].astype(np.int64) // 10**9
        
        # สร้าง OHLCV bars (สมมติ timeframe 1 นาที)
        self.df.set_index('timestamp', inplace=True)
        
        ohlcv = self.df.resample('1T').agg({
            'price': ['first', 'max', 'min', 'last'],
            'amount': 'sum',
            'volume_usdt': 'sum'
        })
        ohlcv.columns = ['open', 'high', 'low', 'close', 'volume', 'volume_usdt']
        ohlcv = ohlcv.dropna()
        
        print(f"   - สร้าง OHLCV 1-minute bars: {len(ohlcv):,} bars")
        
        return ohlcv.reset_index()
    
    def run_pipeline(self, timezone: str = 'UTC') -> tuple:
        """รัน Data Cleaning Pipeline ทั้งหมด"""
        print("=" * 60)
        print("🚀 เริ่ม Data Cleaning Pipeline")
        print("=" * 60)
        
        # รันทุกขั้นตอน
        self.load_data()
        self.clean_timestamp(timezone)
        self.remove_duplicates()
        self.clean_side()
        self.clean_volume()
        self.clean_prices()
        self.handle_missing_data()
        ohlcv = self.add_features()
        
        # สรุปผล
        print("\n" + "=" * 60)
        print("📊 Data Cleaning Report")
        print("=" * 60)
        for key, value in self.cleaning_report.items():
            print(f"   {key}: {value}")
        
        return ohlcv, self.cleaning_report



วิธีใช้งาน

if __name__ == "__main__":
    cleaner = BybitDataCleaner("bybit_btcusdt_trades.csv")
    ohlcv, report = cleaner.run_pipeline(timezone='Asia/Bangkok')
    
    # บันทึกผลลัพธ์
    ohlcv.to_csv("cleaned_btcusdt_ohlcv_1m.csv", index=False)
    print("\n✅ บันทึกไฟล์: cleaned_btcusdt_ohlcv_1m.csv")

ขั้นตอนที่ 3: สร้าง Backtest Framework

หลังจากได้ข้อมูลที่สะอาดแล้ว ต่อไปจะเป็นการสร้าง Simple Backtesting Framework ที่ใช้งานได้จริง: 
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple, Optional

class SimpleBacktester:
    """
    Simple Backtesting Engine สำหรับ Crypto Trading Strategies
    รองรับ: SMA Crossover, RSI, Bollinger Bands
    """
    
    def __init__(self, data: pd.DataFrame, initial_capital: float = 10000):
        self.data = data.copy()
        self.initial_capital = initial_capital
        self.capital = initial_capital
        self.position = 0  # จำนวน BTC ที่ถือ
        self.trades = []
        self.equity_curve = []
        
    def add_sma(self, short_period: int = 10, long_period: int = 50) -> None:
        """เพิ่ม Simple Moving Averages"""
        self.data[f'sma_{short_period}'] = self.data['close'].rolling(window=short_period).mean()
        self.data[f'sma_{long_period}'] = self.data['close'].rolling(window=long_period).mean()
        
    def add_rsi(self, period: int = 14) -> None:
        """เพิ่ม RSI Indicator"""
        delta = self.data['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
        self.data['rsi'] = 100 - (100 / (1 + rs))
        
    def add_bollinger_bands(self, period: int = 20, std_dev: float = 2) -> None:
        """เพิ่ม Bollinger Bands"""
        self.data['bb_middle'] = self.data['close'].rolling(window=period).mean()
        self.data['bb_std'] = self.data['close'].rolling(window=period).std()
        self.data['bb_upper'] = self.data['bb_middle'] + (self.data['bb_std'] * std_dev)
        self.data['bb_lower'] = self.data['bb_middle'] - (self.data['bb_std'] * std_dev)
    
    def strategy_sma_crossover(self) -> None:
        """SMA Crossover Strategy"""
        self.data['signal'] = 0
        self.data.loc[
            self.data[f'sma_10'] > self.data[f'sma_50'], 
            'signal'
        ] = 1  # Buy
        self.data.loc[
            self.data[f'sma_10'] < self.data[f'sma_50'], 
            'signal'
        ] = -1  # Sell
        
    def strategy_rsi(self, oversold: int = 30, overbought: int = 70) -> None:
        """RSI Mean Reversion Strategy"""
        self.data['signal'] = 0
        self.data.loc[self.data['rsi'] < oversold, 'signal'] = 1   # Buy
        self.data.loc[self.data['rsi'] > overbought, 'signal'] = -1  # Sell
        
    def run_backtest(self, strategy_name: str = "Unknown") -> Dict:
        """รัน Backtest"""
        print(f"\n🔄 กำลังรัน Backtest: {strategy_name}")
        
        # Reset state
        self.capital = self.initial_capital
        self.position = 0
        self.trades = []
        
        for idx, row in self.data.iterrows():
            if pd.isna(row.get('signal')):
                continue
                
            current_price = row['close']
            timestamp = row['timestamp']
            
            # Buy Signal
            if row['signal'] == 1 and self.position == 0:
                self.position = self.capital / current_price
                self.trades.append({
                    'timestamp': timestamp,
                    'type': 'BUY',
                    'price': current_price,
                    'quantity': self.position,
                    'capital': self.capital
                })
                self.capital = 0
                
            # Sell Signal
            elif row['signal'] == -1 and self.position > 0:
                self.capital = self.position * current_price
                self.trades.append({
                    'timestamp': timestamp,
                    'type': 'SELL',
                    'price': current_price,
                    'quantity': self.position,
                    'capital': self.capital
                })
                self.position = 0
            
            # บันทึก Equity
            equity = self.capital + (self.position * current_price)
            self.equity_curve.append({
                'timestamp': timestamp,
                'equity': equity
            })
        
        # ปิด position สุดท้าย (ถ้ายังถือ)
        if self.position > 0:
            final_price = self.data['close'].iloc[-1]
            self.capital = self.position * final_price
            self.position = 0
            
        return self.calculate_metrics(strategy_name)
    
    def calculate_metrics(self, strategy_name: str) -> Dict:
        """คำนวณ Performance Metrics"""
        equity_df = pd.DataFrame(self.equity_curve)
        
        total_return = (self.capital - self.initial_capital) / self.initial_capital * 100
        
        # คำนวณ Max Drawdown
        equity_df['peak'] = equity_df['equity'].cummax()
        equity_df['drawdown'] = (equity_df['equity'] - equity_df['peak']) / equity_df['peak'] * 100
        max_drawdown = equity_df['drawdown'].min()
        
        # คำนวณ Sharpe Ratio (สมมติ risk-free rate = 0)
        if len(equity_df) > 1:
            returns = equity_df['equity'].pct_change().dropna()
            sharpe_ratio = returns.mean() / returns.std() * np.sqrt(365 * 24 * 60) if returns.std() > 0 else 0
        else:
            sharpe_ratio = 0
        
        # Win Rate
        winning_trades = sum(1 for i in range(1, len(self.trades), 2) 
                           if i < len(self.trades) and 
                           self.trades[i]['capital'] > self.trades[i-1]['capital'])
        total_trades = len(self.trades) // 2
        win_rate = winning_trades / total_trades * 100 if total_trades > 0 else 0
        
        metrics = {
            'strategy': strategy_name,
            'initial_capital': self.initial_capital,
            'final_capital': self.capital,
            'total_return_%': round(total_return, 2),
            'max_drawdown_%': round(max_drawdown, 2),
            'sharpe_ratio': round(sharpe_ratio, 2),
            'total_trades': total_trades,
            'win_rate_%': round(win_rate, 2)
        }
        
        # Print Report
        print("\n" + "=" * 50)
        print(f"📊 Backtest Results: {strategy_name}")
        print("=" * 50)
        for key, value in metrics.items():
            if isinstance(value, float):
                print(f"   {key}: {value:,.2f}")
            else:
                print(f"   {key}: {value}")
        print("=" * 50)
        
        return metrics



วิธีใช้งาน

if __name__ == "__main__":
    # โหลดข้อมูลที่ Clean แล้ว
    df = pd.read_csv("cleaned_btcusdt_ohlcv_1m.csv")
    
    # สร้าง Backtester
    backtester = SimpleBacktester(df, initial_capital=10000)
    
    # เพิ่ม Indicators
    backtester.add_sma(10, 50)
    backtester.add_rsi(14)
    backtester.add_bollinger_bands(20, 2)
    
    # รัน Strategy
    backtester.strategy_sma_crossover()
    results = backtester.run_backtest("SMA Crossover (10,50)")
    
    # Save trades
    trades_df = pd.DataFrame(backtester.trades)
    trades_df.to_csv("backtest_trades.csv", index=False)
    print(f"\n✅ บันทึก Trades: {len(trades_df)} trades")

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

กรณีที่ 1: MemoryError เมื่อโหลดข้อมูลขนาดใหญ่

ปัญหา: เมื่อดาวน์โหลดข้อมูลหลายเดือน ขนาดไฟล์อาจใหญ่เกิน RAM 
# ❌ วิธีที่ผิด - โหลดทั้งไฟล์เข้า Memory

df = pd.read_csv("huge_file.csv")  # อาจเกิด MemoryError



✅ วิธีที่ถูก - ใช้ Chunking

def load_data_in_chunks(filepath, chunk_size=100000):
    """โหลดข้อมูลเป็น chunks เพื่อประหยัด Memory"""
    chunks = []
    for chunk in pd.read_csv(filepath, chunksize=chunk_size, 
                             parse_dates=['timestamp']):
        # Clean แต่ละ chunk
        chunk = clean_chunk(chunk)
        chunks.append(chunk)
        
    return pd.concat(chunks, ignore_index=True)

def clean_chunk(chunk):
    """Clean แต่ละ chunk ก่อนรวม"""
    # ลบ duplicates ใน chunk
    chunk = chunk.drop_duplicates(subset=['trade_id'])
    
    # ลบ NaN
    chunk = chunk.dropna()
    
    # กรองราคาที่ผิดปกติ
    median = chunk['price'].median()
    std = chunk['price'].std()
    chunk = chunk[
        (chunk['price'] > median - 3*std) &
        (chunk['price'] < median + 3*std)
    ]
    
    return chunk


ใช้งาน

df = load_data_in_chunks("bybit_btcusdt_trades.csv", chunk_size=500000)
print(f"✅ โหลดสำเร็จ: {len(df):,} rows")

กรณีที่ 2: Timezone ผิดทำให้ Backtest ไม่ตรง

ปัญหา: Backtest ผลลัพธ์ไม่ตรงกับเวลาจริง เพราะ Timezone คลาดเคลื่อน 
# ❌ วิธีที่ผิด - ไม่กำหนด Timezone

df['timestamp'] = pd.to_datetime(df['timestamp'])  # ไม่รู้ timezone



✅ วิธีที่ถูก - กำหนด UTC และ Convert เป็น Local Time

import pytz

def standardize_timezone(df, target_tz='Asia/Bangkok'):
    """มาตรฐาน Timezone สำหรับ Backtest"""
    
    # แปลงเป็น UTC ก่อน (ถ้ายังไม่ใช่)
    if df['timestamp'].dt.tz is None:
        df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)
    
    # Convert เป็น Timezone ที่ต้องการ
    target_timezone = pytz.timezone(target_tz)
    df['timestamp_local'] = df['timestamp'].dt.tz_convert(target_timezone)
    df['hour'] = df['timestamp_local'].dt.hour
    df['day_of_week'] = df['timestamp_local'].dt.dayofweek
    
    print(f"Timezone ที่ใช้: {target_tz}")
    print(f"ช่วงเวลา: {df['timestamp_local'].min()} ถึง {df['timestamp_local'].max()}")
    
    return df


ตรวจสอบว่า Session ที่เทรดตรงกับเวลาที่คาดหวังหรือไม่

def check_trading_sessions(df):
    """ตรวจสอบว่าช่วงเวลาเทรดตรงหรือไม่"""
    london_session = df[(df['hour'] >= 8) & (df['hour'] <= 16)]
    asia_session = df[(df['hour'] >= 0) & (df['hour'] <= 8)]
    ny_session = df[(df['hour'] >= 13) & (df['hour'] <= 21)]
    
    print(f"London Session trades: {len(london_session):,}")
    print(f"Asia Session trades: {len(asia_session):,}")
    print(f"New York Session trades: {len(ny_session):,}")
    
    return london_session, asia_session, ny_session


ใช้งาน

df = standardize_timezone(df, target_tz='Asia/Bangkok')
check_trading_sessions(df)

กรณีที่ 3: Look-Ahead Bias ใน Indicators

ปัญหา: ค่า Indicators ถูกคำนวณรวมข้อมูลที่ยังไม่เกิด ทำให้ Backtest ดีเกินจริง 
# ❌ วิธีที่ผิด - ใช้ Future Information (Look-Ahead Bias)

df['future_return'] = df['close'].shift(-1)  # ใช้ราคาพรุ่งนี้!


df['optimal_sma'] = df['close'].rolling(window=best_window).mean()  # ดู results ก่อน



✅ วิธีที่ถูก - ใช้ Expanding Window หรือ Walk-Forward Analysis

def walk_forward_optimization(df, parameter_range, strategy_func):
    """
    Walk-Forward Optimization เพื่อหลีกเลี่ยง Look-Ahead Bias
    """
    results = []
    
    # แบ่งข้อมูลเป็น In-Sample และ Out-of-Sample
    train_size = int(len(df) * 0.7)
    train_df = df.iloc[:train_size].copy()
    test_df = df.iloc[train_size:].copy()
    
    print(f"Train: {len(train_df):,} rows")
    print(f"Test: {len(test_df):,} rows")
    
    # Optimize parameters on train set
    best_params = None
    best_train_return = -999
    
    for params in parameter_range:
        # Train on train set
        train_result = strategy_func(train_df, params)
        
        if train_result['return'] > best_train_return:
            best_train_return = train_result['return']
            best_params = params
    
    print(f"Best params from train: {best_params}")
    
    # Test on out-of-sample set with best params
    test_result = strategy_func(test_df, best_params)
    
    return {
        'train_result': best_train_return,
        'test_result': test_result,
        'best_params': best_params,
        'overfitting_detected': test_result['return'] < best_train_return * 0.5
    }


ตัวอย่าง Parameter Range สำหรับ SMA

sma_parameters = [
    (5, 20), (10, 50), (10, 100), (20, 50), 
    (20, 100), (50, 100), (50, 200)
]

def sma_strategy_with_params(df, params):
    """SMA Strategy พร้อม Parameter"""
    short, long = params
    df = df.copy()
    df['sma_short'] = df['close'].rolling(window=short).mean()
    df['sma_long'] = df['close'].rolling(window=long).mean()
    
    # Signal - ใช้ shift(1) เพื่อไม่ให้ใช้ current bar
    df['signal'] = (df['sma_short'] > df['sma_long']).astype(int).shift(1)
    
    return calculate_strategy_return(df)

def calculate_strategy_return(df):
    """คำนวณ Return �

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