ผมเขียนบทความนี้จากประสบการณ์ตรงในฐานะวิศวกรที่ดูแล data pipeline ของทีม quantitative research มาเกือบ 3 ปี โปรเจกต์ที่จะเล่าคือการย้ายข้อมูล Bybit trades ย้อนหลัง 18 เดือน จาก Tardis Historical Data เข้าสู่ ClickHouse cluster ขนาด 6 โหนด ใช้เวลาจริงตามแผน 14 วันทำการ สิ่งที่ทำให้งานสำเร็จเร็วกว่าที่คาดไว้เกือบ 40% คือการใช้ HolySheep AI เป็นผู้ช่วยเขียน schema ตรวจสอบคุณภาพข้อมูล และวิเคราะห์ความผิดปกติของราคาภายหลังโหลดเสร็จ บทความนี้จะแชร์แผนงาน ความเสี่ยง แผนย้อนกลับ และการคำนวณ ROI แบบละเอียดทุกขั้น

ทำไมต้องย้ายจาก Tardis API ตรงไปเป็นคลังข้อมูล ClickHouse

ก่อนหน้านี้ทีมผมดึงข้อมูลจาก Tardis ผ่าน HTTP API แล้วเก็บเป็นไฟล์ CSV บน NFS ปัญหาที่เจอซ้ำ ๆ มี 4 ข้อหลัก ได้แก่ 1) เวลาดึงข้อมูลย้อนหลัง 1 วันใช้เวลา 12 นาที 2) การ query ย้อนหลัง 6 เดือนใช้เวลาเกือบ 90 วินาที 3) ไม่สามารถทำ concurrent query จากทีม research 4 คนพร้อมกันได้ และ 4) ทุกครั้งที่ schema เปลี่ยนต้องเขียน migration script ใหม่เอง เมื่อย้ายเข้า ClickHouse ทุกปัญหาเหล่านี้หายไป เวลา query เหลือไม่ถึง 800 มิลลิวินาที และ concurrent query รองรับได้ถึง 40 connection พร้อมกัน

อีกเหตุผลสำคัญคือความเร็วในการพัฒนา เมื่อก่อนการเขียน DDL สำหรับ ClickHouse ใช้เวลาเฉลี่ย 45 นาทีต่อตาราง แต่เมื่อใช้ HolySheep AI ช่วยออกแบบ schema เวลาลดลงเหลือ 8 นาที เพราะโมเดลช่วยแนะนำ engine ที่เหมาะสมอย่าง ReplacingMergeTree สำหรับตารางที่มีการ update และ partition key ที่เหมาะกับ trade data

สิ่งที่ต้องเตรียมก่อนเริ่มโปรเจกต์

ขั้นตอนที่ 1 ให้ HolySheep AI ช่วยออกแบบ ClickHouse schema

ขั้นแรกผมใช้ HolySheep AI ช่วยออกแบบ schema โดยส่งตัวอย่างไฟล์ Tardis ไป 10 แถวให้โมเดลวิเคราะห์ แล้วขอให้สร้าง DDL ที่เหมาะกับ trade data ของ Bybit ที่มีปริมาณวันละประมาณ 50 ล้านแถว ใช้เวลาเพียง 8 วินาที latency ของ HolySheep วัดได้ ต่ำกว่า 50 มิลลิวินาที ในการทดสอบ 200 request ติดกัน

import os
import requests

HOLYSHEEP_API_KEY = os.environ["HOLYSHEEP_API_KEY"]
BASE_URL = "https://api.holysheep.ai/v1"

sample_csv = """exchange,symbol,timestamp,local_timestamp,id,side,price,amount
bybit,BTCUSDT,1704067200000000,1704067200123456,123456789,buy,42150.5,0.012
bybit,BTCUSDT,1704067200000123,1704067200200123,123456790,sell,42151.0,0.005
bybit,ETHUSDT,1704067200000500,1704067200500500,987654321,buy,2240.7,0.150"""

payload = {
    "model": "deepseek-v3.2",
    "messages": [
        {
            "role": "system",
            "content": "You are a ClickHouse DBA expert. Generate optimized DDL for trading data."
        },
        {
            "role": "user",
            "content": f"ออกแบบ ClickHouse schema สำหรับข้อมูลนี้ ใช้ MergeTree แบ่ง partition รายเดือน รองรับ 50 ล้านแถวต่อวัน\n\n{sample_csv}"
        }
    ],
    "temperature": 0.1
}

resp = requests.post(
    f"{BASE_URL}/chat/completions",
    headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
    json=payload,
    timeout=30
)
resp.raise_for_status()
print(resp.json()["choices"][0]["message"]["content"])

ขั้นตอนที่ 2 ดาวน์โหลดข้อมูล Bybit trades จาก Tardis S3

Tardis เก็บข้อมูลไว้ในรูปแบบ CSV.gz ที่ s3://tardis-data/bybit/trades/YYYY-MM-DD/ โดยแต่ละไฟล์มีขนาดประมาณ 800 MB ถึง 1.2 GB สำหรับวันที่มีความผันผวนสูง ผมเขียนสคริปต์ดาวน์โหลดแบบขนานโดยใช้ boto3 พร้อม retry logic และ progress logging ขั้นตอนนี้ใช้เวลาเฉลี่ย 18 นาทีต่อวัน

import boto3
from botocore.config import Config
from botocore import UNSIGNED
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import date, timedelta
from pathlib import Path

DOWNLOAD_DIR = Path("/data/tardis_raw/bybit/trades")
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)

s3 = boto3.client(
    "s3",
    config=Config(signature_version=UNSIGNED, retries={"max_attempts": 5})
)
BUCKET = "tardis-data"


def download_day(d: date) -> str:
    prefix = f"bybit/trades/{d.isoformat()}/"
    paginator = s3.get_paginator("list_objects_v2")
    saved = []
    for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            local = DOWNLOAD_DIR / Path(key).relative_to("bybit/trades/")
            local.parent.mkdir(parents=True, exist_ok=True)
            if local.exists() and local.stat().st_size == obj["Size"]:
                continue
            s3.download_file(BUCKET, key, str(local))
            saved.append(str(local))
    return f"{d}: {len(saved)} files"


start = date(2024, 7, 1)
end = date(2026, 1, 1)
days = [start + timedelta(days=i) for i in range((end - start).days)]

with ThreadPoolExecutor(max_workers=8) as ex:
    futures = {ex.submit(download_day, d): d for d in days}
    for fut in as_completed(futures):
        print(fut.result())

ขั้นตอนที่ 3 โหลดเข้า ClickHouse พร้อมตรวจสอบคุณภาพข้อมูลด้วย HolySheep AI

ขั้นตอนนี้สำคัญที่สุด ผมใช้ clickhouse-connect สำหรับ insert แบบ batch และใช้ HolySheep AI ช่วยสร้าง data quality check ที่ครอบคลุม ได้แก่ การตรวจ missing value การตรวจ price outlier การตรวจ timestamp ที่เรียงไม่ถูกต้อง และการตรวจจำนวน trade ต่อนาทีเทียบกับค่าเฉลี่ยย้อนหลัง การใช้ AI ช่วยตรงนี้ทำให้ผมไม่ต้องเขียน SQL ตรวจเอง 40 แบบทดสอ

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

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