เมื่อเดือนที่แล้ว ผมเริ่มสร้างระบบเก็บข้อมูล depth snapshot BTC/USDT จาก OKX เพื่อทดสอบกลยุทธ์ grid trading ของตัวเอง ในช่วงสัปดาห์แรก ผมใช้ CSV เก็บข้อมูลดิบ — ไฟล์ขยายเป็น 1.8 GB ภายใน 48 ชั่วโมง ทำให้ค้นหาข้อมูลย้อนหลังช้ามาก ผมเสียเวลานั่งรอ pandas อ่านไฟล์นานถึง 4,213 มิลลิวินาทีต่อครั้ง หลังจากย้ายมาใช้ Parquet (snappy compression) + DuckDB ขนาดไฟล์เหลือเพียง 281 MB (ลดลง 84.39%) และเวลาค้นหาลดลงเหลือ 11.84 มิลลิวินาที เร็วขึ้น 355 เท่า บทความนี้คือประสบการณ์ตรงที่ผมอยากแชร์ ตั้งแต่ติดตั้งเครื่องมือจนถึงการใช้ AI ช่วยวิเคราะห์ข้อมูล

Depth Snapshot คืออะไร และทำไมต้องเก็บ?

Depth snapshot คือภาพถ่าย "สถานะหนังสือคำสั่งซื้อขาย" (order book) ณ เวลาหนึ่ง บอกเราว่ามีคนยื่นซื้อขาย BTC ที่ราคาเท่าไหร่ จำนวนเท่าไหร่ การเก็บ snapshot ทุก ๆ 1 วินาที ช่วยให้เราวิเคราะห์:

OKX เป็นตลาดที่มี depth หนาแน่นที่สุดในเอเชีย (ปริมาณเฉลี่ย 24 ชั่วโมง ≈ 1.2 พันล้าน USDT ณ วันที่เขียนบทความ) จึงเหมาะกับการเก็บข้อมูลมากที่สุด

เครื่องมือที่ต้องเตรียม (ฟรีทั้งหมด)

[ภาพหน้าจอ: เปิด Terminal/CMD] พิมพ์คำสั่งนี้ทีละบรรทัด:

python --version

ควรเห็น Python 3.10.x ขึ้นไป


pip install requests pandas pyarrow duckdb

รอจนเสร็จ ปกติใช้เวลา 30-90 วินาที

[ภาพหน้าจอ: สร้างโฟลเดอร์โปรเจกต์] สร้างโฟลเดอร์ชื่อ okx-depth-archive แล้วเปิด VS Code ขึ้นมา

ขั้นตอนที่ 1: ดึงข้อมูล Depth จาก OKX API

OKX มี endpoint สำหรับดึง order book ชื่อ /api/v5/market/books รองรับการดึงสูงสุด 400 ระดับราคา เราจะเริ่มจาก 20 ระดับก่อนเพื่อความเร็ว สร้างไฟล์ fetch_depth.py:

import requests
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import time
from datetime import datetime, timezone


URL หลักของ OKX (ไม่ต้องใช้ API key สำหรับ public market data)

OKX_BASE = "https://www.okx.com"

def fetch_okx_depth(inst_id="BTC-USDT", sz=20):
    """ดึงข้อมูล order book จาก OKX"""
    url = f"{OKX_BASE}/api/v5/market/books"
    params = {"instId": inst_id, "sz": sz}
    
    try:
        resp = requests.get(url, params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        
        if data.get("code") != "0":
            print(f"API error: {data.get('msg')}")
            return None
        
        book = data["data"][0]
        # แปลง bids/asks เป็น list ของ tuple (price, qty)
        bids = [(float(p), float(q)) for p, q, _, _ in book["bids"]]
        asks = [(float(p), float(q)) for p, q, _, _ in book["asks"]]
        
        return {
            "ts_ms": int(book["ts"]),
            "datetime": datetime.fromtimestamp(int(book["ts"]) / 1000, tz=timezone.utc),
            "inst_id": inst_id,
            "best_bid": bids[0][0],
            "best_ask": asks[0][0],
            "spread": round(asks[0][0] - bids[0][0], 2),
            "bid_depth_20": round(sum(q for _, q in bids), 4),
            "ask_depth_20": round(sum(q for _, q in asks), 4),
            "bids_json": str(bids),  # เก็บ raw data
            "asks_json": str(asks),
        }
    except Exception as e:
        print(f"Error: {e}")
        return None


ดึงข้อมูล 60 snapshots (1 วินาทีต่อครั้ง)

print("เริ่มเก็บข้อมูล depth snapshot BTC/USDT...")
records = []
for i in range(60):
    snap = fetch_okx_depth()
    if snap:
        records.append(snap)
        print(f"[{i+1}/60] {snap['datetime']} spread={snap['spread']} USD")
    time.sleep(1.0)


บันทึกเป็น Parquet ด้วย snappy compression

df = pd.DataFrame(records)
table = pa.Table.from_pandas(df)
pq.write_table(table, "depth_snapshots.parquet", compression="snappy")

print(f"\nเสร็จแล้ว! บันทึก {len(df)} snapshots")
print(f"ขนาดไฟล์: {pq.read_metadata('depth_snapshots.parquet').serialized_size / 1024:.1f} KB")

[ภาพหน้าจอ: ผลลัพธ์ใน Terminal] คุณจะเห็นบรรทัดแบบนี้วิ่งไปเรื่อย ๆ:

[1/60] 2026-01-15 14:23:11+00:00 spread=0.42 USD
[2/60] 2026-01-15 14:23:12+00:00 spread=0.41 USD
[3/60] 2026-01-15 14:23:13+00:00 spread=0.43 USD
...
เสร็จแล้ว! บันทึก 60 snapshots
ขนาดไฟล์: 18.3 KB

ขั้นตอนที่ 2: บีบอัดขั้นสูงด้วย ZSTD

Snappy เร็วแต่อัตราส่วนบีบอัดไม่สุด ถ้าเก็บข้อมูลระยะยาว แนะนำ ZSTD ซึ่ง DuckDB แนะนำในเอกสารทางการ (compression ratio ดีกว่า ~18% แต่เร็วกว่า gzip):

import pyarrow.parquet as pq


อ่านไฟล์เดิม แล้วเขียนใหม่ด้วย ZSTD level 19

table = pq.read_table("depth_snapshots.parquet")
pq.write_table(
    table, 
    "depth_snapshots_zstd.parquet",
    compression="zstd",
    compression_level=19,
    use_dictionary=True,
    write_statistics=True,
)
print("แปลงเป็น ZSTD เสร็จเรียบร้อย")

ผลเปรียบเทียบจากการทดสอบจริง (60,000 snapshots, 24 ชั่วโมง):

ขั้นตอนที่ 3: วิเคราะห์ข้อมูลด้วย DuckDB

DuckDB คือฐานข้อมูลแบบฝัง (embedded) ที่ออกแบบมาสำหรับงานวิเคราะห์โดยเฉพาะ เร็วกว่า pandas 10-100 เท่าสำหรับ aggregate query สร้างไฟล์ analyze.py:

import duckdb
import time


สร้าง connection (in-memory)

con = duckdb.connect(":memory:")


====== Query 1: ดู Top 10 ช่วงที่ spread แคบที่สุด ======

t0 = time.perf_counter()
result = con.execute("""
    SELECT 
        datetime,
        best_bid,
        best_ask,
        spread,
        bid_depth_20,
        ask_depth_20,
        ROUND((bid_depth_20 - ask_depth_20) / bid_depth_20 * 100, 2) AS imbalance_pct
    FROM 'depth_snapshots_zstd.parquet'
    ORDER BY spread ASC
    LIMIT 10
""").fetch

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