암호화폐 시장에서 시장 제조사(Market Maker)는流動성 제공의 핵심 역할을 담당합니다. OKX 거래소의 시장 제조사 API를 활용하면 자동으로 비트코인과 알트코인의 매수-매도 호가를 관리하고 헤지 포지션을 실시간으로 조정할 수 있습니다. 이 튜토리얼에서는 프로그래밍 경험이 전혀 없는 초보자도 따라할 수 있도록 API 연결부터 자동 헤지 전략 구현, 그리고 리스크 관리까지 단계별로 설명드리겠습니다.
필수 전제 조건: 프로그래밍 경험 불필요, 컴퓨터 기본 조작 가능, USDT 또는 암호화폐 최소 500달러 권장
시장 제조사란 무엇인가요?
시장 제조사는 항상 매수 호가와 매도 호가를 동시에 제시하는 거래자입니다. 예를 들어 비트코인 현재가가 67,000달러일 때, 시장 제조사는 66,995달러에 매수 주문, 67,005달러에 매수 종료 주문을 동시에 걸어두는 것입니다. 이 두 가격 사이의 차이, 즉 스프레드가 제조사의 수익원이 됩니다.
시장 제조사의 수익 원리
- 스프레드 수익: 매수와 매도 가격 차이에서 발생
- 시장이 빈번할 때: 체결 빈도가 높을수록 수익 증가
- volatilidad 활용: 변동성이 높은 시장에서 더 넓은 스프레드로 수익 확보
OKX API 기본 설정
1단계: OKX 계정 생성 및 API 키 발급
OKX 공식 웹사이트에서 계정을 생성한 후, API 키를 발급받아야 합니다. [그림 위치: OKX 웹사이트 우측 상단 프로필 아이콘 → API 클릭 → API 키 생성 버튼]
API 키 생성 시 반드시 다음 권한을 활성화하세요:
- 거래(Trade): 주문 실행 및 취소
- 읽기(Read): 계정 잔고 및 시장 데이터 조회
- 전용 비밀번호(Lookup): 서명 인증용
2단계: 개발 환경 준비
Python 3.9 이상을 설치하고 필요한 라이브러리를 준비합니다. [그림 위치: 명령 프롬프트 또는 터미널 창]
# Python 환경 설정 및 필요한 패키지 설치
pip install requests
pip install websocket-client
pip install hmac
pip install hashlib
pip install time
pip install json필요한 패키지 설명:
- requests: HTTP 요청을 보내는 라이브러리
- websocket-client: 실시간 시장 데이터 수신용
- hmac, hashlib: API 요청 서명 인증용
3단계: API 연결 기본 코드
OKX API에 연결하는 기본 구조를 만들어보겠습니다. 이 코드는 모든 시장 제조사 전략의 뼈대가 됩니다.
import requests
import hmac
import hashlib
import time
import json
========================================
OKX API 기본 설정
========================================
API_KEY = "YOUR_OKX_API_KEY"
API_SECRET = "YOUR_OKX_API_SECRET"
PASSPHRASE = "YOUR_API_PASSPHRASE"
BASE_URL = "https://www.okx.com"
def get_timestamp():
"""현재 시간 가져오기"""
return time.strftime('%Y-%m-%dT%H:%M:%S.000Z', time.gmtime())
def sign(message, secret):
"""API 요청 서명 생성"""
mac = hmac.new(
bytes(secret, encoding='utf8'),
bytes(message, encoding='utf8'),
hashlib.sha256
)
return mac.hexdigest()
def get_request_headers(path, method, body=""):
"""API 요청 헤더 생성"""
timestamp = get_timestamp()
message = timestamp + method + path + body
signature = sign(message, API_SECRET)
return {
'OK-ACCESS-KEY': API_KEY,
'OK-ACCESS-SIGN': signature,
'OK-ACCESS-TIMESTAMP': timestamp,
'OK-ACCESS-PASSPHRASE': PASSPHRASE,
'Content-Type': 'application/json'
}
계정 잔고 조회 테스트
def get_account_balance():
"""내 계정의 잔고를 조회합니다"""
endpoint = "/api/v5/account/balance"
url = BASE_URL + endpoint
response = requests.get(url, headers=get_request_headers(endpoint, "GET"))
return response.json()
테스트 실행
if __name__ == "__main__":
print("OKX API 연결 테스트 중...")
result = get_account_balance()
print(json.dumps(result, indent=2))이 코드를 okx_basic.py로 저장하고 실행하면 API 연결이 정상인지 확인할 수 있습니다. 정상 연결 시 계좌 잔고 정보가 JSON 형태로 출력됩니다.
자동 헤지 전략 구현
헤지(Hedging)란 무엇인가요?
헤지란 위험을 줄이기 위해 반대 방향의 포지션을 취하는 전략입니다. 시장 제조사는 항상 매수와 매도를 동시에 진행하므로, 가격 변동으로 인한 손실을 최소화하기 위해 헤지 포지션을 관리해야 합니다.
단순 양방향 호가 전략
가장 기본적인 시장 제조사 전략은 현재 가격 기준 양쪽에 호가를 배치하는 것입니다. 이 전략의 핵심 파라미터를 먼저 정의합니다:
- 호가 간격: 매수-매도 가격 차이 (기본값: 0.1%)
- 호가 수량: 각 호가의 크기
- 재호가 간격: 기존 호가가 체결되지 않았을 때 새로 호가하는 시간
import requests
import time
import json
from datetime import datetime
========================================
시장 제조사 자동 헤지 전략
========================================
class MarketMaker:
def __init__(self, api_key, api_secret, passphrase):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.base_url = "https://www.okx.com"
self.symbol = "BTC-USDT"
# 전략 파라미터 설정
self.spread_percent = 0.001 # 0.1% 스프레드
self.order_size = 0.01 # BTC 0.01개씩
self.check_interval = 5 # 5초마다 확인
# 현재 활성 주문 추적
self.active_orders = []
def get_current_price(self):
"""현재 시장 가격 조회"""
endpoint = f"/api/v5/market/ticker?instId={self.symbol}"
url = self.base_url + endpoint
response = requests.get(url)
data = response.json()
if data['code'] == '0':
return float(data['data'][0]['last'])
else:
print(f"가격 조회 오류: {data['msg']}")
return None
def calculate_order_price(self, current_price, side):
"""호가 가격 계산"""
if side == "buy":
# 매수 호가는 현재가보다 낮게
return round(current_price * (1 - self.spread_percent), 1)
else:
# 매도 호가는 현재가보다 높게
return round(current_price * (1 + self.spread_percent), 1)
def place_order(self, side, price, size):
"""주문 실행"""
endpoint = "/api/v5/trade/order"
url = self.base_url + endpoint
body = {
"instId": self.symbol,
"tdMode": "cash", # 현물 거래
"side": side, # buy 또는 sell
"ordType": "limit", # 지정가 주문
"px": str(price),
"sz": str(size)
}
headers = self.get_signed_headers(endpoint, "POST", json.dumps(body))
response = requests.post(url, headers=headers, data=json.dumps(body))
return response.json()
def cancel_all_orders(self):
"""모든 활성 주문 취소"""
for order in self.active_orders:
self.cancel_order(order['ordId'])
self.active_orders = []
def cancel_order(self, order_id):
"""개별 주문 취소"""
endpoint = "/api/v5/trade/cancel-order"
url = self.base_url + endpoint
body = {
"instId": self.symbol,
"ordId": order_id
}
headers = self.get_signed_headers(endpoint, "POST", json.dumps(body))
response = requests.post(url, headers=headers, data=json.dumps(body))
return response.json()
def run_strategy(self):
"""시장 제조사 전략 실행 메인 루프"""
print(f"시장 제조사 시작: {datetime.now()}")
print(f"거래 페어: {self.symbol}")
print(f"스프레드: {self.spread_percent * 100}%")
print("-" * 50)
while True:
try:
# 현재 가격 가져오기
current_price = self.get_current_price()
if current_price:
# 매수 호가 계산 및 배치
buy_price = self.calculate_order_price(current_price, "buy")
print(f"매수 호가 배치: {buy_price} USDT")
# 매도 호가 계산 및 배치
sell_price = self.calculate_order_price(current_price, "sell")
print(f"매도 호가 배치: {sell_price} USDT")
# 실제 거래 시 아래 주석 해제
# self.place_order("buy", buy_price, self.order_size)
# self.place_order("sell", sell_price, self.order_size)
time.sleep(self.check_interval)
except KeyboardInterrupt:
print("\n시장 제조사 중지...")
self.cancel_all_orders()
break
except Exception as e:
print(f"오류 발생: {e}")
time.sleep(10)
전략 실행
if __name__ == "__main__":
maker = MarketMaker(
api_key="YOUR_API_KEY",
api_secret="YOUR_API_SECRET",
passphrase="YOUR_PASSPHRASE"
)
maker.run_strategy()실시간 시장 데이터 수신
더 빠른 반응이 필요하면 WebSocket을 사용하여 실시간 시장 데이터를 수신하는 것이 좋습니다.
import websocket
import json
import threading
class RealTimeMarketData:
def __init__(self, symbol="BTC-USDT"):
self.symbol = symbol
self.last_price = None
self.ws = None
self.is_running = False
def on_message(self, ws, message):
"""WebSocket 메시지 수신 처리"""
data = json.loads(message)
if data.get('arg', {}).get('channel') == 'tickers':
ticker_data = data.get('data', [{}])[0]
self.last_price = float(ticker_data.get('last', 0))
print(f"실시간 가격 업데이트: {self.last_price} USDT")
def on_error(self, ws, error):
"""에러 처리"""
print(f"WebSocket 오류: {error}")
def on_close(self, ws):
"""연결 종료 처리"""
print("WebSocket 연결 종료")
def on_open(self, ws):
"""연결 시작 시 구독 요청"""
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "tickers",
"instId": self.symbol
}]
}
ws.send(json.dumps(subscribe_msg))
print(f"{self.symbol} 실시간 데이터 구독 시작")
def start(self):
"""WebSocket 연결 시작"""
self.is_running = True
self.ws = websocket.WebSocketApp(
"wss://ws.okx.com:8443/ws/v5/public",
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
# 별도 스레드에서 WebSocket 실행
ws_thread = threading.Thread(target=self.ws.run_forever)
ws_thread.daemon = True
ws_thread.start()
def stop(self):
"""WebSocket 연결 중지"""
self.is_running = False
if self.ws:
self.ws.close()
def get_latest_price(self):
"""최신 가격 반환"""
return self.last_price
사용 예시
if __name__ == "__main__":
market = RealTimeMarketData("BTC-USDT")
market.start()
try:
while True:
price = market.get_latest_price()
if price:
print(f"현재 BTC 가격: {price} USDT")
time.sleep(1)
except KeyboardInterrupt:
market.stop()
print("실시간 데이터 수신 중지")리스크 관리 전략
왜 리스크 관리가 중요한가요?
시장 제조사 전략에서 리스크 관리는 수익보다 중요합니다. 잘못된 리스크 관리는 단시간에 큰 손실을 초래할 수 있습니다. 실제 사례를 살펴보면, 2022년 FTX 사건 당시 적절한 리스크 관리를 하지 못한 시장 제조사들이 상당한 손실을 입었습니다.
핵심 리스크 관리 원칙
- 포지션 한도 설정: 한 번에 가질 수 있는 최대 포지션 제한
- 손절매 규칙: 특정 손실률 도달 시 모든 포지션 강제 청산
- 分散투자: 여러 거래페어에 분산
- 잔고 비율: 주문 금액이 계좌 잔고의 일정 비율 초과 금지
import time
from datetime import datetime
class RiskManager:
"""리스크 관리자 클래스"""
def __init__(self, max_position_ratio=0.2, max_daily_loss=0.05, max_slippage=0.002):
# 최대 포지션 비율: 계좌 잔고의 20%까지
self.max_position_ratio = max_position_ratio
# 일일 최대 손실률: 5%
self.max_daily_loss = max_daily_loss
# 최대 슬리피지: 0.2%
self.max_slippage = max_slippage
# 일일 손익 추적
self.daily_pnl = 0
self.daily_start_balance = 0
self.last_reset_date = datetime.now().date()
def check_position_limit(self, current_position_value, account_balance):
"""포지션 한도 확인"""
if account_balance <= 0:
return False
position_ratio = current_position_value / account_balance
if position_ratio > self.max_position_ratio:
print(f"경고: 포지션 비율 {position_ratio:.2%} > 제한 {self.max_position_ratio:.2%}")
return False
return True
def check_daily_loss_limit(self, current_balance):
"""일일 손실 한도 확인"""
today = datetime.now().date()
# 날짜가 바뀌면 리셋
if today != self.last_reset_date:
self.daily_pnl = 0
self.daily_start_balance = current_balance
self.last_reset_date = today
if self.daily_start_balance <= 0:
return True
loss_ratio = (self.daily_start_balance - current_balance) / self.daily_start_balance
if loss_ratio > self.max_daily_loss:
print(f"일일 손실 한도 초과: {loss_ratio:.2%} > {self.max_daily_loss:.2%}")
return False
return True
def calculate_safe_order_size(self, account_balance, current_price, risk_ratio=0.01):
"""안전한 주문 수량 계산"""
# 계좌 잔고의 1%만 위험에 노출
max_order_value = account_balance * risk_ratio
order_size = max_order_value / current_price
return round(order_size, 6)
def validate_order(self, order_price, current_price, side):
"""주문 유효성 검증"""
# 슬리피지 체크
if side == "buy":
price_change = (order_price - current_price) / current_price
else:
price_change = (current_price - order_price) / current_price
if price_change > self.max_slippage:
print(f"슬리피지 초과: {price_change:.2%} > {self.max_slippage:.2%}")
return False
return True
def get_risk_report(self, account_balance, position_value):
"""리스크 보고서 생성"""
return {
"timestamp": datetime.now().isoformat(),
"account_balance": account_balance,
"position_value": position_value,
"position_ratio": position_value / account_balance if account_balance > 0 else 0,
"max_position_ratio": self.max_position_ratio,
"daily_pnl": self.daily_pnl,
"daily_loss_ratio": (self.daily_start_balance - account_balance) / self.daily_start_balance if self.daily_start_balance > 0 else 0
}
사용 예시
risk_manager = RiskManager(
max_position_ratio=0.2, # 20% 포지션 제한
max_daily_loss=0.05, # 5% 일일 손실 제한
max_slippage=0.002 # 0.2% 슬리피지 제한
)
주문 전 검증
current_balance = 10000 # USDT
current_price = 67000 # BTC/USDT
order_size = 0.01 # 0.01 BTC
안전한 수량 계산
safe_size = risk_manager.calculate_safe_order_size(current_balance, current_price)
print(f"권장 주문 수량: {safe_size} BTC")
리스크 보고서
report = risk_manager.get_risk_report(current_balance, order_size * current_price)
print(f"리스크 보고서: {report}")시장 제조사 API 서비스 비교
| 특징 | OKX 시장 제조사 API | HolySheep AI 게이트웨이 | Binance API |
|---|---|---|---|
| 주요 용도 | 암호화폐 거래 및 시장 제조 | AI 모델 통합 및 비용 최적화 | 암호화폐 거래 |
| API 연결 방식 | REST + WebSocket | 단일 REST API | REST + WebSocket |
| 지원 모델 | 없음 | GPT-4.1, Claude, Gemini, DeepSeek 등 50+ | 없음 |
| 결제 방식 | 암호화폐만 | 로컬 결제 + 해외 신용카드 | 암호화폐만 |
| 자동 헤지 기능 | 직접 구현 필요 | AI 분석 도구 제공 | 직접 구현 필요 |
| 초보자 친화도 | 중간 (문서 복잡) | 높음 (단일 키 통합) | 중간 |
| 비용 | 거래 수수료만 | GPT-4.1: $8/MTok Claude: $15/MTok Gemini: $2.50/MTok |
거래 수수료만 |
이런 팀에 적합 / 비적합
이 전략이 적합한 경우
- 암호화폐 거래 경험이 있는 개발자
- 자동 거래 시스템을 구축하려는 퀀트 트레이더
- 流动성 제공으로 수동 수익을 자동화하려는 투자자
- API 프로그래밍 기본 지식이 있는 분
- 리스크 관리의 중요성을 인지하고 있는 분
이 전략이 적합하지 않은 경우
- 프로그래밍 경험이 전혀 없는 분
- 높은 레버리지로 단기간 큰 수익을 원하는 분
- 암호화폐市场监管이 불명확한 지역에 거주하는 분
- 투자는 불안정하다는 것을 인지하지 못하는 분
- 감정적 투자를 하는 분
AI 기반 시장 분석으로 HolySheep 활용하기
시장 제조사 전략의 성공은 정확한 시장 분석에 크게 의존합니다. HolySheep AI 게이트웨이를 활용하면 단일 API 키로 GPT-4.1, Claude, Gemini, DeepSeek 등 다양한 AI 모델을 활용하여:
- 시장 분위기 분석: 뉴스를 실시간으로 분석하여 호가 전략 조정
- 변동성 예측: 과거 데이터 기반 변동성 예측으로 스프레드 동적 조절
- 리스크 경고: 비정상적 시장 패턴 감지 시 자동 알림
- 포트폴리오 최적화: 다중 거래페어 분배 비율 AI 추천
# HolySheep AI를 활용한 시장 분석 예시
import requests
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
def analyze_market_sentiment(news_text):
"""AI를 활용한 시장 분위기 분석"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [
{
"role": "system",
"content": "당신은 암호화폐 시장 전문가입니다. 뉴스 내용을 분석하여 시장 분위기를 단어(bullish/bearish/neutral)로만 답변해주세요."
},
{
"role": "user",
"content": f"다음 뉴스에 대한 시장 분위기를 분석해주세요: {news_text}"
}
],
"temperature": 0.3,
"max_tokens": 10
}
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content'].strip()
else:
print(f"분석 실패: {response.status_code}")
return "neutral"
def get_trading_recommendation(symbol, price_data):
"""AI 기반 거래 추천"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "claude-sonnet-4-20250514",
"messages": [
{
"role": "system",
"content": "당신은 보수적인 시장 제조사 어드바이저입니다. 현재 시장 데이터와 리스크 관리를 고려하여 spread_percent(0.001~0.005)와 order_size_multiplier(0.5~1.5)를 JSON으로 추천해주세요."
},
{
"role": "user",
"content": f"거래 페어: {symbol}\n현재 데이터: {price_data}\n적절한 전략 파라미터를 추천해주세요."
}
],
"temperature": 0.2,
"max_tokens": 100
}
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
print(f"추천 실패: {response.status_code}")
return None
사용 예시
if __name__ == "__main__":
# 시장 분석
news = "BTC 대규모 ETF 승인 기대감으로 시장 상승"
sentiment = analyze_market_sentiment(news)
print(f"시장 분위기: {sentiment}")
# 거래 추천
price_data = {
"current_price": 67000,
"volume_24h": 25000000000,
"volatility": 0.02
}
recommendation = get_trading_recommendation("BTC-USDT", price_data)
print(f"AI 추천: {recommendation}")HolySheep AI 게이트웨이를 사용하면 각 모델의 가격을 비교하고 최적의 선택을 할 수 있습니다:
| AI 모델 | 입력 비용 (per 1M 토큰) | 출력 비용 (per 1M 토큰) | 권장 사용 사례 |
|---|---|---|---|
| GPT-4.1 | $2.00 | $8.00 | 복잡한 시장 분석 |
| Claude Sonnet 4.5 | $3.00 | $15.00 | 정교한 리스크 분석 |
| Gemini 2.5 Flash | $0.35 | $2.50 | 실시간 빠른 분석 |
| DeepSeek V3.2 | $0.10 | $0.42 | 대량 데이터 처리 |
자주 발생하는 오류 해결
1. API 인증 오류 (401 Unauthorized)
문제: API 요청 시 401 오류가 발생하며 "signature verification failed" 메시지가 나타납니다.
원인: API 키, 비밀번호, 서명이 일치하지 않거나 타임스탬프가 서버와 다릅니다.
# 해결 방법 1: 타임스탬프 동기화
import ntplib
from datetime import datetime
def sync_server_time():
"""서버 시간과 동기화"""
try:
client = ntplib.NTPClient()
response = client.request('pool.ntp.org')
local_time = datetime.fromtimestamp(response.tx_time)
print(f"현재 로컬 시간: {datetime.now()}")
print(f"NTP 서버 시간: {local_time}")
# Windows의 경우
import subprocess
subprocess.run(['w32tm', '/resync'], capture_output=True)
return True
except Exception as e:
print(f"시간 동기화 실패: {e}")
return False
해결 방법 2: API 키 재발급
OKX 웹사이트에서 기존 API 키를 삭제하고 새로 발급받으세요.
발급 시 반드시 '거래' 권한과 '읽기' 권한을 모두 체크해주세요.
해결 방법 3: 비밀번호 확인
API 키 생성 시 입력한 Passphrase가 코드에서 사용하는 것과 동일한지 확인
2. 주문 실패 오류 (5001: Insufficient balance)
문제: "Insufficient balance" 오류가 발생하며 주문이 체결되지 않습니다.
원인: 계좌 잔고가 주문 금액보다 적거나 주문 금액이 최소 주문 수량에 미달합니다.
# 해결 방법 1: 잔고 확인 및 충분한 금액 확보
def check_balance_before_order(required_amount, symbol="USDT"):
"""주문 전 잔고 확인"""
balance_data = get_account_balance()
for balance in balance_data.get('data', [{}])[0].get('details', []):
if balance.get('ccy') == symbol:
available = float(balance.get('availBal', 0))
print(f"{symbol} 사용 가능 잔고: {available}")
if available < required_amount:
print(f"잔고 부족! 필요 금액: {required_amount}, 현재 잔고: {available}")
return False
return True
return False
해결 방법 2: 최소 주문 수량 확인
def get_min_order_size(symbol):
"""최소 주문 수량 조회"""
endpoint = f"/api/v5/public/instrument?instId={symbol}"
url = f"https://www.okx.com{endpoint}"
response = requests.get(url)
data = response.json()
if data['code'] == '0':
instrument = data['data'][0]
min_size = instrument.get('minSz', '0')
print(f"{symbol} 최소 주문 수량: {min_size}")
return float(min_size)
return None
실제 주문 전 검증
symbol = "BTC-USDT"
current_price = 67000
desired_order_size = 0.001
min_size = get_min_order_size(symbol)
if min_size and desired_order_size >= min_size:
required_usdt = desired_order_size * current_price
if check_balance_before_order(required_usdt):
print("주문 가능!")
else:
print("잔고 부족으로 주문 불가")
else:
print(f"주문 수량이 최소 수량({min_size})보다 작습니다")3. WebSocket 연결 끊김 (1006: Abnormal closure)
문제: WebSocket 연결이 갑자기 종료되며 "connection closed unexpectedly" 오류가 발생합니다.
원인: 서버의 연결 타임아웃, 네트워크 문제, 또는 너무 많은 연결 요청입니다.
# 해결 방법 1: 자동 재연결 로직 구현
import websocket
import time
import threading
class WebSocketReconnector:
def __init__(self, url, on_message, on_error, subscriptions):
self.url = url
self.on_message = on_message
self.on_error = on_error
self.subscriptions = subscriptions
self.ws = None
self.is_running = False
self.reconnect_delay = 1
self.max_reconnect_delay = 30
def connect(self):
"""연결 시도"""
try:
self.ws = websocket.WebSocketApp(
self.url,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
self.is_running = True
ws_thread = threading.Thread(target=self.ws.run_forever)
ws_thread.daemon = True
ws_thread.start()
print("WebSocket 연결 성공")
self.reconnect_delay = 1 # 재연결 딜레이 리셋
except Exception as e:
print(f"연결 실패: {e}")
self.schedule_reconnect()
def on_open(self, ws):
"""연결 성공 시 구독 재요청"""
print("연결 수립, 구독 중...")
for sub in self.subscriptions:
ws.send(json.dumps(sub))
def on_close(self, ws, close_status_code, close_msg):
"""연결 종료 시 처리"""
if self.is_running:
print(f"연결 종료 (코드: {close_status_code})")
self.schedule_reconnect()
def schedule_reconnect(self):
"""재연결 예약"""
print(f"{self.reconnect_delay}초 후 재연결 시도...")
time.sleep(self.reconnect_delay)
# 지수 백오프로 딜레이 증가
self.reconnect_delay = min(
self.reconnect_delay * 2,
self.max_reconnect_delay
)
self.connect()
def disconnect(self):
"""연결 해제"""
self.is_running = False
if self.ws:
self.ws.close()
사용 예시
def handle_message(ws, message):
print(f"수신: {message}")
def handle_error(ws, error):
print(f"오류: {error}")
subscriptions = [
{"op": "subscribe", "args": [{"channel": "tickers", "instId": "BTC-USDT"}]}
]
ws_manager = WebSocketReconnector(
url="wss://ws.okx.com:8443/ws/v5/public",
on_message=handle_message,
on_error=handle_error,
subscriptions=subscriptions
)
ws_manager.connect()
time.sleep(60) # 60초간 수신
ws_manager.disconnect()4. 주문 체결 지연 (Order not filled)
문제: 지정가 주문이 장시간 체결되지 않으며, 시장 가격이 호가를 지나갑니다.
원인: 스프레드가 너무 좁거나 호가 수량이 시장 깊이에 비해 너무 큽니다.
# 해결 방법 1: 시장 깊이 조회 후 호가 배치
def get_market_depth(symbol, limit=10):