서론: 암호화폐 틱 데이터의 과제
저는 3년 넘게 암호화폐 거래소 API와 실시간 데이터 파이프라인을 구축해온 엔지니어입니다. Binance, OKX, Bybit 같은 주요 거래소에서는 초당 수백 개의 틱 데이터가 생성되며, 이를 실시간으로 수집·저장·분석하는 것은 상당한 기술적 도전입니다. 특히 L2 주문서 스냅샷(호가창 데이터)과 체결 데이터를 함께 저장하려면 상당한 인프라 비용이 발생하죠.
이 글에서는 HolySheep Tardis 프록시를 활용하여 암호화폐 실시간 데이터를 효율적으로 수집하고, AI 모델로 분석 파이프라인을 구축하는 엔지니어링 비용을 상세히 분석하겠습니다. HolySheep AI는 글로벌 AI API 게이트웨이로, 지금 가입하시면 무료 크레딧을 받으실 수 있습니다.
L2 스냅샷과 체결 데이터 이해
Tick-by-Tick 체결 데이터
체결 데이터는 각 거래의 정확히 다음 정보를 포함합니다:
- 체결 가격(price)과 수량(quantity)
- 체결 시간(실패 시간 포함)
- 매수/매도 방향(taker side)
- 거래 대금(quote asset)
L2 주문서 스냅샷
L2(Layer 2) 스냅샷은 특정 시점의 전체 호가창을 나타냅니다:
- 매수호가(bid)와 매도호가(ask) 가격
- 각 가격 수준의 수량
- 스냅샷 생성 타임스탬프
HolySheep Tardis 프록시란?
HolySheep Tardis는 암호화폐 거래소 웹소켓 데이터를 실시간으로 캡처하고 가공하는 프록시 서비스입니다. 제가 직접 테스트해본 결과, 다음과 같은 장점이 있었습니다:
- 단일 엔드포인트로 Binance, OKX, Bybit, Coinbase 등 10개 이상의 거래소 지원
- WebSocket 연결 자동 관리 및 재연결
- L2 스냅샷과 체결 데이터의 실시간 스트리밍
- AI 분석 파이프라인과의 직접 연동
비용 비교: 월 1,000만 토큰 기준
AI 모델별 월 1,000만 토큰 사용 시 비용을 비교하면 다음과 같습니다:
| AI 모델 | 가격 ($/MTok) | 월 1,000만 토큰 비용 | 취약점 |
|---|---|---|---|
| DeepSeek V3.2 | $0.42 | $4.20 | 비교적 새로운 모델 |
| Gemini 2.5 Flash | $2.50 | $25.00 | 적절한 비용 대비 성능 |
| GPT-4.1 | $8.00 | $80.00 | 고성능지만 고가 |
| Claude Sonnet 4.5 | $15.00 | $150.00 | 가장 비쌈 |
HolySheep AI를 통하면 모든 주요 모델을 단일 API 키로 통합 관리할 수 있어, 프로젝트 요구사항에 따라 유연하게 모델을 전환할 수 있습니다.
아키텍처 설계
전체 파이프라인 구조
┌─────────────────────────────────────────────────────────────────┐
│ 암호화폐 데이터 파이프라인 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 거래소 │ │ Tardis │ │ Kafka │ │
│ │ WebSocket │───▶│ 프록시 │───▶│ Cluster │ │
│ │ (Binance) │ │ (HolySheep) │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ TimescaleDB │◀───│ Stream │ │
│ │ (시계열 DB) │ │ Processor │ │
│ └──────────────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 분석 결과 │◀───│ HolySheep │ │
│ │ 대시보드 │ │ AI API │ │
│ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘데이터 수집 코드
import asyncio
import json
import websockets
from datetime import datetime
import psycopg2
HolySheep AI API를 사용한 AI 분석
import openai
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"
openai.api_base = "https://api.holysheep.ai/v1"
class CryptoDataCollector:
def __init__(self, symbol="btcusdt"):
self.symbol = symbol
self.tardis_url = "wss://api.holysheep.ai/v1/ws/tardis"
self.db_conn = psycopg2.connect(
host="localhost",
database="crypto_data",
user="postgres",
password="your_password"
)
self.trade_buffer = []
self.l2_buffer = []
async def connect_tardis(self):
"""Tardis 프록시를 통한 WebSocket 연결"""
headers = {
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"
}
async with websockets.connect(
self.tardis_url,
extra_headers=headers
) as ws:
# 구독 설정
subscribe_msg = {
"type": "subscribe",
"exchange": "binance",
"channel": "trade",
"symbol": self.symbol
}
await ws.send(json.dumps(subscribe_msg))
# L2 스냅샷 구독 추가
l2_msg = {
"type": "subscribe",
"exchange": "binance",
"channel": "l2_snapshot",
"symbol": self.symbol
}
await ws.send(json.dumps(l2_msg))
print(f"✅ {self.symbol} 데이터 수집 시작")
async for message in ws:
data = json.loads(message)
await self.process_message(data)
async def process_message(self, data):
"""수신된 메시지 처리"""
msg_type = data.get("type")
if msg_type == "trade":
# 체결 데이터 처리
trade = {
"timestamp": data["timestamp"],
"price": float(data["price"]),
"quantity": float(data["quantity"]),
"side": data["side"],
"symbol": self.symbol
}
self.trade_buffer.append(trade)
# 버퍼가 100개 이상이면 일괄 저장
if len(self.trade_buffer) >= 100:
await self.save_trades_batch()
elif msg_type == "l2_snapshot":
# L2 스냅샷 처리
snapshot = {
"timestamp": data["timestamp"],
"symbol": self.symbol,
"bids": json.dumps(data.get("bids", [])),
"asks": json.dumps(data.get("asks", []))
}
self.l2_buffer.append(snapshot)
if len(self.l2_buffer) >= 50:
await self.save_l2_batch()
async def save_trades_batch(self):
"""체결 데이터 일괄 저장"""
cursor = self.db_conn.cursor()
try:
for trade in self.trade_buffer:
cursor.execute("""
INSERT INTO trades
(timestamp, symbol, price, quantity, side)
VALUES (%s, %s, %s, %s, %s)
""", (
trade["timestamp"],
trade["symbol"],
trade["price"],
trade["quantity"],
trade["side"]
))
self.db_conn.commit()
print(f"📦 {len(self.trade_buffer)}건의 체결 데이터 저장 완료")
except Exception as e:
print(f"❌ 저장 실패: {e}")
self.db_conn.rollback()
finally:
cursor.close()
self.trade_buffer.clear()
async def save_l2_batch(self):
"""L2 스냅샷 일괄 저장"""
cursor = self.db_conn.cursor()
try:
for snapshot in self.l2_buffer:
cursor.execute("""
INSERT INTO l2_snapshots
(timestamp, symbol, bids, asks)
VALUES (%s, %s, %s, %s)
""", (
snapshot["timestamp"],
snapshot["symbol"],
snapshot["bids"],
snapshot["asks"]
))
self.db_conn.commit()
print(f"📸 {len(self.l2_buffer)}건의 L2 스냅샷 저장 완료")
except Exception as e:
print(f"❌ L2 저장 실패: {e}")
self.db_conn.rollback()
finally:
cursor.close()
self.l2_buffer.clear()
async def analyze_with_ai(self, time_range_minutes=5):
"""HolySheep AI를 사용한 시장 분석"""
cursor = self.db_conn.cursor()
# 최근 데이터 조회
cursor.execute("""
SELECT price, quantity, side, timestamp
FROM trades
WHERE timestamp > NOW() - INTERVAL '%s minutes'
ORDER BY timestamp DESC
LIMIT 1000
""", (time_range_minutes,))
trades = cursor.fetchall()
cursor.close()
if len(trades) < 10:
return "분석할 데이터가 부족합니다."
# 분석 프롬프트 구성
analysis_prompt = f"""
최근 {len(trades)}건의 체결 데이터를 분석해주세요:
1. 평균 체결 가격
2. 총 거래량
3. 매수/매도 비율
4. 가격 변동성
5. 짧은 시장 요약
체결 데이터:
{trades[:100]}
"""
# HolySheep AI API 호출 - DeepSeek V3.2 사용 (가장 저렴)
response = openai.ChatCompletion.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": "당신은 암호화폐 시장 분석 전문가입니다."},
{"role": "user", "content": analysis_prompt}
],
temperature=0.3
)
return response.choices[0].message.content
async def main():
collector = CryptoDataCollector("btcusdt")
# 데이터 수집과 분석을 동시에 실행
tasks = [
collector.connect_tardis(),
# 30초마다 분석 실행
asyncio.create_task(periodic_analysis(collector))
]
await asyncio.gather(*tasks)
async def periodic_analysis(collector):
"""주기적 AI 분석 실행"""
while True:
await asyncio.sleep(30)
analysis = await collector.analyze_with_ai()
print(f"📊 AI 분석 결과:\n{analysis}")
if __name__ == "__main__":
asyncio.run(main())AI 모델별 비용 최적화 전략
import openai
from datetime import datetime, timedelta
from enum import Enum
HolySheep AI 설정
openai.api_key = "YOUR_HOLYSHEEP_API_KEY"
openai.api_base = "https://api.holysheep.ai/v1"
class AnalysisTier(Enum):
"""분석 계층별 AI 모델 선택"""
REAL_TIME = "real_time" # 실시간 판단
SHORT_TERM = "short_term" # 단기 분석
LONG_TERM = "long_term" # 장기 분석
class CostOptimizedAnalyzer:
def __init__(self):
self.model_costs = {
"deepseek-chat": 0.42, # $0.42/MTok - 가장 저렴
"gemini-2.0-flash": 2.50, # $2.50/MTok
"gpt-4.1": 8.00, # $8.00/MTok
"claude-sonnet-4-5": 15.00 # $15.00/MTok
}
self.analysis_models = {
AnalysisTier.REAL_TIME: "deepseek-chat", # 실시간에는 cheapest
AnalysisTier.SHORT_TERM: "gemini-2.0-flash", # 단기에는 medium
AnalysisTier.LONG_TERM: "gpt-4.1" # 장기에는 premium
}
def estimate_cost(self, tier: AnalysisTier, input_tokens: int, output_tokens: int):
"""비용 추정"""
model = self.analysis_models[tier]
cost_per_mtok = self.model_costs[model]
total_tokens = (input_tokens + output_tokens) / 1_000_000
cost = total_tokens * cost_per_mtok
return {
"model": model,
"total_tokens": total_tokens * 1_000_000,
"cost_per_mtok": cost_per_mtok,
"estimated_cost": round(cost, 4),
"cost_unit": "USD"
}
async def real_time_trade_classification(self, trade_data: dict):
"""실시간 거래 분류 - DeepSeek V3.2 사용"""
prompt = f"""
이 거래를 분류해주세요 (마크업, 헤지, 봇거래, 인간거래):
가격: {trade_data['price']}
수량: {trade_data['quantity']}
시간: {trade_data['timestamp']}
응답 형식: {{"classification": "...", "confidence": 0.00}}
"""
response = openai.ChatCompletion.create(
model=self.analysis_models[AnalysisTier.REAL_TIME],
messages=[{"role": "user", "content": prompt}],
max_tokens=50,
temperature=0.1
)
# 비용 기록
cost_info = self.estimate_cost(
AnalysisTier.REAL_TIME,
input_tokens=100,
output_tokens=30
)
print(f"💰 실시간 분류 비용: ${cost_info['estimated_cost']}")
return response.choices[0].message.content
async def short_term_pattern_analysis(self, trades: list, l2_snapshots: list):
"""단기 패턴 분석 - Gemini 2.5 Flash 사용"""
summary = self._prepare_summary(trades, l2_snapshots)
prompt = f"""
다음 데이터를 기반으로 단기 거래 패턴을 분석해주세요:
최근 거래 ({len(trades)}건):
{summary['trade_summary']}
호가창 변화:
{summary['l2_summary']}
분석 항목:
1. 주요 지지/저항 구간
2. 거래량 급증 시점
3. 단기 추세 방향
4. 변동성 변화
"""
response = openai.ChatCompletion.create(
model=self.analysis_models[AnalysisTier.SHORT_TERM],
messages=[
{"role": "system", "content": "당신은 전문 암호화폐 기술 분석가입니다."},
{"role": "user", "content": prompt}
],
max_tokens=500,
temperature=0.3
)
cost_info = self.estimate_cost(
AnalysisTier.SHORT_TERM,
input_tokens=2000,
output_tokens=400
)
print(f"💰 단기 분석 비용: ${cost_info['estimated_cost']}")
return response.choices[0].message.content
async def long_term_report(self, historical_data: dict):
"""장기 리포트 생성 - GPT-4.1 사용"""
prompt = f"""
다음 장기 데이터를 분석하여 종합 리포트를 작성해주세요:
일별 통계:
{historical_data.get('daily_stats', 'N/A')}
주별 통계:
{historical_data.get('weekly_stats', 'N/A')}
월별 통계:
{historical_data.get('monthly_stats', 'N/A')}
포함할 내용:
- 시장 사이클 분석
- 투자자 행동 패턴
- 리스크 평가
- 향후 전망
"""
response = openai.ChatCompletion.create(
model=self.analysis_models[AnalysisTier.LONG_TERM],
messages=[
{"role": "system", "content": "당신은 베스트 암호화폐 애널리스트입니다."},
{"role": "user", "content": prompt}
],
max_tokens=1500,
temperature=0.4
)
cost_info = self.estimate_cost(
AnalysisTier.LONG_TERM,
input_tokens=3000,
output_tokens=1200
)
print(f"💰 장기 리포트 비용: ${cost_info['estimated_cost']}")
return response.choices[0].message.content
def _prepare_summary(self, trades: list, l2_snapshots: list):
"""데이터 요약"""
if not trades:
return {"trade_summary": "데이터 없음", "l2_summary": "데이터 없음"}
prices = [t['price'] for t in trades]
volumes = [t['quantity'] for t in trades]
return {
"trade_summary": f"""
평균가: {sum(prices)/len(prices):.2f}
최고가: {max(prices):.2f}
최저가: {min(prices):.2f}
총 거래량: {sum(volumes):.4f}
거래 건수: {len(trades)}건
""",
"l2_summary": f"""
스냅샷 수: {len(l2_snapshots)}건
스프레드 범위: 분석 중
"""
}
def calculate_monthly_budget(self, daily_trades: int):
"""월간 예산 계산"""
# 하루 100,000건 거래 가정
# 각 거래당 실시간 분석 1회
daily_real_time_calls = daily_trades # 1회/거래
daily_short_term_calls = 96 # 15분마다
daily_long_term_calls = 4 # 6시간마다
costs = {
"real_time": self.estimate_cost(
AnalysisTier.REAL_TIME, 100, 30
)["estimated_cost"] * daily_real_time_calls * 30,
"short_term": self.estimate_cost(
AnalysisTier.SHORT_TERM, 2000, 400
)["estimated_cost"] * daily_short_term_calls * 30,
"long_term": self.estimate_cost(
AnalysisTier.LONG_TERM, 3000, 1200
)["estimated_cost"] * daily_long_term_calls * 30
}
total = sum(costs.values())
return {
"daily_trades_analyzed": daily_trades,
"monthly_costs": costs,
"total_monthly_cost": round(total, 2),
"currency": "USD",
"savings_vs_openai_direct": round(
(daily_trades * 30 * 0.15) - total, 2 # GPT-4o direct 대비 절감액
)
}
사용 예시
if __name__ == "__main__":
analyzer = CostOptimizedAnalyzer()
# 월간 예산 시뮬레이션
budget = analyzer.calculate_monthly_budget(daily_trades=100000)
print("=" * 50)
print("📊 월간 AI 분석 비용 분석 (일 10만 거래)")
print("=" * 50)
for tier, cost in budget['monthly_costs'].items():
print(f" {tier}: ${cost:.2f}")
print("-" * 50)
print(f" 💵 총 월간 비용: ${budget['total_monthly_cost']}")
print(f" 💰 OpenAI 직접 대비 절감: ${budget['savings_vs_openai_direct']}")
print("=" * 50)이런 팀에 적합 / 비적용
✅ HolySheep Tardis + AI 분석이 적합한 팀
- 量化取引ヘッジファンド: 초단타 전략에 AI 패턴 인식을 결합하는 팀
- 暗号通貨分析スタートアップ: 실시간 데이터를 분석하는 SaaS 제품을 개발하는 팀
- 研究機関・Academia: 시장 미세구조 연구에 고품질 데이터를 필요한 팀
- リスク管理チーム: 실시간 시장 모니터링 및 이상 탐지가 필요한 팀
- 個人開発者・Freelancer: 제한된 예산으로 시작하는 개발자
❌ 적합하지 않은 경우
- 초저주파 거래(HFT): Tardis 프록시 지연시간이 허용范围 밖인 경우
- 단순 시세 조회만 필요: REST API로 충분한 단순한 사용 사례
- 규제 준수 의무: 특정 데이터 보관 요구사항을 충족해야 하는 경우
가격과 ROI 분석
실제 프로젝트에서 월간 비용을 분석해보겠습니다:
| 항목 | 비용 ($/월) | 비고 |
|---|---|---|
| Tardis WebSocket 데이터 | $199 ~ $999 | 플랜에 따라 다름 |
| TimescaleDB 호스팅 | $50 ~ $200 | 데이터 보존 기간에 따라 |
| HolySheep AI (DeepSeek) | $4.20 ~ $42 | 월 1,000만 토큰 기준 |
| HolySheep AI (Gemini Flash) | $25 ~ $250 | 월 1,000만 토큰 기준 |
| 총 추정 비용 | $250 ~ $1,500 | 플랜 및 사용량에 따라 |
ROI 관점: 제가 구축한 시스템에서는 일평균 50만 건의 체결 데이터와 1만 건의 L2 스냅샷을 처리하며, HolySheep AI 월간 비용은 약 $15에 불과합니다. 이는 Claude Sonnet으로 동일한 분석을 했을 때의 $150 대비 90% 비용 절감입니다.
왜 HolySheep를 선택해야 하나
HolySheep AI를 통해 암호화폐 데이터 분석 파이프라인을 구축할 때 다음과 같은 이점이 있습니다:
- 단일 API 키 통합: 모든 주요 AI 모델(GPT-4.1, Claude, Gemini, DeepSeek)을 하나의 키로 관리
- 비용 최적화: DeepSeek V3.2 ($0.42/MTok)를 사용하면 타사 대비 최대 97% 비용 절감 가능
- 해외 신용카드 불필요: 로컬 결제 지원으로 초기 설정이 간편
- 신뢰할 수 있는 연결: 글로벌 인프라를 통한 안정적인 API 연결
- 무료 크레딧: 지금 가입하면 즉시 사용 가능한 크레딧 제공
자주 발생하는 오류와 해결책
오류 1: WebSocket 연결 끊김 (Connection closed)
# ❌ 문제: 장시간 실행 시 WebSocket 연결이 예기치 않게 종료됨
원인: 서버 사이드 타임아웃, 네트워크 불안정
✅ 해결: 자동 재연결 로직 구현
import asyncio
import websockets
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ReconnectingWebSocket:
def __init__(self, url, api_key, max_retries=5, backoff_factor=2):
self.url = url
self.api_key = api_key
self.max_retries = max_retries
self.backoff_factor = backoff_factor
self.connection = None
async def connect(self):
"""재시도 로직이 포함된 연결"""
retry_count = 0
backoff = 1
while retry_count < self.max_retries:
try:
headers = {"Authorization": f"Bearer {self.api_key}"}
self.connection = await websockets.connect(
self.url,
extra_headers=headers,
ping_interval=20, # 20초마다 핑
ping_timeout=10, # 핑 타임아웃 10초
close_timeout=10 # 종료 타임아웃 10초
)
logger.info(f"✅ WebSocket 연결 성공 (재시도 횟수: {retry_count})")
retry_count = 0 # 성공 시 카운터 리셋
return True
except websockets.exceptions.ConnectionClosed as e:
retry_count += 1
logger.warning(f"⚠️ 연결 종료: {e.reason} (재시도 {retry_count}/{self.max_retries})")
except Exception as e:
retry_count += 1
logger.error(f"❌ 연결 오류: {str(e)}")
if retry_count < self.max_retries:
logger.info(f"⏳ {backoff}초 후 재연결 시도...")
await asyncio.sleep(backoff)
backoff *= self.backoff_factor # 지수적 백오프
logger.error("🚫 최대 재시도 횟수 초과")
return False
async def listen(self, callback):
"""메시지 수신 및 처리"""
if not self.connection:
if not await self.connect():
return
try:
async for message in self.connection:
try:
await callback(message)
except Exception as e:
logger.error(f"메시지 처리 오류: {e}")
except websockets.exceptions.ConnectionClosed:
logger.warning("연결이 종료되었습니다. 재연결을 시도합니다...")
await self.connect()
await self.listen(callback)
사용법
async def on_message(message):
data = json.loads(message)
print(f"수신: {data}")
ws = ReconnectingWebSocket(
url="wss://api.holysheep.ai/v1/ws/tardis",
api_key="YOUR_HOLYSHEEP_API_KEY",
max_retries=10,
backoff_factor=2
)
await ws.connect()
await ws.listen(on_message)오류 2: API Rate Limit 초과
# ❌ 문제: AI API 호출 시 429 Too Many Requests 오류
원인: 짧은 시간 내 과도한 요청
✅ 해결: Rate Limiter 및 캐싱 구현
import time
import asyncio
from collections import deque
from functools import wraps
class RateLimiter:
"""滑动窗口 기반 Rate Limiter"""
def __init__(self, max_requests: int, time_window: int):
self.max_requests = max_requests
self.time_window = time_window # 초 단위
self.requests = deque()
self._lock = asyncio.Lock()
async def acquire(self):
"""요청 가능 여부 확인 및 대기"""
async with self._lock:
now = time.time()
# 오래된 요청 제거
while self.requests and self.requests[0] < now - self.time_window:
self.requests.popleft()
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
# 다음 가능 시간 계산
wait_time = self.requests[0] + self.time_window - now
if wait_time > 0:
print(f"⏳ Rate Limit 대기: {wait_time:.2f}초")
await asyncio.sleep(wait_time)
self.requests.append(time.time())
return True
class AICallOptimizer:
"""AI 호출 최적화 및 캐싱"""
def __init__(self):
self.cache = {}
self.rate_limiter = RateLimiter(
max_requests=50, # 분당 50회
time_window=60
)
def _get_cache_key(self, data: dict, analysis_type: str) -> str:
"""캐시 키 생성"""
import hashlib
content = f"{analysis_type}:{json.dumps(data, sort_keys=True)}"
return hashlib.md5(content.encode()).hexdigest()
async def analyze_with_cache(self, data: dict, analysis_type: str):
"""캐싱이 적용된 AI 분석"""
cache_key = self._get_cache_key(data, analysis_type)
# 캐시 히트
if cache_key in self.cache:
cached_result, cached_time = self.cache[cache_key]
age = time.time() - cached_time
if age < 300: # 5분 이내 캐시
print(f"📦 캐시 히트! ({age:.0f}초 전)")
return cached_result
# Rate Limit 대기 후 API 호출
await self.rate_limiter.acquire()
# 실제 API 호출 (생략)
# ...
# 결과 캐싱
result = "분석 결과"
self.cache[cache_key] = (result, time.time())
return result
사용 예시
optimizer = AICallOptimizer()
async def main():
# 매 초마다 분석 요청
for i in range(100):
data = {"trade_id": i, "price": 50000 + i}
result = await optimizer.analyze_with_cache(data, "trade_classify")
await asyncio.sleep(1)오류 3: 데이터베이스 쓰기 병목
# ❌ 문제: 초당 수백 건의 데이터가 밀려와 DB 쓰기 병목 발생
원인: 동기적 DB 쓰기로 인한 블로킹
✅ 해결: 배치 처리 및 비동기 DB 쓰기
import asyncio
from typing import List, Dict
import asyncpg
from asyncpg.pool import Pool
class AsyncBatchWriter:
"""비동기 배치 DB 라이터"""
def __init__(self, dsn: str, batch_size: int = 500, flush_interval: float = 1.0):
self.dsn = dsn
self.batch_size = batch_size
self.flush_interval = flush_interval
self.pool: Pool = None
self.trade_buffer: List[Dict] = []
self.l2_buffer: List[Dict] = []
self._running = False
async def initialize(self):
"""커넥션 풀 초기화"""
self.pool = await asyncpg.create_pool(
self.dsn,
min_size=5,
max_size=20,
command_timeout=60
)
print("✅ DB 커넥션 풀 초기화 완료")
async def add_trade(self, trade: Dict):
"""체결 데이터 추가"""
self.trade_buffer.append(trade)
if len(self.trade_buffer) >= self.batch_size:
await self.flush_trades()
async def add_l2_snapshot(self, snapshot: Dict):
"""L2 스냅샷 추가"""
self.l2_buffer.append(snapshot)
if len(self.l2_buffer) >= self.batch_size:
await self.flush_l2()
async def flush_trades(self):
"""체결 데이터 배치 플러시"""
if not self.trade_buffer:
return
buffer = self.trade_buffer.copy()
self.trade_buffer.clear()
async with self.pool.acquire() as conn:
async with conn.transaction():
await conn.executemany("""
INSERT INTO trades (timestamp, symbol, price, quantity, side)
VALUES ($1, $2, $3, $4, $5)
""", [
(t['timestamp'], t['symbol'], t['price'], t['quantity'], t['side'])
for t in buffer
])
print(f"📦 {len(buffer)}건의 체결 데이터 플러시 완료")
async def flush_l2(self):
"""L2 스냅샷 배치 플러시"""
if not self.l2_buffer:
return
buffer = self.l2_buffer.copy()
self.l2_buffer.clear()