[실무 강의] **Python API 자동매매: 5분 만에 실시간 시세 모니터링 & 첫 조건부 주문 봇 구현하기** 완벽 마스터 튜토리얼

작성자:

학습 목표

안녕하세요! 2026년, 여러분의 IT/코딩 실무 역량을 한 단계 끌어올릴 수석 IT 교육 강사입니다. 이번 강의에서는 Python을 활용하여 자동매매의 핵심 기반인 실시간 시세 모니터링과 간단한 조건부 주문 봇을 단 5분 만에 구현하는 방법을 완벽하게 마스터할 것입니다. 이 강의를 통해 여러분은 다음을 학습하게 됩니다:

  • Python asyncio와 websockets 라이브러리를 이용한 실시간 데이터 스트리밍 처리 방법.
  • HTTP REST API를 활용하여 가상(또는 실제) 거래소에 주문을 전송하는 기본적인 메커니즘.
  • 특정 가격 조건에 따라 자동으로 매매 신호를 발생시키고 주문을 실행하는 논리 구조 설계.
  • 자동매매 시스템 개발을 위한 기초 환경 설정 및 프로젝트 구조화.

이 튜토리얼은 여러분이 복잡한 자동매매 시스템을 구축하기 위한 첫걸음을 내딛는 데 필요한 핵심 기술과 자신감을 제공할 것입니다. 이제 바로 시작해볼까요?

사전 준비 사항

본 실습을 원활하게 진행하기 위해 다음 환경을 미리 설정해 주시기 바랍니다. 최적의 학습 경험을 위해 아래 권장 사항을 따르는 것을 강력히 추천합니다.

  • 운영체제 (OS): Windows 10/11, macOS, 또는 Linux 배포판 (Ubuntu 20.04+ 권장)
  • 통합 개발 환경 (IDE): Visual Studio Code (VS Code) 최신 버전. 확장 프로그램으로 Python Extension 설치 권장.
  • Python 버전: Python 3.9 이상. (현재 2026년 기준, Python 3.11 또는 3.12를 권장하지만, 3.9 이상이면 무방합니다.)
  • 필수 설치 라이브러리: 터미널 또는 명령 프롬프트에서 아래 명령어를 실행하여 필요한 라이브러리를 설치합니다.
# 가상 환경 생성 (선택 사항이지만 강력 권장)
python -m venv venv

# 가상 환경 활성화
# Windows
.\venv\Scripts\activate
# macOS/Linux
source venv/bin/activate

# 필수 라이브러리 설치
pip install requests websockets asyncio

requests 라이브러리는 HTTP 요청을 처리하는 데 사용되며, websockets는 실시간 데이터를 수신하는 WebSocket 연결을 관리합니다. asyncio는 Python의 비동기 프로그래밍을 위한 내장 라이브러리로, websockets와 함께 사용됩니다.

단계별 실습 과정

이제 본격적으로 Python 자동매매 봇을 구현해 보겠습니다. 각 단계를 차근차근 따라오시면 됩니다.

1단계: 프로젝트 환경 설정 및 필수 라이브러리 설치

먼저, 프로젝트를 위한 새로운 디렉토리를 생성하고 VS Code에서 엽니다. 가상 환경을 활성화한 후, 위에서 안내된 필수 라이브러리를 설치합니다. 이 과정은 여러분의 개발 환경을 깨끗하게 유지하고 프로젝트별 의존성 관리를 용이하게 합니다.

# 봇을 위한 새 디렉토리 생성
mkdir python_autotrade_bot
cd python_autotrade_bot

# VS Code로 프로젝트 열기
code .

# 가상 환경 생성 및 활성화 (이미 했다면 건너뛰기)
python -m venv venv
source venv/bin/activate  # macOS/Linux 기준

# 라이브러리 설치 (이미 했다면 건너뛰기)
pip install requests websockets asyncio

환경 설정이 완료되었다면, 이제 봇의 핵심 기능을 구현할 준비가 된 것입니다.

2단계: 실시간 시세 모니터링 구현 (WebSocket)

실시간 시세 모니터링은 자동매매 봇의 가장 중요한 기능 중 하나입니다. 우리는 websockets 라이브러리를 사용하여 가상의 실시간 시세 스트림에 연결하고 데이터를 수신하는 코드를 작성할 것입니다. 실제 거래소 API는 인증 과정이 복잡하므로, 여기서는 개념 이해를 위해 간단한 테스트 WebSocket 서버 또는 공개된 테스트 스트림을 활용하는 예시를 보여드립니다. (예: Binance의 공개 WebSocket 스트림을 단순화하여 사용하거나, 로컬에서 간단한 Echo 서버를 구성하여 테스트할 수 있습니다.)

price_monitor.py 파일을 생성하고 다음 코드를 작성하세요.

import asyncio
import websockets
import json

async def connect_and_monitor():
    # 실제 거래소 WebSocket URL로 대체해야 합니다.
    # 예시: 'wss://stream.binance.com:9443/ws/btcusdt@trade'
    # 여기서는 간단한 테스트를 위해 가상의 URL을 사용합니다.
    # 실제 사용 시에는 해당 거래소의 API 문서를 반드시 참조하세요.
    uri = "wss://echo.websocket.events" # 간단한 테스트용 에코 서버
    # uri = "wss://stream.binance.com:9443/ws/btcusdt@trade" # 실제 바이낸스 BTCUSDT 실시간 거래 스트림 (비인증)

    print(f"WebSocket 연결 시도: {uri}")
    async with websockets.connect(uri) as websocket:
        print("WebSocket 연결 성공!")
        # 바이낸스 예시의 경우, 특정 심볼 구독 메시지를 보낼 수 있습니다.
        # await websocket.send(json.dumps({"method": "SUBSCRIBE", "params": ["btcusdt@trade"], "id": 1}))
        # print("구독 메시지 전송 완료.")

        while True:
            try:
                message = await websocket.recv()
                data = json.loads(message)
                # 여기서는 수신된 메시지를 단순히 출력합니다.
                # 실제로는 이 데이터를 파싱하여 가격, 거래량 등을 추출합니다.
                if 'p' in data: # 바이낸스 @trade 스트림의 가격 필드 (p)
                    print(f"실시간 가격: {data['p']}")
                else:
                    print(f"수신 데이터: {data}")
            except websockets.exceptions.ConnectionClosed: # 연결 끊김 처리
                print("WebSocket 연결이 종료되었습니다. 재연결 시도...")
                break
            except Exception as e: # 기타 예외 처리
                print(f"데이터 수신 중 오류 발생: {e}")
                await asyncio.sleep(1) # 잠시 대기 후 재시도

async def main():
    await connect_and_monitor()

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

이 코드는 asyncio를 사용하여 비동기적으로 WebSocket 서버에 연결하고, 수신되는 메시지를 지속적으로 출력합니다. uri 변수를 실제 거래소의 WebSocket 스트림 URL로 변경하면 실시간 시세를 받아볼 수 있습니다. 반드시 해당 거래소의 API 문서를 참조하여 올바른 URL과 메시지 형식을 사용해야 합니다.

위 코드를 실행하면 터미널에 실시간으로 수신되는 시세 데이터가 출력되는 것을 확인할 수 있습니다. 이는 여러분의 봇이 시장의 움직임을 실시간으로 감지할 준비가 되었음을 의미합니다.

3단계: 첫 조건부 주문 로직 구현 (REST API)

이제 실시간 시세를 기반으로 특정 조건이 충족되었을 때 주문을 실행하는 로직을 만들어 보겠습니다. 주문 실행은 주로 HTTP REST API를 통해 이루어집니다. 여기서는 ‘특정 가격 이상일 때 매수 주문을 시뮬레이션’하는 간단한 조건을 구현합니다. 실제 주문 API는 인증(API Key, Secret Key를 이용한 서명 등)이 필요하므로, 여기서는 requests 라이브러리를 사용하여 가상의 주문 요청을 보내는 것으로 대체합니다.

order_bot.py 파일을 생성하고 다음 코드를 작성하세요.

import requests
import json
import time

# 실제 거래소 API 엔드포인트와 인증 정보로 대체해야 합니다.
# 예시: Upbit, Binance, Bithumb 등
API_URL = "https://api.example.com/v1/order" # 가상의 주문 API URL
API_KEY = "YOUR_API_KEY"
SECRET_KEY = "YOUR_SECRET_KEY"

def place_conditional_order(current_price, condition_price, order_type, quantity):
    if order_type == "buy" and current_price >= condition_price:
        print(f"[조건 충족] 현재 가격 {current_price}가 조건 가격 {condition_price} 이상입니다. 매수 주문 시도...")
        # 실제 주문 요청을 위한 데이터 구성
        order_payload = {
            "symbol": "BTC/KRW",
            "side": "buy",
            "type": "market", # 시장가 주문
            "quantity": quantity,
            "price": current_price # 시장가 주문이므로 실제 가격은 중요치 않을 수 있으나 예시로 포함
        }
        headers = {
            "Content-Type": "application/json",
            "X-API-KEY": API_KEY, # 실제 API에서는 서명(Signature) 방식이 주로 사용됩니다.
            "X-SECRET-KEY": SECRET_KEY # 보안을 위해 환경 변수로 관리하는 것이 좋습니다.
        }

        try:
            # 실제 API 요청 대신 시뮬레이션
            # response = requests.post(API_URL, headers=headers, data=json.dumps(order_payload))
            # response.raise_for_status() # HTTP 오류 발생 시 예외 발생
            # print(f"주문 성공! 응답: {response.json()}")
            print(f"[주문 시뮬레이션 성공] 매수 주문 완료: {order_payload}")
            return True
        except requests.exceptions.RequestException as e:
            print(f"[주문 시뮬레이션 실패] 주문 요청 중 오류 발생: {e}")
            return False
    elif order_type == "sell" and current_price <= condition_price:
        # 매도 조건 로직 추가 가능
        print(f"[조건 충족] 현재 가격 {current_price}가 조건 가격 {condition_price} 이하입니다. 매도 주문 시도...")
        print(f"[주문 시뮬레이션 성공] 매도 주문 완료: {{'symbol': 'BTC/KRW', 'side': 'sell', 'quantity': {quantity}}})")
        return True
    else:
        print(f"[조건 미충족] 현재 가격 {current_price}, 조건 가격 {condition_price}. 주문 대기 중...")
        return False

# --- 테스트를 위한 임시 실행 부분 (실제 봇에서는 통합됩니다) ---
if __name__ == "__main__":
    test_current_price = 70000000 # 가상 현재 가격
    test_condition_price = 70500000 # 매수 조건 가격
    test_quantity = 0.001 # 주문 수량

    print("\n--- 조건부 주문 로직 테스트 시작 ---")
    # 조건이 충족되지 않을 때
    place_conditional_order(test_current_price, test_condition_price, "buy", test_quantity)
    time.sleep(1)

    # 조건을 충족하는 가격 가정
    test_current_price = 70600000
    place_conditional_order(test_current_price, test_condition_price, "buy", test_quantity)
    print("--- 조건부 주문 로직 테스트 종료 ---")

이 코드는 place_conditional_order 함수를 정의하여 특정 조건(현재 가격이 설정된 조건 가격 이상)을 만족할 때 매수 주문을 시뮬레이션합니다. API_URL, API_KEY, SECRET_KEY는 여러분이 사용할 실제 거래소의 정보로 반드시 변경해야 하며, 보안을 위해 환경 변수나 별도의 설정 파일로 관리하는 것을 강력히 권장합니다.

위 스크립트를 직접 실행하여 조건부 주문 로직이 어떻게 작동하는지 테스트해 볼 수 있습니다. 현재 가격을 변경하면서 조건이 충족될 때와 그렇지 않을 때의 출력을 확인해 보세요. 이 로직은 나중에 실시간 시세 모니터링 봇과 통합될 것입니다.

4단계: 자동매매 봇 통합 및 실행

이제 2단계에서 구현한 실시간 시세 모니터링 기능과 3단계에서 구현한 조건부 주문 로직을 하나의 자동매매 봇으로 통합할 차례입니다. asyncio를 활용하여 비동기적으로 시세를 모니터링하면서, 동시에 조건이 충족되면 주문 함수를 호출하도록 구현합니다. bot.py 파일을 생성하고 다음 코드를 작성하세요.

import asyncio
import websockets
import json
import requests

# --- 설정 변수 (실제 사용 시 외부 파일 또는 환경 변수로 관리) ---
# WebSocket 설정 (실제 거래소의 공개 스트림 URL로 대체)
WS_URI = "wss://stream.binance.com:9443/ws/btcusdt@trade" # 바이낸스 BTCUSDT 실시간 거래 스트림
# WS_URI = "wss://echo.websocket.events" # 테스트용 에코 서버 (위 코드처럼 동작 확인용)
TARGET_SYMBOL = "btcusdt" # 모니터링할 심볼

# REST API 설정 (실제 거래소의 주문 API URL과 인증 정보로 대체)
ORDER_API_URL = "https://api.example.com/v1/order" # 가상의 주문 API URL
API_KEY = "YOUR_API_KEY"
SECRET_KEY = "YOUR_SECRET_KEY"

# 자동매매 조건 설정
CONDITION_PRICE = 70000.0 # 예시: 70,000 USDT 이상일 때 매수
ORDER_QUANTITY = 0.001 # 주문 수량 (예: 0.001 BTC)

# --- 전역 변수 (간단한 예시를 위해 사용, 실제는 더 견고한 상태 관리 필요) ---
last_price = 0.0
order_placed = False # 주문이 이미 실행되었는지 추적

async def price_monitor_task():
    global last_price
    print(f"WebSocket 연결 시도: {WS_URI}")
    while True:
        try:
            async with websockets.connect(WS_URI) as websocket:
                print("WebSocket 연결 성공!")
                # 바이낸스 예시의 경우, 특정 심볼 구독 메시지를 보낼 수 있습니다.
                # await websocket.send(json.dumps({"method": "SUBSCRIBE", "params": [f"{TARGET_SYMBOL}@trade"], "id": 1}))
                # print("구독 메시지 전송 완료.")

                while True:
                    message = await websocket.recv()
                    data = json.loads(message)
                    # 바이낸스 @trade 스트림의 가격 필드 'p'
                    if 'p' in data:
                        current_price = float(data['p'])
                        last_price = current_price
                        # print(f"실시간 가격: {current_price}") # 너무 많은 출력을 피하기 위해 주석 처리
                    await asyncio.sleep(0.1) # 과도한 CPU 사용 방지
        except websockets.exceptions.ConnectionClosed as e:
            print(f"WebSocket 연결 종료: {e}. 5초 후 재연결 시도...")
            await asyncio.sleep(5)
        except Exception as e:
            print(f"WebSocket 모니터링 중 오류 발생: {e}. 5초 후 재시도...")
            await asyncio.sleep(5)

def place_order_sync(current_price, condition_price, order_type, quantity):
    global order_placed
    if order_placed: # 이미 주문이 실행되었다면 다시 실행하지 않음
        return

    if order_type == "buy" and current_price >= condition_price:
        print(f"[조건 충족] 현재 가격 {current_price}가 조건 가격 {condition_price} 이상입니다. 매수 주문 시도...")
        order_payload = {
            "symbol": "BTC/USDT", # 실제 거래소의 심볼 형식에 맞게 변경
            "side": "buy",
            "type": "market",
            "quantity": quantity,
        }
        headers = {
            "Content-Type": "application/json",
            "X-API-KEY": API_KEY,
            "X-SECRET-KEY": SECRET_KEY
        }

        try:
            # 실제 API 요청 대신 시뮬레이션
            # response = requests.post(ORDER_API_URL, headers=headers, data=json.dumps(order_payload))
            # response.raise_for_status()
            # print(f"주문 성공! 응답: {response.json()}")
            print(f"[주문 시뮬레이션 성공] 매수 주문 완료: {order_payload}")
            order_placed = True # 주문 실행 플래그 설정
        except requests.exceptions.RequestException as e:
            print(f"[주문 시뮬레이션 실패] 주문 요청 중 오류 발생: {e}")
        except Exception as e:
            print(f"알 수 없는 오류 발생: {e}")

async def order_logic_task():
    global last_price, order_placed
    while True:
        if last_price > 0 and not order_placed: # 가격 데이터가 있고, 아직 주문이 실행되지 않았다면
            place_order_sync(last_price, CONDITION_PRICE, "buy", ORDER_QUANTITY)
        await asyncio.sleep(1) # 1초마다 조건 확인

async def main():
    # 두 개의 비동기 태스크를 동시에 실행
    await asyncio.gather(
        price_monitor_task(),
        order_logic_task()
    )

if __name__ == "__main__":
    print("자동매매 봇 시작 (2026년 버전)")
    asyncio.run(main())

이 통합된 봇은 price_monitor_task를 통해 실시간 시세를 받아 last_price 변수에 저장하고, order_logic_task는 이 last_price를 주기적으로 확인하여 설정된 CONDITION_PRICE를 만족하면 place_order_sync 함수를 호출하여 주문을 시도합니다. order_placed 플래그를 사용하여 한 번 주문이 실행되면 중복 주문을 방지합니다.

주의: 위 코드는 학습 및 테스트 목적이며, 실제 자동매매 시스템으로 사용하기 위해서는 훨씬 더 많은 예외 처리, 보안 강화, 로깅, 상태 관리, 그리고 실제 거래소 API에 맞는 인증 및 주문 로직 구현이 필요합니다. 특히 API Key와 Secret Key는 절대 코드에 직접 노출하지 않고 환경 변수나 보안 설정 파일을 통해 관리해야 합니다.

이제 터미널에서 python bot.py 명령어를 실행하여 여러분의 첫 번째 자동매매 봇을 가동해 보세요! 실시간 시세가 모니터링되고, 조건이 충족되면 주문 시뮬레이션 메시지가 출력되는 것을 확인할 수 있을 것입니다.

결과 확인

bot.py를 성공적으로 실행했다면, 터미널에 다음과 같은 출력이 나타나는 것을 확인할 수 있습니다.

  1. WebSocket 연결 성공 메시지: 봇이 실시간 시세 스트림에 성공적으로 연결되었음을 나타냅니다.
  2. 실시간 가격 모니터링 (내부): price_monitor_task가 백그라운드에서 지속적으로 최신 가격을 업데이트하고 있습니다. (콘솔 출력은 주석 처리되어 있습니다만, 필요하다면 주석을 해제하여 확인 가능합니다.)
  3. 조건부 주문 시뮬레이션 메시지: order_logic_task가 last_price를 확인하여 CONDITION_PRICE(예: 70000.0)를 초과하는 순간, [조건 충족] 메시지와 함께 [주문 시뮬레이션 성공] 메시지가 출력됩니다. 이는 봇이 설정된 조건에 따라 주문을 실행할 준비가 되었음을 의미합니다.

만약 CONDITION_PRICE를 현재 시장 가격보다 낮게 설정하여 즉시 조건이 충족되도록 테스트할 수도 있습니다. 예를 들어, CONDITION_PRICE = 100.0 등으로 설정하면 봇이 시작되자마자 조건이 충족되어 주문 시뮬레이션이 실행되는 것을 볼 수 있습니다.

이 튜토리얼을 통해 여러분은 Python을 활용한 자동매매 봇의 기본적인 아키텍처와 구현 방법을 익혔습니다. 실시간 데이터 처리, 조건부 로직, 그리고 외부 API 연동의 핵심 원리를 이해하게 된 것입니다. 다음 단계로는 실제 거래소 API 연동, 보다 복잡한 매매 전략 구현, 백테스팅, 그리고 견고한 에러 처리 및 로깅 시스템 구축 등을 고려해 볼 수 있습니다. 여러분의 자동매매 여정에 이 강의가 큰 도움이 되기를 바랍니다!

댓글 남기기