[파이썬] 업비트 자동매매 봇 만들기 5 - asyncio, async, await, 비동기 작업

이제 프로그램과 API 통신, WebSocket이 구체적으로 어떻게 몇 초 몇 분 간격으로 이루어지는지 구상하고 구현해야 한다. 기준 시간은 다음과 같이 정했고, 추가적인 시퀀스 다이어그램을 통해 구조화하였다.

통신 기준

• 분봉 기준의 거래를 구상하고 있기에, 타깃 코인 구분 절차는 1분마다 업데이트 진행
• {"코인명" : "현재가"} 구조인 target_dict는 1초마다 업데이트 진행

1분 반복과 1초 반복은 비동기 함수를 각각 독립적으로 실행해야 한다.

asyncio 활용

비동기 작업과 동시 실행을 위해 아래 2가지 개념을 활용하여 비동기 작업을 구현하였다.

 

1. asyncio란?

Python에서 비동기 프로그래밍을 지원하는 주요 요소로, 단일 스레드 환경에서 비동기 작업을 효율적으로 처리

• asyncio의 핵심은 이벤트 루프로, 비동기 작업들을 동시에 처리하기 위해 사용
• 이벤트 루프는 비동기 함수가 실행될 준비가 되었을 때 작업을 스케줄링하고, 작업이 완료되었을 때 결과를 반환
• 태스크(Task): 비동기 함수는 asyncio.create_task() 또는 asyncio.gather()와 함께 실행되어 태스크로 관리
• 협업적 멀티태스킹: asyncio는 실제 멀티스레딩을 사용하지 않고, 단일 스레드에서 작업들을 비동기적으로 처리

2. await란?

await는 비동기 함수 내에서 비동기 작업이 완료될 때까지 대기하게 하는 키워드
C언어에서의 wait과는 반대로, await 선언된 작업은 완료될 때까지 다른 작업이 실행됨

await를 사용하면, 해당 작업이 완료될 때까지 다른 작업이 실행될 수 있도록 제어를 이벤트 루프에 넘겨주고, 해당 작업이 끝난 후 그 결과를 반환받습니다.

• 비동기 함수에서 await는 다른 비동기 함수가 완료될 때까지 블로킹을 방지하고, 제어를 이벤트 루프에 넘겨줍니다.

• await를 사용하면, 다른 작업이 실행될 수 있기 때문에 효율적인 작업 흐름이 가능하게 됩니다.

---

asyncio.create_task를 통해 두 개의 비동기 작업을 생성하고, asyncio.gather를 통해 동시에 실행시켰다.

하단 코드 실행 결과

print문과 함께 실행시킨 결과, 1초마다 타깃 코인의 현재가를 업데이트하면서 동시에 1분마다 타깃 코인 업데이트를 진행하게 되는 것을 확인할 수 있다. 이제 거래 요청과 거래 완료시 target_dict에서 타겟 코인을 제거하는 거래 도메인만 추가하면 된다.

async def websocket_connect():
    """
    웹소켓에 연결하여 연결 객체를 반환합니다.
    """
    url = "wss://api.upbit.com/websocket/v1"
    ssl_context = ssl.create_default_context(cafile=certifi.where())
    try:
        websocket = await websockets.connect(url, ssl=ssl_context, compression="deflate")
        print(f"[{datetime.now()}] ✅ 웹소켓 연결 성공!")
        return websocket
    except Exception as e:
        print(f"[{datetime.now()}] ❌ 웹소켓 연결 실패: {e}")
        return None

async def update_target_prices(target_dict, websocket):
    """
    웹소켓을 사용해 target_dict에 있는 코인들의 가격(trade_price)을 1초마다 업데이트합니다.
    """
    while True:
        symbols = list(target_dict.keys())
        if symbols:
            subscribe_msg = json.dumps([
                {"ticket": "ticker_websocket"},
                {"type": "ticker", "codes": symbols, "is_only_snapshot": True},
                {"format": "DEFAULT"}
            ])
            try:
                await websocket.send(subscribe_msg)
                print(f"[{datetime.now()}] 구독 메시지 전송: {subscribe_msg}")
            except Exception as e:
                print(f"[{datetime.now()}] ❌ 구독 메시지 전송 실패: {e}")
                await asyncio.sleep(1)
                continue

            # 구독한 코인 수만큼 응답 받기
            for _ in range(len(symbols)):
                try:
                    response = await websocket.recv()
                    data = json.loads(response)
                    code = data.get("code")
                    trade_price = data.get("trade_price")
                    if code and trade_price is not None:
                        target_dict[code] = trade_price
                        print(f"[{datetime.now()}] Updated {code}: {trade_price}")
                except Exception as e:
                    print(f"[{datetime.now()}] ⚠️ 웹소켓 데이터 수신 오류: {e}")
        else:
            print(f"[{datetime.now()}] 업데이트할 타겟 코인이 없습니다.")

        # 1초마다 반복
        await asyncio.sleep(1)

async def update_indicators_periodically(indicators_dict, target_dict):
    """
    매 분(분이 바뀔 때) 동기 함수인 지표 계산 및 타겟 분류를 실행합니다.
    asyncio.to_thread를 사용하여 해당 작업이 진행되는 동안에도 이벤트 루프의 다른 작업(예: target_dict 업데이트)은 계속됩니다.
    """
    previous_minute = datetime.now().minute
    while True:
        # 지표 계산 및 조건에 따른 타겟 분류

async def main():
    # 공유 데이터 구조
    indicators_dict = {}
    target_dict = {}

    # 웹소켓 연결 (한번만 생성)
    websocket = await websocket_connect()
    if websocket is None:
        print("웹소켓 연결 실패. 프로그램 종료.")
        return

    # 두 작업을 각각 별도의 Task로 실행:
    # 1) 매 분마다 지표 계산 및 타겟 분류 (비동기로 실행되며, 별도 스레드에서 동작)
    # 2) 1초마다 웹소켓을 통해 target_dict 업데이트
    task_indicators = asyncio.create_task(update_indicators_periodically(indicators_dict, target_dict))
    task_prices = asyncio.create_task(update_target_prices(target_dict, websocket))

    # 두 작업을 동시에 실행 (두 작업 모두 무한 루프로 실행됨)
    await asyncio.gather(task_indicators, task_prices)

if __name__ == "__main__":
    asyncio.run(main())

2025.02.10 - [자동매매/Upbit] - [파이썬] 업비트 자동매매 봇 만들기 6 - 거래중인 코인 갯수 유지(무분별한 코인 거래 방지)

[파이썬] 업비트 자동매매 봇 만들기 6 - 거래중인 코인 갯수 유지(무분별한 코인 거래 방지)