바이낸스 데이터를 활용한 Python 기반 자동매매 프로그램 개발과 백테스팅 시스템 구축에 대해 도와드리겠습니다.
추천할 만한 매매 전략으로는 다음과 같은 것들이 있습니다:
- 이동평균선 교차(Moving Average Crossover): 단기 이동평균선이 장기 이동평균선을 상향 돌파할 때 매수, 하향 돌파할 때 매도하는 전략입니다. 비교적 구현이 쉽고 트렌드 추종에 효과적입니다.
- RSI(상대강도지수) 기반 전략: RSI가 30 이하로 떨어질 때 매수, 70 이상 올라갈 때 매도하는 과매수/과매도 지표를 활용합니다.
- 볼린저 밴드 전략: 가격이 하단밴드 아래로 떨어질 때 매수, 상단밴드 위로 올라갈 때 매도하는 전략입니다.
- MACD 교차 전략: MACD 라인이 시그널 라인을 상향 돌파할 때 매수, 하향 돌파할 때 매도합니다.
바이낸스 라이브러리를 사용하는 방법의 장점은:
- 공식 지원: 바이낸스에서 직접 관리하는 라이브러리이므로 API 변경 시 업데이트가 빠릅니다.
- 다양한 기능: 시장 데이터 조회, 주문 생성/취소, 계정 잔액 확인 등 거의 모든 바이낸스 기능을 사용할 수 있습니다.
- WebSocket 지원: 실시간 시장 데이터를 WebSocket을 통해 효율적으로 받을 수 있습니다.
- 안정성: 많은 사용자가 검증한 라이브러리입니다.
실시간 자동매매 시스템을 구현하려면 WebSocket을 활용한 방식을 추천합니다. 이를 통해 실시간으로 시장 변화를 감지하고 매매 신호를 포착할 수 있습니다. 백테스팅 시스템을 확장해서 실시간 트레이딩 시스템으로 만들 때 바이낸스 라이브러리의 WebSocket 기능이 특히 유용합니다.
바이낸스의 WebSocket API를 활용한 실시간 자동매매 시스템입니다. 주요 기능과 사용법을 설명해드리겠습니다:
주요 기능
- 실시간 가격 데이터 수신: WebSocket을 통해 실시간으로 캔들 데이터를 수신합니다.
- 다양한 매매 전략: 이동평균선, RSI, 볼린저 밴드, MACD 등 다양한 전략을 선택할 수 있습니다.
- 테스트 모드: 실제 주문 없이 전략을 테스트할 수 있습니다.
- 자동 재연결: WebSocket 연결이 끊어지면 자동으로 재연결을 시도합니다.
바이낸스 API 키와 시크릿 없이도 자동매매 시스템을 개발하고 테스트할 수 있는 몇 가지 방법이 있습니다:
- 공개 데이터 API 사용: 바이낸스는 시장 데이터를 가져오는 공개 API를 제공합니다. 이는 인증이 필요 없으며 시장 데이터를 실시간으로 가져올 수 있습니다. 단, 실제 주문은 불가능합니다.
- 바이낸스 테스트넷 사용: 바이낸스는 테스트 목적으로 테스트넷을 제공합니다. 여기서 API 키를 얻어 실제 돈을 사용하지 않고 거래를 테스트할 수 있습니다.
- 과거 데이터 기반 시뮬레이션: 과거 데이터를 다운로드하여 시뮬레이션을 실행할 수 있습니다.
import websocket
import json
import pandas as pd
import numpy as np
import threading
import time
from datetime import datetime
class BinanceDataCollector:
def __init__(self, symbol='btcusdt', interval='1m'):
self.symbol = symbol.lower()
self.interval = interval
self.ws = None
self.closes = []
self.highs = []
self.lows = []
self.volumes = []
self.timestamps = []
self.df = pd.DataFrame()
self.max_candles = 300
self.socket_running = False
def on_message(self, ws, message):
msg = json.loads(message)
self.process_candle(msg)
def on_error(self, ws, error):
print(f"WebSocket 오류: {error}")
def on_close(self, ws, close_status_code, close_msg):
print("WebSocket 연결이 종료되었습니다.")
self.socket_running = False
def on_open(self, ws):
print("WebSocket 연결이 시작되었습니다.")
self.socket_running = True
def process_candle(self, msg):
"""실시간 캔들 데이터 처리"""
try:
candle = msg['k']
is_candle_closed = candle['x']
if is_candle_closed:
timestamp = datetime.fromtimestamp(candle['t']/1000)
close = float(candle['c'])
high = float(candle['h'])
low = float(candle['l'])
volume = float(candle['v'])
print(f"캔들 종료: {timestamp} | 종가: {close}")
# 데이터 추가
self.timestamps.append(timestamp)
self.closes.append(close)
self.highs.append(high)
self.lows.append(low)
self.volumes.append(volume)
# 최대 캔들 수 유지
if len(self.closes) > self.max_candles:
self.timestamps.pop(0)
self.closes.pop(0)
self.highs.pop(0)
self.lows.pop(0)
self.volumes.pop(0)
# 데이터프레임 업데이트
self.update_dataframe()
self.analyze_data()
except Exception as e:
print(f"데이터 처리 중 오류 발생: {e}")
def update_dataframe(self):
"""데이터프레임 업데이트"""
self.df = pd.DataFrame({
'timestamp': self.timestamps,
'close': self.closes,
'high': self.highs,
'low': self.lows,
'volume': self.volumes
})
self.df.set_index('timestamp', inplace=True)
# 기술적 지표 추가
if len(self.df) > 50:
# 이동평균선
self.df['sma_20'] = self.df['close'].rolling(window=20).mean()
self.df['sma_50'] = self.df['close'].rolling(window=50).mean()
# RSI
delta = self.df['close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
self.df['rsi'] = 100 - (100 / (1 + rs))
# MACD
self.df['ema_12'] = self.df['close'].ewm(span=12, adjust=False).mean()
self.df['ema_26'] = self.df['close'].ewm(span=26, adjust=False).mean()
self.df['macd'] = self.df['ema_12'] - self.df['ema_26']
self.df['signal'] = self.df['macd'].ewm(span=9, adjust=False).mean()
def analyze_data(self):
"""데이터 분석 및 매매 신호 감지"""
if len(self.df) < 2:
return
last_row = self.df.iloc[-1]
prev_row = self.df.iloc[-2]
# 여기에 매매 신호 분석 로직 추가
# 예: 이동평균선 교차 전략
if 'sma_20' in last_row and 'sma_50' in last_row:
if (prev_row['sma_20'] <= prev_row['sma_50']) and (last_row['sma_20'] > last_row['sma_50']):
print("매수 신호: 이동평균선 골든 크로스")
elif (prev_row['sma_20'] >= prev_row['sma_50']) and (last_row['sma_20'] < last_row['sma_50']):
print("매도 신호: 이동평균선 데드 크로스")
# RSI 기반 전략
if 'rsi' in last_row:
if last_row['rsi'] < 30:
print(f"과매도 구간 (RSI: {last_row['rsi']:.2f})")
elif last_row['rsi'] > 70:
print(f"과매수 구간 (RSI: {last_row['rsi']:.2f})")
# MACD 기반 전략
if 'macd' in last_row and 'signal' in last_row:
if (prev_row['macd'] <= prev_row['signal']) and (last_row['macd'] > last_row['signal']):
print("매수 신호: MACD 골든 크로스")
elif (prev_row['macd'] >= prev_row['signal']) and (last_row['macd'] < last_row['signal']):
print("매도 신호: MACD 데드 크로스")
def start_collecting(self):
"""데이터 수집 시작"""
# WebSocket 연결 설정
socket_url = f"wss://streahttp://m.binance.com:9443/ws/{self.symbol}@kline_{self.interval}"
websocket.enableTrace(False)
self.ws = websocket.WebSocketApp(
socket_url,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
print(f"\n===== 데이터 수집 시작 =====")
print(f"심볼: {self.symbol}")
print(f"간격: {self.interval}")
print("========================\n")
# WebSocket 실행 (별도 스레드)
wst = threading.Thread(target=self.ws.run_forever)
wst.daemon = True
wst.start()
# 메인 스레드에서는 간단한 상태 확인
print("실시간 데이터 수집 시작 (종료하려면 Ctrl+C)")
try:
while True:
if not self.socket_running:
print("WebSocket 연결이 종료되었습니다. 재연결 시도 중...")
wst = threading.Thread(target=self.ws.run_forever)
wst.daemon = True
wst.start()
time.sleep(5)
except KeyboardInterrupt:
print("사용자에 의해 프로그램이 종료되었습니다.")
self.ws.close()
print("데이터 수집 종료")
# 메인 실행 코드
if __name__ == "__main__":
# 설정
symbol = "btcusdt" # 거래 심볼
interval = "1m" # 캔들 간격 (1m, 5m, 15m, 1h, 4h, 1d)
# 데이터 수집기 초기화 및 시작
collector = BinanceDataCollector(symbol=symbol, interval=interval)
collector.start_collecting()