업비트에서 실시간 가격 및 호가 데이터 받아오기 (코드 포함)
암호화폐 트레이딩 봇이나 데이터 분석 시스템을 개발할 때 가장 중요한 것 중 하나는 실시간 시장 데이터를 빠르고 안정적으로 수집하는 것입니다. 이번 포스트에서는 업비트(Upbit) 거래소의 WebSocket API를 활용하여 실시간으로 가격과 호가 데이터를 받아오는 방법을 다룹니다.
왜 WebSocket인가?
섹션 제목: “왜 WebSocket인가?”업비트는 REST API와 WebSocket API 두 가지 방식을 제공합니다. 실시간 데이터 수집에는 WebSocket이 훨씬 효율적입니다.
REST API의 단점:
- 주기적으로 요청(polling)해야 하므로 네트워크 오버헤드가 큽니다
- API 호출 제한(rate limit)에 걸리기 쉽습니다
- 실시간성이 떨어집니다
WebSocket의 장점:
- 한 번 연결하면 서버가 데이터를 실시간으로 푸시합니다
- 네트워크 효율성이 높고 지연시간이 적습니다
- 여러 종목을 동시에 구독 가능합니다
필요한 라이브러리
섹션 제목: “필요한 라이브러리”pip install websocket-client keyring keyrings.altwebsocket-client: WebSocket 클라이언트 라이브러리keyring: API 키를 안전하게 관리하기 위한 라이브러리keyrings.alt: keyring의 파일 기반 백엔드
코드 구조 개요
섹션 제목: “코드 구조 개요”이번 예제 코드는 다음과 같은 구조로 설계되었습니다:
- WebSocket 기본 클래스: 스레드 기반으로 동작하며 재사용 가능한 WebSocket 연결 관리
- TickerWebSocket: 실시간 가격 정보 수신용
- OrderbookWebSocket: 실시간 호가 정보 수신용
- 비동기 처리: asyncio를 활용한 데이터 처리 루프
주요 구현 포인트
섹션 제목: “주요 구현 포인트”1. 스레드 기반 WebSocket 연결
섹션 제목: “1. 스레드 기반 WebSocket 연결”class WebSocket(threading.Thread): def __init__(self, subscribe_type: str, codes: list): super().__init__(daemon=True) # 데몬 스레드로 설정 self.ws = websocket.WebSocketApp(...)daemon=True로 설정하면 메인 프로그램이 종료될 때 함께 종료됩니다.
2. WebSocket 구독 메시지 형식
섹션 제목: “2. WebSocket 구독 메시지 형식”업비트 WebSocket API는 다음 형식의 구독 메시지를 요구합니다:
subscribe_message = [ {"ticket": str(uuid.uuid4())}, # 고유 식별자 {"type": "ticker", "codes": ["KRW-BTC", "KRW-ETH"]}, # 구독 정보]ticket: 요청을 구분하는 고유 ID (UUID 사용 권장)type: 데이터 타입 (ticker, orderbook, trade 등)codes: 구독할 마켓 코드 리스트
3. 자동 재연결 기능
섹션 제목: “3. 자동 재연결 기능”self.ws.run_forever(ping_interval=30, ping_timeout=10, reconnect=2)ping_interval=30: 30초마다 ping 메시지 전송ping_timeout=10: ping 응답 타임아웃 10초reconnect=2: 연결 끊김 시 2초 후 자동 재연결
4. 비동기 데이터 처리
섹션 제목: “4. 비동기 데이터 처리”async def process(self): count = 0 last_time = 0 while True: if self.last_data: self.process_data() # 데이터 처리 로직 count += 1 await asyncio.sleep(1 - (time.time() - last_time)) last_time = time.time()매 초마다 최신 데이터를 처리하면서도 정확한 시간 간격을 유지합니다.
실행 방법
섹션 제목: “실행 방법”API 키 설정 (선택사항)
섹션 제목: “API 키 설정 (선택사항)”공개 데이터(가격, 호가)는 API 키 없이도 사용 가능하지만, 개인 거래 정보에는 필요합니다:
import keyringfrom keyrings.alt.file import PlaintextKeyring
keyring.set_keyring(PlaintextKeyring())keyring.set_password("upbit", "access_key", "your_access_key")keyring.set_password("upbit", "secret_key", "your_secret_key")프로그램 실행
섹션 제목: “프로그램 실행”python your_script.py실행하면 다음과 같은 출력을 볼 수 있습니다:
WebSocket created: ticker ['KRW-ETH']WebSocket created: orderbook ['KRW-ETH']Starting WebSocket thread: ticker ['KRW-ETH']Starting WebSocket thread: orderbook ['KRW-ETH']Inserted ticker data at timestamp 1707393600000 | Count: 1Inserted orderbook data at timestamp 1707393600000 | Count: 1활용 방안
섹션 제목: “활용 방안”이 코드를 기반으로 다음과 같은 기능을 추가할 수 있습니다:
- 데이터베이스 저장:
process_data()메서드에서 MongoDB, PostgreSQL 등에 저장 - 실시간 알림: 특정 가격 도달 시 알림 전송
- 차트 시각화: Plotly, Matplotlib 등으로 실시간 차트 업데이트
- 트레이딩 봇: 가격 변동에 따른 자동 매매 로직 구현
- ML 모델 학습: 실시간 데이터를 feature로 활용
전체 코드
섹션 제목: “전체 코드”import asyncioimport uuidimport websocketimport threadingimport jsonimport time
from keyrings.alt.file import PlaintextKeyringimport keyringkeyring.set_keyring(PlaintextKeyring())
# codes# https://docs.upbit.com/kr/reference/list-trading-pairsBITCOIN = "KRW-BTC"ETHEREUM = "KRW-ETH"
access_key = keyring.get_password("upbit", "access_key")secret_key = keyring.get_password("upbit", "secret_key")
class WebSocket(threading.Thread): def __init__(self, subscribe_type: str, codes: list): super().__init__(daemon=True) self.ws = websocket.WebSocketApp( "wss://api.upbit.com/websocket/v1", on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close ) self.last_data = None self.subscribe_type = subscribe_type self.codes = codes
def on_open(self, ws): assert self.subscribe_type != "", "subscribe_type must be set before running the WebSocket" assert self.codes, "codes must be set before running the WebSocket" print(f"WebSocket created: {self.subscribe_type} {self.codes}") subscribe_message = [ {"ticket": str(uuid.uuid4())}, {"type": self.subscribe_type, "codes": self.codes}, ] ws.send(json.dumps(subscribe_message))
def on_message(self, ws, message): data = json.loads(message) self.last_data = data
def on_error(self, ws, err): print(f"Error in WebSocket {self.subscribe_type} {self.codes}:", err)
def on_close(self): print(f"WebSocket closed: {self.subscribe_type} {self.codes}")
def run(self): print(f"Starting WebSocket thread: {self.subscribe_type} {self.codes}") self.ws.run_forever(ping_interval=30, ping_timeout=10, reconnect=2)
async def process(self): count = 0 last_time = 0 while True: if self.last_data: self.process_data() count += 1 print(f"\rInserted {self.subscribe_type} data at timestamp {self.last_data['timestamp']} | Count: {count}", end="", flush=True) await asyncio.sleep(1 - (time.time() - last_time)) last_time = time.time()
def process_data(self): pass
class TickerWebSocket(WebSocket): def __init__(self): super().__init__(subscribe_type="ticker", codes=[ETHEREUM])
def process_data(self): # do something pass
class OrderbookWebSocket(WebSocket): def __init__(self): super().__init__(subscribe_type="orderbook", codes=[ETHEREUM])
def process_data(self): # do something pass
async def main(): ticker_ws = TickerWebSocket() orderbook_ws = OrderbookWebSocket() ticker_ws.start() orderbook_ws.start() ticker_task = asyncio.create_task(ticker_ws.process()) orderbook_task = asyncio.create_task(orderbook_ws.process()) await asyncio.gather(ticker_task, orderbook_task)
if __name__ == "__main__": asyncio.run(main())주의사항
섹션 제목: “주의사항”- 연결 안정성: 네트워크 상태에 따라 연결이 끊길 수 있으므로 재연결 로직이 중요합니다.
- 메모리 관리:
last_data를 계속 덮어쓰므로 메모리 누수는 없지만, 데이터를 저장하는 경우 메모리 관리에 주의해야 합니다. - 에러 처리: 실제 운영 환경에서는 더 상세한 에러 처리와 로깅이 필요합니다.
마무리
섹션 제목: “마무리”이번 포스트에서는 업비트 WebSocket API를 활용한 실시간 데이터 수집 방법을 알아보았습니다.
핵심 포인트:
- WebSocket은 실시간 데이터 수집에 효율적입니다
- 스레드와 asyncio를 조합하여 안정적인 데이터 처리가 가능합니다
- 자동 재연결 기능으로 연결 안정성을 확보할 수 있습니다
이 코드를 기반으로 여러분만의 트레이딩 시스템이나 분석 도구를 만들어보세요!
