บทนำ: ทำไมต้องทำ Backtesting ด้วยข้อมูลคุณภาพสูง

การทำ Quantitative Trading หรือการเทรดเชิงปริมาณที่แท้จริง ไม่ใช่แค่การเขียนโค้ด Python สัก 2-3 บรรทัดแล้วจบ แต่เป็นทั้งระบบนิเวศที่ต้องอาศัย ข้อมูลคุณภาพสูง เครื่องมือวิเคราะห์ที่ทรงพลัง และ โครงสร้างพื้นฐานที่เสถียร ประสบการณ์ 5 ปีของผมในแวดวง Quant ได้สอนว่า "Garbage in, Garbage out" ไม่ใช่แค่สุภาษิต แต่เป็นความจริงที่ทำให้นักลงทุนหลายคนสูญเสียเงินจากการทดสอบย้อนกลับด้วยข้อมูลที่ผิดพลาด บทความนี้จะพาคุณสร้างระบบ Backtesting ที่เชื่อมต่อกับ Tardis.dev สำหรับดึงข้อมูลประวัติศาสตร์ระดับ Tick-by-Tick และ Backtrader สำหรับวิเคราะห์ผลตอบแทน โดยในตอนท้ายจะแนะนำ HolySheep AI ที่สามารถช่วยสร้างสัญญาณการซื้อขายด้วย AI ได้อย่างมีประสิทธิภาพ ลดต้นทุนได้ถึง 85% เมื่อเทียบกับ API ทั่วไป

Tardis.dev คืออะไร และทำไมถึงเป็นตัวเลือกที่ดี

Tardis.dev เป็น Financial Data API ที่รวบรวมข้อมูลตลาดคริปโตจาก Exchange หลายสิบแห่ง ให้บริการข้อมูลแบบ Real-time และ Historical รวมถึง: สิ่งที่ทำให้ Tardis.dev โดดเด่นคือ ความแม่นยำของข้อมูล และ Latency ที่ต่ำมาก ซึ่งเหมาะสำหรับการทำ High-Frequency Backtesting ที่ต้องการความละเอียดระดับ Millisecond

การติดตั้งและเตรียม Environment

ก่อนเริ่มต้น เราต้องติดตั้ง Package ที่จำเป็น: 
# ติดตั้ง Dependencies
pip install backtrader pandas numpy requests asyncio aiohttp


สำหรับ Tardis Client

pip install tardis-client


สำหรับ Visualize ผลลัพธ์

pip install matplotlib bt

ระบบดึงข้อมูลจาก Tardis.dev

ผมได้พัฒนา Async Data Fetcher ที่ทำงานแบบ Non-blocking เพื่อดึงข้อมูลประวัติศาสตร์จาก Tardis.dev อย่างมีประสิทธิภาพ: 
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import backtrader as bt

class TardisDataFetcher:
    """
    Async Data Fetcher สำหรับดึงข้อมูลจาก Tardis.dev
    รองรับ Trade, Order Book และ Candle data
    """
    
    BASE_URL = "https://tardis.dev/api/v1"
    
    def __init__(self, api_token: str):
        self.api_token = api_token
        self.session: Optional[aiohttp.ClientSession] = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {self.api_token}"}
        )
        return self
    
    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
    
    async def fetch_trades(
        self,
        exchange: str,
        symbol: str,
        start_date: datetime,
        end_date: datetime
    ) -> pd.DataFrame:
        """
        ดึงข้อมูล Trade ระหว่างวันที่กำหนด
        
        Args:
            exchange: ชื่อ Exchange เช่น 'binance', 'bybit'
            symbol: สัญลักษณ์ เช่น 'BTCUSDT'
            start_date: วันเริ่มต้น
            end_date: วันสิ้นสุด
        
        Returns:
            DataFrame ที่มี columns: timestamp, price, size, side
        """
        url = f"{self.BASE_URL}/historical/trades/{exchange}/{symbol}"
        
        params = {
            "from": int(start_date.timestamp() * 1000),
            "to": int(end_date.timestamp() * 1000),
            "limit": 100000  # Max records per request
        }
        
        all_trades = []
        
        async with self.session.get(url, params=params) as response:
            if response.status == 200:
                data = await response.json()
                all_trades.extend(data)
            else:
                raise Exception(f"API Error: {response.status}")
        
        # แปลงเป็น DataFrame
        df = pd.DataFrame(all_trades)
        if not df.empty:
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            df = df.sort_values('timestamp')
        
        return df
    
    async def fetch_candles(
        self,
        exchange: str,
        symbol: str,
        timeframe: str,
        start_date: datetime,
        end_date: datetime
    ) -> pd.DataFrame:
        """
        ดึงข้อมูล OHLCV Candle
        
        Timeframe ที่รองรับ: 1m, 5m, 15m, 1h, 4h, 1d
        """
        url = f"{self.BASE_URL}/historical/candles/{exchange}:{symbol}"
        
        params = {
            "from": int(start_date.timestamp()),
            "to": int(end_date.timestamp()),
            "resolution": timeframe
        }
        
        async with self.session.get(url, params=params) as response:
            if response.status == 200:
                data = await response.json()
                
                df = pd.DataFrame(data)
                if not df.empty:
                    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
                    df = df.rename(columns={
                        't': 'timestamp',
                        'o': 'open',
                        'h': 'high',
                        'l': 'low',
                        'c': 'close',
                        'v': 'volume'
                    })
                    df = df.sort_values('timestamp')
                
                return df
            else:
                raise Exception(f"API Error: {response.status}")
    
    async def batch_fetch(
        self,
        requests: List[Dict]
    ) -> Dict[str, pd.DataFrame]:
        """
        ดึงข้อมูลหลาย Symbol/Timeframe พร้อมกัน
        
        Example:
            requests = [
                {"exchange": "binance", "symbol": "BTCUSDT", "timeframe": "1h"},
                {"exchange": "binance", "symbol": "ETHUSDT", "timeframe": "1h"}
            ]
        """
        tasks = []
        keys = []
        
        for req in requests:
            if 'candles' in req:
                task = self.fetch_candles(**req)
            else:
                task = self.fetch_trades(**req)
            tasks.append(task)
            keys.append(f"{req['exchange']}:{req['symbol']}")
        
        results = await asyncio.gather(*tasks)
        return dict(zip(keys, results))



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

async def main():
    async with TardisDataFetcher(api_token="YOUR_TARDIS_TOKEN") as fetcher:
        # ดึงข้อมูล BTCUSDT 1 ชั่วโมงย้อนหลัง 1 เดือน
        end = datetime.now()
        start = end - timedelta(days=30)
        
        df = await fetcher.fetch_candles(
            exchange="binance",
            symbol="BTCUSDT",
            timeframe="1h",
            start_date=start,
            end_date=end
        )
        
        print(f"ดึงข้อมูลได้ {len(df)} candles")
        print(df.tail())

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

เชื่อมต่อ Backtrader กับข้อมูลจาก Tardis

ต่อไปคือการสร้าง Custom Data Feed สำหรับ Backtrader ที่รับข้อมูลจาก Tardis.dev: 
import backtrader as bt
import pandas as pd
from datetime import datetime

class TardisDataFeed(bt.feeds.PandasData):
    """
    Custom Data Feed สำหรับ Backtrader
    รับ DataFrame จาก TardisDataFetcher โดยตรง
    """
    
    params = (
        ('datetime', 'timestamp'),
        ('open', 'open'),
        ('high', 'high'),
        ('low', 'low'),
        ('close', 'close'),
        ('volume', 'volume'),
        ('openinterest', -1),
    )

class MultiTimeFrameDataFeed(bt.feeds.PandasData):
    """
    รองรับข้อมูลหลาย Timeframe ใน Strategy เดียว
    สำหรับการทำ Multi-Timeframe Analysis
    """
    
    params = (
        ('datetime', 'timestamp'),
        ('open', 'open'),
        ('high', 'high'),
        ('low', 'low'),
        ('close', 'close'),
        ('volume', 'volume'),
        ('openinterest', -1),
        (' timeframe', bt.TimeFrame.Minutes),
        ('compression', 1),
    )


class MeanReversionStrategy(bt.Strategy):
    """
    ตัวอย่าง Strategy: Mean Reversion ด้วย Bollinger Bands
    
    Parameters:
        - period: จำนวน Period สำหรับคำนวณ Moving Average
        - devfactor: Standard Deviation Factor
        - printlog: พิมพ์ Log รายการซื้อขาย
    """
    
    params = (
        ('period', 20),
        ('devfactor', 2.0),
        ('printlog', True),
    )
    
    def __init__(self):
        # ใช้ Bollinger Bands
        self.boll = bt.indicators.BollingerBands(
            self.data.close, 
            period=self.params.period,
            devfactor=self.params.devfactor
        )
        
        # Signal สำหรับ Cross-over
        self.buy_signal = bt.indicators.CrossOver(
            self.data.close, 
            self.boll.lines.bot
        )
        self.sell_signal = bt.indicators.CrossOver(
            self.data.close, 
            self.boll.lines.top
        )
        
        # Tracking Orders
        self.order = None
        self.trade_log = []
    
    def log(self, txt, dt=None):
        if self.params.printlog:
            dt = dt or self.datas[0].datetime.date(0)
            print(f'{dt.isoformat()} {txt}')
    
    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return
        
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}')
                self.trade_log.append({
                    'date': self.datas[0].datetime.date(0),
                    'type': 'BUY',
                    'price': order.executed.price,
                    'size': order.executed.size
                })
            elif order.issell():
                self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}')
                self.trade_log.append({
                    'date': self.datas[0].datetime.date(0),
                    'type': 'SELL',
                    'price': order.executed.price,
                    'size': order.executed.size
                })
        
        self.order = None
    
    def next(self):
        # ถ้ามี Order ค้างอยู่ ไม่ต้องทำอะไร
        if self.order:
            return
        
        # ถ้าราคาต่ำกว่า Lower Band = Oversold = ซื้อ
        if self.buy_signal > 0:
            self.log(f'สัญญาณซื้อ: ราคา {self.data.close[0]:.2f} < Lower {self.boll.lines.bot[0]:.2f}')
            self.order = self.buy()
        
        # ถ้าราคาสูงกว่า Upper Band = Overbought = ขาย
        elif self.sell_signal < 0 and self.position:
            self.log(f'สัญญาณขาย: ราคา {self.data.close[0]:.2f} > Upper {self.boll.lines.top[0]:.2f}')
            self.order = self.sell()
    
    def stop(self):
        if self.params.printlog:
            self.log(f'กำไรสุทธิ (Portfolio): {self.broker.getvalue():.2f}')


def run_backtest(
    data_feed: pd.DataFrame,
    initial_cash: float = 100000,
    commission: float = 0.001
) -> bt.Cerebro:
    """
    Run Backtest ด้วย Cerebro
    
    Args:
        data_feed: DataFrame จาก TardisDataFetcher
        initial_cash: เงินทุนเริ่มต้น
        commission: ค่าคอมมิชชั่นต่อรายการ
    
    Returns:
        Cerebro instance พร้อมผลลัพธ์
    """
    # สร้าง Cerebro
    cerebro = bt.Cerebro()
    
    # เพิ่ม Strategy
    cerebro.addstrategy(
        MeanReversionStrategy,
        period=20,
        devfactor=2.0,
        printlog=True
    )
    
    # เพิ่ม Data Feed
    data = TardisDataFeed(dataname=data_feed)
    cerebro.adddata(data)
    
    # ตั้งค่า Broker
    cerebro.broker.setcash(initial_cash)
    cerebro.broker.setcommission(commission=commission)
    
    # เพิ่ม Analyzer สำหรับวิเคราะห์ผล
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
    cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
    
    return cerebro



ตัวอย่างการรัน Backtest

async def full_backtest_example():
    from tardis_fetcher import TardisDataFetcher
    from datetime import timedelta
    
    async with TardisDataFetcher(api_token="YOUR_TARDIS_TOKEN") as fetcher:
        # ดึงข้อมูล 1 ปี
        end = datetime.now()
        start = end - timedelta(days=365)
        
        df = await fetcher.fetch_candles(
            exchange="binance",
            symbol="BTCUSDT",
            timeframe="1d",
            start_date=start,
            end_date=end
        )
        
        # Run Backtest
        cerebro = run_backtest(df, initial_cash=100000)
        
        # แสดงเงินทุนเริ่มต้น
        print(f'เงินทุนเริ่มต้น: {cerebro.broker.getvalue():.2f}')
        
        # รัน Backtest
        results = cerebro.run()
        
        # แสดงเงินทุนสุดท้าย
        final_value = cerebro.broker.getvalue()
        print(f'เงินทุนสุดท้าย: {final_value:.2f}')
        print(f'ผลตอบแทน: {((final_value - 100000) / 100000) * 100:.2f}%')
        
        # ดึงผล Analyzer
        strat = results[0]
        sharpe = strat.analyzers.sharpe.get_analysis()
        drawdown = strat.analyzers.drawdown.get_analysis()
        
        print(f'Sharpe Ratio: {sharpe.get("sharperatio", "N/A")}')
        print(f'Max Drawdown: {drawdown.get("max", {}).get("drawdown", 0):.2f}%')

การ Optimize Strategy ด้วย Multi-Processing

สำหรับการทำ Parameter Optimization ที่มีประสิทธิภาพ ผมแนะนำให้ใช้ Multi-Processing เพื่อใช้ CPU ทั้งหมด: 
from backtrader.optimization import OptimizationSpace
from multiprocessing import Pool, cpu_count
import itertools

class OptimizationEngine:
    """
    Engine สำหรับ Optimization หลาย Parameter พร้อมกัน
    
    Features:
    - Multi-processing support
    - Grid Search และ Random Search
    - บันทึกผลลัพธ์ลง CSV
    """
    
    def __init__(self, data_feed: pd.DataFrame, initial_cash: float = 100000):
        self.data_feed = data_feed
        self.initial_cash = initial_cash
        self.results = []
    
    def evaluate_params(self, params: tuple) -> dict:
        """
        ประเมิน Parameter Set เดียว
        """
        period, devfactor, commission = params
        
        cerebro = bt.Cerebro()
        cerebro.addstrategy(
            MeanReversionStrategy,
            period=period,
            devfactor=devfactor,
            printlog=False
        )
        
        data = TardisDataFeed(dataname=self.data_feed)
        cerebro.adddata(data)
        cerebro.broker.setcash(self.initial_cash)
        cerebro.broker.setcommission(commission=commission)
        
        # เพิ่ม Analyzer
        cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
        cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
        cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
        
        results = cerebro.run()
        strat = results[0]
        
        sharpe = strat.analyzers.sharpe.get_analysis()
        drawdown = strat.analyzers.drawdown.get_analysis()
        returns = strat.analyzers.returns.get_analysis()
        
        final_value = cerebro.broker.getvalue()
        
        return {
            'period': period,
            'devfactor': devfactor,
            'commission': commission,
            'final_value': final_value,
            'return_pct': ((final_value - self.initial_cash) / self.initial_cash) * 100,
            'sharpe_ratio': sharpe.get('sharperatio', 0) or 0,
            'max_drawdown': drawdown.get('max', {}).get('drawdown', 0),
            'total_trades': self._count_trades(strat)
        }
    
    def _count_trades(self, strategy) -> int:
        """นับจำนวนรายการซื้อขาย"""
        trade_analyzer = strategy.analyzers.trades.get_analysis()
        return trade_analyzer.get('total', {}).get('total', 0)
    
    def grid_search(
        self,
        periods: list,
        devfactors: list,
        commissions: list,
        workers: int = None
    ) -> pd.DataFrame:
        """
        Grid Search ด้วย Multi-processing
        
        Args:
            periods: List ของ period ที่จะทดสอบ
            devfactors: List ของ devfactor ที่จะทดสอบ
            commissions: List ของ commission ที่จะทดสอบ
            workers: จำนวน CPU cores ที่จะใช้ (default: all)
        """
        # สร้าง Parameter Grid
        param_combinations = list(itertools.product(
            periods, devfactors, commissions
        ))
        
        print(f"ทดสอบ {len(param_combinations)} combinations...")
        
        workers = workers or cpu_count()
        
        # ใช้ Multi-processing
        with Pool(workers) as pool:
            results = pool.map(self.evaluate_params, param_combinations)
        
        # เก็บผลลัพธ์
        self.results = pd.DataFrame(results)
        self.results = self.results.sort_values('sharpe_ratio', ascending=False)
        
        return self.results
    
    def get_best_params(self, metric: str = 'sharpe_ratio') -> dict:
        """
        ดึง Parameter ที่ดีที่สุดตาม Metric ที่กำหนด
        
        Args:
            metric: 'sharpe_ratio', 'return_pct', หรือ 'final_value'
        """
        if not self.results:
            raise ValueError("ต้องรัน Grid Search ก่อน")
        
        best = self.results.nlargest(1, metric).iloc[0]
        return best.to_dict()



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

def run_optimization(data_feed: pd.DataFrame):
    engine = OptimizationEngine(data_feed, initial_cash=100000)
    
    # กำหนด Parameter Space
    results = engine.grid_search(
        periods=[10, 15, 20, 25, 30],
        devfactors=[1.5, 2.0, 2.5, 3.0],
        commissions=[0.001, 0.002, 0.003],
        workers=4  # ใช้ 4 cores
    )
    
    # แสดง Top 5
    print("\nTop 5 Parameters:")
    print(results.head())
    
    # ดึง Best Parameters
    best = engine.get_best_params('sharpe_ratio')
    print(f"\nBest Parameters: {best}")
    
    # บันทึกผลลัพธ์
    results.to_csv('optimization_results.csv', index=False)
    
    return best

เหมาะกับใคร / ไม่เหมาะกับใคร

เหมาะกับใครไม่เหมาะกับใคร
นักเทรด Quant ที่ต้องการ Backtesting ด้วยข้อมูล Tick-level นักลงทุนมือใหม่ที่ยังไม่เข้าใจหลักการทางการเงิน
ระบบ Trading ที่ต้องการรองรับ High-Frequency หลายสิบ Symbol ผู้ที่มีงบประมาณจำกัดมาก เพราะค่าใช้จ่ายด้านข้อมูล + Compute สูง
ทีม Dev ที่ต้องการ CI/CD สำหรับ Strategy หลายตัว คนที่ต้องการ "รวยเร็ว" ด้วยการเทรดอัตโนมัติ
นักวิจัยที่ต้องทำ Academic Research เกี่ยวกับ Market Microstructure ผู้ที่ไม่มีทักษะ Programming เลย
Fund Manager ที่ต้องการ Validate Strategy ก่อน Live Trade ผู้ที่ไม่มีเวลาศึกษาและปรับปรุงอย่างต่อเนื่อง

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

1. Error: "Exceed Rate Limit" จาก Tardis.dev API

สาเหตุ: เรียก API บ่อยเกินไปหรือดึงข้อมูลเกิน Monthly Limit วิธีแก้ไข: ใช้ Rate Limiter และ Cache ข้อมูลที่ดึงแล้ว 
import time
from functools import wraps
from datetime import datetime
import pickle
import os

class RateLimitedFetcher:
    """
    Fetcher ที่มี Rate Limiting และ Local Caching
    ป้องกันการเรียก API เกิน Limit
    """
    
    def __init__(self, fetcher, requests_per_second: int = 5, cache_dir: str = './cache'):
        self.fetcher = fetcher
        self.requests_per_second = requests_per_second
        self.last_request = 0
        self.min_interval = 1.0 / requests_per_second
        self.cache_dir = cache_dir
        
        # สร้างโฟลเดอร์ Cache
        os.makedirs(cache_dir, exist_ok=True)
    
    def _get_cache_path(self, exchange: str, symbol: str, start: datetime, end: datetime) -> str:
        """สร้าง Path สำหรับ Cache File"""
        cache_key = f"{exchange}_{symbol}_{start.date()}_{end.date()}"
        return os.path.join(self.cache_dir, f"{cache_key}.pkl")
    
    def _is_cache_valid(self, cache_path: str, max_age_days: int = 7) -> bool:
        """ตรวจสอบว่า Cache ยัง valid หรือไม่"""
        if not os.path.exists(cache_path):
            return False
        
        cache_age = time.time() - os.path.getmtime(cache_path)
        return cache_age < (max_age_days * 86400)
    
    def _wait_for_rate_limit(self):
        """รอจนกว่าจะผ่าน Rate Limit"""
        elapsed = time.time() - self.last_request
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self.last_request = time.time()
    
    async def fetch_candles_cached(
        self,
        exchange: str,
        symbol: str,
        timeframe: str,
        start_date: datetime,
        end_date: datetime,
        use_cache: bool = True,
        refresh_cache: bool = False
    ) -> pd.DataFrame:
        """

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