บทนำ: ทำไมข้อมูลถึงต้อง "สะอาด" ก่อน Backtest

การทำ Quantitative Backtesting ที่แม่นยำเริ่มต้นจากข้อมูลที่บริสุทธิ์ หลายคนเริ่มต้นสร้างระบบเทรดด้วย Python แต่พบปัญหาว่าผลลัพธ์จาก Backtest ไม่ตรงกับตลาดจริง — สาเหตุหลักคือ Garbage In, Garbage Out ข้อมูลที่มี NaN, Outlier ผิดปกติ, หรือ Missing Bar จะทำให้ทุกการคำนวณผิดเพี้ยน บทความนี้จะพาทุกท่านสร้าง Data Pipeline สำหรับล้างข้อมูล Tardis อย่างเป็นระบบ โดยใช้ API จาก HolySheep AI ซึ่งมี Latency ต่ำกว่า 50ms และราคาประหยัดกว่า 85% เมื่อเทียบกับ OpenAI API

Tardis API คืออะไร

Tardis คือบริการที่รวบรวมข้อมูล Historical Market Data จากหลาย Exchange ทั่วโลก ให้บริการทั้ง: สำหรับนักพัฒนาระบบเทรดแบบ Quant, Tardis เป็นแหล่งข้อมูลที่น่าเชื่อถือและครอบคลุม Exchange ยอดนิยมอย่าง Binance, Bybit, OKX, Bitget และอื่นๆ

โครงสร้าง Pipeline ภาพรวม


┌─────────────────────────────────────────────────────────────┐
│                    DATA PIPELINE                           │
├─────────────────────────────────────────────────────────────┤
│  1. Extract:   ดึงข้อมูลจาก Tardis API                     │
│       ↓                                                      │
│  2. Validate: ตรวจสอบความครบถ้วน ความถูกต้องของ Timestamp   │
│       ↓                                                      │
│  3. Clean:    จัดการ NaN, Outlier, Missing Bar             │
│       ↓                                                      │
│  4. Transform: คำนวณ Indicator, Resample, Merge            │
│       ↓                                                      │
│  5. Load:     เก็บลง Database / Parquet / CSV              │
└─────────────────────────────────────────────────────────────┘

การตั้งค่าเริ่มต้นและการเชื่อมต่อ API

import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import json
import time


การตั้งค่า HolySheep AI API สำหรับเรียกใช้งาน AI ในการวิเคราะห์ข้อมูล

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

def call_holysheep_ai(prompt: str) -> str:
    """เรียกใช้ AI จาก HolySheep สำหรับวิเคราะห์ข้อมูลอัตโนมัติ"""
    response = requests.post(
        f"{HOLYSHEEP_BASE_URL}/chat/completions",
        headers={
            "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "model": "gpt-4.1",
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3
        }
    )
    return response.json()["choices"][0]["message"]["content"]


การตั้งค่า Tardis API

TARDIS_API_KEY = "YOUR_TARDIS_API_KEY"  # สมัครได้ที่ tardis.dev
TARDIS_BASE_URL = "https://api.tardis.dev/v1"

def get_tardis_headers():
    return {"Authorization": f"Token {TARDIS_API_KEY}"}

print("✅ การตั้งค่าเริ่มต้นเสร็จสมบูรณ์")
print(f"📡 HolySheep API: {HOLYSHEEP_BASE_URL}")
print(f"📡 Tardis API: {TARDIS_BASE_URL}")

ขั้นตอนที่ 1: Extract — ดึงข้อมูล Historical K-line

def fetch_kline_data(
    exchange: str,
    symbol: str,
    start_date: str,
    end_date: str,
    timeframe: str = "1m"
) -> pd.DataFrame:
    """
    ดึงข้อมูล K-line จาก Tardis API
    
    Parameters:
    - exchange: ชื่อ Exchange เช่น 'binance', 'bybit'
    - symbol: สัญลักษณ์คู่เทรด เช่น 'BTCUSDT'
    - start_date: วันที่เริ่มต้น format 'YYYY-MM-DD'
    - end_date: วันที่สิ้นสุด format 'YYYY-MM-DD'
    - timeframe: '1m', '5m', '1h', '4h', '1d'
    """
    
    url = f"{TARDIS_BASE_URL}/ahara/historical/klines"
    params = {
        "exchange": exchange,
        "symbol": symbol,
        "date": f"{start_date},{end_date}",
        "interval": timeframe
    }
    
    print(f"🔄 กำลังดึงข้อมูล {symbol} บน {exchange} ({timeframe})...")
    
    response = requests.get(
        url,
        headers=get_tardis_headers(),
        params=params
    )
    
    if response.status_code != 200:
        raise Exception(f"Tardis API Error: {response.status_code} - {response.text}")
    
    data = response.json()
    
    # แปลงเป็น DataFrame
    df = pd.DataFrame(data)
    
    # ตรวจสอบว่ามีข้อมูลหรือไม่
    if df.empty:
        raise ValueError(f"ไม่พบข้อมูลสำหรับ {symbol} ในช่วงวันที่ที่ระบุ")
    
    # แปลงชื่อคอลัมน์ให้เป็นมาตรฐาน
    df.columns = ['timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote']
    
    # แปลง timestamp เป็น datetime
    df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('datetime', inplace=True)
    
    # แปลงข้อมูลเป็น float
    numeric_cols = ['open', 'high', 'low', 'close', 'volume']
    df[numeric_cols] = df[numeric_cols].astype(float)
    
    print(f"✅ ดึงข้อมูลสำเร็จ: {len(df):,} bars")
    return df


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

df_btc = fetch_kline_data(
    exchange='binance',
    symbol='BTCUSDT',
    start_date='2024-01-01',
    end_date='2024-03-01',
    timeframe='5m'
)

print(df_btc.head())
print(f"\n📊 ข้อมูลครอบคลุม: {df_btc.index.min()} ถึง {df_btc.index.max()}")

ขั้นตอนที่ 2: Extract — ดึงข้อมูล Tick/Trade Data

def fetch_tick_data(
    exchange: str,
    symbol: str,
    start_date: str,
    end_date: str,
    limit: int = 100000
) -> pd.DataFrame:
    """
    ดึงข้อมูล Tick/Trade จาก Tardis API
    
    Parameters:
    - limit: จำนวน records สูงสุดต่อครั้ง (Tardis มี rate limit)
    """
    
    url = f"{TARDIS_BASE_URL}/ahara/historical/trades"
    params = {
        "exchange": exchange,
        "symbol": symbol,
        "date": f"{start_date},{end_date}",
        "limit": limit
    }
    
    print(f"🔄 กำลังดึงข้อมูล Tick สำหรับ {symbol}...")
    
    all_trades = []
    offset = 0
    
    while True:
        params["offset"] = offset
        response = requests.get(
            url,
            headers=get_tardis_headers(),
            params=params
        )
        
        if response.status_code != 200:
            print(f"⚠️ API Error: {response.status_code}, รอ 5 วินาที...")
            time.sleep(5)
            continue
            
        data = response.json()
        if not data:
            break
            
        all_trades.extend(data)
        offset += limit
        print(f"   📥 ดึงได้ {len(all_trades):,} trades แล้ว...")
        
        # หยุดถ้าได้ข้อมูลครบตามช่วงเวลา
        if len(data) < limit:
            break
            
        time.sleep(0.5)  # เคารพ Rate Limit
    
    df_trades = pd.DataFrame(all_trades)
    
    # ตรวจสอบและแปลงข้อมูล
    if df_trades.empty:
        raise ValueError(f"ไม่พบข้อมูล Tick สำหรับ {symbol}")
    
    # แปลง timestamp
    df_trades['datetime'] = pd.to_datetime(df_trades['timestamp'], unit='ms')
    df_trades.set_index('datetime', inplace=True)
    
    # แปลงข้อมูลตัวเลข
    df_trades['price'] = df_trades['price'].astype(float)
    df_trades['amount'] = df_trades['amount'].astype(float)
    
    print(f"✅ ดึงข้อมูล Tick สำเร็จ: {len(df_trades):,} trades")
    return df_trades


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

df_trades = fetch_tick_data(
    exchange='binance',
    symbol='BTCUSDT',
    start_date='2024-01-01',
    end_date='2024-01-02',
    limit=50000
)

print(df_trades.head(10))
print(f"\n📊 ข้อมูล Tick: {df_trades.index.min()} ถึง {df_trades.index.max()}")

ขั้นตอนที่ 3: Validate — ตรวจสอบความครบถ้วนของข้อมูล

def validate_kline_data(df: pd.DataFrame, expected_timeframe: str = "5T") -> Dict:
    """
    ตรวจสอบความครบถ้วนของข้อมูล K-line
    
    - ตรวจหา Missing Bars (เวลาขาดหาย)
    - ตรวจหา Duplicate Bars
    - ตรวจหา Outlier ใน OHLC
    - ตรวจสอบความสมเหตุสมผลของราคา
    """
    
    validation_report = {
        "total_bars": len(df),
        "missing_bars": 0,
        "duplicate_bars": 0,
        "outliers": [],
        "invalid_ohlc": 0,
        "warnings": []
    }
    
    # 1. ตรวจสอบ Missing Bars
    expected_index = pd.date_range(
        start=df.index.min(),
        end=df.index.max(),
        freq=expected_timeframe
    )
    missing_times = expected_index.difference(df.index)
    validation_report["missing_bars"] = len(missing_times)
    
    if len(missing_times) > 0:
        validation_report["warnings"].append(
            f"พบ {len(missing_times)} bars ที่หายไป (Missing)"
        )
        print(f"⚠️ พบ {len(missing_times)} Missing Bars")
        print(f"   ตัวอย่างเวลาที่ขาด: {missing_times[:5].tolist()}")
    
    # 2. ตรวจสอบ Duplicate Bars
    duplicates = df.index[df.index.duplicated()]
    validation_report["duplicate_bars"] = len(duplicates)
    
    if len(duplicates) > 0:
        validation_report["warnings"].append(
            f"พบ {len(duplicates)} bars ที่ซ้ำกัน"
        )
        print(f"⚠️ พบ {len(duplicates)} Duplicate Bars")
    
    # 3. ตรวจสอบความสมเหตุสมผลของ OHLC
    invalid_ohlc = (
        (df['high'] < df['low']) |
        (df['high'] < df['open']) |
        (df['high'] < df['close']) |
        (df['low'] > df['open']) |
        (df['low'] > df['close'])
    )
    validation_report["invalid_ohlc"] = invalid_ohlc.sum()
    
    if invalid_ohlc.sum() > 0:
        print(f"⚠️ พบ {invalid_ohlc.sum()} bars ที่มี OHLC ไม่ถูกต้อง")
    
    # 4. ตรวจหา Outlier (ราคาผิดปกติ)
    # ใช้ IQR Method
    price_change = df['close'].pct_change()
    Q1 = price_change.quantile(0.25)
    Q3 = price_change.quantile(0.75)
    IQR = Q3 - Q1
    
    outlier_threshold_upper = Q3 + 3 * IQR
    outlier_threshold_lower = Q1 - 3 * IQR
    
    outliers = price_change[(price_change > outlier_threshold_upper) | 
                            (price_change < outlier_threshold_lower)]
    
    validation_report["outliers"] = outliers.index.tolist()
    
    if len(outliers) > 0:
        print(f"⚠️ พบ {len(outliers)} Outliers ที่อาจเป็นข้อมูลผิดพลาด")
        print(f"   Outlier ที่ใหญ่ที่สุด: {price_change.max():.2%}")
        print(f"   Outlier ที่เล็กที่สุด: {price_change.min():.2%}")
    
    print("\n" + "="*50)
    print("📋 VALIDATION REPORT")
    print("="*50)
    print(f"Total Bars: {validation_report['total_bars']:,}")
    print(f"Missing Bars: {validation_report['missing_bars']}")
    print(f"Duplicate Bars: {validation_report['duplicate_bars']}")
    print(f"Invalid OHLC: {validation_report['invalid_ohlc']}")
    print(f"Outliers: {len(validation_report['outliers'])}")
    
    return validation_report


ทดสอบการ Validate

validation = validate_kline_data(df_btc, expected_timeframe="5T")

ขั้นตอนที่ 4: Clean — ล้างข้อมูลและแก้ไขปัญหา

def clean_kline_data(df: pd.DataFrame, fill_method: str = "forward") -> pd.DataFrame:
    """
    ล้างข้อมูล K-line โดย:
    1. ลบ Duplicate Bars (เเที่ยงวันที่ซ้ำ)
    2. สร้าง Missing Bars (เติมเวลาที่ขาด)
    3. แก้ไข OHLC ที่ผิดปกติ
    4. เติมค่าที่ขาดหายด้วย Forward Fill
    """
    
    print("🧹 เริ่มล้างข้อมูล...")
    
    df_clean = df.copy()
    
    # 1. ลบ Duplicate ถ้ามี
    before_count = len(df_clean)
    df_clean = df_clean[~df_clean.index.duplicated(keep='first')]
    duplicates_removed = before_count - len(df_clean)
    if duplicates_removed > 0:
        print(f"   ✅ ลบ {duplicates_removed} Duplicate Bars")
    
    # 2. สร้าง Missing Bars (Resample)
    df_clean = df_clean.resample('5T').agg({
        'open': 'first',
        'high': 'max',
        'low': 'min',
        'close': 'last',
        'volume': 'sum'
    })
    
    # 3. แก้ไข OHLC ที่ผิดปกติ
    # High ต้องสูงสุด, Low ต้องต่ำสุด
    df_clean['high'] = df_clean[['open', 'high', 'low', 'close']].max(axis=1)
    df_clean['low'] = df_clean[['open', 'high', 'low', 'close']].min(axis=1)
    
    # 4. เติมค่าที่ขาดด้วย Forward Fill
    if fill_method == "forward":
        df_clean = df_clean.ffill()
    elif fill_method == "interpolate":
        df_clean = df_clean.interpolate(method='time')
    
    # ลบแถวที่ยังมี NaN (ช่วงแรกที่ไม่มีข้อมูล)
    df_clean = df_clean.dropna()
    
    print(f"   ✅ ล้างข้อมูลเสร็จสมบูรณ์: {len(df_clean):,} bars")
    
    return df_clean

def clean_tick_data(df_trades: pd.DataFrame, price_outlier_threshold: float = 0.1) -> pd.DataFrame:
    """
    ล้างข้อมูล Tick/Trade โดย:
    1. ลบ Duplicate Timestamps
    2. กรอง Outlier ที่ราคาผิดปกติ
    3. ตรวจสอบ Side (buy/sell)
    """
    
    print("🧹 ล้างข้อมูล Tick...")
    
    df_clean = df_trades.copy()
    
    # 1. ลบ Duplicate Timestamps
    before = len(df_clean)
    df_clean = df_clean[~df_clean.index.duplicated(keep='first')]
    print(f"   ✅ ลบ {before - len(df_clean)} Duplicate Trades")
    
    # 2. กรอง Outlier (ราคาเปลี่ยนแปลงเกิน threshold)
    if 'price' in df_clean.columns:
        price_change = df_clean['price'].pct_change().abs()
        valid_trades = price_change < price_outlier_threshold
        valid_trades.iloc[0] = True  # Keep first trade
        
        outliers = ~valid_trades
        print(f"   ✅ กรอง {outliers.sum()} Outlier Trades")
        
        df_clean = df_clean[valid_trades]
    
    # 3. ตรวจสอบ Side
    if 'side' in df_clean.columns:
        valid_sides = df_clean['side'].isin(['buy', 'sell'])
        invalid = (~valid_sides).sum()
        if invalid > 0:
            print(f"   ⚠️ พบ {invalid} Trades ที่มี Side ไม่ถูกต้อง")
    
    print(f"   ✅ ล้างข้อมูล Tick เสร็จ: {len(df_clean):,} trades")
    
    return df_clean


ใช้งาน

df_btc_clean = clean_kline_data(df_btc)
df_trades_clean = clean_tick_data(df_trades)

print(f"\n📊 ข้อมูลหลังล้าง: {len(df_btc_clean):,} bars")
print(df_btc_clean.head())

ขั้นตอนที่ 5: Transform — คำนวณ Indicator และ Resample

def add_technical_indicators(df: pd.DataFrame) -> pd.DataFrame:
    """เพิ่ม Technical Indicators สำหรับ Backtesting"""
    
    df_ta = df.copy()
    
    # Simple Moving Averages
    df_ta['sma_20'] = df_ta['close'].rolling(window=20).mean()
    df_ta['sma_50'] = df_ta['close'].rolling(window=50).mean()
    df_ta['sma_200'] = df_ta['close'].rolling(window=200).mean()
    
    # Exponential Moving Averages
    df_ta['ema_12'] = df_ta['close'].ewm(span=12, adjust=False).mean()
    df_ta['ema_26'] = df_ta['close'].ewm(span=26, adjust=False).mean()
    
    # MACD
    df_ta['macd'] = df_ta['ema_12'] - df_ta['ema_26']
    df_ta['macd_signal'] = df_ta['macd'].ewm(span=9, adjust=False).mean()
    df_ta['macd_hist'] = df_ta['macd'] - df_ta['macd_signal']
    
    # RSI
    delta = df_ta['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    df_ta['rsi'] = 100 - (100 / (1 + rs))
    
    # Bollinger Bands
    df_ta['bb_middle'] = df_ta['close'].rolling(window=20).mean()
    bb_std = df_ta['close'].rolling(window=20).std()
    df_ta['bb_upper'] = df_ta['bb_middle'] + (bb_std * 2)
    df_ta['bb_lower'] = df_ta['bb_middle'] - (bb_std * 2)
    
    # ATR (Average True Range)
    high_low = df_ta['high'] - df_ta['low']
    high_close = (df_ta['high'] - df_ta['close'].shift()).abs()
    low_close = (df_ta['low'] - df_ta['close'].shift()).abs()
    true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
    df_ta['atr'] = true_range.rolling(window=14).mean()
    
    # Volume indicators
    df_ta['volume_sma_20'] = df_ta['volume'].rolling(window=20).mean()
    df_ta['volume_ratio'] = df_ta['volume'] / df_ta['volume_sma_20']
    
    print("✅ เพิ่ม Technical Indicators เรียบร้อย")
    
    return df_ta

def resample_to_higher_timeframe(df: pd.DataFrame, timeframe: str) -> pd.DataFrame:
    """Resample ข้อมูลเป็น Timeframe ที่สูงกว่า"""
    
    df_resampled = df.resample(timeframe).agg({
        'open': 'first',
        'high': 'max',
        'low': 'min',
        'close': 'last',
        'volume': 'sum'
    })
    
    df_resampled = df_resampled.dropna()
    
    print(f"✅ Resample เป็น {timeframe}: {len(df_resampled):,} bars")
    
    return df_resampled


ใช้งาน

df_with_ta = add_technical_indicators(df_btc_clean)


Resample เป็น 1 ชั่วโมง

df_hourly = resample_to_higher_timeframe(df_btc_clean, '1H')

print("\n📊 ข้อมูลพร้อมใช้งาน:")
print(df_with_ta[['close', 'sma_20', 'rsi', 'macd']].tail())

ขั้นตอนที่ 6: Load — บันทึกข้อมูลลง Storage

import pyarrow as pa
import pyarrow.parquet as pq
import os

def save_data_to_storage(
    df: pd.DataFrame,
    symbol: str,
    timeframe: str,
    data_type: str = "kline",
    storage_path: str = "./data"
):
    """
    บันทึกข้อมูลลง Storage ในรูปแบบต่างๆ
    
    - Parquet: สำหรับ Backtesting ที่ต้องการความเร็ว
    - CSV: สำหรับการตรวจสอบด้วยสายตา
    """
    
    os.makedirs(storage_path, exist_ok=True)
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    # 1. บันทึกเป็น Parquet
    parquet_path = f"{storage_path}/{symbol}_{timeframe}_{data_type}_{timestamp}.parquet"
    table = pa.Table.from_pandas(df.reset_index())
    pq.write_table(table, parquet_path)
    parquet_size = os.path.getsize(parquet_path) / (1024 * 1024)
    print(f"💾 บันทึก Parquet: {parquet_path} ({parquet_size:.2f} MB)")
    
    # 2. บันทึกเป็น CSV (สำหรับตรวจสอบ)
    csv_path = f"{storage_path}/{symbol}_{timeframe}_{data_type}_{timestamp}.csv"
    df.to_csv(csv_path)
    csv_size = os.path.getsize(csv_path) / (1024 * 1024)
    print(f"💾 บันทึก CSV: {csv_path} ({csv_size:.2f} MB)")
    
    # 3. สร้าง Metadata
    metadata = {
        "symbol": symbol,
        "timeframe": timeframe,
        "data_type": data_type,
        "start_date": str(df.index.min()),
        "end_date": str(df.index.max()),
        "total_bars": len(df),
        "created_at": timestamp,
        "indicators": list(df.columns)
    }
    
    metadata_path = f"{storage_path}/{symbol}_{timeframe}_{data_type}_{timestamp}_metadata.json"
    with open(metadata_path, 'w') as f:
        json.dump(metadata, f, indent=2)
    print(f"💾 บันทึก Metadata: {metadata_path}")
    
    return parquet_path, csv_path


บันทึกข้อมูล

parquet_file, csv_file = save_data_to_storage(
    df_with_ta,
    symbol="BTCUSDT",
    timeframe="5m",
    data_type="kline",
    storage_path="./btc_data"
)

Pipeline สมบูรณ์: รวมทุกขั้นตอนเป็น Automated Workflow

def run_backtest_data_pipeline(
    exchange: str,
    symbol: str,
    start_date: str,
    end_date: str,
    timeframe: str = "5m",
    add_indicators: bool = True,
    storage_path: str = "./backtest_data"
):
    """
    Pipeline สมบูรณ์

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