[Python] 업비트 자동매매 봇 만들기 1 - 업비트의 통신 방식

미국 대선이 끝난 후에 향방을 보니...(중략)... 코인을 해야 함을 느꼈다.
코인은 주식에 비해 펀더멘탈적인 요소가 덜 반영되기에, 차트 기반 기계적인 매수 매도로 돈을 벌어야 한다고 생각한다.
그러나 코인시장은 24시간 265일이다. 예전에 코인 하면서 5분에 한 번씩 확인하러 들어가고 신경 쓰여서 일상생활을 못했던 경험들을 떠올리면,,, 코인을 직접 하는 건 기회비용이 너무 높다고 생각이 들었다.
그래서 든 생각이 일정조건에서 코인을 자동으로 사고 팔게끔 하면 되겠는데??이고, 따라서 이 프로젝트(?)를 진행하게 되었다.
https://docs.upbit.com/reference

[업비트 개발자 센터

docs.upbit.com](https://docs.upbit.com/)

업비트 API 사이트에 접속하면 크게 2가지 방식을 사용한다.
일단 2가지 방법을 알아보자.

---

WebSocket과 Rest API

개요

1. REST API
• 요청할 때마다 새로운 연결을 맺고 응답을 받아야 함
• 비동기 처리 가능하지만 실시간성이 부족함
• 평균 100~200ms (HTTP 오버헤드 포함)
2. WebSocket
• 서버와 클라이언트 간 실시간 양방향 통신이 가능
• 한 번 연결되면 지속적으로 데이터를 주고받을 수 있음
• 서버에서 데이터가 변경되면 즉시 클라이언트에 전송됨
• 평균 1~10ms (지속적 연결)

사용법

네트워크 프로그래밍 수업을 들으며 배운 것인데, WebSocket은 하나의 소켓이다. 소켓을 사용하는 순서는 다음과 같다.

1. 소켓 연결을 시도(handshake)해서 소켓 객체를 가진다
2. 해당 객체를 통해 통신을 시도한다(Upbit는 JSON 객체를 사용)

반면, Rest API는 HTTP요청 GET, POST을 활용해 독립적인 요청을 보낸다.

업비트에서의 요청 제한

WebSocket : 실시간 시세 정보- 초당 5회, 분당 100회 주문 내역
REST API : 주문 - 초당 8회, 일괄취소 - 2초에 1회, 이외 - 초당 30회 (캔들 조회의 경우 초당 10회)

업비트 WebSocket 요청 종류

subscribe_message = [
        {"ticket": "UNIQUE_TICKET"},
        {"type": "ticker", "codes": ["KRW-BTC"]}
    ]    websocket_요청 내용

type의 value값을 오른쪽 6개 값 중 하나로 지정해서 사용한다. 거래 봇이 코인을 2개, 3개,,, 동시에 거래할 때 코인마다 WebSocket을 열어야 하나? 그러면 웹소켓을 무한정 열어야 하나? 이를 해결하기 위해 일단 얼마나 많은 WebSocket 연결을 할 수 있나 알아야 한다.

---

업비트에서 제공하는 값은 요청 수 밖에 없다. 따라서, 직접 한 번에 연결해 보며 몇 개까지 연결이 수월한지 알아보았다. 아래 코드를 작동시키며 직접 num_connections 숫자를 조절해 가면서 테스트해 본 결과 5개 정도 연결이 수월하게 가능했다. 따라서 전역으로 웹소켓 연결을 4~5개 정도 유지하고, n개의 코인에서 같은 websocket를 사용하기로 하였다.

import asyncio
import websockets
import ssl
import certifi
import json

# 업비트 WebSocket URL
UPBIT_WS_URL = "wss://api.upbit.com/websocket/v1"

# SSL 컨텍스트 설정
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.load_verify_locations(cafile=certifi.where())

# WebSocket을 통해 데이터 수신
async def connect_to_websocket(connection_id):
    try:
        async with websockets.connect(UPBIT_WS_URL, ssl=ssl_context) as websocket:
            # WebSocket에 구독 요청 보내기
            subscribe_msg = [
                {"ticket": f"test-{connection_id}"},
                {"type": "ticker", "codes": ["KRW-BTC"]}  # 비트코인 가격을 구독
            ]
            await websocket.send(json.dumps(subscribe_msg))
            print(f"WebSocket {connection_id} connected and subscribed.")

            # 응답 수신 대기
            while True:
                response = await websocket.recv()
                print(f"WebSocket {connection_id} received data: {response}")
    except Exception as e:
        print(f"WebSocket {connection_id} error: {e}")

# 여러 WebSocket 연결 실행
async def main():
    # 연결할 WebSocket 개수 설정
    num_connections = 5  # 원하는 개수로 조정
    tasks = [connect_to_websocket(i) for i in range(num_connections)]
    await asyncio.gather(*tasks)

# 이벤트 루프 실행
if __name__ == "__main__":
    asyncio.run(main())

하지만 여기서 든 또 다른 생각..

병렬적으로 사용하는 코인들에 전부 같은 웹 소켓을 사용하면
요청-응답 상황에서 Race Condition이 발생하지는 않을까??

파이썬과 해당 언어의 라이브러리의 특성에 대해선 아직 잘 알지 못한다. 따라서 이 부분에 대한 해답을 찾기 위해 AI의 추론 모델들을 활용해 1. 단일 소켓과 다중 소켓이 유리한지, 2. 코인들을 탐색해서 병렬적으로 거래를 처리하려고 할때 요청-응답 순서 등이 일치하지 않을 경우를 물어봤고 응답들을 종합해 요약한 결과는 다음과 같다.
(여담으로, 2025년 1월 29일 기준 DeepSeek가 장안의 화제여서 사용해보았다. 그러나 이번 질문을 했을때 o1이 가장 나에게 도움이 되는 답변을 해주는 걸 보니 ChatGPT가 아직은 가장 좋은 것 같다)

1. 단일소켓 vs 다중소켓

• 단일 소켓
  • 단일 소켓이 끊기면 모든 접속이 불가능
  • 수많은 구독(수많은 코인 종목)에 대한 데이터가 집중, 한쪽에서 지연이 발생하면 다른 기능도 처리 지연이 발생
• 다중 소켓
  • 서버나 네트워크 리소스 사용량이 상대적으로 증가

2. 레이스 컨디션 여부 (병렬적으로 작동되는 매매 로직에 전부 같은 웹 소켓으로 서버와 통신할 때)

질문에 대한 부연 설명을 하자면 다음과 같다. 예를들어 BTC와 ETH에 대해 병렬 스레드(프로세스)가 동시에 작동한다고 생각하자. 이때 현재가 소켓과 캔들 소켓을 두 스레드(프로세스)가 같은 걸 사용하면, 요청을 보내고 응답을 보낼 때 서로 섞이는 것이 아닐까?라는 의문이다.

A : 일반적으로 라이브러리(또는 네트워크 세션) 단에서 소켓 인스턴스별로 이벤트 처리가 구분되어 일어나므로, 각 소켓의 이벤트가 다른 소켓에 섞여 들어가는 일은 거의 없다.

라이브러리 혹은 네트워크 단에서 WebSocket Multiplexing 을 구현한다고 한다(한 소켓을 나누는 것). 따라서 요청-응답이 섞이는 일을 고려할 필요는 없어 보인다. 웹소켓 멀티플렉싱에 대해 추가적으로 공부해 보아야겠다.

추가적으로, 아래 코드를 통해 업비트 웹소켓이 멀티플렉싱을 직접 구현해보았다.

import asyncio
import json
import threading
import websockets
from queue import Queue

UPBIT_WS_URL = "wss://api.upbit.com/websocket/v1"

class WebSocketManager:
    """ 단일 웹소켓 연결을 유지하며 모든 데이터를 수신 및 분배하는 클래스 """
    def __init__(self, tickers, data_queue):
        self.tickers = tickers
        self.data_queue = data_queue  # 공용 큐 (스레드와 공유)
        self.loop = asyncio.get_event_loop()
        self.ws_task = None  # 웹소켓 실행 태스크

    async def connect_ws(self):
        """ 웹소켓을 연결하고 데이터 수신 후 큐에 삽입 """
        async with websockets.connect(UPBIT_WS_URL) as websocket:
            subscribe_msg = [
                {"ticket": "single_ws"},
                {"type": "ticker", "codes": self.tickers},
                {"format": "DEFAULT"}
            ]
            await websocket.send(json.dumps(subscribe_msg))
            print(f"[WebSocket] Subscribed to: {self.tickers}")

            while True:
                try:
                    data = await websocket.recv()
                    parsed_data = json.loads(data)
                    self.data_queue.put(parsed_data)  # 모든 데이터를 큐에 삽입

                except websockets.exceptions.ConnectionClosed:
                    print("[WebSocket] Connection lost, reconnecting...")
                    await self.connect_ws()  # 자동 재연결

    def start(self):
        """ 웹소켓을 실행하는 스레드 생성 """
        self.ws_task = threading.Thread(target=self.loop.run_until_complete, args=(self.connect_ws(),))
        self.ws_task.start()

class DataProcessor(threading.Thread):
    """ 개별 코인 티커를 필터링하여 처리하는 클래스 """
    def __init__(self, coin_list, data_queue):
        super().__init__()
        self.coin_list = set(coin_list)  # 해당 스레드가 처리할 코인 목록
        self.data_queue = data_queue

    def run(self):
        """ 큐에서 데이터를 가져와 자신이 처리할 코인만 필터링 """
        while True:
            data = self.data_queue.get()
            ticker = data.get("code")
            if ticker in self.coin_list:
                print(f"[Processor-{self.name}] {ticker}: {data['trade_price']}")  # 필요한 데이터만 출력

if __name__ == "__main__":
    # 1. 구독할 전체 코인 목록
    all_tickers = ["KRW-BTC", "KRW-ETH", "KRW-XRP", "KRW-ADA", "KRW-DOT", "KRW-LINK"]

    # 2. 각 스레드가 처리할 코인 그룹
    coin_groups = [
        ["KRW-BTC", "KRW-ETH"],  # 첫 번째 스레드 그룹
        ["KRW-XRP", "KRW-ADA"],  # 두 번째 스레드 그룹
        ["KRW-DOT", "KRW-LINK"]  # 세 번째 스레드 그룹
    ]

    # 3. 공용 큐 생성 (웹소켓과 프로세서 스레드 간 공유)
    shared_queue = Queue()

    # 4. 단일 웹소켓을 실행하는 WebSocketManager 시작
    ws_manager = WebSocketManager(all_tickers, shared_queue)
    ws_manager.start()

    # 5. 여러 개의 프로세서 스레드 실행 (필요한 데이터만 가져감)
    processors = [DataProcessor(coins, shared_queue) for coins in coin_groups]
    for processor in processors:
        processor.start()

 

2025.01.29 - [자동매매/Upbit] - [Python] 업비트 자동매매 봇 만들기 2 - JSON통신 압축

[Python] 업비트 자동매매 봇 만들기 2 - JSON통신 압축