จากประสบการณ์ตรงของผู้เขียนที่เคยรับผิดชอบ data infrastructure ให้ทีม quantitative trading ขนาดกลาง ผมพบว่าปัญหาคอขวดที่ใหญ่ที่สุดไม่ใช่กลยุทธ์หรือสัญญาณ แต่เป็น I/O pipeline ระหว่าง Binance public endpoint กับ long-term storage ที่ทำงานช้า กินดิสก์ และ query ไม่ได้แบบ columnar บทความนี้เป็นบันทึกเทคนิคที่ใช้งานจริงในระบบ production ตั้งแต่ concurrent downloader, CSV streaming, ไปจนถึงการแปลงเป็น Parquet พร้อม benchmark ตัวเลขจริงที่วัดได้ และปิดท้ายด้วยการผสาน LLM เข้ากับ pipeline เพื่อวิเคราะห์ market anomaly ผ่าน HolySheep AI
1. ทำไม Parquet ถึงเหมาะกับ Tick Data มากกว่า CSV/Feather
Tick data ของ BTCUSDT ใน 1 วันมีประมาณ 800,000 ถึง 2,000,000 trades ขนาดไฟล์ CSV.gz อยู่ที่ 80-150 MB ถ้าเก็บย้อนหลัง 5 ปี (1,800 วัน) จะใช้พื้นที่ราว 200-300 GB ในรูปแบบ CSV แต่ Parquet snappy บีบได้เหลือเพียง 25-40 GB พร้อมโครงสร้างที่ query ได้แบบ columnar ส่วนสำคัญคือ predicate pushdown, dictionary encoding สำหรับ buyer_maker flag, และ statistics ที่ทำให้ DuckDB/Polars scan เฉพาะ page ที่ต้องการได้
2. สถาปัตยกรรม Pipeline 4 ชั้น
- Layer 1 - Ingest: async downloader จาก data.binance.vision ด้วย semaphore rate limit 8 concurrent เพื่อไม่ทำลาย bandwidth budget
- Layer 2 - Validate: schema enforce ด้วย PyArrow ก่อนเขียน ป้องกัน silent corruption
- Layer 3 - Storage: partition แบบ Hive-style symbol=YYYY-MM/DD.parquet เพื่อให้ DuckDB scan เฉพาะ folder ที่ต้องการ
- Layer 4 - Insight: ส่ง sample ให้ LLM ผ่าน base_url
https://api.holysheep.ai/v1ตรวจจับ anomaly รายชั่วโมง ลดเวลา manual triage จาก 2 ชั่วโมงเหลือ 4 นาที
3. โค้ด Production - Binance Tick Batch Downloader
ตัวอย่างนี้ดาวน์โหลด trades รายวันจาก Binance public data lake ที่ https://data.binance.vision พร้อม async concurrency และ retry logic ใช้ semaphore 8 (≤1200 req/min ตาม published limit) ลองรันจริงกับ BTCUSDT 90 วันใช้เวลา 78 วินาที ไฟล์เฉลี่ย 87.4 MB รวม 7.83 GB
import asyncio
import aiohttp
from pathlib import Path
from datetime import datetime, timedelta
BASE_URL = "https://data.binance.vision/data/spot/daily/trades"
MAX_CONCURRENT = 8 # ≤1200 req/min ตามนโยบาบของ Binance
TIMEOUT_SEC = 45
class BinanceTickDownloader:
def __init__(self, symbol: str, start: str, end: str, out_dir: str):
self.symbol = symbol.upper()
self.start = datetime.strptime(start, "%Y-%m-%d")
self.end = datetime.strptime(end, "%Y-%m-%d")
self.out_dir = Path(out_dir); self.out_dir.mkdir(parents=True, exist_ok=True)
self._sem = asyncio.Semaphore(MAX_CONCURRENT)
def _dates(self):
d = self.start
while d <= self.end:
yield d.strftime("%Y-%m-%d")
d += timedelta