저는 3년 넘게 글로벌 금융 데이터 파이프라인을 설계해 온 시니어 엔지니어입니다. 2022년 11월 FTX 붕괴는 암호화폐 역사상 가장 급격한流动性消失事件이었습니다. 이 튜토리얼에서는 Tardis Machine提供的 高頻度 市场数据를 활용하여 FTX 거래 정지 직전 Order Book의 微細構造를 재구성하고,流动性枯渇가 가속화되는 과정을 분석하는 프로덕션 레벨 파이프라인을 구축합니다.
1. 아키텍처 개요
FTX 붕괴 시점(2022-11-11 02:00 UTC)의 Order Book 재구성을 위해 다음과 같은 아키텍처를 설계했습니다:
- 데이터 소스: Tardis.dev Historical Market Data (WebSocket 캡처)
- 전처리: Rust 기반 고속 파싱 + Apache Arrow 형식 변환
- 저장소: ClickHouse (OLAP 최적화 Order Book 스냅샷)
- 분석 엔진: Python + Polars (메모리 효율적 DataFrame)
- 시각화: D3.js + HolySheep AI Claude 모델 (자동 분석 리포트)
2. 프로젝트 설정
# requirements.txt
tardis-client==1.6.3
clickhouse-driver==0.2.6
polars==0.19.19
pyarrow==14.0.1
websockets==12.0
asyncio-locks==0.2.1
numpy==1.26.2
Rust 종속성 (Cargo.toml)
[dependencies]
tokio = { version = "1.34", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
csv = "1.2"
# docker-compose.yml
version: '3.8'
services:
clickhouse:
image: clickhouse/clickhouse-server:23.11
ports:
- "8123:8123"
- "9000:9000"
environment:
CLICKHOUSE_DB: ftx_orderbook
volumes:
- ./data:/var/lib/clickhouse
tardis-fetcher:
build: ./rust_parser
depends_on:
- clickhouse
environment:
TARDIS_API_KEY: ${TARDIS_API_KEY}
CH_HOST: clickhouse
3. Rust 기반 고속 데이터 파서
Order Book 업데이트는 초당 수천 건 발생하므로, Rust로 작성된 파서를 통해 처리량을 확보합니다:
// src/main.rs
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OrderBookUpdate {
pub exchange: String,
pub symbol: String,
pub timestamp: i64,
pub asks: Vec,
pub bids: Vec,
pub sequence: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PriceLevel {
pub price: f64,
pub size: f64,
}
#[derive(Default)]
pub struct OrderBookState {
asks: HashMap, // price -> size
bids: HashMap,
last_sequence: u64,
}
impl OrderBookState {
pub fn apply_update(&mut self, update: &OrderBookUpdate) -> Result<(), String> {
// 시퀀스 검증으로 데이터 무결성 확인
if update.sequence <= self.last_sequence {
return Err(format!(
"Sequence rollback detected: {} -> {}",
self.last_sequence, update.sequence
));
}
// Ask 업데이트 적용
for level in &update.asks {
if level.size == 0.0 {
self.asks.remove(&level.price);
} else {
self.asks.insert(level.price, level.size);
}
}
// Bid 업데이트 적용
for level in &update.bids {
if level.size == 0.0 {
self.bids.remove(&level.price);
} else {
self.bids.insert(level.price, level.size);
}
}
self.last_sequence = update.sequence;
Ok(())
}
pub fn get_spread(&self) -> Option {
let best_ask = self.asks.keys().cloned().fold(f64::INFINITY, f64::min);
let best_bid = self.bids.keys().cloned().fold(0.0, f64::max);
if best_ask == f64::INFINITY || best_bid == 0.0 {
return None;
}
Some(best_ask - best_bid)
}
pub fn calculate_depth(&self, levels: usize) -> (f64, f64) {
let mut ask_depth = 0.0;
let mut bid_depth = 0.0;
let mut sorted_asks: Vec<_> = self.asks.iter().collect();
let mut sorted_bids: Vec<_> = self.bids.iter().collect();
sorted_asks.sort_by(|a, b| a.0.partial_cmp(b.0).unwrap());
sorted_bids.sort_by(|a, b| b.0.partial_cmp(a.0).unwrap());
for (i, (_, &size)) in sorted_asks.iter().enumerate() {
if i >= levels { break; }
ask_depth += size;
}
for (i, (_, &size)) in sorted_bids.iter().enumerate() {
if i >= levels { break; }
bid_depth += size;
}
(ask_depth, bid_depth)
}
}
// ClickHouse 쓰기용 직렬화
impl OrderBookUpdate {
pub fn to_csv_row(&self, state: &OrderBookState) -> String {
let spread = state.get_spread().unwrap_or(0.0);
let (ask_depth, bid_depth) = state.calculate_depth(10);
format!(
"{},{},{},{},{},{},{}\n",
self.timestamp,
self.symbol,
state.last_sequence,
spread,
ask_depth,
bid_depth,
bid_depth / ask_depth.max(f64::MIN_POSITIVE) // Bid/Ask Ratio
)
}
}
4. Python 분석 파이프라인
# analyze_ftx_collapse.py
import polars as pl
from datetime import datetime, timezone
from clickhouse_driver import Client
import asyncio
from typing import Optional
import httpx
HolySheep AI 설정 - 단일 API 키로 다중 모델 통합
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class FTXOrderBookAnalyzer:
"""FTX 붕괴 시점 Order Book 분석기"""
def __init__(self, ch_host: str = "localhost"):
self.client = Client(host=ch_host, database="ftx_orderbook")
self._init_schema()
def _init_schema(self):
"""ClickHouse 스키마 초기화"""
self.client.execute("""
CREATE TABLE IF NOT EXISTS orderbook_snapshots (
timestamp DateTime64(3),
symbol String,
sequence UInt64,
spread Float64,
ask_depth_10 Float64,
bid_depth_10 Float64,
bid_ask_ratio Float64,
event_type Enum8('normal' = 1, 'spread_widening' = 2, 'liquidity_crisis' = 3)
) ENGINE = MergeTree()
ORDER BY (symbol, timestamp)
""")
def load_critical_period(self, start: datetime, end: datetime) -> pl.DataFrame:
"""FTX 붕괴 직전 6시간 데이터 로드"""
query = """
SELECT
timestamp,
symbol,
spread,
ask_depth_10,
bid