저는 지난 3개월간 고빈도 매매 봇 개발 과정에서 가장 큰瓶颈이었던 것이 바로 시장 데이터 수집 문제였습니다. 빗썸 USDT-M 선물과 Hyperliquid의 tick 단위 데이터를 어떻게 안정적으로 확보하고 백테스팅 파이프라인에 통합하느냐가 핵심 과제였죠. Tardis는 잘 알려져 있지만, 월 $500 이상의 비용과 함께 데이터 가용성에 제약이 있어 실제로 여러 대안을 검증해 보았습니다. 이 글에서는 제가 직접 검증한 Tardis 대안 5가지와 HolySheep AI를 활용한 지능형 백테스팅 아키텍처를 상세히 다룹니다.
왜 Tick 레벨 데이터인가: 백테스팅 품질의 핵심
고빈도 트레이딩에서 Tick 레벨 데이터는 전략 품질을 결정짓는 가장 중요한 요소입니다. 1분봉이나 5분봉 기반 백테스팅은 실제 주문 执行 시간(Execution Latency)과 슬리피지를 정확히 반영하지 못해, 논문상 수익률이 실제 환경과 30-50% 이상 괴리되는 경우가 흔합니다.
백테스팅 품질 비교
| 데이터 해상도 | 슬리피지 반영 | 호가창 반영 | 백테스팅 정확도 | 적합 전략 유형 |
|---|---|---|---|---|
| 1분봉 | 불가 | 불가 | ~60% | 스윙, 포지션 |
| 초단위(OHLCV) | 부분 반영 | 불가 | ~75% | 스크얼핑, 데이트레이딩 |
| Tick 레벨 | 완전 반영 | 완전 반영 | ~95% | 모든 고빈도 전략 |
Tardis 소개와 한계점
Tardis는 암호화폐 실시간 및 이력 데이터를 제공하는 대표 서비스입니다. Hyperliquid를 포함한 주요 거래소의 Tick 데이터를 websocket과 REST로 제공합니다.
Tardis 주요 사양
- Hyperliquid 지원: 선물 거래 데이터 제공
- 데이터 유형: Trade, Orderbook, Funding Rate, liquidated orders
- 가격: 스타터플랜 $450/월에서 엔터프라이즈 $2,000+/월
- 가용성: 실시간 스트리밍 + 최대 2년 이력 데이터
Tardis의 주요 한계
제가 Tardis를 사용하면서 불편했던 점은 세 가지입니다. 첫째, 월 $450의 스타터플랜은 소규모 개발자나 독립 트레이더에게 부담스러운 비용입니다. 둘째, Hyperliquid의 L1 Orderbook 데이터는 지원하지만, L2 거래 실행 내역의 깊이 있는 분석에는 제약이 있습니다. 셋째, webhook 기반 알림이 지연될 수 있어 초저지연 트레이딩에는 부적합합니다.
주요 경쟁 서비스 비교
| 서비스 | Hyperliquid 지원 | 시작가 | Tick 데이터 | Websocket 지연 | 백테스팅 내장 | AI 분석 연동 |
|---|---|---|---|---|---|---|
| Tardis | 선물のみ | $450/월 | 있음 | ~50ms | 없음 | 제한적 |
| CoinAPI | 선물 + 현물 | $79/월 | 있음 | ~100ms | 없음 | 제한적 |
| Nexus | 선물 | $299/월 | 있음 | ~30ms | 기본 | 제한적 |
| Hyperliquid API | 전체 | 무료 | 있음 | ~10ms | 없음 | 불가 |
| HolySheep AI + 커스텀 파이프라인 | 전체 | $0~ | 있음 | ~10ms | AI 기반 | 완전 지원 |
HolySheep AI 기반 지능형 백테스팅 아키텍처
제가 구축한 백테스팅 파이프라인의 핵심은 HolySheep AI입니다. Hyperliquid에서 직접 Tick 데이터를 수집하고, HolySheep AI의 GPT-4.1과 Claude Sonnet을 활용하여 패턴 인식과 전략 최적화를 수행합니다. 이 조합의 장점은 데이터 수집 비용이 거의 제로라는 점과, AI 기반 전략 분석이 내장되어 있다는 것입니다.
아키텍처 개요
┌─────────────────────────────────────────────────────────────┐
│ HolySheep AI 백테스팅 파이프라인 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Hyperliquid │ │ 데이터 수집 │ │ Redis │ │
│ │ Websocket │───▶│ 서비스 │───▶│ Buffer │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 백테스팅 │◀───│ HolySheep │◀───│ Tick │ │
│ │ 엔진 │ │ AI 분석 │ │ 스토어 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 리포트 │ │ 전략 │ │
│ │ 생성 │ │ 최적화 │ │
│ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1단계: Hyperliquid Tick 데이터 수집
먼저 Hyperliquid의 공짜 API를 활용하여 Tick 레벨 데이터를 수집합니다. HolySheep AI와 결합하면, 수집된 데이터를 실시간으로 AI 분석하여 패턴을 감지할 수 있습니다.
# hyperliquid_tick_collector.py
HolySheep AI를 활용한 Hyperliquid Tick 데이터 수집기
import asyncio
import json
import hmac
import hashlib
import time
import redis
from datetime import datetime
from typing import Dict, List, Optional
import requests
class HyperliquidTickCollector:
"""
Hyperliquid에서 Tick 레벨 데이터를 수집하고
HolySheep AI를 통한 실시간 분석을 수행하는 수집기
"""
def __init__(self, holysheep_api_key: str, redis_client: redis.Redis):
self.base_url = "https://api.hyperliquid.xyz"
self.ws_url = "wss://api.hyperliquid.xyz/ws"
self.holysheep_api_key = holysheep_api_key
self.redis = redis_client
self.buffer_size = 1000 # 배치 처리 단위
async def get_orderbook_snapshot(self, symbol: str) -> Dict:
"""호가창 스냅샷 조회"""
payload = {
"type": "orderbookL2",
"coin": symbol,
"limit": 50
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/info",
json=payload
) as response:
return await response.json()
async def subscribe_trades(self, symbols: List[str]) -> None:
"""Trade 데이터 구독 및 Tick 저장"""
subscribe_msg = {
"method": "subscribe",
"subscription": {
"type": "trades",
"coin": symbols[0] if symbols else "BTC"
}
}
async with aiohttp.ClientSession() as session:
async with session.ws_connect(self.ws_url) as ws:
await ws.send_json(subscribe_msg)
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
await self.process_tick(data)
async def process_tick(self, tick_data: Dict) -> None:
"""Tick 데이터 처리 및 HolySheep AI 분석 트리거"""
timestamp = time.time()
# Redis 버퍼에 Tick 저장
tick_key = f"tick:{tick_data.get('coin', 'UNKNOWN')}:{timestamp}"
self.redis.lpush("tick_buffer", json.dumps({
'data': tick_data,
'timestamp': timestamp,
'collected_at': datetime.now().isoformat()
}))
# 버퍼가 일정 크기 도달 시 HolySheep AI로 분석
if self.redis.llen("tick_buffer") >= self.buffer_size:
await self.analyze_with_holysheep()
async def analyze_with_holysheep(self) -> Optional[Dict]:
"""HolySheep AI를 활용한 Tick 패턴 분석"""
# 버퍼에서 데이터 꺼내기
raw_ticks = []
for _ in range(self.buffer_size):
tick_json = self.redis.rpop("tick_buffer")
if tick_json:
raw_ticks.append(json.loads(tick_json))
if not raw_ticks:
return None
# HolySheep AI API 호출 - GPT-4.1로 패턴 분석
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {self.holysheep_api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{
"role": "system",
"content": """당신은 고빈도 트레이딩 데이터 분석 전문가입니다.
Tick 레벨 거래 데이터를 분석하여:
1. 비정상적인 거래 패턴 감지
2. 유동성 변화 분석
3. 잠재적 arbitrage 기회 식별
JSON 형식으로 결과를 반환하세요."""
},
{
"role": "user",
"content": f"최근 {len(raw_ticks)}개의 Tick 데이터를 분석하세요:\n{json.dumps(raw_ticks[:10], indent=2)}"
}
],
"temperature": 0.3,
"max_tokens": 2000
}
)
if response.status_code == 200:
result = response.json()
analysis = result['choices'][0]['message']['content']
# 분석 결과를 Redis에 저장
self.redis.setex(
f"analysis:{int(time.time())}",
3600, # 1시간 TTL
analysis
)
return json.loads(analysis)
return None
사용 예시
async def main():
import aiohttp
import aioredis
redis_client = await aioredis.create_redis_pool('redis://localhost:6379')
collector = HyperliquidTickCollector(
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY", # HolySheep API 키
redis_client=redis_client
)
# BTC-PERP tick 데이터 수집 시작
await collector.subscribe_trades(["BTC"])
await asyncio.sleep(3600) # 1시간 수집
redis_client.close()
if __name__ == "__main__":
asyncio.run(main())2단계: HolySheep AI 백테스팅 분석기
수집된 Tick 데이터를 기반으로 백테스팅을 수행하고, HolySheep AI의 Claude Sonnet으로 전략 성능을 분석합니다.
# holysheep_backtester.py
HolySheep AI를 활용한 Tick 레벨 백테스팅 및 전략 최적화
import json
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import numpy as np
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
@dataclass
class BacktestConfig:
"""백테스팅 설정"""
initial_capital: float = 10_000 # 초기 자본 $10,000
commission_rate: float = 0.0004 # 수수료 0.04%
slippage_bps: float = 2.0 # 슬리피지 2 basis points
position_size_pct: float = 0.1 # 포지션 크기 10%
max_positions: int = 3 # 최대 동시 포지션 수
@dataclass
class Trade:
timestamp: float
side: OrderSide
price: float
size: float
fee: float
slippage: float
@dataclass
class BacktestResult:
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
max_drawdown: float
sharpe_ratio: float
avg_trade_duration: float
holy_analysis: str
class HolySheepBacktester:
"""
HolySheep AI 기반 Tick 레벨 백테스팅 엔진
- Tardis 대안으로 직접 수집한 데이터 활용
- GPT-4.1로 시그널 생성
- Claude Sonnet으로 전략 분석 및 최적화
"""
def __init__(
self,
holysheep_api_key: str,
config: BacktestConfig = None
):
self.api_key = holysheep_api_key
self.base_url = "https://api.holysheep.ai/v1"
self.config = config or BacktestConfig()
self.trades: List[Trade] = []
self.equity_curve: List[float] = [self.config.initial_capital]
def load_tick_data(self, tick_file: str) -> List[Dict]:
"""파일에서 Tick 데이터 로드"""
with open(tick_file, 'r') as f:
return json.load(f)
async def generate_signals(
self,
recent_ticks: List[Dict],
historical_window: int = 100
) -> List[Dict]:
"""HolySheep GPT-4.1로 거래 시그널 생성"""
# 최근 N개 tick을 분석용으로 구성
analysis_window = recent_ticks[-historical_window:]
prompt = f"""다음은 Hyperliquid BTC-PERP Tick 데이터입니다.
현재 시장 상황을 분석하고 매수/매도/관망 시그널을 생성하세요.
데이터 조건:
- 최근 {len(analysis_window)}개 Tick
- 마지막 가격: {analysis_window[-1].get('price', 'N/A')}
- 볼륨 가중 평균: {sum(t.get('size', 0) for t in analysis_window) / len(analysis_window) if analysis_window else 0}
응답 형식 (JSON):
{{
"signal": "BUY" | "SELL" | "HOLD",
"confidence": 0.0 ~ 1.0,
"reason": "분석 근거",
"suggested_entry": 가격,
"stop_loss": 가격,
"take_profit": 가격
}}"""
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "당신은 전문 트레이딩 봇입니다. 정확한 JSON만 반환하세요."},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 500
}
)
if response.status_code == 200:
content = response.json()['choices'][0]['message']['content']
# JSON 파싱
try:
# 마크다운 코드 블록 제거
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
elif "```" in content:
content = content.split("``")[1].split("``")[0]
return json.loads(content.strip())
except json.JSONDecodeError:
return {"signal": "HOLD", "confidence": 0, "reason": "파싱 실패"}
return {"signal": "HOLD", "confidence": 0, "reason": "API 오류"}
def execute_backtest(
self,
tick_data: List[Dict],
initial_capital: float = None
) -> BacktestResult:
"""백테스팅 실행"""
if initial_capital:
self.config.initial_capital = initial_capital
self.equity_curve = [initial_capital]
capital = self.config.initial_capital
position = None
trades = []
for i, tick in enumerate(tick_data):
price = float(tick.get('price', 0))
size = float(tick.get('size', 0))
timestamp = tick.get('timestamp', 0)
# HolySheep AI 시그널 생성 (N tick마다)
if i % 10 == 0:
signal = asyncio.run(
self.generate_signals(tick_data[:i+1])
)
if signal.get('signal') == 'BUY' and not position and capital > 0:
# 매수 실행
position_size = (capital * self.config.position_size_pct) / price
commission = capital * self.config.position_size_pct * self.config.commission_rate
slippage_cost = price * self.config.slippage_bps / 10000
trade = Trade(
timestamp=timestamp,
side=OrderSide.BUY,
price=price + slippage_cost,
size=position_size,
fee=commission,
slippage=slippage_cost
)
trades.append(trade)
position = {
'entry_price': trade.price,
'size': position_size,
'entry_time': timestamp
}
capital -= (position_size * trade.price + commission)
elif signal.get('signal') == 'SELL' and position:
# 매도 실행
commission = position['size'] * price * self.config.commission_rate
slippage_cost = price * self.config.slippage_bps / 10000
trade = Trade(
timestamp=timestamp,
side=OrderSide.SELL,
price=price - slippage_cost,
size=position['size'],
fee=commission,
slippage=slippage_cost
)
trades.append(trade)
pnl = (trade.price - position['entry_price']) * position['size']
capital += (position['size'] * trade.price - commission)
position = None
# 에쿼티 곡선 업데이트
if position:
unrealized_pnl = (price - position['entry_price']) * position['size']
self.equity_curve.append(capital + unrealized_pnl)
else:
self.equity_curve.append(capital)
# 결과 분석
return self.calculate_metrics(trades)
def calculate_metrics(self, trades: List[Trade]) -> BacktestResult:
"""성과 지표 계산 및 HolySheep AI 분석"""
if not trades:
return BacktestResult(
total_trades=0, winning_trades=0, losing_trades=0,
win_rate=0, total_pnl=0, max_drawdown=0,
sharpe_ratio=0, avg_trade_duration=0,
holy_analysis="거래 없음"
)
# PnL 계산
pnls = []
for i in range(0, len(trades)-1, 2):
if i+1 < len(trades):
buy_trade = trades[i]
sell_trade = trades[i+1]
pnl = (sell_trade.price - buy_trade.price) * sell_trade.size - buy_trade.fee - sell_trade.fee
pnls.append(pnl)
winning_trades = sum(1 for p in pnls if p > 0)
losing_trades = sum(1 for p in pnls if p <= 0)
total_pnl = sum(pnls)
# 최대 낙폭 계산
equity = np.array(self.equity_curve)
running_max = np.maximum.accumulate(equity)
drawdowns = (running_max - equity) / running_max
max_drawdown = np.max(drawdowns) * 100
# 샤프 비율 (간단 버전)
returns = np.diff(equity) / equity[:-1]
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252 * 24) if np.std(returns) > 0 else 0
# HolySheep AI 전략 분석 요청
analysis = self._request_holy_analysis(pnls, trades)
# 평균 거래 시간
durations = []
for i in range(0, len(trades)-1, 2):
if i+1 < len(trades):
durations.append(trades[i+1].timestamp - trades[i].timestamp)
avg_duration = sum(durations) / len(durations) if durations else 0
return BacktestResult(
total_trades=len(pnls),
winning_trades=winning_trades,
losing_trades=losing_trades,
win_rate=winning_trades / len(pnls) * 100 if pnls else 0,
total_pnl=total_pnl,
max_drawdown=max_drawdown,
sharpe_ratio=sharpe,
avg_trade_duration=avg_duration,
holy_analysis=analysis
)
def _request_holy_analysis(
self,
pnls: List[float],
trades: List[Trade]
) -> str:
"""Claude Sonnet으로 전략 분석"""
summary = {
'total_trades': len(pnls),
'win_rate': sum(1 for p in pnls if p > 0) / len(pnls) * 100 if pnls else 0,
'avg_pnl': sum(pnls) / len(pnls) if pnls else 0,
'max_win': max(pnls) if pnls else 0,
'max_loss': min(pnls) if pnls else 0,
'total_pnl': sum(pnls)
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "claude-sonnet-4-5",
"messages": [
{
"role": "system",
"content": "당신은 퀀트 트레이딩 전문가입니다. 백테스팅 결과를 분석하고 개선점을 제시하세요."
},
{
"role": "user",
"content": f"""백테스팅 결과를 분석하고 개선점을 제시하세요:
{json.dumps(summary, indent=2)}
응답 형식:
1. 핵심 인사이트 (3가지)
2. 주요 문제점
3. 최적화 제안
4. 리스크 평가"""
}
],
"temperature": 0.5,
"max_tokens": 1500
}
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
return "분석 요청 실패"
사용 예시
async def run_backtest():
import asyncio
# HolySheep API 키로 초기화
backtester = HolySheepBacktester(
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY",
config=BacktestConfig(
initial_capital=10_000,
commission_rate=0.0004,
slippage_bps=2.0,
position_size_pct=0.1
)
)
# Tick 데이터 로드
tick_data = backtester.load_tick_data('hyperliquid_btcpperp_ticks.json')
print(f"Loaded {len(tick_data)} tick data points")
# 백테스팅 실행
result = backtester.execute_backtest(tick_data)
# 결과 출력
print(f"""
╔══════════════════════════════════════════╗
║ 백테스팅 결과 리포트 ║
╠══════════════════════════════════════════╣
║ 총 거래 횟수: {result.total_trades}회 ║
║ 승률: {result.win_rate:.2f}% ║
║ 총 수익/손실: ${result.total_pnl:,.2f} ║
║ 최대 낙폭: {result.max_drawdown:.2f}% ║
║ 샤프 비율: {result.sharpe_ratio:.3f} ║
║ 평균 거래 시간: {result.avg_trade_duration:.1f}초 ║
╚══════════════════════════════════════════╝
""")
print("HolySheep AI 전략 분석:")
print(result.holy_analysis)
return result
if __name__ == "__main__":
asyncio.run(run_backtest())자주 발생하는 오류와 해결책
오류 1: WebSocket 연결 끊김 및 재연결 실패
# 문제: Hyperliquid WebSocket이 간헐적으로 연결 끊김
해결: 자동 재연결 로직 구현
import asyncio
import aiohttp
import json
from typing import Optional, Callable
import logging
logger = logging.getLogger(__name__)
class WebSocketReconnector:
"""WebSocket 자동 재연결 관리자"""
def __init__(
self,
url: str,
max_retries: int = 10,
base_delay: float = 1.0,
max_delay: float = 60.0
):
self.url = url
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.ws: Optional[aiohttp.ClientWebSocketResponse] = None
self.session: Optional[aiohttp.ClientSession] = None
self.is_running = False
self.retry_count = 0
async def connect(self) -> bool:
"""연결 시도"""
try:
if self.session is None:
self.session = aiohttp.ClientSession()
self.ws = await self.session.ws_connect(
self.url,
timeout=aiohttp.ClientTimeout(total=30),
autoclose=False
)
self.retry_count = 0
logger.info(f"WebSocket 연결 성공: {self.url}")
return True
except aiohttp.ClientError as e:
logger.error(f"WebSocket 연결 실패: {e}")
return False
async def reconnect(self) -> bool:
"""지수 백오프와 함께 재연결"""
delay = min(
self.base_delay * (2 ** self.retry_count),
self.max_delay
)
logger.warning(
f"재연결 시도 {self.retry_count + 1}/{self.max_retries} "
f"{delay:.1f}초 후..."
)
await asyncio.sleep(delay)
# 기존 연결 정리
if self.ws and not self.ws.closed:
await self.ws.close()
self.retry_count += 1
if self.retry_count >= self.max_retries:
logger.error("최대 재연결 횟수 초과")
return False
return await self.connect()
async def listen(self, handler: Callable):
"""메시지 수신 및 처리"""
self.is_running = True
while self.is_running:
if self.ws is None or self.ws.closed:
if not await self.reconnect():
break
try:
msg = await self.ws.receive(timeout=30)
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
await handler(data)
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.error(f"WebSocket 에러: {msg.data}")
await self.reconnect()
elif msg.type == aiohttp.WSMsgType.CLOSE:
logger.warning("서버 의해 연결 종료")
await self.reconnect()
except asyncio.TimeoutError:
# Keep-alive ping送信
if self.ws and not self.ws.closed:
await self.ws.send_str(json.dumps({"type": "ping"}))
except Exception as e:
logger.error(f"예상치 못한 에러: {e}")
await self.reconnect()
async def close(self):
"""연결 종료"""
self.is_running = False
if self.ws and not self.ws.closed:
await self.ws.close()
if self.session:
await self.session.close()
사용
async def main():
reconnector = WebSocketReconnector(
url="wss://api.hyperliquid.xyz/ws",
max_retries=10,
base_delay=1.0
)
async def handle_message(data):
# 메시지 처리 로직
pass
await reconnector.connect()
await reconnector.listen(handle_message)
asyncio.run(main())
오류 2: HolySheep API Rate Limit 초과
# 문제: AI 분석 요청 시 rate limit 초과
해결: 요청 batching 및 지수 백오프 구현
import time
import asyncio
from collections import deque
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import requests
@dataclass
class RateLimitConfig:
requests_per_minute: int = 60
requests_per_second: int = 10
max_batch_size: int = 20
backoff_base: float = 2.0
max_retries: int = 5
class HolySheepRateLimitedClient:
"""Rate limit이 적용된 HolySheep AI 클라이언트"""
def __init__(
self,
api_key: str,
config: RateLimitConfig = None
):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.config = config or RateLimitConfig()
self.request_timestamps = deque(maxlen=self.config.requests_per_minute)
self._lock = asyncio.Lock()
async def _check_rate_limit(self):
"""Rate limit 체크 및 대기"""
async with self._lock:
now = time.time()
# 1분 이내 요청 필터링
while self.request_timestamps and \
self.request_timestamps[0] < now - 60:
self.request_timestamps.popleft()
# Rate limit 체크
if len(self.request_timestamps) >= self.config.requests_per_minute:
wait_time = 60 - (now - self.request_timestamps[0])
if wait_time > 0:
await asyncio.sleep(wait_time)
self.request_timestamps.append(time.time())
async def analyze_batch(
self,
items: List[Dict],
model: str = "gpt-4.1"
) -> List[Optional[Dict]]:
"""배치 단위로 분석 (rate limit 준수)"""
results = []
for i in range(0, len(items), self.config.max_batch_size):
batch = items[i:i + self.config.max_batch_size]
# Rate limit 체크
await self._check_rate_limit()
# 배치 분석 요청
response = await self._send_batch_request(batch, model)
results.extend(response)
# 배치 간 최소 간격
if i + self.config.max_batch_size < len(items):
await asyncio.sleep(0.1) # 100ms 간격
return results
async def _send_batch_request(
self,
batch: List[Dict],
model: str
) -> List[Optional[Dict]]:
"""배치 요청 전송 및 재시도 로직"""
for retry in range(self.config.max_retries):
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"