สรุปคำตอบสำคัญ

บทความนี้จะอธิบายวิธีดาวน์โหลดข้อมูล Bybit incremental_book_L2 สำหรับงาน Quantitative Backtesting ด้วยไฟล์ CSV อย่างละเอียด พร้อมแนะนำวิธีใช้ HolySheep AI เพื่อประมวลผลข้อมูล Order Book ด้วย AI ที่มีความหน่วงต่ำกว่า 50 มิลลิวินาที และประหยัดค่าใช้จ่ายได้มากกว่า 85% เมื่อเทียบกับการใช้ API ทางการ

incremental_book_L2 คืออะไร และทำไมต้องใช้ในการ Backtesting

ข้อมูล incremental_book_L2 คือข้อมูล Order Book แบบอัปเดตทีละระดับราคา (Level 2) ที่ส่งมาจาก Bybit WebSocket API โดยจะส่งเฉพาะการเปลี่ยนแปลงของราคาและปริมาณคำสั่งซื้อขาย (Delta Update) แทนที่จะส่งข้อมูลทั้งหมดทุกครั้ง ทำให้ประหยัดแบนด์วิดท์และเหมาะสำหรับการทำ Quantitative Backtesting ที่ต้องการข้อมูลราคาละเอียดระดับ Tick-by-Tick

ข้อดีของ incremental_book_L2 สำหรับ Backtesting

วิธีดาวน์โหลด Bybit incremental_book_L2 สำหรับ CSV Export

ขั้นตอนที่ 1: เตรียมข้อมูล Order Book จาก Bybit

#!/usr/bin/env python3
"""
Bybit incremental_book_L2 Data Fetcher
สำหรับดาวน์โหลดข้อมูล Order Book เพื่อใช้ใน Quantitative Backtesting
"""

import json
import csv
import time
import asyncio
from datetime import datetime
from websockets.client import connect
from typing import Dict, List, Optional

class BybitBookL2Downloader:
    """ดาวน์โหลดข้อมูล incremental_book_L2 จาก Bybit WebSocket"""
    
    BYBIT_WS_URL = "wss://stream.bybit.com/v5/public/linear"
    
    def __init__(self, symbol: str = "BTCUSDT"):
        self.symbol = symbol
        self.order_book: Dict = {"b": [], "a": []}  # bids, asks
        self.snapshots: List[Dict] = []
        self.delta_updates: List[Dict] = []
        self.is_connected = False
    
    async def fetch_snapshot(self) -> bool:
        """ดึงข้อมูล Snapshot ครั้งแรก (Order Book ณ ช่วงเวลาหนึ่ง)"""
        try:
            # ใช้ REST API เพื่อดึง Snapshot
            import aiohttp
            url = f"https://api.bybit.com/v5/market/orderbook"
            params = {"category": "linear", "symbol": self.symbol, "limit": 500}
            
            async with aiohttp.ClientSession() as session:
                async with session.get(url, params=params) as response:
                    if response.status == 200:
                        data = await response.json()
                        if data.get("retCode") == 0:
                            result = data["result"]
                            self.order_book = {
                                "b": result.get("b", []),  # Bids
                                "a": result.get("a", [])   # Asks
                            }
                            return True
            return False
        except Exception as e:
            print(f"Error fetching snapshot: {e}")
            return False
    
    async def subscribe_incremental(self):
        """Subscribe เพื่อรับข้อมูล Delta Update ผ่าน WebSocket"""
        subscribe_msg = {
            "op": "subscribe",
            "args": [f"orderbook.500.{self.symbol}"]
        }
        await self.ws.send(json.dumps(subscribe_msg))
        print(f"✅ Subscribed to incremental_book_L2 for {self.symbol}")
    
    def process_delta(self, data: Dict):
        """ประมวลผล Delta Update และอัปเดต Order Book"""
        update_data = data.get("data", {})
        
        # อัปเดต Bids
        if "b" in update_data:
            for price, qty, *_ in update_data["b"]:
                self._update_level("b", price, qty)
        
        # อัปเดต Asks
        if "a" in update_data:
            for price, qty, *_ in update_data["a"]:
                self._update_level("a", price, qty)
        
        # บันทึก Delta Update
        self.delta_updates.append({
            "timestamp": datetime.now().isoformat(),
            "type": "delta",
            "b": update_data.get("b", []),
            "a": update_data.get("a", [])
        })
    
    def _update_level(self, side: str, price: float, qty: float):
        """อัปเดตระดับราคาเฉพาะ (Price Level)"""
        levels = self.order_book[side]
        
        # หา Index ของราคานี้
        for i, level in enumerate(levels):
            if float(level[0]) == float(price):
                if float(qty) == 0:
                    # ลบราคานี้ออก (Qty = 0 หมายถึงลบ)
                    levels.pop(i)
                else:
                    # อัปเดต Qty
                    levels[i][1] = qty
                return
        
        # ถ้าไม่มีราคานี้ใน Order Book ให้เพิ่มใหม่
        if float(qty) > 0:
            levels.append([price, qty])
            # เรียงลำดับใหม่
            if side == "b":
                levels.sort(key=lambda x: float(x[0]), reverse=True)
            else:
                levels.sort(key=lambda x: float(x[0]))
    
    def export_to_csv(self, filename: str = "bybit_book_l2.csv"):
        """ส่งออกข้อมูลเป็นไฟล์ CSV สำหรับ Backtesting"""
        with open(filename, "w", newline="") as f:
            writer = csv.writer(f)
            
            # Header
            writer.writerow([
                "timestamp", "type", "side", "price", "qty", 
                "mid_price", "spread", "bid_volume", "ask_volume"
            ])
            
            # ส่งออกทั้ง Snapshots และ Deltas
            for update in self.delta_updates:
                writer.writerow([
                    update["timestamp"],
                    update["type"],
                    "",
                    "",
                    "",
                    self._calc_mid_price(),
                    self._calc_spread(),
                    self._calc_total_volume("b"),
                    self._calc_total_volume("a")
                ])
                
                # รายละเอียดของแต่ละ Update
                for side in ["b", "a"]:
                    for level in update.get(side, []):
                        writer.writerow([
                            update["timestamp"],
                            update["type"],
                            side,
                            level[0],
                            level[1],
                            "",
                            "",
                            "",
                            ""
                        ])
        
        print(f"✅ ส่งออกไฟล์ CSV แล้ว: {filename}")
        return filename
    
    def _calc_mid_price(self) -> Optional[float]:
        """คำนวณราคา Mid (Average ของ Bid และ Ask สูงสุด)"""
        if self.order_book["b"] and self.order_book["a"]:
            best_bid = float(self.order_book["b"][0][0])
            best_ask = float(self.order_book["a"][0][0])
            return (best_bid + best_ask) / 2
        return None
    
    def _calc_spread(self) -> Optional[float]:
        """คำนวณ Spread"""
        if self.order_book["b"] and self.order_book["a"]:
            best_bid = float(self.order_book["b"][0][0])
            best_ask = float(self.order_book["a"][0][0])
            return best_ask - best_bid
        return None
    
    def _calc_total_volume(self, side: str) -> float:
        """คำนวณ Total Volume ของ Bid หรือ Ask"""
        return sum(float(level[1]) for level in self.order_book[side])


วิธีใช้งาน

async def main():
    downloader = BybitBookL2Downloader(symbol="BTCUSDT")
    
    # ดึง Snapshot ครั้งแรก
    snapshot_ok = await downloader.fetch_snapshot()
    if snapshot_ok:
        print("✅ ดึงข้อมูล Snapshot สำเร็จ")
    
    # เริ่ม Subscribe เพื่อรับ Delta Updates
    # (โค้ดเต็มดูในภาคต่อ...)
    
    # ส่งออก CSV
    downloader.export_to_csv("btcusdt_orderbook_2026.csv")

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

ขั้นตอนที่ 2: ดาวน์โหลดข้อมูลย้อนหลัง (Historical Data)

#!/usr/bin/env python3
"""
Bybit Historical Data Downloader
ดาวน์โหลดข้อมูล Order Book ในอดีตสำหรับ Backtesting
"""

import requests
import csv
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional

class BybitHistoricalDownloader:
    """ดาวน์โหลดข้อมูล Order Book ในอดีตจาก Bybit"""
    
    REST_BASE_URL = "https://api.bybit.com"
    
    def __init__(self, api_key: str = None, api_secret: str = None):
        self.api_key = api_key
        self.api_secret = api_secret
    
    def download_klines(
        self, 
        symbol: str = "BTCUSDT", 
        interval: str = "1",  # 1, 3, 5, 15, 30, 60, 120, 240, 300, 900, 1800, 3600, 7200, 14400, 28800, 43200, 86400, 604800, 2592000
        start_time: int = None,
        end_time: int = None,
        limit: int = 1000
    ) -> List[Dict]:
        """
        ดาวน์โหลดข้อมูล Candlestick (Klines) ในอดีต
        สำหรับใช้ร่วมกับ Order Book ในการ Backtesting
        """
        endpoint = f"{self.REST_BASE_URL}/v5/market/kline"
        
        params = {
            "category": "linear",
            "symbol": symbol,
            "interval": interval,
            "limit": min(limit, 1000),
        }
        
        if start_time:
            params["start"] = start_time
        if end_time:
            params["end"] = end_time
        
        response = requests.get(endpoint, params=params)
        data = response.json()
        
        if data.get("retCode") == 0:
            return data.get("result", {}).get("list", [])
        else:
            raise Exception(f"API Error: {data.get('retMsg')}")
    
    def download_orderbook_snapshot(
        self,
        symbol: str = "BTCUSDT",
        limit: int = 500
    ) -> Dict:
        """
        ดาวน์โหลด Order Book Snapshot ปัจจุบัน
        """
        endpoint = f"{self.REST_BASE_URL}/v5/market/orderbook"
        
        params = {
            "category": "linear",
            "symbol": symbol,
            "limit": limit
        }
        
        response = requests.get(endpoint, params=params)
        data = response.json()
        
        if data.get("retCode") == 0:
            return data.get("result", {})
        else:
            raise Exception(f"API Error: {data.get('retMsg')}")
    
    def download_funding_rate(
        self,
        symbol: str = "BTCUSDT",
        start_time: int = None,
        end_time: int = None,
        limit: int = 200
    ) -> List[Dict]:
        """
        ดาวน์โหลดข้อมูล Funding Rate สำหรับ Backtesting
        สำคัญสำหรับกลยุทธ์ที่เกี่ยวกับ Funding Arbitrage
        """
        endpoint = f"{self.REST_BASE_URL}/v5/market/funding/history"
        
        params = {
            "category": "linear",
            "symbol": symbol,
            "limit": min(limit, 200)
        }
        
        if start_time:
            params["startTime"] = start_time
        if end_time:
            params["endTime"] = end_time
        
        response = requests.get(endpoint, params=params)
        data = response.json()
        
        if data.get("retCode") == 0:
            return data.get("result", {}).get("list", [])
        else:
            raise Exception(f"API Error: {data.get('retMsg')}")
    
    def download_with_pagination(
        self,
        symbol: str = "BTCUSDT",
        interval: str = "1",
        days_back: int = 30
    ) -> List[List]:
        """
        ดาวน์โหลดข้อมูลย้อนหลังหลายๆ หน้า
        """
        all_data = []
        
        # คำนวณเวลาเริ่มต้น
        end_time = int(datetime.now().timestamp() * 1000)
        start_time = int(
            (datetime.now() - timedelta(days=days_back)).timestamp() * 1000
        )
        
        current_end = end_time
        
        while current_end > start_time:
            try:
                data = self.download_klines(
                    symbol=symbol,
                    interval=interval,
                    start_time=start_time,
                    end_time=current_end,
                    limit=1000
                )
                
                if not data:
                    break
                
                all_data.extend(data)
                print(f"📥 ดาวน์โหลดได้ {len(data)} records, รวม: {len(all_data)}")
                
                # ใช้เวลาของ Record สุดท้ายเป็น End Time สำหรับหน้าถัดไป
                current_end = int(data[-1][0]) - 1
                
                # รอเพื่อไม่ให้ถูก Rate Limit
                time.sleep(0.2)
                
            except Exception as e:
                print(f"❌ Error: {e}")
                break
        
        # กลับลำดับ (เก่าสุดไปใหม่สุด)
        all_data.reverse()
        return all_data
    
    def export_to_csv(
        self, 
        data: List, 
        filename: str,
        include_headers: bool = True
    ):
        """ส่งออกข้อมูลเป็น CSV"""
        
        headers = [
            "timestamp", "open", "high", "low", "close", 
            "volume", "turnover"
        ]
        
        with open(filename, "w", newline="") as f:
            writer = csv.writer(f)
            
            if include_headers:
                writer.writerow(headers)
            
            for record in data:
                # record format: [startTime, open, high, low, close, volume, turnover]
                writer.writerow([
                    datetime.fromtimestamp(int(record[0]) / 1000).isoformat(),
                    record[1],  # open
                    record[2],  # high
                    record[3],  # low
                    record[4],  # close
                    record[5],  # volume
                    record[6]   # turnover
                ])
        
        print(f"✅ ส่งออก CSV: {filename} ({len(data)} records)")
        return filename


วิธีใช้งาน

if __name__ == "__main__":
    downloader = BybitHistoricalDownloader()
    
    # ดาวน์โหลดข้อมูลย้อนหลัง 30 วัน
    print("📥 เริ่มดาวน์โหลดข้อมูล BTCUSDT ย้อนหลัง 30 วัน...")
    
    klines = downloader.download_with_pagination(
        symbol="BTCUSDT",
        interval="1",  # 1 นาที
        days_back=30
    )
    
    # ส่งออก CSV
    downloader.export_to_csv(klines, "btcusdt_klines_30d.csv")
    
    # ดาวน์โหลด Order Book Snapshot
    print("📥 ดาวน์โหลด Order Book Snapshot...")
    snapshot = downloader.download_orderbook_snapshot("BTCUSDT")
    print(f"Best Bid: {snapshot['b'][0]}, Best Ask: {snapshot['a'][0]}")

ขั้นตอนที่ 3: ใช้ HolySheep AI วิเคราะห์ข้อมูล Order Book

#!/usr/bin/env python3
"""
HolySheep AI - Order Book Analysis for Backtesting
ใช้ HolySheep AI เพื่อวิเคราะห์ข้อมูล Order Book และสร้าง Signals
"""

import os
import json
import csv
import httpx
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass


========================


HolySheep AI Configuration


========================


Base URL ของ HolySheep: https://api.holysheep.ai/v1


สมัครได้ที่: https://www.holysheep.ai/register


อัตราแลกเปลี่ยน: ¥1 = $1 (ประหยัด 85%+)


รองรับ: WeChat, Alipay


HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("YOUR_HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

@dataclass
class OrderBookLevel:
    """โครงสร้างข้อมูล Order Book Level"""
    price: float
    qty: float
    orders: int = 1

@dataclass
class OrderBookSnapshot:
    """โครงสร้างข้อมูล Order Book Snapshot"""
    timestamp: str
    bids: List[OrderBookLevel]
    asks: List[OrderBookLevel]
    mid_price: float
    spread: float

class HolySheepOrderBookAnalyzer:
    """ใช้ HolySheep AI เพื่อวิเคราะห์ Order Book Patterns"""
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or HOLYSHEEP_API_KEY
        self.base_url = HOLYSHEEP_BASE_URL
        self.client = httpx.Client(timeout=30.0)
    
    def analyze_orderbook_imbalance(
        self, 
        orderbook: OrderBookSnapshot
    ) -> Dict:
        """
        วิเคราะห์ Order Book Imbalance
        คำนวณอัตราส่วนระหว่าง Bid Volume และ Ask Volume
        """
        total_bid_vol = sum(l.qty for l in orderbook.bids)
        total_ask_vol = sum(l.qty for l in orderbook.asks)
        
        imbalance = (total_bid_vol - total_ask_vol) / (total_bid_vol + total_ask_vol + 1e-10)
        
        return {
            "timestamp": orderbook.timestamp,
            "bid_volume": total_bid_vol,
            "ask_volume": total_ask_vol,
            "imbalance": imbalance,
            "signal": "bullish" if imbalance > 0.1 else "bearish" if imbalance < -0.1 else "neutral"
        }
    
    def analyze_with_ai(
        self, 
        orderbook: OrderBookSnapshot,
        model: str = "deepseek-chat"  # $0.42/MTok - ประหยัดที่สุด
    ) -> Dict:
        """
        ใช้ HolySheep AI เพื่อวิเคราะห์ Order Book Pattern
        เลือก DeepSeek V3.2 ($0.42/MTok) สำหรับงาน Technical Analysis
        """
        
        prompt = f"""
        Analyze this Order Book snapshot and provide trading insights:
        
        Timestamp: {orderbook.timestamp}
        Mid Price: ${orderbook.mid_price:.2f}
        Spread: ${orderbook.spread:.2f}
        
        Top 5 Bids (Price, Qty):
        {self._format_levels(orderbook.bids[:5])}
        
        Top 5 Asks (Price, Qty):
        {self._format_levels(orderbook.asks[:5])}
        
        Please provide:
        1. Order Book Imbalance Analysis
        2. Potential Support/Resistance Levels
        3. Short-term Price Movement Prediction
        4. Risk Assessment
        """
        
        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": "You are an expert quantitative analyst specializing in Order Book analysis."},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 1000
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        response = self.client.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload
        )
        
        if response.status_code == 200:
            result = response.json()
            return {
                "analysis": result["choices"][0]["message"]["content"],
                "usage": result.get("usage", {}),
                "model": model
            }
        else:
            raise Exception(f"HolySheep API Error: {response.status_code} - {response.text}")
    
    def _format_levels(self, levels: List[OrderBookLevel]) -> str:
        """จัดรูปแบบ Order Book Levels สำหรับ Prompt"""
        return "\n".join([
            f"  ${l.price:.2f}: {l.qty:.4f}" for l in levels
        ])
    
    def batch_analyze(
        self, 
        snapshots: List[OrderBookSnapshot],
        model: str = "deepseek-chat"
    ) -> List[Dict]:
        """
        วิเคราะห์ Order Book Snapshots หลายตัวพร้อมกัน
        ใช้ DeepSeek V3.2 ($0.42/MTok) เพื่อประหยัดค่าใช้จ่าย
        """
        results = []
        total_cost = 0
        
        for i, snapshot in enumerate(snapshots):
            try:
                # คำนวณ Imbalance ก่อน
                imbalance_data = self.analyze_orderbook_imbalance(snapshot)
                
                # วิเคราะห์ด้วย AI (ใช้ DeepSeek ประหยัดสุด)
                if i % 10 == 0:  # วิเคราะห์ AI ทุก 10 snapshots
                    ai_result = self.analyze_with_ai(snapshot, model)
                    total_cost += ai_result["usage"].get("total_tokens", 0) * 0.00042  # DeepSeek rate
                    results.append({
                        **imbalance_data,
                        "ai_analysis": ai_result["analysis"]
                    })
                else:
                    results.append(imbalance_data)
                
                print(f"✅ วิเคราะห์ {i+1}/{len(snapshots)} snapshots")
                
            except Exception as e:
                print(f"❌ Error at {i}: {e}")
                continue
        
        print(f"\n💰 ค่าใช้จ่ายโดยประมาณ: ${total_cost:.4f}")
        return results
    
    def export_analysis(self, results: List[Dict], filename: str):
        """ส่งออกผลการวิเคราะห์เป็น CSV"""
        with open(filename, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["timestamp", "imbalance", "signal", "ai_analysis"])
            
            for r in results:
                writer.writerow([
                    r["timestamp"],
                    f"{r['imbalance']:.4f}",
                    r["signal"],
                    r.get("ai_analysis", "")[:500]  # ตัดข้อความยาว
                ])
        
        print(f"✅ ส่งออกผลวิเคราะห์: {filename}")


========================


วิธีใช้งาน


========================

if __name__ == "__main__":
    # สมัคร HolySheep ที่: https://www.holysheep.ai/register
    # รับ API Key และเริ่มใช้งาน
    
    analyzer = HolySheepOrderBookAnalyzer()
    
    # ตัวอย่าง Order Book Snapshot
    sample_snapshot = OrderBookSnapshot(
        timestamp=datetime.now().isoformat(),
        bids=[
            OrderBookLevel(price=65000.0, qty=2.5),
            OrderBookLevel(price=64999.5, qty=1.8),
            OrderBookLevel(price=64999.0, qty=3.2),
        ],
        asks=[
            OrderBookLevel(price=65000.5, qty=1.5),
            OrderBookLevel(price=65001.0, qty=2.0),
            OrderBookLevel(price=65001.5, qty=1.2),
        ],
        mid_price=65000.25,
        spread=0.5
    )
    
    # วิเคราะห์ Imbalance
    result = analyzer.analyze_orderbook_imbalance(sample_snapshot)
    print(f"📊 Order Book Imbalance: {result['imbalance']:.4f}")
    print(f"📈 Signal: {result['signal']}")
    
    # วิเคราะห์

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