การดึงข้อมูล K线 (Candlestick) จาก Binance API เป็นพื้นฐานสำคัญสำหรับนักพัฒนา trading bot และนักวิเคราะห์ทางเทคนิค แต่ในทางปฏิบัติ หลายคนเจอปัญหาเช่น ConnectionError: timeout ขณะดึงข้อมูลราคาย้อนหลัง หรือ 401 Unauthorized เมื่อใช้ API key ผิดวิธี บทความนี้จะสอนวิธีดึงข้อมูล K线 Binance แต่ละ timeframe (1m, 5m, 1h, 1d) อย่างมีประสิทธิภาพ พร้อมกลยุทธ์จัดเก็บที่ถูกต้อง และวิธีแก้ปัญหาที่พบบ่อย

ทำไมต้องเก็บข้อมูล K线 Binance เอง?

Binance มี API ให้ใช้ฟรี แต่มีข้อจำกัด:

การเก็บข้อมูลไว้เองช่วยให้:

การดึงข้อมูล K线 Binance ด้วย Python

1. ติดตั้ง Library ที่จำเป็น

pip install pandas requests python-binance schedule sqlalchemy

2. สคริปต์ดึงข้อมูล K线 ทุก Timeframe

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


การดึงข้อมูล K线 จาก Binance Public API

BASE_URL = "https://api.binance.com"

def get_klines(symbol, interval, start_time=None, limit=1000):
    """
    ดึงข้อมูล K线 จาก Binance
    
    Parameters:
    - symbol: เช่น 'BTCUSDT', 'ETHUSDT'
    - interval: '1m', '5m', '15m', '1h', '4h', '1d', '1w'
    - start_time: timestamp เริ่มต้น (milliseconds)
    - limit: จำนวน candles สูงสุด 1000 ต่อครั้ง
    
    Returns:
    - DataFrame พร้อมข้อมูล OHLCV
    """
    endpoint = "/api/v3/klines"
    params = {
        "symbol": symbol.upper(),
        "interval": interval,
        "limit": limit
    }
    
    if start_time:
        params["startTime"] = start_time
    
    try:
        response = requests.get(f"{BASE_URL}{endpoint}", params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        # แปลงเป็น DataFrame
        df = pd.DataFrame(data, columns=[
            "open_time", "open", "high", "low", "close", "volume",
            "close_time", "quote_volume", "trades", "taker_buy_base",
            "taker_buy_quote", "ignore"
        ])
        
        # แปลงประเภทข้อมูล
        numeric_cols = ["open", "high", "low", "close", "volume", "quote_volume"]
        for col in numeric_cols:
            df[col] = pd.to_numeric(df[col], errors='coerce')
        
        # แปลง timestamp
        df["open_time"] = pd.to_datetime(df["open_time"], unit="ms")
        df["close_time"] = pd.to_datetime(df["close_time"], unit="ms")
        
        return df
    
    except requests.exceptions.Timeout:
        print(f"⏰ Timeout error: {symbol} {interval}")
        return None
    except requests.exceptions.HTTPError as e:
        print(f"❌ HTTP Error: {e}")
        return None
    except Exception as e:
        print(f"❌ Error: {e}")
        return None


ทดสอบการดึงข้อมูล

if __name__ == "__main__":
    # ดึงข้อมูล BTCUSDT รายชั่วโมง 1000 แท่งล่าสุด
    df = get_klines("BTCUSDT", "1h", limit=1000)
    
    if df is not None:
        print(f"✅ ดึงข้อมูลสำเร็จ: {len(df)} candles")
        print(df.tail())

3. ระบบดึงข้อมูลย้อนหลังแบบต่อเนื่อง

import sqlite3
from datetime import datetime, timedelta
import time
import schedule


การเชื่อมต่อ SQLite สำหรับจัดเก็บข้อมูล

DB_PATH = "binance_klines.db"

def init_database():
    """สร้างตารางถ้ายังไม่มี"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS klines (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            symbol TEXT NOT NULL,
            interval TEXT NOT NULL,
            open_time TEXT NOT NULL,
            open REAL,
            high REAL,
            low REAL,
            close REAL,
            volume REAL,
            quote_volume REAL,
            trades INTEGER,
            UNIQUE(symbol, interval, open_time)
        )
    """)
    
    conn.commit()
    conn.close()

def save_to_database(df, symbol, interval):
    """บันทึกข้อมูลลงฐานข้อมูล"""
    if df is None or df.empty:
        return False
    
    conn = sqlite3.connect(DB_PATH)
    
    for _, row in df.iterrows():
        try:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT OR REPLACE INTO klines 
                (symbol, interval, open_time, open, high, low, close, volume, quote_volume, trades)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                symbol.upper(),
                interval,
                row["open_time"].isoformat(),
                row["open"],
                row["high"],
                row["low"],
                row["close"],
                row["volume"],
                row["quote_volume"],
                int(row["trades"])
            ))
        except Exception as e:
            print(f"⚠️ Insert error: {e}")
    
    conn.commit()
    conn.close()
    return True

def get_latest_timestamp(symbol, interval):
    """ดึง timestamp ล่าสุดที่มีในฐานข้อมูล"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    cursor.execute("""
        SELECT open_time FROM klines 
        WHERE symbol = ? AND interval = ?
        ORDER BY open_time DESC LIMIT 1
    """, (symbol.upper(), interval))
    
    result = cursor.fetchone()
    conn.close()
    
    if result:
        return int(datetime.fromisoformat(result[0]).timestamp() * 1000)
    return None

def fetch_all_historical(symbol, interval, months=12):
    """
    ดึงข้อมูลย้อนหลังหลายเดือน
    Binance limit: 1000 candles ต่อ request
    """
    init_database()
    
    # หาเวลาเริ่มต้น
    latest_ts = get_latest_timestamp(symbol, interval)
    if latest_ts is None:
        # ถ้าไม่มีข้อมูล ใช้เวลา 12 เดือนก่อน
        start_time = int((datetime.now() - timedelta(days=months * 30)).timestamp() * 1000)
    else:
        start_time = latest_ts
    
    all_data = []
    current_start = start_time
    total_fetched = 0
    
    print(f"📥 เริ่มดึงข้อมูล {symbol} {interval} ตั้งแต่ {datetime.fromtimestamp(current_start/1000)}")
    
    while True:
        df = get_klines(symbol, interval, start_time=current_start, limit=1000)
        
        if df is None or df.empty:
            break
        
        all_data.append(df)
        total_fetched += len(df)
        
        # ใช้เวลา close ของแท่งสุดท้ายเป็น start ครั้งต่อไป
        last_close_time = int(df["close_time"].iloc[-1].timestamp() * 1000) + 1
        current_start = last_close_time
        
        print(f"   ✅ ดึงได้ {len(df)} candles, รวม: {total_fetched}")
        
        # รอเพื่อไม่ให้ rate limit
        time.sleep(0.3)
        
        # ถ้าได้น้อยกว่า 1000 แปลว่าถึงข้อมูลล่าสุดแล้ว
        if len(df) < 1000:
            break
        
        # ป้องกัน infinite loop
        if total_fetched > 50000:
            print("⚠️ ถึงขีดจำกัดการดึงข้อมูล")
            break
    
    # รวมข้อมูลทั้งหมดและบันทึก
    if all_data:
        combined_df = pd.concat(all_data, ignore_index=True)
        combined_df = combined_df.drop_duplicates(subset=["open_time"])
        combined_df = combined_df.sort_values("open_time")
        
        save_to_database(combined_df, symbol, interval)
        print(f"✅ บันทึกสำเร็จ: {len(combined_df)} candles")
        
        return combined_df
    
    return None


ทดสอบการดึงข้อมูล 12 เดือน

if __name__ == "__main__":
    symbols = ["BTCUSDT", "ETHUSDT"]
    intervals = ["1h", "4h", "1d"]
    
    for symbol in symbols:
        for interval in intervals:
            fetch_all_historical(symbol, interval, months=12)
            time.sleep(2)  # รอระหว่างแต่ละคู่

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

กลุ่มผู้ใช้เหมาะกับไม่เหมาะกับ
นักพัฒนา Trading Bot ต้องการ backtest หลายพันครั้ง, ดึงข้อมูลเร็วไม่รอ API ผู้เริ่มต้นที่มีข้อมูลไม่มาก
นักวิเคราะห์ทางเทคนิค วิเคราะห์กราฟหลาย timeframe, เปรียบเทียบ indicator ผู้ที่ใช้แค่ chart บนเว็บ Binance
Data Scientist สร้างโมเดล ML ทำนายราคา, ต้องการ dataset ขนาดใหญ่ ผู้ที่ต้องการข้อมูลแค่ real-time
สถาบันการเงิน ต้องการข้อมูล backup, compliance, audit trail ผู้ใช้รายบุคคลทั่วไป

ราคาและ ROI

การเก็บข้อมูล K线 ด้วยวิธีข้างต้นใช้ได้กับทุก API แต่ถ้าต้องการประมวลผลข้อมูลด้วย AI/ML แนะนำใช้ HolySheep AI ซึ่งมีข้อดีด้านค่าใช้จ่าย:

ผู้ให้บริการราคา GPT-4.1ราคา Claude 4.5ราคา DeepSeek V3.2ความเร็ว
HolySheep AI $8/MTok $15/MTok $0.42/MTok <50ms
OpenAI โดยตรง $15/MTok ไม่มี ไม่มี 200-500ms
Anthropic โดยตรง ไม่มี $18/MTok ไม่มี 300-600ms

ROI จากการใช้ HolySheep:

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

สำหรับนักพัฒนาที่ต้องการประมวลผลข้อมูล K线 ด้วย AI เพื่อ:

API ของ HolySheep AI รองรับทุกโมเดลยอดนิยม (GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2) ด้วยราคาที่ถูกที่สุดในตลาด

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

1. ConnectionError: timeout - ดึงข้อมูลไม่ได้

สาเหตุ: Binance API มี rate limit หรือ server ช้าในบางช่วงเวลา

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

def create_session_with_retry():
    """สร้าง session ที่มี retry mechanism"""
    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=["HEAD", "GET", "OPTIONS"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session


ใช้งาน

session = create_session_with_retry()
response = session.get(f"{BASE_URL}/api/v3/klines", params=params, timeout=30)

2. 401 Unauthorized - API Key ไม่ถูกต้อง

สาเหตุ: สำหรับ private API (เช่น account info) ต้องมี signature

import hmac
import hashlib
import requests

BINANCE_API_KEY = "YOUR_BINANCE_API_KEY"
BINANCE_SECRET_KEY = "YOUR_BINANCE_SECRET_KEY"

def get_account_info():
    """ดึงข้อมูลบัญชี Binance (ต้องใช้ signature)"""
    endpoint = "/api/v3/account"
    timestamp = int(time.time() * 1000)
    
    # สร้าง query string
    query_string = f"timestamp={timestamp}"
    
    # สร้าง signature
    signature = hmac.new(
        BINANCE_SECRET_KEY.encode('utf-8'),
        query_string.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    
    # รวม query + signature
    full_url = f"{BASE_URL}{endpoint}?{query_string}&signature={signature}"
    
    headers = {
        "X-MBX-APIKEY": BINANCE_API_KEY,
        "Content-Type": "application/json"
    }
    
    response = requests.get(full_url, headers=headers)
    
    if response.status_code == 200:
        return response.json()
    elif response.status_code == 401:
        raise Exception("❌ 401 Unauthorized: ตรวจสอบ API Key และ Secret Key")
    else:
        raise Exception(f"❌ Error {response.status_code}: {response.text}")


หมายเหตุ: Public API (klines) ไม่ต้องมี signature


แค่ดึงข้อมูล OHLCV สามารถใช้วิธีด้านบนได้เลย

3. ข้อมูลซ้ำหรือหายหลังจากรันซ้ำ

สาเหตุ: การดึงข้อมูลซ้อนทับกันหรือ timestamp ไม่ตรง

def safe_fetch_and_save(symbol, interval, lookback_days=7):
    """
    ดึงข้อมูลอย่างปลอดภัยพร้อมป้องกันข้อมูลซ้ำ
    """
    init_database()  # ตรวจสอบตาราง
    
    # ดึง timestamp ล่าสุดที่มี
    latest_ts = get_latest_timestamp(symbol, interval)
    
    # กำหนดเวลาเริ่มต้น (7 วันก่อน timestamp ล่าสุด หรือ 7 วันก่อนปัจจุบัน)
    if latest_ts:
        start_time = latest_ts - (lookback_days * 24 * 60 * 60 * 1000)
        start_dt = datetime.fromtimestamp(start_time / 1000)
    else:
        start_time = int((datetime.now() - timedelta(days=lookback_days)).timestamp() * 1000)
        start_dt = datetime.fromtimestamp(start_time / 1000)
    
    print(f"📥 ดึงข้อมูล {symbol} {interval} ตั้งแต่ {start_dt}")
    
    # ดึงข้อมูล
    df = get_klines(symbol, interval, start_time=start_time, limit=1000)
    
    if df is not None and not df.empty:
        # ลบข้อมูลเก่าที่ซ้ำก่อน insert
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        
        cursor.execute("""
            DELETE FROM klines 
            WHERE symbol = ? AND interval = ? AND open_time >= ?
        """, (symbol.upper(), interval, df["open_time"].iloc[0].isoformat()))
        
        conn.commit()
        conn.close()
        
        # บันทึกข้อมูลใหม่ (UNIQUE constraint จะป้องกันข้อมูลซ้ำ)
        save_to_database(df, symbol, interval)
        print(f"✅ บันทึก {len(df)} candles สำเร็จ")
        
    return df


การใรงานทุก 5 นาที

def job():
    for symbol in ["BTCUSDT", "ETHUSDT"]:
        for interval in ["1m", "5m", "15m"]:
            safe_fetch_and_save(symbol, interval, lookback_days=1)
            time.sleep(1)

schedule.every(5).minutes.do(job)

สรุป

การดึงและจัดเก็บข้อมูล K线 Binance อย่างมีประสิทธิภาพต้องอาศัย:

  1. การจัดการ Rate Limit: ใช้ retry mechanism และรอระหว่าง request
  2. การจัดเก็บที่ถูกต้อง: ใช้ UNIQUE constraint ป้องกันข้อมูลซ้ำ
  3. การดึงแบบ incremental: เก็บเฉพาะข้อมูลใหม่แทนดึงทั้งหมดทุกครั้ง
  4. การ backup: เก็บข้อมูลไว้เองไม่พึ่ง API 100%

สำหรับนักพัฒนาที่ต้องการนำข้อมูล K线 ไปประมวลผลด้วย AI เพื่อวิเคราะห์หรือสร้างรายงานอัตโนมัติ HolySheep AI เป็นตัวเลือกที่คุ้มค่าที่สุดในตลาด ด้วยราคาประหยัดกว่า 85% และความเร็วตอบสนองต่ำกว่า 50ms

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน

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

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