ในโลกของการเทรดคริปโต การเข้าถึงข้อมูลประวัติการซื้อขายที่แม่นยำเป็นสิ่งจำเป็นอย่างยิ่งสำหรับนักพัฒนา อัลกอริทึมเทรด และนักวิเคราะห์ข้อมูล บทความนี้จะพาคุณเรียนรู้วิธีการดึงข้อมูล 逐笔交易 (แท่งเทียนทีละแท่ง) จาก Binance API พร้อมทั้งเทคนิคการจัดเก็บข้อมูลให้มีประสิทธิภาพสูงสุด

ทำไมต้องดึงข้อมูล Binance Historical Trades?

จากประสบการณ์การพัฒนาระบบเทรดอัตโนมัติมาหลายปี พบว่าข้อมูลประวัติการซื้อขายมีความสำคัญอย่างยิ่งในหลายกรณี:

Binance Historical Trades API คืออะไร?

Binance Historical Trades API เป็น endpoint ที่ให้คุณดึงข้อมูลการซื้อขายแต่ละรายการ (Individual trades) จาก Binance โดยสามารถระบุได้ทั้ง symbol และช่วงเวลาที่ต้องการ ซึ่งแตกต่างจาก Klines (Candlestick) ที่รวมข้อมูลหลายรายการไว้ในแท่งเทียนเดียว

วิธีการดึงข้อมูลด้วย Python

การใช้งาน Binance API โดยตรงมีข้อจำกัดเรื่อง rate limit และความเร็วในการดึงข้อมูลปริมาณมาก ด้านล่างคือตัวอย่างโค้ดที่ใช้งานได้จริง:

import requests
import time
import pandas as pd
from datetime import datetime, timedelta

class BinanceHistoricalData:
    """คลาสสำหรับดึงข้อมูลประวัติการซื้อขายจาก Binance"""
    
    BASE_URL = "https://api.binance.com/api/v3"
    
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
    
    def get_historical_trades(self, symbol: str, limit: int = 1000, 
                              from_id: int = None) -> list:
        """
        ดึงข้อมูลประวัติการซื้อขาย
        
        Args:
            symbol: คู่เทรด เช่น BTCUSDT, ETHBUSD
            limit: จำนวนรายการที่ต้องการ (สูงสุด 1000)
            from_id: trade ID เริ่มต้น (optional)
        
        Returns:
            list: รายการข้อมูลการซื้อขาย
        """
        endpoint = f"{self.BASE_URL}/historicalTrades"
        params = {
            'symbol': symbol.upper(),
            'limit': min(limit, 1000)  # API limit
        }
        
        if from_id:
            params['fromId'] = from_id
        
        try:
            response = self.session.get(endpoint, params=params)
            response.raise_for_status()
            data = response.json()
            
            # จัดเรียงข้อมูลจากเก่าไปใหม่
            return sorted(data, key=lambda x: x['id'])
            
        except requests.exceptions.RequestException as e:
            print(f"❌ เกิดข้อผิดพลาด: {e}")
            return []
    
    def get_trades_in_range(self, symbol: str, start_time: int, 
                           end_time: int) -> pd.DataFrame:
        """ดึงข้อมูลการซื้อขายในช่วงเวลาที่กำหนด"""
        all_trades = []
        last_id = None
        
        while True:
            trades = self.get_historical_trades(symbol, limit=1000, 
                                                from_id=last_id)
            
            if not trades:
                break
                
            # กรองข้อมูลตามช่วงเวลา
            filtered = [t for t in trades 
                       if start_time <= t['time'] <= end_time]
            all_trades.extend(filtered)
            
            # หยุดถ้าเลยช่วงเวลาที่ต้องการ
            if trades[-1]['time'] > end_time:
                break
                
            last_id = trades[-1]['id']
            
            # รอตาม rate limit (ประมาณ 1 วินาที)
            time.sleep(1.2)
        
        return pd.DataFrame(all_trades)


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

if __name__ == "__main__":
    client = BinanceHistoricalData()
    
    # ดึงข้อมูล 1 ชั่วโมงล่าสุด
    end_time = int(datetime.now().timestamp() * 1000)
    start_time = int((datetime.now() - timedelta(hours=1)).timestamp() * 1000)
    
    df = client.get_trades_in_range("BTCUSDT", start_time, end_time)
    print(f"✅ ดึงข้อมูลสำเร็จ: {len(df)} รายการ")
    print(df.head())

การจัดเก็บข้อมูลอย่างมีประสิทธิภาพ

เมื่อดึงข้อมูลมาแล้ว การจัดเก็บที่เหมาะสมจะช่วยให้การ query และวิเคราะห์เร็วขึ้นมาก ด้านล่างคือสถาปัตยกรรมการจัดเก็บที่แนะนำ:

import sqlite3
import pandas as pd
from typing import Generator
import gzip
import json
from pathlib import Path

class TradeDataStore:
    """ระบบจัดเก็บข้อมูลการซื้อขายแบบมีประสิทธิภาพ"""
    
    def __init__(self, db_path: str = "trades.db"):
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """สร้างตารางถ้ายังไม่มี"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS trades (
                    id INTEGER PRIMARY KEY,
                    symbol TEXT NOT NULL,
                    price REAL NOT NULL,
                    qty REAL NOT NULL,
                    quote_qty REAL NOT NULL,
                    time INTEGER NOT NULL,
                    is_buyer_maker INTEGER,
                    is_best_match INTEGER,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    UNIQUE(symbol, id)
                )
            """)
            
            # สร้าง index เพื่อความเร็วในการ query
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_symbol_time 
                ON trades(symbol, time)
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_time 
                ON trades(time)
            """)
            
            conn.commit()
    
    def bulk_insert(self, df: pd.DataFrame, batch_size: int = 5000):
        """เพิ่มข้อมูลเป็นชุดเพื่อความเร็ว"""
        if df.empty:
            return 0
        
        # เปลี่ยนชื่อคอลัมน์ให้ตรงกับตาราง
        df = df.rename(columns={
            'id': 'id',
            'symbol': 'symbol',
            'price': 'price',
            'qty': 'qty',
            'quoteQty': 'quote_qty',
            'time': 'time',
            'isBuyerMaker': 'is_buyer_maker',
            'isBestMatch': 'is_best_match'
        })
        
        inserted = 0
        with sqlite3.connect(self.db_path) as conn:
            for i in range(0, len(df), batch_size):
                batch = df.iloc[i:i+batch_size]
                try:
                    batch.to_sql('trades', conn, 
                                if_exists='append', 
                                index=False)
                    inserted += len(batch)
                except sqlite3.IntegrityError:
                    # ข้อมูลซ้ำ - ข้าม
                    pass
        
        return inserted
    
    def query_by_time_range(self, symbol: str, 
                           start_time: int, 
                           end_time: int) -> pd.DataFrame:
        """ดึงข้อมูลตามช่วงเวลา (เร็วมากเพราะใช้ index)"""
        with sqlite3.connect(self.db_path) as conn:
            query = """
                SELECT * FROM trades 
                WHERE symbol = ? AND time >= ? AND time <= ?
                ORDER BY time ASC
            """
            return pd.read_sql_query(query, conn, 
                                    params=[symbol, start_time, end_time])
    
    def get_aggregated_klines(self, symbol: str, 
                              interval: str = '1h',
                              start_time: int = None,
                              end_time: int = None) -> pd.DataFrame:
        """
        รวมข้อมูล trades เป็น klines (OHLCV)
        
        Args:
            interval: '1m', '5m', '15m', '1h', '4h', '1d'
        """
        # คำนวณ interval เป็น milliseconds
        interval_ms = {
            '1m': 60000, '5m': 300000, '15m': 900000,
            '1h': 3600000, '4h': 14400000, '1d': 86400000
        }
        
        if interval not in interval_ms:
            raise ValueError(f"ไม่รองรับ interval: {interval}")
        
        interval_ms = interval_ms[interval]
        
        # สร้าง SQL query สำหรับรวมข้อมูล
        sql = """
            SELECT 
                (time / ?) * ? as interval_time,
                MIN(price) as low,
                MAX(price) as high,
                SUM(CASE WHEN (time / ?) % 2 = 0 THEN price END) as open,
                SUM(CASE WHEN (time / ?) % 2 = 1 THEN price END) as close,
                SUM(quote_qty) as volume,
                COUNT(*) as trade_count
            FROM trades
            WHERE symbol = ?
        """
        params = [interval_ms, interval_ms, 
                  interval_ms, interval_ms, symbol]
        
        if start_time:
            sql += " AND time >= ?"
            params.append(start_time)
        if end_time:
            sql += " AND time <= ?"
            params.append(end_time)
        
        sql += " GROUP BY interval_time ORDER BY interval_time"
        
        with sqlite3.connect(self.db_path) as conn:
            return pd.read_sql_query(sql, conn, params=params)


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

if __name__ == "__main__":
    store = TradeDataStore("crypto_trades.db")
    
    # เพิ่มข้อมูลจาก DataFrame
    # inserted = store.bulk_insert(trades_df)
    # print(f"✅ เพิ่มข้อมูล {inserted} รายการ")
    
    # ดึงข้อมูล OHLCV 1 ชั่วโมง
    klines = store.get_aggregated_klines(
        "BTCUSDT", 
        interval='1h',
        start_time=int((pd.Timestamp.now() - pd.Timedelta(days=7)).timestamp() * 1000)
    )
    print(klines.head())

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

กลุ่มผู้ใช้ เหมาะสม หมายเหตุ
นักพัฒนา Crypto Trading Bot ✅ เหมาะมาก ต้องการข้อมูล real-time และ historical สำหรับ backtesting
นักวิเคราะห์ข้อมูล (Data Analyst) ✅ เหมาะมาก ต้องการข้อมูลคุณภาพสูงสำหรับวิเคราะห์แนวโน้มตลาด
นักพัฒนา AI/ML Models ✅ เหมาะมาก ต้องการ dataset ขนาดใหญ่สำหรับ train model
ผู้เริ่มต้นศึกษา Crypto ⚠️ พอได้ ต้องมีความรู้ Python และ API พื้นฐาน
ผู้ใช้ทั่วไปที่ต้องการดูราคา ❌ ไม่เหมาะ ควรใช้แอป Binance โดยตรงแทน

ราคาและ ROI

การใช้งาน Binance API โดยตรงฟรี แต่มีข้อจำกัดด้าน rate limit หากต้องการประมวลผลข้อมูลด้วย AI เพื่อวิเคราะห์หรือสร้างรายงาน สมัครที่นี่ HolySheep AI นำเสนอบริการ API ราคาประหยัดมากเมื่อเทียบกับ OpenAI หรือ Anthropic:

ผู้ให้บริการ GPT-4.1 ($/MTok) Claude Sonnet 4.5 ($/MTok) Gemini 2.5 Flash ($/MTok) DeepSeek V3.2 ($/MTok)
HolySheep AI $8 $15 $2.50 $0.42
OpenAI / Anthropic $15-60 $15-30 $5-15 ไม่มี
ประหยัดได้ 85%+ 50%+ 80%+ -

ตัวอย่าง ROI: หากคุณใช้ GPT-4.1 ประมวลผลข้อมูล 1 ล้าน tokens ต่อเดือน จะประหยัดได้ถึง $52 ต่อเดือนเมื่อใช้ HolySheep แทน OpenAI

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

# ตัวอย่าง: ใช้ HolySheep วิเคราะห์ข้อมูลการซื้อขายด้วย AI

import requests

def analyze_trading_patterns(trades_data: list):
    """
    วิเคราะห์รูปแบบการซื้อขายด้วย AI
    ใช้ HolySheep API แทน OpenAI (ประหยัด 85%+)
    """
    
    # เตรียมข้อมูลสำหรับวิเคราะห์
    prompt = f"""วิเคราะห์ข้อมูลการซื้อขายต่อไปนี้:
    
    จำนวนรายการ: {len(trades_data)}
    
    1. ราคาสูงสุด: {max(t['price'] for t in trades_data)}
    2. ราคาต่ำสุด: {min(t['price'] for t in trades_data)}
    3. ปริมาณรวม: {sum(t['quoteQty'] for t in trades_data)}
    
    รายการตัวอย่าง (5 รายการแรก):
    {trades_data[:5]}
    
    กรุณาวิเคราะห์:
    - แนวโน้มตลาด (ขาขึ้น/ขาลง/เคลื่อนไหวช่วง)
    - ความผันผวนของราคา
    - รูปแบบการซื้อขายที่น่าสนใจ
    """
    
    # ใช้ HolySheep API
    response = requests.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers={
            "Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
            "Content-Type": "application/json"
        },
        json={
            "model": "gpt-4.1",
            "messages": [
                {"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้านการวิเคราะห์ตลาดคริปโต"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.7,
            "max_tokens": 1000
        }
    )
    
    if response.status_code == 200:
        result = response.json()
        return result['choices'][0]['message']['content']
    else:
        raise Exception(f"API Error: {response.status_code}")


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

if __name__ == "__main__":
    sample_trades = [
        {"id": 1, "price": 45000.5, "qty": 0.5, "quoteQty": 22500.25},
        {"id": 2, "price": 45100.0, "qty": 0.3, "quoteQty": 13530.0},
        {"id": 3, "price": 44980.0, "qty": 0.8, "quoteQty": 35984.0},
        {"id": 4, "price": 45050.0, "qty": 0.2, "quoteQty": 9010.0},
        {"id": 5, "price": 45120.0, "qty": 0.6, "quoteQty": 27072.0},
    ]
    
    analysis = analyze_trading_patterns(sample_trades)
    print("📊 ผลการวิเคราะห์:")
    print(analysis)

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

ข้อผิดพลาดที่ 1: Rate Limit Exceeded (429)

# ❌ ปัญหา: เรียก API บ่อยเกินไปทำให้ถูก block


✅ วิธีแก้ไข: ใช้ exponential backoff


import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_resilient_session():
    """สร้าง session ที่ทนต่อการถูก block"""
    session = requests.Session()
    
    # ตั้งค่า retry strategy
    retry_strategy = Retry(
        total=5,
        backoff_factor=2,  # รอ 2, 4, 8, 16, 32 วินาที
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    
    return session

def get_trades_with_retry(symbol: str, max_retries: int = 5):
    """ดึงข้อมูลพร้อม retry เมื่อถูก rate limit"""
    session = create_resilient_session()
    
    for attempt in range(max_retries):
        try:
            response = session.get(
                f"https://api.binance.com/api/v3/historicalTrades",
                params={"symbol": symbol, "limit": 1000}
            )
            
            if response.status_code == 429:
                wait_time = int(response.headers.get('Retry-After', 60))
                print(f"⏳ รอ {wait_time} วินาที (attempt {attempt + 1}/{max_retries})")
                time.sleep(wait_time)
                continue
                
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise
            print(f"⚠️ Attempt {attempt + 1} failed: {e}")
            time.sleep(2 ** attempt)  # Exponential backoff
            
    return None

ข้อผิดพลาดที่ 2: ข้อมูลซ้ำหรือขาดหายเมื่อดึงข้อมูลจำนวนมาก

# ❌ ปัญหา: ดึงข้อมูลแล้วมี ID ซ้ำหรือขาดหาย


✅ วิธีแก้ไข: ใช้การตรวจสอบและ deduplicate


import pandas as pd
from typing import Set, Tuple

class DataIntegrityChecker:
    """ตรวจสอบความสมบูรณ์ของข้อมูลการซื้อขาย"""
    
    def __init__(self):
        self.seen_ids: Set[int] = set()
        self.expected_gap = 1  # trade ID ควรต่างกัน 1
    
    def validate_trades(self, trades: list) -> Tuple[list, list]:
        """
        ตรวจสอบความสมบูรณ์ของข้อมูล
        
        Returns:
            (valid_trades, issues) - รายการที่ถูกต้องและปัญหาที่พบ
        """
        valid = []
        issues = []
        last_id = None
        
        for trade in sorted(trades, key=lambda x: x['id']):
            trade_id = trade['id']
            
            # ตรวจสอบ ID ซ้ำ
            if trade_id in self.seen_ids:
                issues.append({
                    'type': 'duplicate',
                    'trade_id': trade_id,
                    'message': f'Trade ID {trade_id} ซ้ำ'
                })
                continue
            
            # ตรวจสอบ gap
            if last_id is not None:
                gap = trade_id - last_id
                if gap > self.expected_gap:
                    issues.append({
                        'type': 'missing',
                        'from_id': last_id,
                        'to_id': trade_id,
                        'missing_count': gap - 1,
                        'message': f'ขาดข้อมูล {gap - 1} รายการ ระหว่าง {last_id} - {trade_id}'
                    })
            
            self.seen_ids.add(trade_id)
            last_id = trade_id
            valid.append(trade)
        
        return valid, issues
    
    def reset(self):
        """รีเซ็ตการติดตาม IDs"""
        self.seen_ids.clear()


การใช้งาน

checker = DataIntegrityChecker()


ดึงข้อมูลหลายช่วง

all_trades = []
for start_id in range(1, 10000, 1000):
    trades = get_trades_from_api(start_id=start_id, limit=1000)
    all_trades.extend(trades)


ตรวจสอบความสมบูรณ์

valid_trades, issues = checker.validate_trades(all_trades)

print(f"✅ ข้อมูลที่ถูกต้อง: {len(valid_trades)} รายการ")
if issues:
    print(f"⚠️ พบปัญหา {len(issues)} รายการ:")
    for issue in issues[:5]:  # แสดง 5 รายการแรก
        print(f"  - {issue['message']}")

ข้อผิดพลาดที่ 3: Memory Error เมื่อ

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

บทความที่เกี่ยวข้อง