[파이썬] 업비트 자동매매 봇 만들기 7 - private websocket, Critical Section, 비동기 Lock (Mutex)

거래 정보를 실시간으로 받기 위한 웹소켓을 알아보던 중, private는 realtime밖에 지원을 안 한다는 것을 알게 되었다. 이를 간략하게 정리하면 다음과 같다.

---

Snapshot과 Realtime 데이터

업비트 웹소켓으로 데이터를 전달받는 방식은 2가지가 있다.
이 옵션을 설정할 수 있는 데이터 타입들과 설정할 수 없는 데이터 영역이 나뉘어있다.

스냅샷(Snapshot)

• 요청 당시의 상태를 의미하는 일회성 데이터
• 최초 연결 시 모든 요청 코인에 대한 정보를 한 번에 수신
• 현재가, 체결, 호가, 캔들 영역에서 스냅숏과 실시간 요청을 설정할 수 있다.

실시간(Realtime)

• 요청 정보가 스트림 형태로 지속적으로 제공
• 시세가 변동된 코인이나 거래가 발생한 코인에 대해서만 데이터가 전송
• 내 주문 및 체결, 내 자산의 구독형식은 모두 private형식에 실시간만 지원하게 되어있다.

따라서 주문을 넣고(RestAPI만 지원하므로, RestAPI로 주문), 체결 결과를 확인하는 과정에서 WebSocket이 또 필요하다고 판단됐다.

이 부분은 내 주문 및 체결(MyOrder)로 구독해야 한다. API 레퍼런스 예제 2번에 다음과 같이 명시되어 있는 것을 알 수 있다.

예제 2. 모든 마켓 정보 수신 (codes에 빈 배열)

[
     { "ticket": "test-myOrder" },
     {
       "type": "myOrder",
       "codes": []
     }
]

이를 활용한다면, 모든 거래의 실시간 상황을 한 개의 웹소켓을 활용해서 JSON으로 받을 수 있을 것 같다.

WebSocket - MyOrder 테스트

실제 거래를 통해 어떻게 반환되는지 테스트해 보았다. 다음 조건으로 구매했고, 3개의 JSON이 수신됐다.

구매시간 : 25.02.10.12:12

구매 대상 : DOGE/KRW

구매 가격 : 372.7원

구매 수량 : 25.08990494개

총 구매금액 : 16,805원

아래와 같이 3개의 상태 변화마다 JSON을 수신받을 수 있었다.

1️⃣ wait: 주문이 접수됨 (대기 상태)

2️⃣ trade: 주문이 체결되기 시작함 (부분 체결 또는 완전 체결)

3️⃣ done: 주문이 완전히 체결되고 거래가 완료됨

---

구매 이후, 구매 결과를 공유 자원에 모두 몰아서 관리하면 웹소켓 하나로 구현이 가능할 것이라고 생각했다. 이를 위해 wallet_dict 공유자원을 추가하였다. 구조는 다음과 같다 : {코인명:[구매 평균 가격, 구매 수량]}

MyOrder를 전담하는 websocket에서, status : done인 JSON들만 가져와서 평균 가격과 구매 수량을 저장만 하면 된다(제거는 거래 함수에서 진행).

 

구현을 하던 도중, 예기치 못한 문제가 발생하였다. 파일 구성을 웹소켓 반환 따로, 웹소켓 사용 따로 를 구상했었는데 private 요청은 WebSocketApp으로만 구현이 가능하다는 사실을 깨달았다. 어떤 수를 써도(extra_headers) 구현이 불가능했다. 따라서 websocket을 생성하는 동시에 wallet_dict를 관리하도록 해야만 했다.

구현을 하던 도중, 아주 중요한 문제가 생겼다. WebSocketApp 다음 코드로 안 넘어간다는 것. 스레드로 처리하면 될 것 같다.

또한 데이터가 오직 한 Task에서만 변경된다면 락이 필요 없다는 것을 깨달았다. 하지만 내 코드의 공유자원들 중 한 Task 이상에서 변경이 일어나는 자원은 trading_dict, target_dict, wallet_dict, active_trades(읽기, 쓰기 별개로 이루어짐)이다. 이에 비동기 락을 적용해야 한다. 이를 위해 shared_resources.py파일을 새로 만들었다. (자바에서 이런 문제를 해결한 자료형이 있는데, 파이썬에는 없다고 한다..).

 

2025.02.11.15:30 업비트 측에 문의한 결과, extra_headers가 아니라 additional_headers로 이름이 바뀌었다고 한다. 실제로 바꿔서 접속을 시도하니 바로 연결되는 모습을 볼 수 있다. 하지만 웹소켓을 통한 거래 내역을 확인해 보니, 시장가 거래 1건도 실제로는 2건 3건 이상으로 나뉘어서 거래된다. 이 내용이 웹소켓으로 정확히 오기에, RestAPI를 활용해서 1초마다 내 계좌를 조회해서 매수평균가격과 거래 volume을 가져오는 것이 정확한 정보일 것 같으므로

거래 내역 조회(wallet_dict 업데이트)는 RestAPI를 활용

하기로 하였다. 웹소켓이 빠르긴 하지만, Rest API 또한 충분히 빠르다는 것을 측정을 통해 확인했다.

업비트 API를 활용한 통신 속도 테스트

또한 아래에서 고민한 공유 자원의 크리티컬 섹션에 관한 고민은 그대로기에, 락을 구현하는 것이 더욱 안정적일 것 같다.

아래는 문의한 내용에 대한 답변이다.

async with websockets.connect(url, additional_headers=headers, ssl=ssl_context) as ws

WebsocketApp과 비동기 락 동시 사용 불가, Condition 적용 

(위의 문제 해결로 인해 비동기 접속 가능, 이 부분을 프로그램에 적용하지 않음)

파이썬 비동기 락과 멀티스레드 락을 같이 사용하는 것이 불가능하다는 것을 알았다. indicatior_dict는 쓰기와 읽기가 명확하기에 멀티 스레드를 사용할 수 있지만, 다른 공유자원은 구조상 불가능하다는 것을 깨달았다. 따라서 WebsocketApp을 사용하지 않고, Rest API를 활용해서 1초마다 wallet_dict를 업데이트하기로 하였다. API 요청 제한(주문이 아닌 요청)이 초당 30회이므로, 초당 1번씩 업데이트하는 것은 충분할 것이라고 판단했다.

또한, Condition을 활용하게 된다면 의도적으로 1초간 제한시간을 둔 await asyncio.sleep(1)을 적당히 수정하거나 제거해야 한다.

 

공유 자원 접근 구조와 Lock 적용

공유 자원에 대한 읽기 쓰기 정리

삭제를 하는 부분은 무조건 Lock가 필요하다. 업데이트만 이루어지는 부분(indicator_dict)은 무조건은 필요할 것 같지 않다. 따라서, Lock은 다음 위치에 적용하였다.

• target_dict를 업데이트하거나 삭제하는 부분
• trading_dict의 코인을 새로 등록하거나, 가격 정보를 업데이트하거나, trading_dict에서 코인을 제거할 때
• active_trade set 자료형에 코인을 등록하거나 제거할 때
• wallet_dict 작성할 때

거래 요청

매수와 매도 모두 같은 엔드포인트(https://api.upbit.com/v1/orders)로 POST 요청을 보내기에, order() 주문 함수를 따로 만들었다(하단 코드 참조). 다음 편은 금액이 부족한 경우나 혹은 구매 금액조건 등 실제로 부딪힐 예외사항들을 처리하고 서버에 배포하는 과정이 될 것 같다.

async def order(ACCESS_KEY, SECRET_KEY, type, volume):
    """
    Upbit 시장가 주문 함수.

    매개변수:
      - ACCESS_KEY: Upbit API Access Key
      - SECRET_KEY: Upbit API Secret Key
      - type: 주문 타입 ("bid"이면 매수, "ask"이면 매도)
      - volume: 주문 금액(매수 시) 또는 코인 수량(매도 시)

    리턴:
      - API 응답 JSON (dict)
    """
    # Upbit 주문 API 엔드포인트
    url = "https://api.upbit.com/v1/orders"
    # nonce 생성 (고유값)
    nonce = str(uuid.uuid4())
    # 거래할 마켓 (필요에 따라 변경)
    market = "KRW-BTC"

    # 주문 파라미터 설정 (주문 종류에 따라 price 혹은 volume 사용)
    if type == "bid":
        # 매수(시장가) 주문: 'price' 파라미터에 주문 금액을 지정
        data = {
            "market": market,
            "side": "bid",
            "price": str(volume),  # 주문 금액 (원화)
            "ord_type": "price"  # 시장가 매수 주문은 ord_type이 "price"
        }
    elif type == "ask":
        # 매도(시장가) 주문: 'volume' 파라미터에 주문 수량을 지정
        data = {
            "market": market,
            "side": "ask",
            "volume": str(volume),  # 주문 코인 수량
            "ord_type": "market"  # 시장가 매도 주문은 ord_type이 "market"
        }
    else:
        raise ValueError("유효하지 않은 주문 타입입니다. 'bid' 또는 'ask'여야 합니다.")

    # JWT 페이로드에 필요한 값 설정 (쿼리 스트링 포함)
    payload = {
        "access_key": ACCESS_KEY,
        "nonce": nonce,
        "query": urllib.parse.urlencode(data)
    }

    # JWT 토큰 생성 (HS256 알고리즘 사용)
    token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")
    headers = {"Authorization": f"Bearer {token}"}
    # ssl 적용
    ssl_context = ssl.create_default_context(cafile=certifi.where())
    # aiohttp를 사용하여 비동기로 POST 요청 전송
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=data, headers=headers, ssl=ssl_context) as response:
            # API 응답 JSON 반환
            return await response.json()

2025.02.13 - [자동매매/Upbit] - [파이썬] 업비트 자동매매 봇 만들기 8 - 1차 완료 후 정리. 예외 사항 처리 및 추후 구현 가능 사항들.

[파이썬] 업비트 자동매매 봇 만들기 8 - 1차 완료 후 정리. 예외 사항 처리 및 추후 구현 가능 사