การทำ Quantitative Trading โดยเฉพาะอย่างยิ่งการ Backtest กลยุทธ์ที่เราออกแบบนั้น จำเป็นต้องมีข้อมูล Orderbook ย้อนหลังที่มีความถูกต้องและครบถ้วน หลายคนมักประสบปัญหาในการเข้าถึงข้อมูล Historical Orderbook ของ Binance วันนี้เราจะมาดูวิธีการเข้าถึงข้อมูลเหล่านี้อย่างละเอียด พร้อมโค้ดตัวอย่างที่พร้อมใช้งานจริง

ทำไมต้องใช้ข้อมูล Historical Orderbook?

ข้อมูล Orderbook เป็นหัวใจสำคัญในการวิเคราะห์ตลาดและออกแบบกลยุทธ์การเทรด เพราะมันบอกรายละเอียดเกี่ยวกับ:

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

Binance มี Public API สำหรับดึงข้อมูล Historical Orderbook ผ่าน endpoint /api/v3/depth แต่ต้องระวังว่า Binance ไม่เก็บข้อมูล Orderbook ย้อนหลังไว้นาน ปกติจะเก็บเพียงไม่กี่วันเท่านั้น

ติดตั้งและใช้งาน Binance Connector

pip install binance-connector pandas numpy
import time
import pandas as pd
from binance.spot import Spot as Client
from binance.error import ClientError

class BinanceOrderbookFetcher:
    """คลาสสำหรับดึงข้อมูล Historical Orderbook จาก Binance"""
    
    def __init__(self, symbol='BTCUSDT', limit=1000):
        self.client = Client(base_url='https://api.binance.com')
        self.symbol = symbol.upper()
        self.limit = limit  # จำนวนระดับราคาที่ต้องการ (1-5000)
        
    def get_orderbook_snapshot(self, timestamp=None):
        """
        ดึงข้อมูล Orderbook Snapshot ณ ปัจจุบัน
        ข้อมูลจะถูกจำกัดไว้ที่ 500 ระดับราคาต่อครั้ง
        """
        try:
            params = {
                'symbol': self.symbol,
                'limit': min(self.limit, 1000)  # สูงสุด 1000 ระดับ
            }
            
            response = self.client.depth(**params)
            
            orderbook_data = {
                'timestamp': response.get('lastUpdateId'),
                'bids': response.get('bids', []),
                'asks': response.get('asks', []),
                'lastUpdateId': response.get('lastUpdateId')
            }
            
            return orderbook_data
            
        except ClientError as error:
            print(f"❌ Client Error: {error.error_code} - {error.error_message}")
            return None
    
    def get_orderbook_dataframe(self):
        """แปลงข้อมูล Orderbook เป็น DataFrame สำหรับวิเคราะห์"""
        orderbook = self.get_orderbook_snapshot()
        
        if orderbook is None:
            return None
        
        # สร้าง DataFrame สำหรับ Bids (คำสั่งซื้อ)
        bids_df = pd.DataFrame(orderbook['bids'], 
                               columns=['price', 'quantity'],
                               dtype=float)
        
        # สร้าง DataFrame สำหรับ Asks (คำสั่งขาย)
        asks_df = pd.DataFrame(orderbook['asks'], 
                               columns=['price', 'quantity'],
                               dtype=float)
        
        return {
            'timestamp': orderbook['timestamp'],
            'bids': bids_df,
            'asks': asks_df
        }
    
    def calculate_market_depth(self, levels=10):
        """คำนวณ Market Depth สำหรับการวิเคราะห์"""
        data = self.get_orderbook_dataframe()
        
        if data is None:
            return None
        
        bids = data['bids'].head(levels)
        asks = data['asks'].head(levels)
        
        total_bid_volume = (bids['price'] * bids['quantity']).sum()
        total_ask_volume = (asks['price'] * asks['quantity']).sum()
        
        return {
            'total_bid_volume': total_bid_volume,
            'total_ask_volume': total_ask_volume,
            'bid_ask_ratio': total_bid_volume / total_ask_volume if total_ask_volume > 0 else 0,
            'spread': float(asks['price'].min() - bids['price'].max()) if len(asks) > 0 and len(bids) > 0 else 0,
            'spread_percentage': float((asks['price'].min() - bids['price'].max()) / asks['price'].min() * 100) if len(asks) > 0 and len(bids) > 0 else 0
        }


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

fetcher = BinanceOrderbookFetcher(symbol='BTCUSDT', limit=500)
orderbook_data = fetcher.get_orderbook_dataframe()

if orderbook_data:
    print(f"📊 ข้อมูล Orderbook ณ เวลา: {orderbook_data['timestamp']}")
    print(f"\n💚 Top 5 Bids (คำสั่งซื้อ):")
    print(orderbook_data['bids'].head())
    print(f"\n❤️ Top 5 Asks (คำสั่งขาย):")
    print(orderbook_data['asks'].head())
    
    depth = fetcher.calculate_market_depth(levels=10)
    print(f"\n📈 Market Depth Analysis:")
    print(f"   Total Bid Volume: ${depth['total_bid_volume']:,.2f}")
    print(f"   Total Ask Volume: ${depth['total_ask_volume']:,.2f}")
    print(f"   Bid/Ask Ratio: {depth['bid_ask_ratio']:.4f}")
    print(f"   Spread: ${depth['spread']:.2f} ({depth['spread_percentage']:.4f}%)")

วิธีที่ 2: ใช้ Binance WebSocket สำหรับ Real-time + Historical Data

สำหรับการทำ Backtest ที่ต้องการข้อมูลย้อนหลังหลายวัน วิธีที่ดีที่สุดคือการใช้ WebSocket ร่วมกับการเก็บข้อมูลอย่างต่อเนื่อง หรือใช้บริการจากผู้ให้บริการข้อมูล third-party

import websocket
import json
import pandas as pd
import threading
import sqlite3
from datetime import datetime, timedelta
import time

class BinanceOrderbookRecorder:
    """
    คลาสสำหรับบันทึกข้อมูล Orderbook ผ่าน WebSocket
    เหมาะสำหรับการเก็บข้อมูลย้อนหลังเพื่อใช้ใน Backtesting
    """
    
    def __init__(self, symbol='btcusdt', db_path='orderbook_data.db'):
        self.symbol = symbol.lower()
        self.stream_url = f"wss://stream.binance.com:9443/ws/{self.symbol}@depth20@100ms"
        self.db_path = db_path
        self.is_running = False
        self.recorded_data = []
        self._setup_database()
        
    def _setup_database(self):
        """สร้างตารางใน SQLite สำหรับเก็บข้อมูล"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS orderbook_snapshots (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp INTEGER,
                datetime TEXT,
                bid_price REAL,
                bid_quantity REAL,
                ask_price REAL,
                ask_quantity REAL,
                update_id INTEGER
            )
        ''')
        
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_timestamp 
            ON orderbook_snapshots(timestamp)
        ''')
        
        conn.commit()
        conn.close()
        print(f"✅ Database setup complete: {self.db_path}")
    
    def on_message(self, ws, message):
        """จัดการเมื่อได้รับข้อความใหม่จาก WebSocket"""
        try:
            data = json.loads(message)
            
            if 'bids' in data and 'asks' in data:
                timestamp = data.get('lastUpdateId', int(time.time() * 1000))
                dt = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')
                
                # บันทึก top 5 levels ของ orderbook
                for i, (bid, ask) in enumerate(zip(data['bids'][:5], data['asks'][:5])):
                    self._save_to_database(
                        timestamp=timestamp,
                        datetime_str=dt,
                        bid_price=float(bid[0]),
                        bid_quantity=float(bid[1]),
                        ask_price=float(ask[0]),
                        ask_quantity=float(ask[1]),
                        update_id=timestamp
                    )
                    
        except Exception as e:
            print(f"❌ Error processing message: {e}")
    
    def _save_to_database(self, timestamp, datetime_str, bid_price, 
                          bid_quantity, ask_price, ask_quantity, update_id):
        """บันทึกข้อมูลลง SQLite"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                INSERT INTO orderbook_snapshots 
                (timestamp, datetime, bid_price, bid_quantity, 
                 ask_price, ask_quantity, update_id)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (timestamp, datetime_str, bid_price, bid_quantity,
                  ask_price, ask_quantity, update_id))
            
            conn.commit()
            conn.close()
            
        except Exception as e:
            print(f"❌ Database error: {e}")
    
    def on_error(self, ws, error):
        """จัดการเมื่อเกิดข้อผิดพลาด"""
        print(f"❌ WebSocket Error: {error}")
    
    def on_close(self, ws, close_status_code, close_msg):
        """จัดการเมื่อ WebSocket ถูกปิด"""
        print(f"🔴 WebSocket closed: {close_status_code} - {close_msg}")
    
    def on_open(self, ws):
        """จัดการเมื่อ WebSocket เปิด"""
        print(f"🟢 WebSocket connected to {self.stream_url}")
        self.is_running = True
    
    def start_recording(self, duration_seconds=None):
        """
        เริ่มบันทึกข้อมูล Orderbook
        duration_seconds: ระยะเวลาการบันทึก (ถ้าไม่ระบุจะบันทึกต่อเนื่อง)
        """
        ws = websocket.WebSocketApp(
            self.stream_url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )
        
        ws_thread = threading.Thread(target=ws.run_forever)
        ws_thread.daemon = True
        ws_thread.start()
        
        print(f"📍 เริ่มบันทึกข้อมูล Orderbook สำหรับ {self.symbol}")
        
        try:
            if duration_seconds:
                time.sleep(duration_seconds)
                ws.close()
            else:
                # บันทึกจนกว่าจะกด Ctrl+C
                while self.is_running:
                    time.sleep(1)
        except KeyboardInterrupt:
            print("\n⏹️ หยุดบันทึกข้อมูล")
            ws.close()
    
    def load_data_to_dataframe(self, start_time=None, end_time=None):
        """โหลดข้อมูลจาก database เป็น DataFrame"""
        conn = sqlite3.connect(self.db_path)
        
        query = "SELECT * FROM orderbook_snapshots WHERE 1=1"
        params = []
        
        if start_time:
            query += " AND timestamp >= ?"
            params.append(start_time)
        if end_time:
            query += " AND timestamp <= ?"
            params.append(end_time)
        
        df = pd.read_sql_query(query, conn, params=params if params else None)
        conn.close()
        
        return df
    
    def get_orderbook_statistics(self):
        """สร้างสถิติจากข้อมูลที่บันทึกไว้"""
        conn = sqlite3.connect(self.db_path)
        
        stats_query = '''
            SELECT 
                COUNT(*) as total_records,
                MIN(datetime) as start_time,
                MAX(datetime) as end_time,
                AVG(bid_quantity) as avg_bid_qty,
                AVG(ask_quantity) as avg_ask_qty,
                AVG(ask_price - bid_price) as avg_spread
            FROM orderbook_snapshots
        '''
        
        stats = pd.read_sql_query(stats_query, conn).iloc[0]
        conn.close()
        
        return stats


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

recorder = BinanceOrderbookRecorder(symbol='btcusdt', db_path='btc_orderbook.db')


เริ่มบันทึกข้อมูล 60 วินาที

print("🔄 เริ่มบันทึกข้อมูล Orderbook สำหรับ 60 วินาที...")
recorder.start_recording(duration_seconds=60)


โหลดข้อมูลและแสดงสถิติ

stats = recorder.get_orderbook_statistics()
print(f"\n📊 สถิติข้อมูลที่บันทึก:")
print(f"   จำนวน records: {stats['total_records']:,}")
print(f"   ช่วงเวลา: {stats['start_time']} ถึง {stats['end_time']}")
print(f"   Spread เฉลี่ย: ${stats['avg_spread']:.2f}")

วิธีที่ 3: ใช้บริการ Third-Party Data Provider

สำหรับนักพัฒนาที่ต้องการข้อมูล Orderbook ย้อนหลังนานกว่า 7 วัน การใช้บริการจาก Data Provider เฉพาะทางเป็นทางเลือกที่ดีกว่า เนื่องจาก Binance ไม่ได้เก็บข้อมูลราคาละเอียดระดับ Orderbook ไว้ให้นานเกินไป

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

กลุ่มผู้ใช้งาน เหมาะกับ ไม่เหมาะกับ
นักเทรดรายวัน (Day Trader) ใช้ Binance API โดยตรง เพราะต้องการข้อมูลปัจจุบันและไม่ต้องการข้อมูลย้อนหลังนาน ผู้ที่ต้องการข้อมูลย้อนหลังหลายเดือน
Quantitative Researcher ใช้ WebSocket + Local Database หรือ Third-party Provider เพื่อเก็บข้อมูลละเอียด ผู้ที่มีงบประมาณจำกัดและต้องการแค่ข้อมูล OHLCV
สถาบันหรือกองทุน ใช้บริการ premium เช่น CoinAPI, Kaiko ที่ให้ข้อมูลระดับ Tick-by-Tick พร้อมคุณภาพ guaranteed ผู้ที่เริ่มต้นศึกษาหรือทำโปรเจกต์ขนาดเล็ก
นักพัฒนา AI/ML ใช้ HolySheep AI สำหรับวิเคราะห์ข้อมูลและสร้างโมเดล prediction ด้วย LLM ราคาถูก ผู้ที่ต้องการข้อมูล Real-time สดๆ ตลอดเวลา

ราคาและ ROI

สำหรับการทำ Quantitative Research การเลือกเครื่องมือที่เหมาะสมจะช่วยประหยัดต้นทุนได้มาก โดยเฉพาะค่า API และค่า Compute สำหรับการประมวลผลข้อมูลขนาดใหญ่

บริการ ราคา/เดือน (USD) ปริมาณข้อมูล ความคุ้มค่า
Binance API (Free) $0 500 requests/นาที ⭐⭐⭐ ดีสำหรับเริ่มต้น
CoinAPI $79 - $399 10K - 100K requests/วัน ⭐⭐ เหมาะกับองค์กร
Kaiko $500+ Unlimited historical ⭐⭐ แพงแต่ครบ
HolySheep AI เริ่มต้นฟรี API + Compute ⭐⭐⭐⭐⭐ ประหยัด 85%+

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

สำหรับนักพัฒนา Quantitative Trading ที่ต้องการใช้ LLM ในการวิเคราะห์ข้อมูล Orderbook และสร้างกลยุทธ์ สมัครที่นี่ HolySheep AI เป็นตัวเลือกที่คุ้มค่าที่สุด:

ราคา LLM 2026 (ราคาต่อล้าน Token):

โมเดล ราคาปกติ (USD) ราคา HolySheep (USD) ประหยัด
GPT-4.1 $15-60 $8 ~47-87%
Claude Sonnet 4.5 $30 $15 50%
Gemini 2.5 Flash $10 $2.50 75%
DeepSeek V3.2 $2.80 $0.42 85%

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

1. Rate Limit Error 429

ปัญหา: เมื่อส่ง request ไปยัง Binance API เกินจำนวนที่กำหนด จะได้รับ error 429

import time
from binance.error import ClientError

def safe_api_call(func, max_retries=3, backoff_factor=2):
    """
    ฟังก์ชันสำหรับเรียก API อย่างปลอดภัยพร้อม retry logic
    """
    for attempt in range(max_retries):
        try:
            result = func()
            return result
            
        except ClientError as e:
            if e.error_code == -1003:  # Rate limit error
                wait_time = backoff_factor ** attempt
                print(f"⚠️ Rate limit hit, waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                print(f"❌ API Error: {e.error_message}")
                return None
                
        except Exception as e:
            print(f"❌ Unexpected error: {e}")
            return None
    
    print(f"❌ Max retries ({max_retries}) exceeded")
    return None


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

fetcher = BinanceOrderbookFetcher()
result = safe_api_call(lambda: fetcher.get_orderbook_snapshot())

2. WebSocket Disconnection บ่อยครั้ง

ปัญหา: WebSocket หลุดการเชื่อมต่อบ่อยทำให้ข้อมูลขาดหาย

import websocket
import threading
import time

class RobustWebSocketConnection:
    """WebSocket ที่มีระบบ reconnect อัตโนมัติ"""
    
    def __init__(self, url, reconnect_delay=5, max_reconnects=10):
        self.url = url
        self.reconnect_delay = reconnect_delay
        self.max_reconnects = max_reconnects
        self.ws = None
        self.reconnect_count = 0
        self.is_running = False
        
    def start(self):
        """เริ่มเชื่อมต่อพร้อม auto-reconnect"""
        self.is_running = True
        
        while self.is_running and self.reconnect_count < self.max_reconnects:
            try:
                print(f"🔄 พยายามเชื่อมต่อครั้งที่ {self.reconnect_count + 1}")
                
                self.ws = websocket.WebSocketApp(
                    self.url,
                    on_message=self.on_message,
                    on_error=self.on_error,
                    on_close=self.on_close,
                    on_open=self.on_open
                )
                
                # ใช้ run_forever พร้อม ping_interval เพื่อรักษาการเชื่อมต่อ
                self.ws.run_forever(ping_interval=30, ping_timeout=10)
                
            except Exception as e:
                print(f"❌ Connection error: {e}")
                self.reconnect_count += 1
                
                if self.reconnect_count < self.max_reconnects:
                    print(f"⏰ รอ {self.reconnect_delay} วินาทีก่อนเชื่อมต่อใหม่...")
                    time.sleep(self.reconnect_delay)
                    # เพิ่ม delay ทุกครั้งที่ reconnect
                    self.reconnect_delay = min(self.reconnect_delay * 1.5, 60)
    
    def on

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