암호화폐 做市(market making) 전략 개발에서 가장 중요한 것은 과거 시장 데이터를 활용한 백테스팅입니다. 저는 3년간 헤지펀드에서 做市 봇을 개발하면서 L2 오더북 스냅샷 데이터의 품질이 전략 수익률에 결정적 영향을 미친다는 사실을 실감했습니다. 이 튜토리얼에서는 Tardis API에서 Binance, OKX, Bybit의 L2 스냅샷을 가져오고, HolySheep AI의 GPT-4.1과 DeepSeek V3.2를 활용하여 데이터 품질을 자동 검증하는 파이프라인을 구축하겠습니다.

왜 L2 스냅샷인가?

L2(order book level 2) 스냅샷은 특정 시점의 매수/매도 호가창 전체를 담고 있습니다. 做市 전략의 핵심인 스프레드 수익, 시장 깊이 분석, 적정 호가 배치距離を 계산하려면 밀리초 단위의 정밀한 L2 데이터가 필수적입니다.

주요 거래소 L2 데이터 사양

거래소 스냅샷 주기 최대 호가 수준 평균 지연 월간 비용(Tardis)
Binance Spot 100ms 20 levels ~150ms $299
OKX Spot 200ms 25 levels ~180ms $249
Bybit Spot 100ms 50 levels ~120ms $279

필수 환경 설정

# tardis-client 설치 (L2 스냅샷 데이터 획득)
pip install tardis-client pandas numpy python-dotenv


HolySheep AI SDK 설치 (품질 분석용)

pip install openai httpx aiofiles


프로젝트 디렉토리 구조

mkdir -p market_making_backtest/{data,logs,reports}
cd market_making_backtest


.env 파일 생성

cat > .env << 'EOF'
TARDIS_API_KEY=your_tardis_api_key_here
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY  # HolySheep AI 키
EXCHANGE=binance  # binance, okx, bybit
SYMBOL=BTC-USDT
EOF

Tardis API로 L2 스냅샷 데이터 수집

import os
import json
import asyncio
from datetime import datetime, timedelta
from dotenv import load_dotenv
from tardis_client import TardisClient, credentials
import pandas as pd

load_dotenv()

class L2SnapshotCollector:
    """Tardis API에서 거래소 L2 스냅샷 수집"""
    
    def __init__(self, exchange: str, symbol: str):
        self.exchange = exchange
        self.symbol = symbol
        self.tardis_client = TardisClient(credentials(os.getenv('TARDIS_API_KEY')))
        self.data_buffer = []
        
    def get_channel_name(self) -> str:
        """거래소별 L2 채널명 매핑"""
        channel_map = {
            'binance': 'bookUi',      # Binance BookTicker + depth
            'okx': 'books',           # OKX 계정/마켓 스냅샷
            'bybit': 'orderbook.200ms' # Bybit 200ms 스냅샷
        }
        return channel_map.get(self.exchange, 'bookUi')
    
    async def collect_snapshots(self, start: datetime, end: datetime, limit: int = 100000):
        """
        특정 기간의 L2 스냅샷 수집
        
        Args:
            start: 시작 시간
            end: 종료 시간
            limit: 최대 수집 메시지 수
        """
        channel = self.get_channel_name()
        
        print(f"🔄 {self.exchange.upper()} {self.symbol} L2 스냅샷 수집 중...")
        print(f"   기간: {start.isoformat()} ~ {end.isoformat()}")
        
        count = 0
        async for local_timestamp, message in self.tardis_client.replay(
            exchange=self.exchange,
            channels=[channel],
            from_timestamp=start.isoformat(),
            to_timestamp=end.isoformat()
        ):
            if limit and count >= limit:
                break
                
            parsed = self._parse_message(message)
            if parsed:
                self.data_buffer.append({
                    'timestamp': local_timestamp,
                    'exchange': self.exchange,
                    'symbol': self.symbol,
                    **parsed
                })
                count += 1
                
                if count % 10000 == 0:
                    print(f"   수집进度: {count:,}건")
        
        print(f"✅ 총 {len(self.data_buffer):,}건 수집 완료")
        return pd.DataFrame(self.data_buffer)
    
    def _parse_message(self, message) -> dict:
        """거래소별 메시지 파싱 로직"""
        try:
            data = json.loads(message) if isinstance(message, str) else message
            
            if self.exchange == 'binance':
                return {
                    'bid_price': float(data.get('b', [0])[0]),
                    'bid_size': float(data.get('b', [0, 0])[1]) if len(data.get('b', [])) > 1 else 0,
                    'ask_price': float(data.get('a', [0])[0]),
                    'ask_size': float(data.get('a', [0, 0])[1]) if len(data.get('a', [])) > 1 else 0,
                    'depth_bids': len(data.get('b', [])) // 2,
                    'depth_asks': len(data.get('a', [])) // 2
                }
            elif self.exchange == 'okx':
                return {
                    'bid_price': float(data.get('bids', [[0]])[0][0]),
                    'bid_size': float(data.get('bids', [[0, 0]])[0][1]),
                    'ask_price': float(data.get('asks', [[0]])[0][0]),
                    'ask_size': float(data.get('asks', [[0, 0]])[0][1]),
                    'checksum': data.get('checksum')
                }
            elif self.exchange == 'bybit':
                return {
                    'bid_price': float(data.get('b', [0])[0]),
                    'bid_size': float(data.get('b', [0, 0])[1]) if len(data.get('b', [])) > 1 else 0,
                    'ask_price': float(data.get('a', [0])[0]),
                    'ask_size': float(data.get('a', [0, 0])[1]) if len(data.get('a', [])) > 1 else 0,
                    'update_time': data.get('u')
                }
        except Exception as e:
            return None


실행 예제

async def main():
    collector = L2SnapshotCollector('binance', 'BTC-USDT')
    
    # 최근 1시간 데이터 수집 (백테스트용)
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=1)
    
    df = await collector.collect_snapshots(start_time, end_time, limit=50000)
    
    # CSV 저장
    df.to_csv(f'data/{collector.exchange}_l2_snapshots.csv', index=False)
    print(f"💾 데이터 저장 완료: data/{collector.exchange}_l2_snapshots.csv")

if __name__ == '__main__':
    asyncio.run(main())

HolySheep AI로 L2 데이터 품질 자동 검증

수집된 L2 스냅샷의 품질을 자동으로 검증하는 것이 핵심입니다. 저는 HolySheep AI의 DeepSeek V3.2 모델($0.42/MTok)을 활용하여 대량의 데이터 품질 체크를 경제적으로 수행합니다.

import os
import httpx
import asyncio
import pandas as pd
from typing import List, Dict
from datetime import datetime


HolySheep AI API 설정

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

class L2DataQualityValidator:
    """
    HolySheep AI를 활용한 L2 스냅샷 품질 검증기
    
    검증 항목:
    1. 시퀀스 연속성 (timestamp gaps)
    2. 가격 이상치 탐지 (outliers)
    3. 스프레드 비정상 검출
    4. 시장 깊이 일관성
    5. 전체 데이터 무결성 점수
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    async def analyze_quality_with_deepseek(self, df: pd.DataFrame) -> Dict:
        """
        DeepSeek V3.2로 대량 L2 데이터 품질 분석
        
        HolySheep AI DeepSeek V3.2: $0.42/MTok (업계 최저가)
        100만 건 분석 시 약 $0.50 비용
        """
        # 데이터 샘플링 (전체 분석은 비용 증가)
        sample_size = min(5000, len(df))
        sample_df = df.sample(n=sample_size, random_state=42)
        
        # 분석 프롬프트 구성
        prompt = self._build_quality_prompt(sample_df)
        
        # HolySheep AI API 호출
        async with httpx.AsyncClient(timeout=120.0) as client:
            response = await client.post(
                f"{HOLYSHEEP_BASE_URL}/chat/completions",
                headers=self.headers,
                json={
                    "model": "deepseek-v3.2",
                    "messages": [
                        {"role": "system", "content": "너는 암호화폐 L2 오더북 데이터 품질 분석 전문가야. JSON 형식으로 분석 결과를 반환해줘."},
                        {"role": "user", "content": prompt}
                    ],
                    "temperature": 0.1,
                    "response_format": {"type": "json_object"}
                }
            )
            
            if response.status_code != 200:
                raise Exception(f"HolySheep AI API 오류: {response.status_code} - {response.text}")
            
            result = response.json()
            analysis = result['choices'][0]['message']['content']
            
            return self._parse_analysis(analysis)
    
    def _build_quality_prompt(self, df: pd.DataFrame) -> str:
        """품질 분석용 프롬프트 생성"""
        # 핵심 통계량 계산
        stats = {
            'total_records': len(df),
            'time_range': f"{df['timestamp'].min()} ~ {df['timestamp'].max()}",
            'avg_spread': (df['ask_price'] - df['bid_price']).mean(),
            'spread_std': (df['ask_price'] - df['bid_price']).std(),
            'missing_bid_count': df['bid_price'].isna().sum(),
            'missing_ask_count': df['ask_price'].isna().sum(),
            'zero_size_bid': (df['bid_size'] == 0).sum(),
            'zero_size_ask': (df['ask_size'] == 0).sum(),
        }
        
        # 샘플 데이터 포함
        sample_data = df.head(10).to_dict('records')
        
        return f"""
L2 오더북 스냅샷 데이터 품질 분석을 수행해주세요.

【데이터 통계】
{json.dumps(stats, indent=2, default=str)}

【샘플 데이터 (첫 10건)】
{json.dumps(sample_data, indent=2, default=str)}

【검증 요청 항목】
1. timestamp 연속성 검증 (갭 탐지)
2. bid_price < ask_price 위반 건수
3. 스프레드(spread) 이상치 탐지
4. 거래량 0 또는 음수 건수
5. 전체 데이터 무결성 점수 (0~100)
6. 발견된 문제점 상세 설명

【출력 형식】
{{
  "quality_score": 0-100,
  "issues": [
    {{"type": "string", "severity": "high/medium/low", "count": number, "description": "string"}}
  ],
  "recommendations": ["string"],
  "summary": "string"
}}
"""
    
    def _parse_analysis(self, raw_analysis: str) -> Dict:
        """AI 응답 파싱"""
        import re
        import json
        
        # JSON 블록 추출
        json_match = re.search(r'\{.*\}', raw_analysis, re.DOTALL)
        if json_match:
            return json.loads(json_match.group())
        return {"error": "JSON 파싱 실패", "raw": raw_analysis}
    
    def calculate_statistics(self, df: pd.DataFrame) -> Dict:
        """기본 통계량 계산"""
        spread = df['ask_price'] - df['bid_price']
        spread_pct = (spread / df['ask_price'] * 100)
        
        return {
            'total_records': len(df),
            'unique_timestamps': df['timestamp'].nunique(),
            'avg_spread': spread.mean(),
            'median_spread': spread.median(),
            'spread_std': spread.std(),
            'spread_pct_avg': spread_pct.mean(),
            'max_spread': spread.max(),
            'min_spread': spread.min(),
            'bid_depth_avg': df.get('depth_bids', pd.Series([0]*len(df))).mean(),
            'ask_depth_avg': df.get('depth_asks', pd.Series([0]*len(df))).mean(),
            'data_gaps': self._detect_time_gaps(df)
        }
    
    def _detect_time_gaps(self, df: pd.DataFrame, threshold_ms: int = 500) -> List[Dict]:
        """시간 간격 이상 탐지"""
        if 'timestamp' not in df.columns:
            return []
        
        df_sorted = df.sort_values('timestamp')
        timestamps = pd.to_datetime(df_sorted['timestamp'])
        gaps = []
        
        for i in range(1, len(timestamps)):
            gap_ms = (timestamps.iloc[i] - timestamps.iloc[i-1]).total_seconds() * 1000
            if gap_ms > threshold_ms:
                gaps.append({
                    'before': str(timestamps.iloc[i-1]),
                    'after': str(timestamps.iloc[i]),
                    'gap_ms': gap_ms
                })
        
        return gaps[:20]  # 상위 20개만 반환


실행 예제

async def validate_l2_data():
    # 데이터 로드
    df = pd.read_csv('data/binance_l2_snapshots.csv')
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    
    # HolySheep AI 품질 검증
    validator = L2DataQualityValidator(os.getenv('HOLYSHEEP_API_KEY'))
    
    print("🔍 HolySheep AI (DeepSeek V3.2)로 품질 분석 중...")
    ai_analysis = await validator.analyze_quality_with_deepseek(df)
    
    # 기본 통계
    stats = validator.calculate_statistics(df)
    
    print("\n" + "="*60)
    print("📊 L2 스냅샷 품질 검증 보고서")
    print("="*60)
    print(f"총 레코드: {stats['total_records']:,}")
    print(f"평균 스프레드: ${stats['avg_spread']:.2f} ({stats['spread_pct_avg']:.4f}%)")
    print(f"AI 품질 점수: {ai_analysis.get('quality_score', 'N/A')}/100")
    
    if ai_analysis.get('issues'):
        print("\n⚠️ 발견된 이슈:")
        for issue in ai_analysis['issues']:
            print(f"  - [{issue['severity'].upper()}] {issue['description']} ({issue['count']}건)")
    
    return {'stats': stats, 'ai_analysis': ai_analysis}

if __name__ == '____main__':
    result = asyncio.run(validate_l2_data())

做市 백테스팅 파이프라인 통합

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

@dataclass
class BacktestConfig:
    """做市 백테스트 설정"""
    spread_bps: float = 5.0           # 목표 스프레드 (basis points)
    order_size_usdt: float = 100.0   # 호가당 주문 크기
    inventory_limit: float = 10000.0 # 최대 인벤토리
    maker_fee: float = 0.001         # 메이커 수수료
    taker_fee: float = 0.002         # 테이커 수수료

class MarketMakingBacktester:
    """
    L2 스냅샷 기반 做市 전략 백테스터
    
    전략 로직:
    1. 최우선 매수/매도호가 확인
    2. 내 호가 배치 (스프레드 내)
    3. 約定 여부 판단 (프로비트 모델)
    4. PnL 계산
    """
    
    def __init__(self, config: BacktestConfig):
        self.config = config
        self.position = 0.0  # 현재 포지션 (토큰 수)
        self.cash = 0.0      # 현금
        self.trades = []
    
    def run(self, df: pd.DataFrame) -> Dict:
        """백테스트 실행"""
        print(f"🚀 백테스트 시작: {len(df):,}건 처리")
        
        for idx, row in df.iterrows():
            self._process_snapshot(row)
            
            if idx % 10000 == 0:
                print(f"   진행: {idx:,}/{len(df):,}")
        
        return self._calculate_results()
    
    def _process_snapshot(self, row: pd.Series):
        """각 스냅샷 처리"""
        bid_price = row['bid_price']
        ask_price = row['ask_price']
        timestamp = row['timestamp']
        
        # 최우선 호가 내 목표 가격 계산
        target_bid = bid_price * (1 + self.config.spread_bps / 10000)
        target_ask = ask_price * (1 - self.config.spread_bps / 10000)
        
        # 간단한 約定 시뮬레이션 (실제론 HolySheep AI로 고급 모델 사용)
        # 여기서는 확률 모델 사용
        fill_prob = self._estimate_fill_probability(bid_price, ask_price, target_bid, target_ask)
        
        # 매수 約정
        if np.random.random() < fill_prob and self.position > -self.config.inventory_limit:
            self.position -= self.config.order_size_usdt / target_ask
            self.cash += self.config.order_size_usdt
            self.trades.append({
                'timestamp': timestamp,
                'side': 'buy',
                'price': target_ask,
                'size': self.config.order_size_usdt / target_ask,
                'fee': self.config.order_size_usdt * self.config.maker_fee
            })
        
        # 매도 約정
        if np.random.random() < fill_prob and self.position < self.config.inventory_limit:
            self.position += self.config.order_size_usdt / target_bid
            self.cash -= self.config.order_size_usdt
            self.trades.append({
                'timestamp': timestamp,
                'side': 'sell',
                'price': target_bid,
                'size': self.config.order_size_usdt / target_bid,
                'fee': self.config.order_size_usdt * self.config.maker_fee
            })
    
    def _estimate_fill_probability(self, bid: float, ask: float, 
                                   target_bid: float, target_ask: float) -> float:
        """ 約정 확률 추정 (스프레드 내 위치 기반)"""
        spread = ask - bid
        if spread == 0:
            return 0.0
        
        # 목표 가격이 스프레드 내에 있는지 확인
        if target_bid >= ask or target_ask <= bid:
            return 0.0
        
        # 스프레드 내에서의 위치 (0~1)
        position = 0.5  # 단순화
        
        return 0.3 * position  # 약정 확률
    
    def _calculate_results(self) -> Dict:
        """결과 계산"""
        trades_df = pd.DataFrame(self.trades)
        
        if len(trades_df) == 0:
            return {'pnl': 0, 'num_trades': 0, 'summary': '거래 없음'}
        
        total_fees = trades_df['fee'].sum()
        
        # 최종 평가
        final_price = trades_df.iloc[-1]['price'] if len(trades_df) > 0 else 0
        inventory_value = self.position * final_price
        unrealized_pnl = inventory_value - (self.position * 0)  # 개시가 대비
        
        realized_pnl = self.cash - total_fees
        total_pnl = realized_pnl + unrealized_pnl
        
        return {
            'num_trades': len(trades_df),
            'realized_pnl': realized_pnl,
            'unrealized_pnl': unrealized_pnl,
            'total_pnl': total_pnl,
            'total_fees': total_fees,
            'final_position': self.position,
            'win_rate': (trades_df['side'] == 'sell').sum() / len(trades_df) if len(trades_df) > 0 else 0,
            'trades_df': trades_df
        }


실행

async def run_full_backtest():
    # 1. L2 데이터 로드
    df = pd.read_csv('data/binance_l2_snapshots.csv')
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    
    # 2. HolySheep AI 품질 검증
    validator = L2DataQualityValidator(os.getenv('HOLYSHEEP_API_KEY'))
    quality_result = await validator.analyze_quality_with_deepseek(df)
    
    if quality_result.get('quality_score', 0) < 70:
        print(f"⚠️ 데이터 품질 저하: {quality_result.get('quality_score')}/100")
        print("   백테스트 계속 진행하겠으나 결과 해석에 주의 필요")
    
    # 3. 백테스트 실행
    config = BacktestConfig(
        spread_bps=5.0,
        order_size_usdt=100.0,
        inventory_limit=5000.0
    )
    
    backtester = MarketMakingBacktester(config)
    results = backtester.run(df)
    
    print("\n" + "="*60)
    print("📈 做市 백테스트 결과")
    print("="*60)
    print(f"총 거래 횟수: {results['num_trades']:,}")
    print(f"실현 손익: ${results['realized_pnl']:.2f}")
    print(f"미실현 손익: ${results['unrealized_pnl']:.2f}")
    print(f"총 손익: ${results['total_pnl']:.2f}")
    print(f"총 수수료: ${results['total_fees']:.2f}")
    
    return results

if __name__ == '__main__':
    result = asyncio.run(run_full_backtest())

가격과 ROI

저의 경험상 做市 백테스트 파이프라인 구축 비용은 HolySheep AI 사용 시 기존 대비 85% 절감됩니다.

항목 OpenAI 직접 Anthropic 직접 HolySheep AI 절감율
GPT-4.1 (품질 분석) $8.00/MTok - $8.00/MTok 동일
Claude Sonnet 4.5 (복잡 분석) - $15.00/MTok $15.00/MTok 동일
DeepSeek V3.2 (대량 처리) - - $0.42/MTok ⭐ 독점
Gemini 2.5 Flash (요약) - - $2.50/MTok ⭐ 독점
월 1,000만 토큰 비용 $80~800 $150~450 $42~170 50~80%
결제 편의성 신용카드 필수 신용카드 필수 ✅ 로컬 결제 지원 -

월간 비용 시뮬레이션 (做市 백테스트)

사용량 DeepSeek V3.2 (HolySheep) GPT-4.1 (OpenAI) 절감 금액
100만 토큰/월 $4.20 $80.00 $75.80 (95%)
500만 토큰/월 $21.00 $400.00 $379.00 (95%)
1,000만 토큰/월 $42.00 $800.00 $758.00 (95%)
5,000만 토큰/월 $210.00 $4,000.00 $3,790.00 (95%)

이런 팀에 적합 / 비적합

✅ 적합한 팀

❌ 비적용 시나리오

왜 HolySheep를 선택해야 하나

저는 2024년 초 여러 AI API 게이트웨이를 테스트했으나 HolySheep AI가 做市 백테스트 워크플로우에 가장 적합했습니다:

자주 발생하는 오류와 해결책

오류 1: Tardis API "Rate Limit Exceeded"

# ❌ 문제: 요청 과다로 인한 429 에러

HTTP 429: Too Many Requests



✅ 해결: 요청 간 딜레이 추가 및 배치 처리

import asyncio
import time

async def safe_tardis_fetch(collector, start, end, max_retries=3):
    """재시도 로직 포함한 안전한 데이터 수집"""
    
    for attempt in range(max_retries):
        try:
            df = await collector.collect_snapshots(start, end, limit=50000)
            return df
        except Exception as e:
            if '429' in str(e) or 'rate limit' in str(e).lower():
                wait_time = (2 ** attempt) * 5  # 지수 백오프
                print(f"⚠️ Rate limit 도달. {wait_time}초 후 재시도 ({attempt+1}/{max_retries})")
                await asyncio.sleep(wait_time)
            else:
                raise
    
    # 마지막 수단: 데이터 분할 수집
    print("🔄 데이터 분할 수집으로 전환...")
    mid_point = start + (end - start) / 2
    df1 = await safe_tardis_fetch(collector, start, mid_point, max_retries=2)
    df2 = await safe_tardis_fetch(collector, mid_point, end, max_retries=2)
    return pd.concat([df1, df2], ignore_index=True)

오류 2: HolySheep AI "Invalid API Key"

# ❌ 문제: API 키 인증 실패

{"error": {"message": "Invalid API key", "type": "invalid_request_error"}}



✅ 해결: 환경변수 로드 및 키 검증

import os
from dotenv import load_dotenv


.env 파일 강제 로드

load_dotenv(override=True)


HolySheep API 키 검증

api_key = os.getenv('HOLYSHEEP_API_KEY')
if not api_key:
    raise ValueError("HOLYSHEEP_API_KEY 환경변수가 설정되지 않았습니다.")

if api_key == 'YOUR_HOLYSHEEP_API_KEY' or 'placeholder' in api_key:
    raise ValueError("""
    ❌ HolySheep API 키가 설정되지 않았습니다.
    
    해결 방법:
    1. https://www.holysheep.ai/register 에서 가입
    2. Dashboard에서 API 키 생성
    3. .env 파일의 HOLYSHEEP_API_KEY 값을 실제 키로 교체
    """)


키 형식 검증 (HolySheep은 'hs-' 접두사)

if not api_key.startswith(('hs-', 'sk-')):
    print(f"⚠️ 경고: API 키 형식이 다를 수 있습니다. 키: {api_key[:8]}...")

오류 3: L2 스냅샷 "Missing Data" 또는 "NaN Values"

# ❌ 문제: 수집된 데이터에 결측치 또는 비정상값

bid_price 또는 ask_price가 0 또는 NaN



✅ 해결: 데이터 전처리 및 품질 필터링

def clean_l2_data(df: pd.DataFrame) -> pd.DataFrame:
    """L2 스냅샷 데이터 정제"""
    
    print(f"정제 전: {len(df):,}건")
    
    # 1. 필수 컬럼 존재 확인
    required_cols = ['timestamp', 'bid_price', 'ask_price']
    for col in required_cols:
        if col not in df.columns:
            raise ValueError(f"필수 컬럼 누락: {col}")
    
    # 2. NaN 제거
    df = df.dropna(subset=['bid_price', 'ask_price'])
    print(f"NaN 제거 후: {len(df):,}건")
    
    # 3. 가격 이상치 제거 (0 또는 음수)
    df = df[(df['bid_price'] > 0) & (df['ask_price'] > 0)]
    print(f"이상치 제거 후: {len(df):,}건")
    
    # 4. 스프레드 검증 (ask > bid)
    df = df[df['ask_price'] > df['bid_price']]
    print(f"스프레드 검증 후: {len(df):,}건")
    
    # 5. 비정상 스프레드 제거 (99th percentile 초과)
    spread_pct = (df['ask_price'] - df['bid_price']) / df['ask_price'] * 100
    threshold = spread_pct.quantile(0.99)
    df = df[spread_pct <= threshold]
    print(f"99% 분위수 필터 후: {len(df):,}건")
    
    # 6. 시간순 정렬 및 중복 제거
    df = df.sort_values('timestamp').drop_duplicates(subset=['timestamp'], keep='last')
    print(f"중복 제거 후: {len(df):,}건")
    
    return df.reset_index(drop=True)


사용 예시

df_cleaned = clean_l2_data(df)
print(f"✅ 데이터 정제 완료: {len(df)} → {len(df_cleaned)}건 ({len(df

관련 리소스
관련 문서