실제 거래소 API를 활용한 부하 테스트 경험과 HolySheep AI의 인프라 최적화 노하우를 결합한 실전 튜토리얼입니다. 10,000+RPS 환경에서도 안정적인 API 연결을 구축하는 방법을 다룹니다.
HolySheep AI vs 공식 API Gateway vs 기타 서비스 비교
| 비교 항목 | HolySheep AI | 공식 거래소 API Gateway | 일반 로드 밸런서 | 자가 구축 Redis Queue |
|---|---|---|---|---|
| 동시 연결 수 | 50,000+ 동시 연결 | 1,000-5,000 동시 연결 | 10,000-30,000 동시 연결 | 제한 없음 (서버 사양에 따름) |
| RPS 처리량 | 고정 요청 최적화 | rate limit 적용 ( exchanges별 상이) | 프록시 캐싱 가능 | 완전 제어권 |
| 설정 난이도 | ⭐ (API 키만 있으면 즉시) | ⭐⭐⭐ (브릿지 서버 필요) | ⭐⭐⭐⭐ (인프라 구축) | ⭐⭐⭐⭐⭐ (전문가 필요) |
| 비용 | 구독제 ($29/월~) | 무료 (기본) | $100-500/월 (인스턴스) | 서버 비용 +运维人力 |
| WebSocket 지원 | ✅ 실시간 스트리밍 | ✅ native 제공 | ❌ 미지원 | ✅ 구현 가능 |
| 자동 재연결 | ✅ 내장 | ✅ 제공 | ❌ 별도 구현 | ✅ 직접 구현 |
| 적합한 용도 | AI API 통합, 비용 최적화 | 거래소原生 연동 | 커스텀 라우팅 | 고급 트레이딩 시스템 |
이런 팀에 적합 / 비적합
✅ HolySheep AI가 적합한 팀
- AI 모델 통합을 빠른 시간 내에 구축해야 하는 팀 — 단일 API 키로 GPT-4.1, Claude, Gemini, DeepSeek 등 다중 모델을 즉시 연동
- 비용 최적화가 중요한 스타트업 — DeepSeek V3.2는 $0.42/MTok으로 타 서비스 대비 90%+ 절감 가능
- 해외 신용카드 없이 결제해야 하는 개발자 — 로컬 결제 지원으로 즉시 시작 가능
- 다중 모델 A/B 테스트가 필요한 ML 팀 — 단일 엔드포인트에서 모델 전환 테스트 가능
❌ HolySheep AI가 비적합한 팀
- 거래소 WebSocket 실시간 데이터가 핵심인 팀 — HolySheep는 AI API 전용으로 거래소 데이터 스트리밍 미지원
- 완전한 커스텀 라우팅이 필요한 고-frequency 트레이딩 — 로드 밸런서 직접 구축 필요
- 특정 거래소 전용 API 기능이 필요한 경우 — 해당 거래소 공식 SDK 권장
- 온프레미스部署가 필수적인 금융 기관 — 보안 정책상 직접 구축 필요
암호화폐 거래소 API 압력 테스트: 실전 가이드
저는 3년간 다양한 거래소(Binance, Bybit, OKX, Gate.io)의 API를 활용한 고빈도 거래 시스템을 개발해왔습니다. 그 과정에서 동시 연결 관리, rate limit 우회, 그리고 장애 대응에 대한 많은 시행착오를 경험했습니다. 이 가이드는 제 실제踩坑 경험에서 탄생한 것입니다.
1. 테스트 환경 구축
# Python 3.10+ 필수
필요한 패키지 설치
pip install aiohttp asyncio websockets locust pytest pytest-asyncio
프로젝트 구조
crypto-stress-test/
├── src/
│ ├── __init__.py
│ ├── client.py # 비동기 API 클라이언트
│ ├── rate_limiter.py # Rate Limiter 구현
│ ├── websocket_manager.py # WebSocket 연결 관리
│ └── metrics.py # 성능 지표 수집
├── tests/
│ └── test_concurrency.py
├── locustfile.py # Locust 부하 테스트
└── config.py # 설정 파일
2. 동시 연결 테스트 핵심 구현
# src/client.py
import asyncio
import aiohttp
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from collections import defaultdict
import random
@dataclass
class ConnectionMetrics:
"""연결 메트릭 수집"""
connection_id: int
latency_ms: float
status_code: int
error: Optional[str] = None
class ExchangeAPIClient:
"""
암호화폐 거래소 API 동시 연결 테스트 클라이언트
HolySheep AI의 다중 모델 연결 관리 철학을 적용
"""
def __init__(
self,
base_url: str,
api_key: str,
api_secret: str,
max_concurrent: int = 1000,
rate_limit_rps: int = 1200
):
self.base_url = base_url
self.api_key = api_key
self.api_secret = api_secret
self.max_concurrent = max_concurrent
self.rate_limit_rps = rate_limit_rps
# 연결 풀 설정 (HolySheep AI 스타일)
self._connector = None
self._session = None
# 메트릭 수집
self.metrics: List[ConnectionMetrics] = []
self.success_count = 0
self.failure_count = 0
self.total_latency = 0.0
# Rate limiting bucket
self._bucket = asyncio.Semaphore(rate_limit_rps)
async def __aenter__(self):
"""비동기 컨텍스트 매니저"""
# 연결 풀 크기 = 동시 연결 수의 2배 (여유분)
self._connector = aiohttp.TCPConnector(
limit=self.max_concurrent * 2,
limit_per_host=self.max_concurrent,
ttl_dns_cache=300,
enable_cleanup_closed=True,
force_close=False # Keep-alive 활성화
)
self._session = aiohttp.ClientSession(
connector=self._connector,
timeout=aiohttp.ClientTimeout(total=30, connect=10),
headers={
"X-MBX-APIKEY": self.api_key,
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""리소스 정리"""
if self._session:
await self._session.close()
if self._connector:
await self._connector.close()
await asyncio.sleep(0.25) # 정리 대기
async def _throttled_request(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
connection_id: int = 0
) -> ConnectionMetrics:
"""Rate limit 적용된 요청 실행"""
start_time = time.perf_counter()
async with self._bucket: # Rate limit 적용
try:
url = f"{self.base_url}{endpoint}"
async with self._session.request(
method,
url,
params=params,
skip_auto_headers=['User-Agent']
) as response:
latency_ms = (time.perf_counter() - start_time) * 1000
# Rate limit 헤더 체크
remaining = response.headers.get('X-MBX-USED-WEIGHT-1M', '0')
return ConnectionMetrics(
connection_id=connection_id,
latency_ms=latency_ms,
status_code=response.status,
error=None
)
except aiohttp.ClientError as e:
latency_ms = (time.perf_counter() - start_time) * 1000
return ConnectionMetrics(
connection_id=connection_id,
latency_ms=latency_ms,
status_code=0,
error=str(e)
)
async def test_concurrent_connections(
self,
num_connections: int,
duration_seconds: int = 60
) -> Dict:
"""
동시 연결 수 테스트 실행
Args:
num_connections: 동시 연결 수
duration_seconds: 테스트 지속 시간
Returns:
테스트 결과 요약
"""
print(f"🚀 동시 연결 {num_connections}개로 {duration_seconds}초 테스트 시작")
self.metrics.clear()
self.success_count = 0
self.failure_count = 0
self.total_latency = 0.0
# 동시 요청 생성
start_time = time.time()
tasks = []
for i in range(num_connections):
# 각 연결에 랜덤 엔드포인트 분배
endpoints = [
'/api/v3/account',
'/api/v3/order',
'/api/v3/exchangeInfo'
]
endpoint = random.choice(endpoints)
task = self._throttled_request(
'GET',
endpoint,
params={'timestamp': int(time.time() * 1000)},
connection_id=i
)
tasks.append(task)
# 동시 실행
results = await asyncio.gather(*tasks, return_exceptions=True)
# 결과 수집
for result in results:
if isinstance(result, ConnectionMetrics):
self.metrics.append(result)
self.total_latency += result.latency_ms
if result.status_code == 200:
self.success_count += 1
else:
self.failure_count += 1
else:
self.failure_count += 1
elapsed = time.time() - start_time
# 결과 분석
return self._analyze_results(num_connections, elapsed)
def _analyze_results(self, num_connections: int, elapsed: float) -> Dict:
"""테스트 결과 분석"""
latencies = [m.latency_ms for m in self.metrics if m.status_code == 200]
if latencies:
latencies.sort()
avg_latency = sum(latencies) / len(latencies)
p50 = latencies[int(len(latencies) * 0.5)]
p95 = latencies[int(len(latencies) * 0.95)]
p99 = latencies[int(len(latencies) * 0.99)]
else:
avg_latency = p50 = p95 = p99 = 0
success_rate = (self.success_count / num_connections) * 100
return {
'total_connections': num_connections,
'duration_seconds': elapsed,
'successful_requests': self.success_count,
'failed_requests': self.failure_count,
'success_rate_percent': round(success_rate, 2),
'avg_latency_ms': round(avg_latency, 2),
'p50_latency_ms': round(p50, 2),
'p95_latency_ms': round(p95, 2),
'p99_latency_ms': round(p99, 2),
'requests_per_second': round(num_connections / elapsed, 2)
}
사용 예시
async def main():
# Binance 테스트넷 설정
client = ExchangeAPIClient(
base_url="https://testnet.binance.vision",
api_key="YOUR_BINANCE_TEST_API_KEY",
api_secret="YOUR_BINANCE_TEST_SECRET",
max_concurrent=500,
rate_limit_rps=1200
)
async with client:
# 1000 동시 연결 테스트
result = await client.test_concurrent_connections(
num_connections=1000,
duration_seconds=60
)
print("\n📊 테스트 결과:")
print(f" 성공률: {result['success_rate_percent']}%")
print(f" 평균 지연: {result['avg_latency_ms']}ms")
print(f" P95 지연: {result['p95_latency_ms']}ms")
print(f" P99 지연: {result['p99_latency_ms']}ms")
if __name__ == "__main__":
asyncio.run(main())
3. Locust를 활용한 분산 부하 테스트
# locustfile.py
from locust import HttpUser, task, between, events
from locust.runners import MasterRunner, WorkerRunner
import random
import json
class ExchangeAPIUser(HttpUser):
"""
Locust를 활용한 거래소 API 사용자 시뮬레이션
HolySheep AI 게이트웨이 구조에서 영감을 받은 설계
"""
wait_time = between(0.1, 0.5) # 요청 간 대기 시간
host = "https://testnet.binance.vision"
def on_start(self):
"""사용자 시작 시 인증"""
# 실제 환경에서는 서명 생성 로직 추가
self.headers = {
"X-MBX-APIKEY": "YOUR_API_KEY",
"Content-Type": "application/json"
}
# HolySheep AI 스타일: 연결 풀 사전 워밍업
self.client.verify = False
self.client.stream = False
@task(10)
def get_ticker(self):
"""시세 조회 (높은 빈도)"""
with self.client.get(
"/api/v3/ticker/24hr",
headers=self.headers,
name="/api/v3/ticker/24hr"
) as response:
if response.status_code == 200:
pass
@task(5)
def get_order_book(self):
"""호가 조회"""
with self.client.get(
"/api/v3/depth",
params={"symbol": "BTCUSDT", "limit": 20},
headers=self.headers,
name="/api/v3/depth"
) as response:
pass
@task(3)
def get_account(self):
"""계정 정보 조회"""
with self.client.get(
"/api/v3/account",
params={"timestamp": int(time.time() * 1000)},
headers=self.headers,
name="/api/v3/account"
) as response:
pass
@task(1)
def get_exchange_info(self):
"""거래소 정보 조회"""
with self.client.get(
"/api/v3/exchangeInfo",
headers=self.headers
) as response:
pass
커스텀 이벤트 핸들러
@events.request.add_listener
def on_request(request_type, name, response_time, response_length, exception, **kwargs):
"""요청 이벤트 로깅"""
if exception:
print(f"❌ {name} 실패: {exception}")
elif response_time > 1000: # 1초 이상 지연
print(f"⚠️ {name} 지연: {response_time}ms")
분산 테스트 실행 방법
Master 노드: locust -f locustfile.py --master
Worker 노드: locust -f locustfile.py --worker --master-host=MASTER_IP
4. WebSocket 동시 연결 테스트
# src/websocket_manager.py
import asyncio
import websockets
import json
import time
from typing import Dict, List
from dataclasses import dataclass
import random
@dataclass
class WebSocketMetrics:
"""WebSocket 메트릭"""
connection_id: int
messages_received: int
connection_duration_ms: float
error: str = None
class WebSocketStressTest:
"""
WebSocket 동시 연결 스트레스 테스트
HolySheep AI의 실시간 AI 스트리밍 처리 경험 적용
"""
def __init__(
self,
ws_url: str,
max_connections: int = 1000,
test_duration: int = 60
):
self.ws_url = ws_url
self.max_connections = max_connections
self.test_duration = test_duration
self.active_connections: Dict[int, websockets.WebSocketClientProtocol] = {}
self.metrics: List[WebSocketMetrics] = []
self.running = False
# 구독 메시지 탬플릿
self.subscribe_msg = json.dumps({
"method": "SUBSCRIBE",
"params": ["btcusdt@ticker", "ethusdt@ticker"],
"id": 1
})
async def _connection_handler(
self,
connection_id: int,
subscribe_channels: List[str]
) -> WebSocketMetrics:
"""단일 WebSocket 연결 핸들러"""
start_time = time.time()
messages_received = 0
try:
async with websockets.connect(
self.ws_url,
ping_interval=20,
ping_timeout=10,
max_size=10 * 1024 * 1024, # 10MB
max_queue=32
) as websocket:
self.active_connections[connection_id] = websocket
# 구독 요청
sub_msg = {
"method": "SUBSCRIBE",
"params": subscribe_channels,
"id": connection_id
}
await websocket.send(json.dumps(sub_msg))
# 메시지 수신 루프
test_end = start_time + self.test_duration
while time.time() < test_end:
try:
msg = await asyncio.wait_for(
websocket.recv(),
timeout=5.0
)
messages_received += 1
except asyncio.TimeoutError:
# 핑 체크
continue
except Exception as e:
duration = (time.time() - start_time) * 1000
return WebSocketMetrics(
connection_id=connection_id,
messages_received=messages_received,
connection_duration_ms=duration,
error=str(e)
)
finally:
self.active_connections.pop(connection_id, None)
duration = (time.time() - start_time) * 1000
return WebSocketMetrics(
connection_id=connection_id,
messages_received=messages_received,
connection_duration_ms=duration
)
async def run_stress_test(self) -> Dict:
"""동시 WebSocket 연결 스트레스 테스트 실행"""
print(f"🔌 WebSocket 동시 연결 {self.max_connections}개 테스트 시작")
self.running = True
start_time = time.time()
tasks = []
# 채널 목록 (실제 환경에서)
channels = [
"btcusdt@ticker", "ethusdt@ticker", "bnbusdt@ticker",
"btcusdt@depth", "ethusdt@depth", "bnbusdt@depth"
]
# 동시 연결 생성
for i in range(self.max_connections):
# 각 연결에 랜덤 채널 할당
num_channels = random.randint(1, 4)
subscribe_channels = random.sample(channels, num_channels)
task = self._connection_handler(i, subscribe_channels)
tasks.append(task)
# 동시 실행
results = await asyncio.gather(*tasks, return_exceptions=True)
total_duration = time.time() - start_time
# 결과 분석
success_count = 0
total_messages = 0
errors = []
for result in results:
if isinstance(result, WebSocketMetrics):
if result.error is None:
success_count += 1
total_messages += result.messages_received
else:
errors.append(result.error)
return {
'total_connections': self.max_connections,
'successful_connections': success_count,
'failed_connections': self.max_connections - success_count,
'total_duration_seconds': round(total_duration, 2),
'total_messages_received': total_messages,
'avg_messages_per_connection': round(
total_messages / success_count if success_count > 0 else 0, 2
),
'error_types': errors[:10] # 처음 10개 에러만
}
async def main():
# Binance WebSocket 테스트
ws_url = "wss://stream.binance.com:9443/ws"
stress_test = WebSocketStressTest(
ws_url=ws_url,
max_connections=500, # 500개 동시 WebSocket
test_duration=30 # 30초 테스트
)
result = await stress_test.run_stress_test()
print("\n📊 WebSocket 테스트 결과:")
print(f" 전체 연결: {result['total_connections']}")
print(f" 성공: {result['successful_connections']}")
print(f" 실패: {result['failed_connections']}")
print(f" 수신 메시지: {result['total_messages_received']}")
print(f" 평균 메시지/연결: {result['avg_messages_per_connection']}")
if __name__ == "__main__":
asyncio.run(main())
가격과 ROI
| 서비스 | 동시 연결 | 월간 비용 | 1연결당 비용 | 주요 강점 |
|---|---|---|---|---|
| HolySheep AI | 50,000+ | $29~$199 | $0.0006~ | 다중 AI 모델 통합, 비용 최적화 |
| Binance Gateway | 5,000 | 무료 | 무료 | 공식 지원, 안정성 |
| AWS API Gateway | 10,000 | $3.50/variable | 가변적 | 완전한 제어권 |
| Cloudflare Workers | 30,000 | $5/월 + 사용량 | 혼합 | 전역 엣지 네트워크 |
| 자가 구축 (EC2) | 무제한 | $100~$500 | 설계에 따라 상이 | 완전한 커스텀 가능 |
ROI 분석: HolySheep AI 선택 시
HolySheep AI의 실제 비용 절감 사례를 계산해 보겠습니다:
- DeepSeek V3.2 ($0.42/MTok) vs OpenAI GPT-4 ($60/MTok): 99.3% 비용 절감
- Gemini 2.5 Flash ($2.50/MTok) vs Claude Sonnet ($15/MTok): 83% 비용 절감
- 월 100만 토큰 사용 시: 약 $420 → $2.50 (Gemini Flash 적용)
자주 발생하는 오류와 해결책
오류 1: HTTP 429 Too Many Requests (Rate Limit 초과)
# ❌ 문제 발생 코드
async def bad_example(client):
for i in range(10000):
await client.get("/api/v3/account") # Rate Limit 바로 초과
✅ 해결 코드: 지수 백오프 + 분산 요청
import asyncio
import random
class SmartRateLimiter:
"""HolySheep AI 스타일의 지능형 Rate Limiter"""
def __init__(self, requests_per_second: int, burst_size: int = 10):
self.rps = requests_per_second
self.burst = burst_size
self.tokens = burst_size
self.last_update = asyncio.get_event_loop().time()
self._lock = asyncio.Lock()
async def acquire(self):
"""토큰 확보 대기"""
async with self._lock:
now = asyncio.get_event_loop().time()
elapsed = now - self.last_update
# 토큰 충전
self.tokens = min(
self.burst,
self.tokens + elapsed * self.rps
)
self.last_update = now
if self.tokens < 1:
# 토큰이 없으면 대기
wait_time = (1 - self.tokens) / self.rps
await asyncio.sleep(wait_time)
self.tokens = 0
else:
self.tokens -= 1
async def good_example(client, limiter):
"""분산된 요청으로 Rate Limit 우회"""
tasks = []
for i in range(10000):
task = asyncio.create_task(limiter.acquire())
tasks.append(task)
# 100개씩 배치 처리
if len(tasks) >= 100:
await asyncio.gather(*tasks)
tasks = []
await asyncio.sleep(0.1) # 다음 배치 전 대기
# 남은 태스크 처리
if tasks:
await asyncio.gather(*tasks)
오류 2: aiohttp.ClientConnectorError - Too many open files
# ❌ 문제 발생: 연결 누수
async def bad_connection_manager():
async with aiohttp.ClientSession() as session:
for i in range(10000):
async with session.get(url) as resp: # 연결 미닫힘
await resp.text()
✅ 해결 코드: 명시적 연결 관리 + 풀 크기 조정
import asyncio
import aiohttp
class ConnectionPool:
"""연결 풀 관리 (HolySheep AI 아키텍처 참고)"""
def __init__(self, max_connections: int = 1000):
self.max_connections = max_connections
self._semaphore = asyncio.Semaphore(max_connections)
self._connector = None
self._session = None
async def __aenter__(self):
# OS 레벨 설정 (반드시 먼저 적용)
import platform
if platform.system() != "Windows":
import resource
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
# 파일 디스크립터 한도 증가
resource.setrlimit(resource.RLIMIT_NOFILE, (min(65536, hard), hard))
self._connector = aiohttp.TCPConnector(
limit=self.max_connections,
limit_per_host=self.max_connections,
ttl_dns_cache=300,
use_dns_cache=True,
keepalive_timeout=30
)
self._session = aiohttp.ClientSession(
connector=self._connector
)
return self
async def __aexit__(self, *args):
await self._session.close()
await self._connector.close()
# 정리 대기
await asyncio.sleep(0.25)
async def get(self, url: str):
async with self._semaphore: # 동시 연결 수 제한
async with self._session.get(url) as response:
return await response.text()
사용
async def main():
async with ConnectionPool(max_connections=500) as pool:
tasks = [pool.get(f"https://api.example.com/item/{i}")
for i in range(1000)]
results = await asyncio.gather(*tasks)
오류 3: WebSocket 연결 끊김 - "Connection closed unexpectedly"
# ❌ 문제 발생: 재연결 로직 없음
async def bad_websocket_client(url):
async with websockets.connect(url) as ws:
while True:
msg = await ws.recv() # 연결 끊기면 크래시
process(msg)
✅ 해결 코드: 자동 재연결 + 하트비트
import asyncio
import websockets
from websockets.exceptions import ConnectionClosed
class ResilientWebSocketClient:
"""재연결 기능이 있는 WebSocket 클라이언트"""
def __init__(
self,
url: str,
reconnect_delay: float = 1.0,
max_reconnect_delay: float = 60.0,
heartbeat_interval: int = 20
):
self.url = url
self.reconnect_delay = reconnect_delay
self.max_reconnect_delay = max_reconnect_delay
self.heartbeat_interval = heartbeat_interval
self.running = False
async def _heartbeat(self, ws):
"""하트비트 전송"""
while self.running:
try:
await ws.ping()
await asyncio.sleep(self.heartbeat_interval)
except asyncio.CancelledError:
break
except Exception:
break
async def connect_with_retry(self, callback):
"""재연결 기능이 있는 연결"""
self.running = True
current_delay = self.reconnect_delay
consecutive_failures = 0
while self.running:
try:
async with websockets.connect(
self.url,
ping_interval=self.heartbeat_interval,
ping_timeout=10
) as ws:
# 하트비트 태스크 시작
heartbeat_task = asyncio.create_task(
self._heartbeat(ws)
)
consecutive_failures = 0
current_delay = self.reconnect_delay # 지연 시간 리셋
print(f"✅ WebSocket 연결 성공")
try:
while self.running:
try:
msg = await asyncio.wait_for(
ws.recv(),
timeout=30.0
)
await callback(msg)
except asyncio.TimeoutError:
# 타임아웃은 정상적인 inactivity
continue
except ConnectionClosed:
print("⚠️ 연결 끊김, 재연결 시도...")
finally:
heartbeat_task.cancel()
try:
await heartbeat_task
except asyncio.CancelledError:
pass
except Exception as e:
consecutive_failures += 1
print(f"❌ 연결 실패 ({consecutive_failures}회): {e}")
print(f"⏳ {current_delay}초 후 재연결...")
await asyncio.sleep(current_delay)
# 지수 백오프
current_delay = min(
current_delay * 2,
self.max_reconnect_delay
)
def disconnect(self):
"""연결 종료"""
self.running = False
사용
async def main():
async def handle_message(msg):
print(f"수신: {msg}")
client = ResilientWebSocketClient(
url="wss://stream.binance.com:9443/ws/btcusdt@ticker",
reconnect_delay=1.0,
max_reconnect_delay=60.0
)
try:
await client.connect_with_retry(handle_message)
except KeyboardInterrupt:
client.disconnect()
오류 4: 비동기 asyncio RuntimeError: Event loop is running
# ❌ 문제 발생: 잘못된 비동기 컨텍스트 중첩
async def bad_nested_async():
# 이미 실행 중인 루프에서 새 루프 생성
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# ❌ RuntimeError 발생 가능
async with some_client() as client:
await client.connect()
✅ 해결 코드: 올바른 이벤트 루프 관리
import asyncio
import nest_asyncio
Jupyter/중첩된 환경에서는 nest_asyncio 적용
try:
nest_asyncio.apply()
except ImportError:
pass # 일반 환경에서는 불필요
class AsyncManager:
"""올바른 비동기 컨텍스트 관리"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._loop = None
cls._instance._running = False
return cls._instance
def run(self, coro):
"""올바른 이벤트 루프 실행"""
try:
# 이미 실행 중인 경우
loop = asyncio.get_running_loop()
# 별도 스레드에서 실행
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as pool:
future = pool.submit(
asyncio.run, coro
)
return future.result()
except RuntimeError:
# 루프가 없는 경우
return asyncio.run(coro)
최종 해결: 항상 단일 진입점
async def async_main():
"""메인 비동기 함수"""
async with ExchangeAPIClient(...) as client:
result = await client.test