저자 경험: 저는 수년간加密货币 거래소 API와 다양한 데이터 프로바이더를 활용한 실시간 시장 데이터 파이프라인 구축 프로젝트를 진행해왔습니다. 그 과정에서 Binance, Coinbase, Bybit 등 주요 거래소의 원시 데이터를 직접 수집·분석하면서 겪었던 수많은 벽들을 이번 튜토리얼에서 모두 공유하려 합니다.
이 글은 API 경험이 전혀 없는 완전 초보자를 위해, Python 환경 설정부터 Binance 호가창 데이터를 실시간으로 가져오는 방법까지 단계별로 설명합니다. HolySheep AI의 게이트웨이 서비스를 활용하면 여러 데이터 소스와 AI 모델을 단일 API 키로 통합 관리할 수 있어 개발 생산성이 크게 향상됩니다.
📋 사전 준비물
시작하기 전에 다음 도구들이 설치되어 있어야 합니다:
- Python 3.8 이상 — Python 공식 웹사이트에서 다운로드하거나 터미널에서
python --version명령어로 확인 - pip — Python 패키지 설치 도구 (대부분 Python 설치 시 함께 포함)
- IDE 또는 코드 에디터 — VSCode, PyCharm, 또는 메모장도 가능
1단계: Python 환경 설정
먼저 프로젝트 폴더를 만들고 가상 환경을 구성합니다. 가상 환경을 사용하면 프로젝트마다 다른 패키지 버전을 독립적으로 관리할 수 있어 충돌을 방지할 수 있습니다.
# 프로젝트 폴더 생성 및 이동
mkdir binance-orderbook-tutorial
cd binance-orderbook-tutorial
Python 가상 환경 생성 (Windows의 경우)
python -m venv venv
venv\Scripts\activate
Python 가상 환경 생성 (Mac/Linux의 경우)
python3 -m venv venv
source venv/bin/activate
가상 환경 활성화 확인 (프롬프트 앞에 (venv) 표시됨)
(venv) C:\Users\YourName\binance-orderbook-tutorial>
2단계: 필수 라이브러리 설치
Binance 호가창 데이터를websocket 실시간으로 수신하기 위해 python-binance 라이브러리를 사용합니다. 이 라이브러리는 Binance 공식 SDK로 안정적인 데이터 연결을 보장합니다.
# pip 업그레이드 (선택사항이지만 권장)
pip install --upgrade pip
Binance SDK 및 데이터 처리를 위한 라이브러리 설치
pip install python-binance pandas websockets
설치 확인
pip list | grep -E "python-binance|pandas|websockets"3단계: Binance API 키 발급
Binance에서 실시간 호가창 데이터를 받으려면 API 키가 필요합니다. 테스트넷 키를 사용하면 실제 거래 없이 무료로 연습할 수 있습니다.
테스트넷 API 키 발급 순서:
- Binance 테스트넷 웹사이트(
testnet.binance.vision) 접속 - 우측 상단 "Login" 버튼 클릭
- "Get Some Test Coins" 클릭하여 테스트 코인 자동 충전
- API Keys 메뉴에서 키 생성 (보안 인증 완료)
⚠️ 화면 설명: API 키 발급 페이지에서 "Show Secret Key"를 클릭하면 API Key와 Secret Key 두 개의 문자열이 표시됩니다. 이 값들을 안전한 곳에 메모하세요. Secret Key는 다시 확인할 수 없습니다.
4단계: Binance 호가창 데이터 수신 코드 작성
# binance_orderbook.py
Binance 실시간 호가창(Orderbook) 데이터 수신 예제
import time
from binance.client import Client
from binance.depthcache import DepthCacheManager
========================================
HolySheep AI Gateway 연동 정보
========================================
HolySheep를 사용하면 단일 API 키로
GPT-4.1, Claude, Gemini, DeepSeek 등
모든 주요 AI 모델 통합 가능
https://www.holysheep.ai/register에서 가입
========================================
Binance 테스트넷 API 키 설정
⚠️ 실제 거래소 사용 시 환경 변수로 관리하세요
BINANCE_API_KEY = "your_api_key_here"
BINANCE_SECRET_KEY = "your_secret_key_here"
def on_update(depth_cache):
"""
호가창 데이터 갱신 시 호출되는 콜백 함수
depth_cache: 현재까지 누적된 호가창 데이터
"""
bids = depth_cache.get_bids() # 매수 호가 (Bid)
asks = depth_cache.get_asks() # 매도 호가 (Ask)
print("\n" + "="*60)
print(f"📊 BTC/USDT 실시간 호가창 — {time.strftime('%H:%M:%S')}")
print("="*60)
print(f"{'매수호가 (Bid)':<30} | {'매도호가 (Ask)':<30}")
print("-"*60)
# 상위 5단계 호가만 표시
for i in range(min(5, len(bids), len(asks))):
bid_price = f"{bids[i][0]:,.2f}"
bid_qty = f"{bids[i][1]:.4f}"
ask_price = f"{asks[i][0]:,.2f}"
ask_qty = f"{asks[i][1]:.4f}"
print(f"{bid_price} ({bid_qty}) | {ask_price} ({ask_qty})")
# 스프레드(매수-매도 차이) 계산
if bids and asks:
spread = asks[0][0] - bids[0][0]
spread_pct = (spread / bids[0][0]) * 100
print("-"*60)
print(f"💰 현재 스프레드: {spread:.2f} USDT ({spread_pct:.4f}%)")
def main():
"""메인 함수: Binance WebSocket 연결 및 호가창 수신"""
try:
# 테스트넷 클라이언트 생성
client = Client(BINANCE_API_KEY, BINANCE_SECRET_KEY, testnet=True)
print("🔗 Binance 테스트넷에 연결 중...")
# Depth Cache Manager를 사용하여 실시간 호가창 갱신
dcm = DepthCacheManager(
client=client,
symbol='BTCUSDT', # Bitcoin/USDT 페어
callback=on_update,
interval=100 # 100ms 간격 갱신
)
print("✅ 연결 성공! 실시간 호가창 데이터 수신 중...")
print(" Ctrl+C를 눌러 종료하세요\n")
# 30초 동안 데이터 수신 후 자동 종료
time.sleep(30)
except KeyboardInterrupt:
print("\n\n⛔ 사용자가 종료 요청함")
except Exception as e:
print(f"\n❌ 오류 발생: {type(e).__name__}: {e}")
finally:
print("👋 프로그램을 종료합니다.")
if __name__ == "__main__":
main()5단계: HolySheep AI를 활용한 호가창 데이터 분석
Binance에서 수집한 호가창 데이터를 HolySheep AI의 게이트웨이서비스를 통해 AI 모델로 분석할 수 있습니다. HolySheep는 지금 가입 시 무료 크레딧을 제공하며, 해외 신용카드 없이 로컬 결제가 가능합니다.
# analyze_orderbook.py
수집된 호가창 데이터를 AI로 분석하는 예제
import requests
import json
========================================
HolySheep AI Gateway 설정
========================================
HolySheep AI — 단일 API 키로 모든 주요 모델 통합
https://api.holysheep.ai/v1
========================================
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # HolySheep 대시보드에서 발급
BASE_URL = "https://api.holysheep.ai/v1"
분석할 호가창 데이터 (예시)
orderbook_data = """
현재 BTC/USDT 호가창 상태:
- 최우선 매수호가: 67,450.00 USDT (수량: 2.1543 BTC)
- 최우선 매도호가: 67,455.00 USDT (수량: 1.8921 BTC)
- 2번째 매수호가: 67,448.50 USDT (수량: 3.2100 BTC)
- 2번째 매도호가: 67,457.80 USDT (수량: 2.5600 BTC)
- 스프레드: 5 USDT (0.0074%)
"""
def analyze_market_with_ai(orderbook_data: str) -> str:
"""
HolySheep AI Gateway를 통해 호가창 데이터를 AI 모델로 분석
지원 모델 (2025년 기준):
- GPT-4.1: $8/MTok (컨텍스트 128K)
- Claude Sonnet 4.5: $15/MTok (컨텍스트 200K)
- Gemini 2.5 Flash: $2.50/MTok (가장 비용 효율적)
- DeepSeek V3.2: $0.42/MTok (최저가 옵션)
"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
prompt = f"""다음은 Binance BTC/USDT 거래쌍의 실시간 호가창 데이터입니다.
이 데이터를 기반으로 시장 상황을 분석하고 거래 시그널을 제시해주세요.
{orderbook_data}
분석 항목:
1. 현재 시장 유동성 상태
2. 매수/매도圧力 분석
3. 스프레드 상태 평가
4. 단기 거래 시그널 (매수/매도/관찰)
"""
payload = {
"model": "gpt-4.1", # 다양한 모델中选择 가능
"messages": [
{
"role": "user",
"content": prompt
}
],
"temperature": 0.7,
"max_tokens": 1000
}
try:
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result = response.json()
return result["choices"][0]["message"]["content"]
elif response.status_code == 401:
return "❌ API 키가 유효하지 않습니다. HolySheep 대시보드에서 키를 확인하세요."
elif response.status_code == 429:
return "⏳ 요청 제한 초과. 잠시 후 다시 시도해주세요."
else:
return f"❌ 오류 발생: {response.status_code} — {response.text}"
except requests.exceptions.Timeout:
return "⏰ 요청 시간 초과. 네트워크 연결을 확인해주세요."
except requests.exceptions.ConnectionError:
return "🌐 연결 오류. 인터넷 연결과 API 엔드포인트를 확인해주세요."
def main():
print("="*60)
print("🤖 HolySheep AI 호가창 분석기")
print("="*60)
print("\n분석할 데이터:")
print(orderbook_data)
print("-"*60)
print("\n🔄 AI 분석 중...\n")
analysis = analyze_market_with_ai(orderbook_data)
print("📝 AI 분석 결과:")
print("="*60)
print(analysis)
print("="*60)
if __name__ == "__main__":
main()6단계: 코드 실행 및 결과 확인
설정한 파일들을 실행하여 실제로 호가창 데이터를 수신해보세요.
# 터미널에서 실행
cd binance-orderbook-tutorial
Binance 호가창 데이터 실시간 수신
python binance_orderbook.py
출력 예시:
🔗 Binance 테스트넷에 연결 중...
✅ 연결 성공! 실시간 호가창 데이터 수신 중...
Ctrl+C를 눌러 종료하세요
============================================================
📊 BTC/USDT 실시간 호가창 — 16:32:05
============================================================
매수호가 (Bid) | 매도호가 (Ask)
------------------------------------------------------------
67,450.00 (2.1543) | 67,455.00 (1.8921)
67,448.50 (3.2100) | 67,457.80 (2.5600)
67,445.00 (1.8900) | 67,460.25 (4.1200)
67,442.30 (5.6700) | 67,463.50 (1.3400)
67,440.00 (2.3300) | 67,466.00 (3.7800)
------------------------------------------------------------
💰 현재 스프레드: 5.00 USDT (0.0074%)
성능 최적화:高频 호가창 업데이트 처리
고주파 거래나 알고리즘 트레이딩 시스템에서는 100ms 간격보다 더 빠른 업데이트가 필요할 수 있습니다. 이때는 네이티브 WebSocket 연결을 사용하면 됩니다.
# fast_orderbook.py
네이티브 WebSocket을 사용한 고속 호가창 수신
import asyncio
import json
from websockets.client import connect
BINANCE_WS_URL = "wss://stream.binance.com:9443/ws/btcusdt@depth10@100ms"
async def receive_orderbook():
"""
Binance WebSocket을 통해 100ms 간격으로
상위 10단계 호가창 데이터를 수신합니다.
데이터 구조:
- lastUpdateId: 마지막 업데이트 ID
- bids: 매수 호가 배열 [[가격, 수량], ...]
- asks: 매도 호가 배열 [[가격, 수량], ...]
"""
try:
async with connect(BINANCE_WS_URL) as websocket:
print("🔌 WebSocket 연결 성공 (100ms 간격)")
print(" Ctrl+C를 눌러 종료\n")
message_count = 0
async for message in websocket:
data = json.loads(message)
# 데이터 파싱
bids = data.get('b', [])
asks = data.get('a', [])
update_id = data.get('u', 0)
message_count += 1
# 10번째 메시지마다 표시 (과도한 출력을 방지)
if message_count % 10 == 0:
print(f"\n📊 업데이트 #{message_count} (ID: {update_id})")
print(f" 매수호가 {len(bids)}개, 매도호가 {len(asks)}개")
# 최우선 호가 표시
if bids and asks:
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
spread = best_ask - best_bid
print(f" 최우선 매수: {best_bid:,.2f} | 최우선 매도: {best_ask:,.2f}")
print(f" 스프레드: {spread:.2f} USDT")
# CPU 절약을 위해 짧은 대기
await asyncio.sleep(0.01)
except asyncio.CancelledError:
print("\n⛔ WebSocket 연결 종료됨")
except Exception as e:
print(f"❌ 오류: {e}")
if __name__ == "__main__":
print("="*50)
print("⚡ Binance 고속 호가창 수신 (100ms)")
print("="*50)
asyncio.run(receive_orderbook())실전 활용: 호가창 기반 거래 시그널 생성
수집한 호가창 데이터를 활용하여 실제 거래 시그널을 생성하는 예제입니다. HolySheep AI의 DeepSeek V3.2 모델($0.42/MTok)은 비용 효율적이므로高频 분석에 적합합니다.
# trading_signal.py
호가창 데이터를 기반으로 거래 시그널 생성
import time
import requests
from datetime import datetime
HolySheep AI 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
BASE_URL = "https://api.holysheep.ai/v1"
def calculate_orderbook_metrics(bids: list, asks: list) -> dict:
"""
호가창에서 시장 지표 계산
- 총 매수/매도 liquidity
- 加权平均価格 (VWAP)
- 호가 불균형 (Order Imbalance)
"""
bid_volumes = [float(b[1]) for b in bids]
ask_volumes = [float(a[1]) for a in asks]
bid_prices = [float(b[0]) for b in bids]
ask_prices = [float(a[0]) for a in asks]
total_bid_volume = sum(bid_volumes)
total_ask_volume = sum(ask_volumes)
# VWAP 계산
bid_vwap = sum(p * v for p, v in zip(bid_prices, bid_volumes)) / total_bid_volume if total_bid_volume > 0 else 0
ask_vwap = sum(p * v for p, v in zip(ask_prices, ask_volumes)) / total_ask_volume if total_ask_volume > 0 else 0
# 호가 불균형: (-1 ~ 1) 범위, 음수면 매도 우위, 양수면 매수 우위
total_volume = total_bid_volume + total_ask_volume
imbalance = (total_bid_volume - total_ask_volume) / total_volume if total_volume > 0 else 0
return {
"bid_volume": total_bid_volume,
"ask_volume": total_ask_volume,
"bid_vwap": bid_vwap,
"ask_vwap": ask_vwap,
"imbalance": imbalance,
"mid_price": (float(bids[0][0]) + float(asks[0][0])) / 2 if bids and asks else 0,
"spread": float(asks[0][0]) - float(bids[0][0]) if bids and asks else 0
}
def generate_signal_with_ai(metrics: dict) -> str:
"""
HolySheep AI를 사용하여 호가창 지표 기반 거래 시그널 생성
사용 모델: DeepSeek V3.2 ($0.42/MTok)
- 비용 최적화가 필요한高频 분석에 적합
"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
prompt = f"""호가창 지표 분석 결과:
[mid_price] 현재 중립가격: ${metrics['mid_price']:,.2f}
[bid_volume] 총 매수 유동성: {metrics['bid_volume']:.4f} BTC
[ask_volume] 총 매도 유동성: {metrics['ask_volume']:.4f} BTC
[imbalance] 호가 불균형: {metrics['imbalance']:.4f} (매수 우세: +, 매도 우세: -)
[bid_vwap] 매수 加权平均가격: ${metrics['bid_vwap']:,.2f}
[ask_vwap] 매도 加权平均가격: ${metrics['ask_vwap']:,.2f}
[spread] 스프레드: ${metrics['spread']:.2f}
이 지표를 기반으로:
1. 시장 Pressure 분석 (단기)
2. 거래 시그널 (매수/매도/중립)
3. 리스크 수준 (높음/중간/낮음)
을 3줄 이내로 요약해주세요."""
payload = {
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3, # 일관된 분석을 위해 낮은 temperature
"max_tokens": 200
}
try:
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=15
)
if response.status_code == 200:
return response.json()["choices"][0]["message"]["content"]
else:
return f"분석 실패: {response.status_code}"
except Exception as e:
return f"연결 오류: {str(e)}"
테스트 실행
if __name__ == "__main__":
# 샘플 호가창 데이터
sample_bids = [
["67450.00", "2.1543"],
["67448.50", "3.2100"],
["67445.00", "1.8900"],
["67442.30", "5.6700"],
["67440.00", "2.3300"]
]
sample_asks = [
["67455.00", "1.8921"],
["67457.80", "2.5600"],
["67460.25", "4.1200"],
["67463.50", "1.3400"],
["67466.00", "3.7800"]
]
print("📊 호가창 지표 계산 중...")
metrics = calculate_orderbook_metrics(sample_bids, sample_asks)
print("\n📈 계산된 지표:")
print(f" 중립가격: ${metrics['mid_price']:,.2f}")
print(f" 호가 불균형: {metrics['imbalance']:.4f}")
print(f" 총 유동성: {metrics['bid_volume'] + metrics['ask_volume']:.4f} BTC")
print("\n🤖 AI 시그널 생성 (DeepSeek V3.2 사용)...")
signal = generate_signal_with_ai(metrics)
print(f"\n📝 AI 분석 결과:\n{signal}")📊 HolySheep AI vs 기타 서비스 비교
| 비교 항목 | HolySheep AI | OpenAI 직접 | Anthropic 직접 | Google AI Studio |
|---|---|---|---|---|
| 결제 방식 | 로컬 결제 (신용카드 불필요) | 해외 신용카드 필수 | 해외 신용카드 필수 | 해외 신용카드 필수 |
| GPT-4.1 | $8/MTok | $8/MTok | — | — |
| Claude Sonnet 4.5 | $15/MTok | — | $15/MTok | — |
| Gemini 2.5 Flash | $2.50/MTok | — | — | $2.50/MTok |
| DeepSeek V3.2 | $0.42/MTok ✅ | — | — | — |
| 단일 API 키 | 모든 모델 통합 ✅ | 단일 모델만 | 단일 모델만 | 단일 모델만 |
| 무료 크레딧 | 가입 시 제공 ✅ | $5 initially | 제한적 | 제한적 |
| 한국어 지원 | 완벽 지원 ✅ | 제한적 | 제한적 | 제한적 |
이런 팀에 적합 / 비적용
✅ HolySheep AI가 적합한 팀
- 한국 개발자 팀 — 해외 신용카드 없이 즉시 결제 및 서비스 이용 가능
- 다중 모델 활용 팀 — GPT, Claude, Gemini, DeepSeek 등을 번갈아 사용해야 하는 프로젝트
- 비용 최적화 관심 팀 — DeepSeek V3.2($0.42/MTok)를 활용하여高频 AI 분석 비용 절감
- 빠른 프로토타이핑 — 단일 API 키로 다양한 모델 테스트 가능
- 암호화폐/거래소 개발자 — Binance, Coinbase 등 데이터와 AI 분석을 통합하고 싶은 경우
❌ HolySheep AI가 적합하지 않은 경우
- 기업 자체 모델 사용 — 자체 학습된 프라이빗 모델만 사용하는 경우
- 극단적 커스텀 필요 — 특정 모델의 세밀한 파라미터 조정이 필수적인 경우
- 엄격한 데이터 주권 요구 — 모든 데이터 처리를 자국 내에서만 완료해야 하는 규제 환경
가격과 ROI
HolySheep AI의 가격 경쟁력을 실제 시나리오와 비교해보겠습니다.
| 시나리오 | 월간 토큰량 | HolySheep 비용 | 개별 서비스 비용 | 절감액 |
|---|---|---|---|---|
| 개인 개발자 (Light) | 1M 토큰 | $2.50 (Gemini Flash) | $2.50 | 동일 + 로컬 결제 편의 |
| 스타트업 (Medium) | 50M 토큰 | $21.00 (DeepSeek) | $30.00+ | 30%+ 절감 |
| 중견 기업 (Heavy) | 500M 토큰 | $210.00 | $350.00+ | 40%+ 절감 |
| 거래소 분석 파이프라인 | 1B 토큰 | $420.00 | $800.00+ | 47%+ 절감 |
ROI 분석: 월 $100 예산으로 HolySheep를 사용하면 DeepSeek V3.2 기준 약 238M 토큰을 처리할 수 있습니다. 이는 일평균 8M 토큰으로, Binance 호가창 데이터 분석을 수초 단위로 수천 번 수행할 수 있는 용량입니다.
왜 HolySheep를 선택해야 하나
- 단일 API 키로 모든 모델 통합
프로젝트마다 다른 API 키를 관리하는 번거로움 제거. GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2를 하나의 키로 모두 호출 가능. - 로컬 결제 지원
해외 신용카드 없이银行卡,、国内汇款等多种方式로 결제 가능. 한국 개발자에게 가장 큰 진입장벽 해소. - 비용 최적화
DeepSeek V3.2 $0.42/MTok는 경쟁 서비스 대비大幅 절감.高频 분석 워크로드에서 특히 효과적. - 신속한 통합
기존 OpenAI SDK와 完全 호환되는 API 구조. 코드 변경 최소화. - 한국어 기술 지원
HolySheep 공식 웹사이트에서 한국어로 바로 문의 가능.
자주 발생하는 오류와 해결책
오류 1: Binance API 인증 실패 (Invalid API Key)
# ❌ 오류 메시지
binance.exceptions.BinanceAPIException: APIError(code=-2015):
Invalid API-key, IP, or permissions for action.
✅ 해결 방법 1: 테스트넷 키 확인
Binance 테스트넷(testnet.binance.vision)에서 발급받은 키 사용
본넷 키는 테스트넷에서 작동하지 않음
client = Client(
"테스트넷_API_KEY", # ← 반드시 테스트넷 키 사용
"테스트넷_SECRET_KEY",
testnet=True # ← testnet=True 필수
)
✅ 해결 방법 2: 환경 변수로 안전하게 관리
import os
client = Client(
os.environ.get("BINANCE_TESTNET_API_KEY"),
os.environ.get("BINANCE_TESTNET_SECRET_KEY"),
testnet=True
)
터미널에서 설정:
Windows: set BINANCE_TESTNET_API_KEY=your_key
Mac/Linux: export BINANCE_TESTNET_API_KEY=your_key
오류 2: HolySheep API 키 인증 실패 (401 Unauthorized)
# ❌ 오류 메시지
{"error": {"message": "Invalid API Key", "type": "invalid_request_error"}}
✅ 해결 방법: 올바른 엔드포인트와 키 형식 확인
1. 올바른 base_url 사용 (반드시 https://api.holysheep.ai/v1)
BASE_URL = "https://api.holysheep.ai/v1"
2. API 키 형식 확인 (Bearer 토큰 방식으로 요청)
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}", # ← "Bearer " 필수
"Content-Type": "application/json"
}
3. HolySheep 대시보드에서 키 재발급
https://www.holysheep.ai/register → Dashboard → API Keys → Create New Key
4. 키 값 앞뒤 공백 제거
HOLYSHEEP_API_KEY = HOLYSHEEP_API_KEY.strip()
5. 연결 테스트 코드
import requests
response = requests.get(
"https://api.holysheep.ai/v1/models", # 모델 목록 확인
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"}
)
print(f"상태 코드: {response.status_code}")
print(f"사용 가능한 모델: {response.json()}")오류 3: WebSocket 연결 끊김 (ConnectionClosed)
# ❌ 오류 메시지
websockets.exceptions.ConnectionClosed: WebSocket connection is closed
✅ 해결 방법: 자동 재연결 로직 구현
import asyncio
import websockets
import json
MAX_RECONNECT_ATTEMPTS = 5
RECONNECT_DELAY = 3 # 초
async def websocket_with_reconnect(url: str, callback):
"""
자동 재연결 기능이 있는 WebSocket 클라이언트
"""
for attempt in range(MAX_RECONNECT_ATTEMPTS):
try:
async with websockets.connect(url) as websocket:
print(f"✅ 연결 성공 (시도 {attempt + 1})")
async for message in websocket:
await callback(message)
except websockets.exceptions.ConnectionClosed as e:
print(f"⚠️ 연결 끊김: {e.reason}")
print(f"🔄 {RECONNECT_DELAY}초 후 재연결 시도... ({attempt + 1}/{MAX_RECONNECT_ATTEMPTS})")
await asyncio.sleep(RECONNECT_DELAY)
except Exception as e:
print(f"❌ 예상치 못한 오류: {e}")
break
else:
print("❌ 최대 재연결 횟수 초과. 연결을 확인해주세요.")
사용 예시
async def handle_message(message):
data = json.loads(message)
print(f"수신: {data}")
asyncio.run(websocket_with_reconnect(BINANCE_WS