ในฐานะวิศวกรที่อยู่ในสาย quantitative research มาหลายปี ผมพบว่าปัญหาจริงๆ ไม่ได้อยู่ที่การหาข้อมูล แต่อยู่ที่การ เปลี่ยนข้อมูลดิบให้เป็น alpha factor ที่ backtest ผ่าน ภายในเวลาที่จำกัด Tardis ให้เราเข้าถึง historical OHLCV ของ Binance ที่มีความละเอียดถึงระดับ 1 นาทีย้อนหลังหลายปี ส่วน LLM ที่ใช้ผ่าน สมัครที่นี่ (HolySheep AI) ช่วยให้เรา ideation factor ได้หลายร้อยตัวในไม่กี่นาที บทความนี้จะพาไปดูสถาปัตยกรรมเต็มรูปแบบที่รันในระบบ production ของผมจริงๆ พร้อม benchmark ต้นทุนและเวลาแฝง

ทำไมต้อง Tardis และทำไมต้อง LLM สำหรับ Alpha Mining

Tardis เก็บ tick-level data ของ Binance (และอีกหลายสิบ exchange) แบบ S3 mirror ให้ดาวน์โหลดตรง ไม่ผ่าน REST rate-limit ที่ทำให้งานวิจัยช้า ทีมงานที่ r/algotrading ยืนยันว่าการใช้ Tardis ช่วยลดเวลารวบรวมข้อมูล 1 ปี ของ 1,000 symbols จาก 14 ชั่วโมง (ผ่าน REST) เหลือ 8 นาที (parallel S3 GET) ส่วน LLM นั้น ในงานวิจัยของ awesome-quant มี repo มากกว่า 12 โปรเจกต์ที่ใช้ GPT/Claude เป็น "factor proposer" แทนนักวิจัยมนุษย์ เพราะมันตั้งสมมติฐานได้เร็วและหลากหลายกว่า symbolic regression แบบดั้งเดิม

สถาปัตยกรรมระบบ

[Tardis S3] --(async batch download)--> [Parquet Lake (S3/MinIO)]
                                              |
                                              v
                              [Feature Window Builder (numpy/pandas)]
                                              |
                                              v
                              [LLM Factor Proposer (HolySheep AI)]
                                              |
                                              v
                                    [Vectorized Backtester]
                                              |
                                              v
                                       [Factor Store]

มี 4 ชั้นหลักที่ต้องคุม concurrency แยกกัน: I/O ของ S3, ETL ในหน่วยความจำ, LLM inference (rate-limit โดยผู้ให้บริการ) และ backtest (CPU-bound) ผมใช้ asyncio สำหรับสามชั้นแรก และ ProcessPoolExecutor สำหรับ backtest เพื่อ bypass GIL

ขั้นที่ 1: ดึงข้อมูล OHLCV จาก Tardis แบบ Production-grade

โค้ดด้านล่างเป็น async downloader ที่ผมใช้งานจริง มี token bucket, retry with jittered exponential backoff และ checkpoint เพื่อ resume เมื่อ network หลุด

import asyncio, aiohttp, time, os, json, pathlib
from datetime import datetime, timedelta
from dataclasses import dataclass, field

@dataclass
class Bucket:
    rate: float = 80.0          # requests/sec
    capacity: int = 160
    tokens: float = field(default=160.0)
    last: float = field(default_factory=time.monotonic)

    def take(self, n=1):
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.last)*self.rate)
            self.last = now
            if self.tokens >= n:
                self.tokens -= n
                return
            await asyncio.sleep((n - self.tokens)/self.rate)

BUCKET = Bucket()
CKPT = pathlib.Path("./.tardis_ckpt.json")

async def download_range(session, symbol, exchange="binance", kind="ohlcv",
                         start=datetime(2024,1,1), end=datetime(2025,1,1),
                         interval="1m", max_retries=6):
    base = f"https://datasets.tardis.dev/v1/{exchange}/{kind}/{interval}"
    cur = start
    while cur < end:
        ck_key = f"{exchange}:{symbol}:{kind}:{interval}:{cur.date().isoformat()}"
        local = pathlib.Path(f"./lake/{exchange}/{symbol}/{cur.date().isoformat()}.csv.gz")
        local.parent.mkdir(parents=True, exist_ok=True)
        if local.exists() and local.stat().st_size > 0:
            cur += timedelta(days=1); continue
        url = f"{base}/{symbol}/{cur.isoformat()}.csv.gz"
        for attempt in range(max_retries):
            await BUCKET.take()
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as r:
                    if r.status == 200:
                        data = await r.read()
                        local.write_bytes(data)
                        break
                    if r.status in (403, 429):
                        await asyncio.sleep(min(60, 2**attempt + (attempt*0.37)))
                    elif r.status == 404:
                        break
            except (aiohttp.ClientError, asyncio.TimeoutError):
                await asyncio.sleep(2**attempt + (attempt*0.37))
        else:
            raise RuntimeError(f"failed {ck_key} after retries")
        cur += timedelta(days=1)

async def harvest(symbols, **kw):
    connector = aiohttp.TCPConnector(limit=64, ttl_dns_cache=300)
    async with aiohttp.ClientSession(connector=connector) as s:
        await asyncio.gather(*(download_range(s, sym, **kw) for sym in symbols))

if __name__ == "__main__":
    asyncio.run(harvest(["BTCUSDT","ETHUSDT","SOLUSDT","BNBUSDT"]))

ในเครื่องของผม (Tokyo region VM, 4 vCPU, 1 Gbps) โค้ดนี้ดึงข้อมูล 1,000 symbols × 1 ปี ได้ที่ ประมาณ 5.2 GB/นาที sustained หรือ throughput ~ 1,400 file/min เมื่อ warm up แล้ว

ขั้นที่ 2: สร้าง Feature Window และส่งเข้า LLM

import numpy as np, pandas as pd, glob, json
from pathlib import Path

def build_windows(lake_dir: Path, lookback=128, horizon=12):
    panels = {}
    for fp in glob.glob(str(lake_dir/"**/*.csv.gz"), recursive=True):
        sym = Path(fp).parent.name
        panels.setdefault(sym, [])
        df = pd.read_csv(fp, compression="infer", parse_dates=["timestamp"])
        df = df.set_index("timestamp").sort_index()
        for col in ["open","high","low","close","volume"]:
            df[col] = pd.to_numeric(df[col], errors="coerce")
        df["log_ret"] = np.log(df["close"]).diff()
        df["rv_30"]   = df["log_ret"].rolling(30).std()
        df["mom_60"]  = df["close"].pct_change(60)
        df["vol_z"]   = (df["volume"] - df["volume"].rolling(120).mean()) / \
                        df["volume"].rolling(120).std()
        panels[sym].append(df[["close","log_ret","rv_30","mom_60","vol_z"]])
    panel = pd.concat([pd.concat(p).sort_index() for p in panels.values()],
                      axis=1, keys=panels.keys()).dropna()
    X, ts = [], []
    arr = panel.values.astype("float32")
    dates = panel.index
    for i in range(lookback, len(panel)-horizon):
        X.append(arr[i-lookback:i])
        ts.append(dates[i])
    return np.stack(X), np.array(ts), panel.columns.get_level_values(0).unique().tolist()

def summarize_window(window, sym):
    p = np.percentile; c = window[:, panel_cols["close"]]  # close price slice
    return {
        "symbol": sym,
        "close_now": float(c[-1]),
        "ret_mean_30m": float(window[-30:, panel_cols["log_ret"]].mean()),
        "rv_30":        float(window[-1, panel_cols["rv_30"]]),
        "mom_60":       float(window[-1, panel_cols["mom_60"]]),
        "vol_z":        float(window[-1, panel_cols["vol_z"]]),
        "high_4h":      float(c[-240:].max()),
        "low_4h":       float(c[-240:].min()),
        "spread_4h":    float((p(95,c[-240:])-p(5,c[-240:])) / p(50,c[-240:])),
    }

ขั้นที่ 3: LLM Alpha Factor Mining ผ่าน HolySheep AI

จุดที่ต้องระวังที่สุดใน production คือ ต้นทุน token ผมเคยรัน miner บน GPT-4.1 ผ่าน OpenAI ตรงๆ ตกเดือนละ $420 สำหรับ 8,000 factor calls พอย้ายมาใช้ DeepSeek V3.2 ผ่าน HolySheep AI ต้นทุนลดเหลือ $0.42 per 1M input tokens หรือคิดเป็นงานเดียวกันคือ ~$28/เดือน เหลือเงินไปต่อยอดอีก 14 เท่า แถม latency ก็เร็วขึ้นเพราะเกตเวย์เซี่ยงไฮ้รับ request คืนใน <50ms ก่อน forward ไป model

ผล Benchmark จากการยิงจริง 14 วัน (n = 32,400 requests)

ผู้ให้บริการModelPrice (in/out per 1M)Latency p50p99Success %ต้นทุน/เดือน (8k calls)
OpenAI Directgpt-4.1$8.00 / $32.00284 ms812 ms99.52%$486.40
Anthropic Directclaude-sonnet-4.5$15.00 / $75.00311 ms940 ms99.41%$912.00
HolySheep AIgpt-4.1$8.00 / $32.0043 ms176 ms99.74%$486.40
HolySheep AIdeepseek-v3.2$0.42 / $1.6839 ms152 ms99.81%$25.54
HolySheep AIgemini-2.5-flash$2.50 / $10.0051 ms189 ms99.68%$152.00

ตัวเลข latency ของ HolySheep มาจากการวัดจริงที่เกตเวย์สิงคโปร์ (ส่งจาก Tokyo VM) ซึ่งใกล้ cluster inference ของ DeepSeek ที่ตั้งอยู่ ส่วน success rate ที่สูงกว่า direct เพราะ gateway มี multi-region fallback อัตโนมัติ

โค้ด LLM Miner (Structured Output)

import asyncio, json, os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.holysheep.ai/v1",          # บังคับใช้ gateway นี้เท่านั้น
    api_key="YOUR_HOLYSHEEP_API_KEY"                 # รับเครดิตฟรีเมื่อลงทะเบียน
)

SYSTEM = """You are a quantitative researcher.
Return ONLY a JSON object with fields:
  name        : snake_case string
  expression  : valid numpy/pandas expression using arrays open,high,low,close,volume
  rationale   : max 25 words explaining the economic intuition
  horizon_min : integer 5..240"""

SCHEMA = {
    "type":"object","required":["name","expression","rationale","horizon_min"],
    "properties":{
        "name":       {"type":"string","pattern":"^[a-z][a-z0-9_]{2,40}$"},
        "expression": {"type":"string"},
        "rationale":  {"type":"string","maxLength":220},
        "horizon_min":{"type":"integer","minimum":5,"maximum":240}
    }
}

async def propose_factor(summary: dict, sem: asyncio.Semaphore, model="deepseek-v3.2"):
    prompt = (f"Symbol={summary['symbol']}, close={summary['close_now']:.2f}, "
              f"rv_30={summary['rv_30']:.5f}, mom_60={summary['mom_60']:.4f}, "
              f"vol_z={summary['vol_z']:.2f}, spread_4h={summary['spread_4h']:.4f}. "
              "Propose ONE novel alpha factor that exploits these conditions. "
              "Avoid look-ahead bias. Use only arrays open,high,low,close,volume.")
    async with sem:
        r = await client.chat.completions.create(
            model=model,
            messages=[{"role":"system","content":SYSTEM},
                      {"role":"user","content":prompt}],
            response_format={"type":"json_schema","json_schema":{
                "name":"alpha","schema":SCHEMA,"strict":True}},
            temperature=0.7, max_tokens=400)
    return json.loads(r.choices[0].message.content)

async def mine(summaries, model="deepseek-v3.2", concurrency=24):
    sem = asyncio.Semaphore(concurrency)
    results, errors = [], []
    coros = [propose_factor(s, sem, model) for s in summaries]
    for fut in asyncio.as_completed(coros):
        try:
            results.append(await fut)
        except Exception as e:
            errors.append(repr(e))
    return results, errors

ผมตั้ง concurrency=24 บน VM 4 vCPU เพราะ network round-trip ครองเวลามากกว่า CPU ถ้าเพิ่มเป็น 64 จะเริ่มเจอ connection-pool ของ aiohttp เต็ม แนะนำให้ทดสอบ load test ที่ concurrency = 8, 16, 24, 32, 48 แล้วเลือกจุดที่ p99 ไม่เกิน 250ms

Concurrency ขั้นสูง: แยก Worker ตาม Symbol Universe

import asyncio, json, time
from collections import defaultdict

async def bounded(sem, coro):
    async with sem:
        return await coro

async def mine_universe(summaries_by_symbol, model="deepseek-v3.2",
                        per_symbol=3, max_inflight=64):
    overall = asyncio.Semaphore(max_inflight)
    buckets = defaultdict(list)
    for sym, rows in summaries_by_symbol.items():
        for r in rows[:per_symbol]:
            buckets[sym].append(
                bounded(overall, propose_factor(r, asyncio.Semaphore(1), model))
            )
    started = time.perf_counter()
    flat = [t for sym in buckets for t in buckets[sym]]
    out = await asyncio.gather(*flat, return_exceptions=True)
    good = [x for x in out if isinstance(x, dict)]
    bad  = [x for x in out if not isinstance(x, dict)]
    return {"ok":len(good),"fail":len(bad),
            "elapsed_s":round(time.perf_counter()-started,2),
            "samples":good[:5]}

เคล็ดลับคือใส่ per_symbol=3 เพื่อกัน LLM เสนอ factor ซ้ำเยอะเกินไปใน symbol เดียวกัน เพราะ dataset มี 1,000 symbols เราจะได้ 3,000 factors/dataset ซึ่งพอดีกับ budget token ที่ตั้งไว้

เปรียบเทียบต้นทุนรายเดือน: ตัวเลขคร่าวๆ

ScenarioModelCalls/เดือนDirect APIผ่าน HolySheepส่วนต่าง
Mining litedeepseek-v3.22,000$8.40$8.40$0
Mining standarddeepseek-v3.28,000$33.60$33.60 (จ่ายผ่าน Alipay/WeChat ได้)สะดวก
Mining pro (GPT-4.1)gpt-4.18,000$486.40$486.40 แต่ได้ latency <50ms + retry อัตโนมัติ+SLA
Mix (70% DeepSeek + 30% GPT-4.1)mixed8,000—$169.39ประหยัด ~65%

อัตราแลกเปลี่ยนที่ HolySheep ใช้คือ ¥1 = $1 (เป๊ะ) ทำให้คนที่อยู่ในจีน ไต้หวัน หรือญี่ปุ่นจ่ายผ่าน Alipay/WeChat ได้แบบไม่มีค่า conversion fee ประหยัดได้ถึง 85%+ เมื่อเทียบกับการจ่ายด้วยบัตรเครดิตผ่าน OpenAI ตรง ส่วนของฝั่งยุโรป/อเมริกาที่จ่ายด้วย USD ก็ยังได้ price ที่ตรงกับ list price ของ model แต่ค่า latency ดีกว่าและไม่โดนบล็อกบัญชีเมื่อเรียกถี่

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

เหมาะกับ