[알고리즘 트레이딩] 다이버전스 감지 알고리즘

트레이딩에서 다이버전스(Divergence)는 가격과 보조지표(예: RSI, MACD 등)의 움직임이 일치하지 않을 때 발생하는 현상으로, 추세 반전의 신호로 활용됩니다. 이번 글에서는 RSI를 이용한 다이버전스 감지 알고리즘을 소개하고, 이를 트레이딩 봇에 적용하는 방법을 설명하겠습니다.

---

🔹 다이버전스란?

✅ 강세 다이버전스 (Bullish Divergence)

• 가격이 새로운 저점을 형성하지만, RSI는 이전보다 높은 저점을 형성할 때 발생
• 매수 신호로 해석될 수 있음

✅ 약세 다이버전스 (Bearish Divergence)

• 가격이 새로운 고점을 형성하지만, RSI는 이전보다 낮은 고점을 형성할 때 발생
• 매도 신호로 해석될 수 있음

---

🔹 RSI 기반 다이버전스 감지 알고리즘

아래는 업비트(Upbit)에서 데이터를 가져와 RSI를 계산하고, 다이버전스를 감지하는 코드입니다.

import pandas as pd
import numpy as np
import ccxt

def fetch_data(symbol, timeframe, limit=200):
    exchange = ccxt.upbit()  # 업비트 사용
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

def calculate_rsi(df, period=14):
    delta = df['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
    rs = gain / loss
    df['rsi'] = 100 - (100 / (1 + rs))
    return df

def detect_divergence(df):
    df = calculate_rsi(df)
    
    for i in range(2, len(df)-1):
        # 최근 3개의 저점 비교 (강세 다이버전스 감지)
        if (df['close'][i-2] > df['close'][i] and df['rsi'][i-2] < df['rsi'][i]):
            print(f"🔹 강세 다이버전스 감지! {df['timestamp'][i]} 가격: {df['close'][i]:.2f}, RSI: {df['rsi'][i]:.2f}")
        
        # 최근 3개의 고점 비교 (약세 다이버전스 감지)
        if (df['close'][i-2] < df['close'][i] and df['rsi'][i-2] > df['rsi'][i]):
            print(f"🔻 약세 다이버전스 감지! {df['timestamp'][i]} 가격: {df['close'][i]:.2f}, RSI: {df['rsi'][i]:.2f}")

# 테스트 실행
symbol = "BTC/KRW"
df = fetch_data(symbol, "1h", 200)
detect_divergence(df)

---

🔹 알고리즘 설명

1. 업비트에서 OHLCV 데이터(캔들 데이터) 가져오기
2. RSI 계산: 14일 기준 RSI 지표를 계산
3. 다이버전스 감지:
   • 최근 3개의 저점/고점을 비교하여 가격과 RSI가 반대 방향으로 움직이는지 확인
   • 강세 다이버전스 발생 시 매수 신호
   • 약세 다이버전스 발생 시 매도 신호

---

🔹 트레이딩 봇에 적용하는 방법

위 코드를 트레이딩 봇에 적용하려면 detect_signals() 함수 내부에서 다이버전스 감지 기능을 추가하면 됩니다.

📌 적용 예시 (기존 봇 코드에 추가)

def detect_signals(df_10m, df_1h):
    latest_10m = df_10m.iloc[-1]
    latest_1h = df_1h.iloc[-1]
    
    # 기존 트렌드 분석 코드
    trend_10m = latest_10m['ema_60'] > latest_10m['ema_120']
    trend_1h = latest_1h['ema_60'] > latest_1h['ema_120']
    
    # 다이버전스 감지
    detect_divergence(df_10m)
    detect_divergence(df_1h)

---

🔹 확장 가능성

✅ MACD, CCI 등 다른 보조지표 추가 ✅ 머신러닝을 활용한 다이버전스 패턴 학습 ✅ 자동 매매 시스템과 연동하여 실시간 트레이딩 적용

이제 다이버전스 개념과 이를 활용한 알고리즘을 이해하셨다면, 실제 트레이딩 전략에 적용해 보세요! 🚀