การทำ Backtest ระบบเทรดคริปโตที่ดีต้องอาศัยข้อมูลราคาและ Order Book ย้อนหลังที่ครบถ้วนและแม่นยำ หลายคนอาจสงสัยว่าจะหาไฟล์ข้อมูล Binance Historical Order Book ได้จากที่ไหน เพื่อนำไปทดสอบกลยุทธ์การเทรดก่อนลงมือจริง

บทความนี้จะพาคุณทำความรู้จักกับวิธีการดาวน์โหลดข้อมูล Order Book จาก Binance อย่างละเอียดทีละขั้นตอน โดยไม่ต้องมีความรู้ด้านโค้ดมาก่อนก็สามารถทำได้

Order Book คืออะไร และทำไมต้องใช้สำหรับ Backtest

Order Book คือตารางที่แสดงคำสั่งซื้อและคำสั่งขายที่รอการจับคู่ในตลาด ณ แต่ละระดับราคา โดยแบ่งเป็น:

สำหรับการ Backtest ข้อมูล Order Book ช่วยให้คุณเห็น:

วิธีที่ 1: ใช้ Binance API โดยตรง

วิธีนี้เหมาะสำหรับผู้ที่มีความรู้พื้นฐาน Python เล็กน้อย และต้องการข้อมูลแบบ Real-time หรือ Historical ระยะสั้น

ขั้นตอนที่ 1: สมัคร Binance Account

หากยังไม่มีบัญชี ให้สมัครที่เว็บไซต์ Binance.com และเปิดใช้งาน API Key สำหรับการเข้าถึงข้อมูล

ขั้นตอนที่ 2: ติดตั้ง Python และ Library ที่จำเป็น

pip install python-binance pandas requests

ขั้นตอนที่ 3: ดาวน์โหลด Order Book Snapshot

import requests
import time
import json

def get_historical_orderbook(symbol, limit=1000, startTime=None, endTime=None):
    """
    ดึงข้อมูล Order Book ย้อนหลังจาก Binance
    symbol: คู่เทรด เช่น BTCUSDT
    limit: จำนวนระดับราคา (สูงสุด 1000)
    """
    
    base_url = "https://api.binance.com"
    endpoint = "/api/v3/depth"
    
    params = {
        "symbol": symbol,
        "limit": limit
    }
    
    if startTime:
        params["startTime"] = startTime
    if endTime:
        params["endTime"] = endTime
    
    response = requests.get(f"{base_url}{endpoint}", params=params)
    
    if response.status_code == 200:
        data = response.json()
        return {
            "lastUpdateId": data["lastUpdateId"],
            "bids": data["bids"],
            "asks": data["asks"],
            "timestamp": time.time()
        }
    else:
        print(f"Error: {response.status_code}")
        return None

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

orderbook = get_historical_orderbook("BTCUSDT", limit=500) if orderbook: print(f"Last Update ID: {orderbook['lastUpdateId']}") print(f"จำนวน Bid Levels: {len(orderbook['bids'])}") print(f"จำนวน Ask Levels: {len(orderbook['asks'])}") print(f"Top 5 Bids: {orderbook['bids'][:5]}")

ข้อจำกัดของวิธีนี้

วิธีที่ 2: ดาวน์โหลดจาก Kaggle Dataset

Kaggle มี Dataset สำหรับ Binance Order Book ที่ผู้ใช้รวบรวมไว้ให้ดาวน์โหลดฟรี

แหล่งข้อมูลที่แนะนำ

ขั้นตอนการดาวน์โหลด

  1. ไปที่ Kaggle.com และสร้าง Account
  2. ค้นหาคำว่า "Binance Order Book"
  3. เลือก Dataset ที่ต้องการ
  4. กดปุ่ม Download (อาจต้องติดตั้ง Kaggle CLI ก่อน)
# ติดตั้ง Kaggle CLI
pip install kaggle

ดาวน์โหลด Dataset

วางไฟล์ kaggle.json (API Token) ไว้ในโฟลเดอร์ ~/.kaggle/

kaggle datasets download -d username/dataset-name -p ./data/

แตกไฟล์

unzip ./data/dataset-name.zip -d ./data/orderbook/

ข้อจำกัดของวิธีนี้

วิธีที่ 3: ใช้บริการ Data Provider อื่น

หากต้องการข้อมูลที่สมบูรณ์และพร้อมใช้งานทันที สามารถใช้บริการ Data Provider ที่รวบรวมข้อมูล Order Book ไว้ให้

บริการ ราคา ความละเอียด ความเร็ว
Binance API (ฟรี) ฟรี 100ms จำกัด Rate Limit
Kaggle Dataset ฟรี - $50/เดือน 1-60 วินาที ต้องดาวน์โหลด
CCXT Library ฟรี ขึ้นอยู่กับ Exchange ปานกลาง
HolySheep AI เริ่มต้น $0.42/MTok ข้อมูลคุณภาพสูง <50ms

วิธีที่ 4: ใช้ CCXT Library (แนะนำสำหรับมือใหม่)

CCXT เป็น Library ยอดนิยมที่รวม API ของ Exchange หลายตัวไว้ในที่เดียว ทำให้การดึงข้อมูล Order Book ทำได้ง่ายมาก

pip install ccxt pandas
import ccxt
import pandas as pd
import time

class BinanceDataFetcher:
    def __init__(self):
        self.exchange = ccxt.binance({
            'enableRateLimit': True,
            'options': {'defaultType': 'spot'}
        })
    
    def fetch_orderbook(self, symbol, limit=1000):
        """
        ดึงข้อมูล Order Book ปัจจุบัน
        """
        try:
            orderbook = self.exchange.fetch_order_book(symbol, limit)
            return orderbook
        except Exception as e:
            print(f"Error fetching orderbook: {e}")
            return None
    
    def fetch_historical_books(self, symbol, since, limit=100):
        """
        ดึงข้อมูล Order Book ย้อนหลัง
        symbol: คู่เทรด เช่น BTC/USDT
        since: timestamp เริ่มต้น (milliseconds)
        limit: จำนวนครั้งที่ต้องการดึง
        """
        all_books = []
        current_time = since
        
        for i in range(limit):
            try:
                # ดึงข้อมูล
                orderbook = self.exchange.fetch_order_book(symbol, 1000)
                orderbook['timestamp'] = current_time
                all_books.append(orderbook)
                
                # เลื่อนเวลาไป 1 นาที (60000ms)
                current_time += 60000
                
                # รอรับ Rate Limit
                time.sleep(self.exchange.rateLimit / 1000)
                
                if i % 10 == 0:
                    print(f"ดึงข้อมูลแล้ว {i+1}/{limit} ครั้ง")
                    
            except Exception as e:
                print(f"Error at iteration {i}: {e}")
                time.sleep(5)
        
        return all_books
    
    def save_to_csv(self, orderbooks, filename):
        """
        บันทึกข้อมูล Order Book ลง CSV
        """
        rows = []
        for book in orderbooks:
            for bid in book['bids'][:10]:  # เอาเฉพาะ 10 อันดับแรก
                rows.append({
                    'timestamp': book.get('timestamp'),
                    'side': 'bid',
                    'price': bid[0],
                    'volume': bid[1]
                })
            for ask in book['asks'][:10]:
                rows.append({
                    'timestamp': book.get('timestamp'),
                    'side': 'ask',
                    'price': ask[0],
                    'volume': ask[1]
                })
        
        df = pd.DataFrame(rows)
        df.to_csv(filename, index=False)
        print(f"บันทึก {len(rows)} rows ลง {filename}")

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

if __name__ == "__main__": fetcher = BinanceDataFetcher() # ดึง Order Book ปัจจุบัน current_book = fetcher.fetch_orderbook("BTC/USDT", 100) print(f"Bid สูงสุด: {current_book['bids'][0]}") print(f"Ask ต่ำสุด: {current_book['asks'][0]}")

การเตรียมข้อมูล Order Book สำหรับ Backtest

เมื่อได้ข้อมูล Order Book มาแล้ว ต้องผ่านขั้นตอนการเตรียมข้อมูลก่อนนำไปใช้

import pandas as pd
import numpy as np

def prepare_orderbook_for_backtest(df):
    """
    เตรียมข้อมูล Order Book สำหรับ Backtest
    """
    # เรียงลำดับตาม timestamp
    df = df.sort_values('timestamp')
    
    # คำนวณ Volume สะสม (Cumulative Volume)
    df['cum_volume'] = df.groupby(['timestamp', 'side'])['volume'].cumsum()
    
    # คำนวณ Mid Price
    df['mid_price'] = (df['price'] * df['volume']).groupby(df['timestamp']).transform('sum') / df['cum_volume'].groupby(df['timestamp']).transform('max')
    
    # คำนวณ Spread
    df_pivot = df.pivot(index='timestamp', columns='side', values='price')
    df_pivot['spread'] = df_pivot['ask'] - df_pivot['bid']
    df_pivot['spread_pct'] = (df_pivot['spread'] / df_pivot['mid']) * 100
    
    # คำนวณ Market Depth
    df_depth = df.groupby(['timestamp', 'side'])['volume'].sum().unstack()
    df_depth['depth_ratio'] = df_depth['bid'] / df_depth['ask']
    
    return df_pivot, df_depth

โหลดข้อมูลที่บันทึกไว้

df = pd.read_csv('orderbook_data.csv')

เตรียมข้อมูล

pivot_data, depth_data = prepare_orderbook_for_backtest(df) print("ตัวอย่าง Spread:") print(pivot_data[['spread', 'spread_pct']].head(10)) print("\nตัวอย่าง Market Depth Ratio:") print(depth_data['depth_ratio'].describe())

การใช้ AI ช่วยวิเคราะห์ Order Book

หากต้องการให้การวิเคราะห์ Order Book ง่ายขึ้น สามารถใช้ AI ช่วยได้ โดย HolySheep AI เป็นอีกหนึ่งทางเลือกที่น่าสนใจ ราคาเริ่มต้นเพียง $0.42/MTok รองรับการประมวลผลข้อมูลจำนวนมากด้วยความเร็วต่ำกว่า 50ms

import requests
import json

def analyze_orderbook_with_ai(orderbook_data, api_key):
    """
    ใช้ AI วิเคราะห์ Order Book Pattern
    """
    base_url = "https://api.holysheep.ai/v1"
    
    prompt = f"""
    วิเคราะห์ Order Book ด้านล่างและระบุ:
    1. Order Book Imbalance (ถ้า bid > ask = bullish, ถ้า ask > bid = bearish)
    2. ระดับราคาที่มี Wall ขนาดใหญ่
    3. ความน่าจะเป็นที่ราคาจะเคลื่อนไหวไปทิศทางใด
    
    Order Book Data:
    Top 5 Bids: {orderbook_data['bids'][:5]}
    Top 5 Asks: {orderbook_data['asks'][:5]}
    """
    
    payload = {
        "model": "deepseek-v3.2",
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.3
    }
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    response = requests.post(
        f"{base_url}/chat/completions",
        headers=headers,
        json=payload
    )
    
    if response.status_code == 200:
        result = response.json()
        return result['choices'][0]['message']['content']
    else:
        return f"Error: {response.status_code}"

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

orderbook = { 'bids': [['94500', '2.5'], ['94450', '1.8'], ['94400', '3.2'], ['94350', '1.0'], ['94300', '0.5']], 'asks': [['94550', '0.3'], ['94600', '1.2'], ['94650', '2.0'], ['94700', '4.5'], ['94750', '1.0']] }

analysis = analyze_orderbook_with_ai(orderbook, "YOUR_HOLYSHEEP_API_KEY")

print(analysis)

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

เหมาะกับใคร ไม่เหมาะกับใคร
  • นักเทรดที่ต้องการทดสอบกลยุทธ์ด้วยข้อมูลจริง
  • นักพัฒนา Bot เทรดรายใหม่
  • ผู้ที่ต้องการข้อมูล Order Book ฟรี
  • นักวิจัยที่ต้องการข้อมูลระยะสั้น
  • ผู้ที่ต้องการข้อมูลย้อนหลังหลายปี
  • นักเทรดรายวันที่ต้องการข้อมูล Real-time
  • ผู้ที่ไม่มีเวลาประมวลผลข้อมูลเอง
  • องค์กรที่ต้องการข้อมูลคุณภาพระดับ Institutional

ราคาและ ROI

วิธีการ ค่าใช้จ่าย เวลาที่ใช้ คุณภาพข้อมูล
Binance API ฟรี (Rate Limit จำกัด) 2-4 ชั่วโมง/วัน ★★★★☆
Kaggle Dataset ฟรี - $50/เดือน 30 นาที - 2 ชั่วโมง ★★★☆☆
CCXT + ดึงเอง ฟรี 4-8 ชั่วโมง/สัปดาห์ ★★★★☆
HolySheep AI เริ่มต้น $0.42/MTok นาที ★★★★★

หมายเหตุ: HolySheep AI มีอัตราแลกเปลี่ยน ¥1=$1 ทำให้ประหยัดได้มากกว่า 85% เมื่อเทียบกับบริการอื่น พร้อมรองรับ WeChat/Alipay และให้เครดิตฟรีเมื่อลงทะเบียน

ทำไมต้องเลือก HolySheep

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

1. Error 429: Rate Limit Exceeded

# ปัญหา: ส่ง request เร็วเกินไป ถูก Binance บล็อก

import time
import requests

def safe_request(url, params, max_retries=5):
    """
    ส่ง request อย่างปลอดภัย โดยรอเมื่อเจอ Rate Limit
    """
    headers = {
        "X-MBX-APIKEY": "YOUR_API_KEY"
    }
    
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, headers=headers)
            
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 429:
                # รอ 60 วินาที แล้วลองใหม่
                wait_time = 60 * (attempt + 1)
                print(f"Rate Limited! รอ {wait_time} วินาที...")
                time.sleep(wait_time)
            else:
                print(f"HTTP Error: {response.status_code}")
                return None
                
        except Exception as e:
            print(f"Request Error: {e}")
            time.sleep(5)
    
    print("ล้มเหลวหลังจากลอง 5 ครั้ง")
    return None

วิธีแก้ไข: เพิ่ม delay และ retry logic

ลดจำนวน request ต่อวินาที

ใช้ WebSocket แทน REST API สำหรับข้อมูล Real-time

2. ข้อมูล Order Book ไม่ครบถ้วนหรือมี Gap

# ปัญหา: ดาวน์โหลดข้อมูลมาแล้ว มีช่วงเวลาหายไป

import pandas as pd
from datetime import datetime, timedelta

def validate_orderbook_completeness(df, expected_interval_ms=60000):
    """
    ตรวจสอบว่าข้อมูล Order Book มีความต่อเนื่องหรือไม่
    """
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df = df.sort_values('timestamp')
    
    # คำนวณช่วงเวลาที่คาดหวัง
    df['time_diff'] = df['timestamp'].diff().dt.total_seconds() * 1000
    
    # หาช่วงที่มี Gap
    gaps = df[df['time_diff'] > expected_interval_ms * 1.5]
    
    if len(gaps) > 0:
        print(f"พบ Gap {len(gaps)} จุดในข้อมูล")
        print("ตัวอย่าง Gap:")
        print(gaps[['timestamp', 'time_diff']].head(10))
        return gaps
    else:
        print("ข้อมูลมีความต่อเนื่อง")
        return pd.DataFrame()

def fill_gaps(df, gaps, fetcher):
    """
    เติมข้อมูลที่ขาดหายไป
    """
    filled_data = []
    
    for idx, row in gaps.iterrows():
        gap_start = row['timestamp'] - timedelta(minutes=1)
        gap_end = row['timestamp'] + timedelta(minutes=1)
        
        # ดึงข้อมูลชดเชย
        new_data = fetcher.fetch_orderbook(
            "BTC/USDT",
            limit=500
        )
        
        if new_data:
            new_data['timestamp'] = gap_start
            filled_data.append(new_data)
    
    # รวมข้อมูล
    all_data = pd.concat([df, pd.DataFrame(filled_data)])
    return all_data.sort_values('timestamp')

วิธีแก้ไข:

1. ตรวจสอบความต่อเนื่องของข้อมูลก่อนใช้งานเสมอ

2. ดาวน์โหลดข้อมูลสำรองเผื่อไว้

3. ใช้ WebSocket สำหรับข้อมูล Real-time ที่ต้องการความต่อเนื่อง

3. ไฟล์ CSV ใหญ่เกินจนเปิดไม่ได้

# ป