https://www.youtube.com/watch?v=06KeUDR6Y-0
위 영상을 보고 해당 전략에 대해서 백테스팅을 해보고 싶어서 백테스팅을 진행했다.
간단하게 설명하자면
RSI는 Relative Strength Index.
RSI의 계산은
$RSI = 100 - \frac{100}{1 + RS}$
$RS = AU / DU$
여기서 AU, DU는 14일(이 전략은 5분봉 기준이니까 70분) 동안의 평균 상승폭과 평균 하락폭이다.
주로 RSI를 30, 70선을 긋고 이하를 과매도, 과매수 영역이라고 선정하고 그에 따라서 매수, 매도를 진행한다.
Stochastic RSI는 좀더 RSI를 민감하게 만들었다고 생각하면 된다.
지정한 기간(보통 14개) 내의 최고 RSI와 최저 RSI 값을 기준으로 현재 RSI의 상대적 위치를 백분율로 환산한다.
$%K = \frac{RSI_{current} - RSI_{lowest}}{RSI_{higest} - RSI_{lowest}} * 100$
MACD는 (Moving Average Covergence Divergence)로 장기 이평선, 단기 이평선이 수렴과 발산을 계속해서 한다는거를
MACD선은 12EMA - 26EMA로 계산한다. 시그널 선은 MACD의 9EMA로 계산한다.
MACD선이 시그널 선을 돌파하는 것을 골든 크로스로 매수포인트로 잡고 거꾸로 시그널 선밑으로 박는거를 데드 크로스로 매도 포인트로 잡는다.
이 전략에서는 Stochastic RSI를 과매수, 과매도의 기준으로 잡고 RSI를 추세 기준으로 잡는다.
거기에 더 조건을 강화하기 위해서 MACD 골든 크로스를 기준으로 잡는거 같은데.
대충 무슨 느낌인지는 알겠다. Stochastic으로 과매도지만 RSI가 50인 상승 추세라면 다시 재반등을 한다. 거기에 MACD를 사용해서 기준 이평선에서 벗어난 부분만큼 수익을 낸다. 이런생각인거 같은데.
일단 빗썸 api 써서 이더리움 5년치 데이터를 가지고와서 테스트 한 결과
import requests
import pandas as pd
import numpy as np
import mplfinance as mpf
from ta.momentum import RSIIndicator, StochRSIIndicator
from ta.trend import MACD
import matplotlib.pyplot as plt
import os
import time
# 1. 데이터 수집 함수 (Bithumb API 사용 - yfinance는 5분봉 5년치 제공 불가)
def get_bithumb_candles_long_term(market, interval, days_back):
"""
최근 N일치 데이터를 역순으로 수집하여 합치는 함수
"""
url = f"https://api.bithumb.com/v1/candles/minutes/{interval}"
all_candles = []
# 5분봉 기준 하루 288개. 5년이면 약 525,600개
total_needed = (1440 // interval) * days_back
current_to = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
print(f"총 {total_needed:,}개의 캔들을 요청합니다... (약 {days_back}일)")
while len(all_candles) < total_needed:
params = {
"market": market,
"to": current_to,
"count": 200
}
try:
resp = requests.get(url, params=params).json()
if not resp or 'error' in resp:
print(f"더 이상 데이터를 가져올 수 없거나 에러 발생: {resp}")
break
all_candles.extend(resp)
# 가장 오래된 데이터의 시간으로 current_to 갱신
current_to = resp[-1]['candle_date_time_kst'].replace('T', ' ')
if len(all_candles) % 2000 == 0:
print(f"현재 {len(all_candles):,}개 수집 완료...")
# API 과부하 방지 (잠시 대기)
time.sleep(0.05)
except Exception as e:
print(f"API 요청 중 예외 발생: {e}")
break
df = pd.DataFrame(all_candles)
if not df.empty:
df['datetime'] = pd.to_datetime(df['candle_date_time_kst'])
df = df.set_index('datetime').sort_index()
df = df.rename(columns={
'opening_price': 'Open', 'high_price': 'High', 'low_price': 'Low',
'trade_price': 'Close', 'candle_acc_trade_volume': 'Volume'
})
return df[['Open', 'High', 'Low', 'Close', 'Volume']].astype(float)
return df
# 2. 실행 설정
market = "KRW-ETH"
interval = 5
days = 1825 # 5년 (365 * 5)
csv_file = "eth_krw_5m_5y.csv"
# 로컬 캐시 확인
if os.path.exists(csv_file):
print(f"로컬 파일 '{csv_file}'에서 데이터를 로드합니다.")
df = pd.read_csv(csv_file, index_col=0, parse_dates=True)
else:
df = get_bithumb_candles_long_term(market, interval, days)
if not df.empty:
df.to_csv(csv_file)
print(f"데이터 수집 완료 및 '{csv_file}' 저장 완료.")
if df.empty:
print("데이터를 불러오지 못했습니다.")
exit()
# 3. 지표 계산
print("지표 계산 중...")
df['RSI'] = RSIIndicator(close=df['Close'], window=14).rsi()
stoch = StochRSIIndicator(close=df['Close'], window=14, smooth1=3, smooth2=3)
df['StochRSI_k'] = stoch.stochrsi_k() * 100
macd_ind = MACD(close=df['Close'], window_slow=26, window_fast=12, window_sign=9)
df['MACD'] = macd_ind.macd()
df['MACD_signal'] = macd_ind.macd_signal()
df['Golden_Cross_Flag'] = (df['MACD'] > df['MACD_signal']) & (df['MACD'].shift(1) <= df['MACD_signal'].shift(1))
# 4. 진입 신호 (최근 5봉 내 골든크로스 발생 시)
cond_mask = (df['RSI'] >= 50) & (df['StochRSI_k'] <= 30)
df['Recent_Golden_Cross'] = df['Golden_Cross_Flag'].rolling(window=5).max()
df['Entry_Signal'] = np.where(cond_mask & (df['Recent_Golden_Cross'] == 1), 1, 0)
# 5. 백테스팅
print("백테스팅 진행 중...")
initial_capital = 1_000_000
capital, position = initial_capital, 0
trades = []
for i in range(len(df)):
if position == 0 and df['Entry_Signal'].iloc[i] == 1:
lookback = 5
curr_close = df['Close'].iloc[i]
sl = df['Low'].iloc[max(0, i-lookback):i+1].min()
if sl >= curr_close: sl = curr_close * 0.99
risk = curr_close - sl
tp = curr_close + (risk * 1.5)
position = capital / curr_close
capital = 0
trades.append({'type': 'Buy', 'time': df.index[i], 'price': curr_close, 'sl': sl, 'tp': tp})
elif position > 0:
last_trade = trades[-1]
sl, tp = last_trade['sl'], last_trade['tp']
exit_p = 0
if df['Low'].iloc[i] <= sl: exit_p = sl
elif df['High'].iloc[i] >= tp: exit_p = tp
if exit_p > 0:
capital = position * exit_p
position = 0
trades.append({'type': 'Sell', 'time': df.index[i], 'price': exit_p})
final_value = capital + (position * df['Close'].iloc[-1] if position > 0 else 0)
total_return = (final_value - initial_capital) / initial_capital * 100
print("\n" + "="*50)
print(f"--- 5년 백테스트 결과 (5분봉) ---")
print(f"시작일: {df.index[0]}")
print(f"종료일: {df.index[-1]}")
print(f"총 데이터 개수: {len(df):,}개")
print(f"최종 자산: {final_value:,.0f}원")
print(f"누적 수익률: {total_return:.2f}%")
print(f"총 거래 횟수: {len([t for t in trades if t['type'] == 'Sell'])}회")
print("="*50)
일단 현물 롱을 기준으로만 수행하기는 했는데 승률이 그렇게 좋은것 같지는 않다. 내가 뭘 잘못했나? 유튜브 영상이랑 결과가 너무 다른데..
일단 이거는 여기까지 해야겠다.