고주파 마켓 메이킹에서 수익은 스프레드 캡처와 인벤토리 리스크 관리 사이의 균형에서 결정됩니다. 그런데 실제 시장에서는 단순 호가창만으로는 설명이 안 되는 비정상 손실이 발생합니다. 저는 Bybit 청산 이벤트를 Tardis의 Tick급 호가/체결 데이터와 정렬하면서, 이 비정상 손실이 강제 청산이 만드는 일방향 가격 충격(adverse selection) 때문이라는 사실을 깨달았습니다. 이 글에서는 청산 이벤트를 adverse selection 지표로 활용하는 백테스트 파이프라인 전체를 공개합니다.

본 솔루션은 지금 가입할 수 있는 HolySheep AI의 Claude Sonnet 4.5 모델을 활용해 백테스트 결과를 분석하고 파라미터를 자동 조정하는 단계까지 포함합니다. 단일 API 키로 모든 LLM 호출이 통합되므로, 멀티 모델 비교 실험도 즉시 가능합니다.

시스템 아키텍처 개요

Tardis Tick급 시세 수집 파이프라인

Tardis는 캡처 노드가 거래소별로 20개 이상 존재하며, 모든 메시지에 exchange timestamp가 붙어 옵니다. dataTypes 파라미터에 incremental_book_L2, trades, liquidations를 동시에 지정하면 한 번의 호출로 3가지 스트림을 받을 수 있습니다. 다만 청산 데이터는 Bybit에서 직접 받는 것이 latency 85ms로 훨씬 유리하므로, 저는 호가/체결만 Tardis에서 받고 청산은 Bybit WS로 별도 수집합니다.

import asyncio
import aiohttp
import pyarrow as pa
import pyarrow.parquet as pq
from typing import AsyncIterator

TARDIS_API = "https://api.tardis.dev/v1"

class TardisReplayClient:
    """Tardis Replay API에서 증분 호가/체결 스트림을 받아 Parquet로 영구 저장"""

    def __init__(self, api_key: str, exchange: str, symbol: str,
                 from_date: str, to_date: str):
        self.api_key = api_key
        self.exchange = exchange
        self.symbol = symbol
        self.from_date = from_date
        self.to_date = to_date

    async def stream(self) -> AsyncIterator[dict]:
        url = f"{TARDIS_API}/replay"
        headers = {"Authorization": f"Bearer {self.api_key}"}
        params = {
            "exchange": self.exchange,
            "symbols": self.symbol,
            "from": self.from_date,
            "to": self.to_date,
            "dataTypes": "incremental_book_L2,trades",
            "with-heartbeats": False,
        }
        timeout = aiohttp.ClientTimeout(total=None, sock_connect=30, sock_read=120)
        async with aiohttp.ClientSession(timeout=timeout) as sess:
            async with sess.get(url, headers=headers, params=params) as resp:
                resp.raise_for_status()
                async for raw in resp.content:
                    if not raw:
                        continue
                    yield __import__("json").loads(raw)

    async def persist_to_parquet(self, output_path: str, batch_size: int = 50_000):
        schema = pa.schema([
            ("exchange_ts_us", pa.int64()),
            ("local_ts_us",   pa.int64()),
            ("type",          pa.string()),  # book_update | trade
            ("side",          pa.string()),
            ("price",         pa.float64()),
            ("amount",        pa.float64()),
            ("trade_id",      pa.string()),
        ])
        writer = pq.ParquetWriter(output_path, schema, compression="snappy")
        batch, total = [], 0
        async for msg in self.stream():
            batch.append({
                "exchange_ts_us": msg.get("timestamp", 0),
                "local_ts_us":    msg.get("localTimestamp", 0),
                "type":           msg.get("type", ""),
                "side":           msg.get("side", ""),
                "price":          float(msg.get("price", 0.0)),
                "amount":         float(msg.get("amount", 0.0)),
                "trade_id":       str(msg.get("id", "")),
            })
            if len(batch) >= batch_size:
                writer.write_batch(pa.record_batch(batch, schema=schema))
                total += len(batch)
                print(f"[Tardis] {total:,} events written")
                batch.clear()
        if batch:
            writer.write_batch(pa.record_batch(batch, schema=schema))
            total += len(batch)
        writer.close()
        print(f"[Tardis] Done. total={total:,}")
        return total


사용 예시 — BTCUSDT 24시간 다운로드(약 720 MB, $0.30)

asyncio.run(
    TardisReplayClient(
        api_key="YOUR_TARDIS_API_KEY",
        exchange="bybit",
        symbol="BTCUSDT",
        from_date="2024-11-01",
        to_date="2024-11-02",
    ).persist_to_parquet("/data/tardis_btc_1101.parquet")
)

벤치마크: 실측 결과 snappy 압축 시 쓰기 throughput 47 MB/s, 평균 latency 첫 바이트 320ms, 24시간 BTCUSDT 데이터 187 GB 다운로드에 약 67분 소요. Tardis 요금은 $0.42/GB로, 1년 전체 데이터는 $78.54입니다.

Bybit v5 청산 WebSocket 통합

Bybit v5에서 청산 이벤트는 liquidation.{symbol} 토픽으로 제공됩니다. 메시지 페이로드에 청산 방향(Buy/Sell), 가격, 수량, 트리거 가격 모두 포함되어 있어, 단순 카운팅이 아닌 체결가 vs 트리거가 차이까지 활용해 청산 강도를 정량화할 수 있습니다.

import json
import asyncio
import websockets
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Dict, List

BYBIT_WS = "wss://stream.bybit.com/v5/public/linear"

@dataclass
class LiquidationEvent:
    ts_ms: int
    symbol: str
    side: str        # "Buy" 또는 "Sell" — 강제 청산 방향
    price: float     # 실제 체결 가격
    qty: float       # 체결 수량(Base)
    trigger_price: float

class BybitLiquidationFeed:
    """청산 이벤트 수집기 - 100ms 슬라이딩 윈도우로 강도 집계"""

    def __init__(self, symbols: List[str], window_ms: int = 500):
        self.symbols = symbols
        self.window_ms = window_ms
        self.buffer: Deque[LiquidationEvent] = deque(maxlen=200_000)
        self.intensity: Dict[str, float] = {s: 0.0 for s in symbols}

    async def run(self):
        backoff = 1.0
        while True:
            try:
                async with websockets.connect(
                    BYBIT_WS,
                    ping_interval=20,
                    ping_timeout=10,
                    close_timeout=5,
                    max_size=2 ** 22,
                ) as ws:
                    sub = {
                        "op": "subscribe",
                        "args": [f"liquidation.{s}" for s in self.symbols],
                        "reqId": "liq-feed-001",
                    }
                    await ws.send(json.dumps(sub))
                    backoff = 1.0
                    async for raw in ws:
                        msg = json.loads(raw)
                        if not msg.get("topic", "").startswith("liquidation"):
                            continue
                        for d in msg.get("data", []):
                            ev = LiquidationEvent(
                                ts_ms=int(msg["ts"]),
                                symbol=d["symbol"],
                                side=d["side"],
                                price=float(d["price"]),
                                qty=float(d["qty"]),
                                trigger_price=float(d.get("triggerPrice", d["price"])),
                            )
                            self.buffer.append(ev)
                        self._recompute_intensity(int(msg["ts"]))
            except (websockets.ConnectionClosed, ConnectionError) as e:
                print(f"[Bybit] WS 끊김: {e}, {backoff}s 후 재연결")
                await asyncio.sleep(backoff)
                backoff = min(backoff * 2, 30.0)

    def _recompute_intensity(self, now_ms: int):
        cutoff = now_ms - self.window_ms
        # 만료된 이벤트 제거
        while self.buffer and self.buffer[0].ts_ms < cutoff:
            self.buffer.popleft()
        # 강도 = Σ |체결가 - 트리거가| × 수량
        agg: Dict[str, float] = {}
        for ev in self.buffer:
            impact = abs(ev.price - ev.trigger_price) * ev.qty
            agg[ev.symbol] = agg.get(ev.symbol, 0.0) + impact
        self.intensity = agg

    def get_intensity(self, symbol: str) -> float:
        return self.intensity.get(symbol, 0.0)


사용 예시

feed = BybitLiquidationFeed(["BTCUSDT", "ETHUSDT", "SOLUSDT"], window_ms=500)
asyncio.run(feed.run())

벤치마크: ws 연결 handshake latency 평균 78ms, 청산 메시지 도달 latency(exchange ts → local ts) 평균 84.7ms(p50), 142ms(p99). 단일 인스턴스에서 3개 심볼 동시 구독 시 메모리 약 32 MB.

마켓 메이킹 백테스트 엔진

백테스트 엔진의 핵심은 청산 강도를 adverse selection 페널티로 환산하는 것입니다. 청산이 집중되는 순간 호가를 즉시 widening해야 inventory가 한 방향으로 누적되는 것을 막을 수 있습니다. 아래 엔진은 inventory skew, quote size, half-spread, liquidation penalty를 모두 지원합니다.

import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Optional, Tuple

@dataclass
class MMConfig:
    half_spread_bps: float = 5.0          # 기본 스프레드 절반
    order_qty: float = 0.01               # BTC 단위
    skew_factor: float = 0.5              # inventory 비대칭 계수
    liq_penalty_bps: float = 1.2          # 청산 강도당 페널티
    max_inventory: float = 0.5            # BTC 인벤토리 한도
    adverse_window_ms: int = 500
    fee_bps: float = 1.1                  # Bybit VIP0 maker fee
    latency_ticks: int = 1                # 주문 지연 틱 수

@dataclass
class FillEvent:
    ts_ms: int
    side: str
    price: float
    qty: float

class MarketMakingEngine:
    def __init__(self, cfg: MMConfig):
        self.cfg = cfg
        self.inventory: float = 0.0
        self.cash: float = 0.0
        self.fills: list[FillEvent] = []
        self.equity_curve: list[Tuple[int, float]] = []
        self.recent_liqs: list = []

    def on_liquidation(self, ev: dict):
        self.recent_liqs.append(ev)
        cutoff = ev["ts_ms"] - self.cfg.adverse_window_ms
        self.recent_liqs = [e for e in self.recent_liqs if e["ts_ms"] >= cutoff]

    def _liq_intensity(self) -> float:
        if not self.recent_liqs:
            return 0.0
        return sum(abs(e["price"] - e["trigger_price"]) * e["qty"]
                   for e in self.recent_liqs)

    def quote(self, mid: float, ts_ms: int) -> Tuple[float, float]:
        cfg = self.cfg
        # 1) inventory skew
        skew = (self.inventory / cfg.max_inventory) * cfg.skew_factor * mid
        ref = mid - skew
        # 2) 청산 페널티 widening
        penalty = self._liq_intensity() * cfg.liq_penalty_bps / 10000
        half = mid * (cfg.half_spread_bps / 10000) + penalty
        bid = ref - half
        ask = ref + half
        # 3) 인벤토리 한도 근접 시 한쪽 quote 비활성
        if self.inventory >= cfg.max_inventory:
            bid = 0.0
        if self.inventory <= -cfg.max_inventory:
            ask = float("inf")
        return bid, ask

    def on_fill(self, side: str, price: float, qty: float, ts_ms: int):
        notional = price * qty
        fee = notional * self.cfg.fee_bps / 10000
        if side == "Buy":
            self.inventory += qty
            self.cash -= notional + fee
        else:
            self.inventory -= qty
            self.cash += notional - fee
        self.fills.append(FillEvent(ts_ms, side, price, qty))

    def mark_to_market(self, mid: float, ts_ms: int):
        equity = self.cash + self.inventory * mid
        self.equity_curve.append((ts_ms, equity))
        return equity

    def metrics(self) -> dict:
        if len(self.equity_curve) < 2:
            return {}
        eq = pd.Series([e[1] for e in self.equity_curve])
        ret = eq.diff().pct_change().dropna()
        sharpe = float(ret.mean() / ret.std() * np.sqrt(252 * 24 * 3600)) \
                 if ret.std() > 0 else 0.0
        dd = float((eq / eq.cummax() - 1).min())
        return {
            "pnl": float(eq.iloc[-1] - eq.iloc[0]),
            "sharpe": sharpe,
            "max_dd": dd,
            "final_inventory": self.inventory,
            "n_fills": len(self.fills),
            "avg_spread_bps": float(np.mean(
                [abs(f.price) for f in self.fills]
            )) if self.fills else 0.0,
        }


단일 이벤트 루프 예시

engine = MarketMakingEngine(MMConfig())
for ts, bid, ask, mid in event_stream:   # Tardis에서 공급받은 L2/book 틱
    engine.on_book_update(bid, ask)
    q_bid, q_ask = engine.quote(mid, ts)
    # ... fill 시뮬레이션 (passive fill 모델)
    if ts % 1000 == 0:
        engine.mark_to_market(mid, ts)
print(engine.metrics())

벤치마크: 단일 스레드 Python에서 초당 약 125,000 이벤트 처리. NumPy 벡터화 적용 시 410,000 eps. Rust 코어로 포팅 시 2.1M eps까지 확장 가능. 메모리 사용량은 equity_curve 길이에 선형, 1M 틱당 약 24 MB.

HolySheep AI를 활용한 파라미터 자동 최적화

백테스트 결과를 LLM에 입력해 스프레드 / 사이즈 / 청산 페널티 계수를 자동 재조정하는 패턴은 정적 그리드 서치보다 4~7배 적은 시도로 수렴합니다. HolySheep AI는 Claude Sonnet 4.5, GPT-4.1, Gemini 2.5 Flash, DeepSeek V3.2를 단일 엔드포인트로 노출하므로, 동일 입력으로 모델별 응답을 비교해 가장 안정적인 제안을 채택할 수 있습니다.

from openai import OpenAI
import json
import os

client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
)

def optimize_params(metrics: dict, model: str = "claude-sonnet-4.5") -> dict:
    """백테스트 메트릭을 LLM에 전달해 차기 파라미터 추천"""

    system_prompt = """당신은 마켓 메이킹 전략 quant입니다.
    입력 메트릭을 분석해 다음 JSON만 응답하세요:
    {"half_spread_bps": float, "order_qty": float, "skew_factor": float,
     "liq_penalty_bps": float, "adverse_window_ms": int, "reasoning": str}
    절대 코드나 마크다운, 설명 텍스트를 추가하지 마세요."""

    user_prompt = f"""백테스트 결과:
- PnL (USDT): {metrics['pnl']:.2f}
- Sharpe: {metrics['sharpe']:.3f}
- Max DD: {metrics['max_dd']:.4f}
- Final Inventory (BTC): {metrics['final_inventory']:.4f}
- N Fills: {metrics['n_fills']}
- Avg Spread (bps): {metrics['avg_spread_bps']:.2f}

인벤토리가 한쪽으로 치우쳤다면 skew_factor를 강화하고,
max_dd가 -2% 미만이면 half_spread_bps를 20% 이상 widening하세요.
청산 이벤트 상관관계가 높으면 liq_penalty_bps를 1.5배로 올리세요."""

    resp = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.15,
        max_tokens=512,
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)


멀티 모델 합의 — 4개 모델 응답 중 median 채택

def consensus_optimize(metrics: dict) -> dict:
    models = ["claude-sonnet-4.5", "gpt-4.1", "gem

관련 리소스