업비트에서 실시간 가격 및 호가 데이터 받아오기 (코드 포함)

Feb 8, 2026

업비트 WebSocket 실시간 데이터 대표 이미지

암호화폐 트레이딩 봇이나 데이터 분석 시스템을 개발할 때 가장 중요한 것 중 하나는 실시간 시장 데이터를 빠르고 안정적으로 수집하는 것입니다. 이번 포스트에서는 업비트(Upbit) 거래소의 WebSocket API를 활용하여 실시간으로 가격과 호가 데이터를 받아오는 방법을 다룹니다.

왜 WebSocket인가?

업비트는 REST API와 WebSocket API 두 가지 방식을 제공합니다. 실시간 데이터 수집에는 WebSocket이 훨씬 효율적입니다.

REST API의 단점:

주기적으로 요청(polling)해야 하므로 네트워크 오버헤드가 큽니다
API 호출 제한(rate limit)에 걸리기 쉽습니다
실시간성이 떨어집니다

WebSocket의 장점:

한 번 연결하면 서버가 데이터를 실시간으로 푸시합니다
네트워크 효율성이 높고 지연시간이 적습니다
여러 종목을 동시에 구독 가능합니다

필요한 라이브러리

pip install websocket-client keyring keyrings.alt

websocket-client: WebSocket 클라이언트 라이브러리
keyring: API 키를 안전하게 관리하기 위한 라이브러리
keyrings.alt: keyring의 파일 기반 백엔드

코드 구조 개요

이번 예제 코드는 다음과 같은 구조로 설계되었습니다:

WebSocket 기본 클래스: 스레드 기반으로 동작하며 재사용 가능한 WebSocket 연결 관리
TickerWebSocket: 실시간 가격 정보 수신용
OrderbookWebSocket: 실시간 호가 정보 수신용
비동기 처리: asyncio를 활용한 데이터 처리 루프

주요 구현 포인트

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 구독 메시지 형식

업비트 WebSocket API는 다음 형식의 구독 메시지를 요구합니다:

subscribe_message = [
    {"ticket": str(uuid.uuid4())},  # 고유 식별자
    {"type": "ticker", "codes": ["KRW-BTC", "KRW-ETH"]},  # 구독 정보
]

ticket: 요청을 구분하는 고유 ID (UUID 사용 권장)
type: 데이터 타입 (ticker, orderbook, trade 등)
codes: 구독할 마켓 코드 리스트

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. 비동기 데이터 처리

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 키 없이도 사용 가능하지만, 개인 거래 정보에는 필요합니다:

import keyring
from 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: 1
Inserted orderbook data at timestamp 1707393600000 | Count: 1

활용 방안

이 코드를 기반으로 다음과 같은 기능을 추가할 수 있습니다:

데이터베이스 저장: process_data() 메서드에서 MongoDB, PostgreSQL 등에 저장
실시간 알림: 특정 가격 도달 시 알림 전송
차트 시각화: Plotly, Matplotlib 등으로 실시간 차트 업데이트
트레이딩 봇: 가격 변동에 따른 자동 매매 로직 구현
ML 모델 학습: 실시간 데이터를 feature로 활용

전체 코드

import asyncio
import uuid
import websocket
import threading
import json
import time

from keyrings.alt.file import PlaintextKeyring
import keyring
keyring.set_keyring(PlaintextKeyring())


# codes
# https://docs.upbit.com/kr/reference/list-trading-pairs
BITCOIN = "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를 조합하여 안정적인 데이터 처리가 가능합니다
자동 재연결 기능으로 연결 안정성을 확보할 수 있습니다

이 코드를 기반으로 여러분만의 트레이딩 시스템이나 분석 도구를 만들어보세요!