[알고리즘 트레이딩] Multiple Time Frame Momentum Strategy를 활용한 트레이딩 봇 개발
Multiple Time Frame Momentum Strategy를 활용한 트레이딩 봇 개발
최근 암호화폐 시장에서 다중 타임프레임을 활용한 전략이 주목받고 있습니다. 이번 글에서는 Multiple Time Frame Momentum Strategy를 기반으로 한 트레이딩 봇을 개발하는 과정을 공유합니다.
1. 다중 타임프레임 전략이란?
다중 타임프레임(Multiple Time Frame) 분석은 서로 다른 시간대의 차트를 함께 고려하여 매매 신호를 더욱 정교하게 분석하는 기법입니다.
우리는 1시간봉과 10분봉을 활용하여 장기적인 방향성을 확인한 후, 단기적인 매수/매도 타이밍을 포착하는 전략을 구현할 것입니다.
2. 트레이딩 봇의 핵심 로직
본 트레이딩 봇은 Upbit 거래소에서 데이터를 가져와 다음과 같은 기준으로 매매 신호를 생성합니다:
- 1시간봉(60분): 장기적인 트렌드를 확인 (EMA 60과 EMA 120의 관계 분석)
- 10분봉: 단기적인 진입 타이밍 결정 (EMA 20과 EMA 60의 관계 + 볼린저 밴드 %b 활용)
매수 조건:
- 1시간봉에서 EMA(60) > EMA(120) (상승 추세 확인)
- 10분봉에서 EMA(20) > EMA(60)
- 볼린저 밴드의 %b 값이 0.2 이하 (하단 근접)
매도 조건:
- 10분봉에서 EMA(20) < EMA(60) (하락 추세 진입)
3. 코드 구현
아래는 위 전략을 구현한 Python 코드입니다:
import ccxt
import pandas as pd
import time
import logging
import argparse
import numpy as np
from datetime import datetime
from colorama import init, Fore
import winsound
symbol = None
# API 키 불러오기
with open("upbit.txt") as f:
api_key, secret_key = [line.strip() for line in f.readlines()]
exchange = ccxt.upbit({
'apiKey': api_key,
'secret': secret_key,
})
init(autoreset=True)
def setup_logging(symbol):
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()]
)
logging.info(f"{symbol}에 대한 로깅 시작.")
def fetch_data(symbol, timeframe, limit=200):
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_indicators(df):
df['ema_20'] = df['close'].ewm(span=20, adjust=False).mean()
df['ema_60'] = df['close'].ewm(span=60, adjust=False).mean()
df['ema_120'] = df['close'].ewm(span=120, adjust=False).mean()
df['bol_upper'] = df['close'].rolling(window=20).mean() + 2 * df['close'].rolling(window=20).std()
df['bol_lower'] = df['close'].rolling(window=20).mean() - 2 * df['close'].rolling(window=20).std()
df['percent_b'] = (df['close'] - df['bol_lower']) / (df['bol_upper'] - df['bol_lower'])
return df
def detect_signals(df_10m, df_1h):
latest_10m = df_10m.iloc[-1]
latest_1h = df_1h.iloc[-1]
trend_1h = latest_1h['ema_60'] > latest_1h['ema_120']
buy_condition = trend_1h and latest_10m['ema_20'] > latest_10m['ema_60'] and latest_10m['percent_b'] < 0.2
sell_condition = latest_10m['ema_20'] < latest_10m['ema_60']
logging.info(f"[{symbol}] 1시간봉 EMA(60): {latest_1h['ema_60']:.6f}, EMA(120): {latest_1h['ema_120']:.6f}, 트렌드: {'상승' if trend_1h else '하락'}")
logging.info(f"[{symbol}] 10분봉 EMA(20): {latest_10m['ema_20']:.6f}, EMA(60): {latest_10m['ema_60']:.6f}, %b: {latest_10m['percent_b']:.6f}")
if buy_condition:
logging.info(f"[{symbol}] {Fore.GREEN}매수 신호 발생!")
winsound.Beep(1000, 500)
elif sell_condition:
logging.info(f"[{symbol}] {Fore.RED}매도 신호 발생!")
winsound.Beep(500, 500)
def main(symbol):
full_symbol = f"{symbol}/KRW"
try:
df_10m = fetch_data(full_symbol, timeframe='10m')
df_1h = fetch_data(full_symbol, timeframe='1h')
df_10m = calculate_indicators(df_10m)
df_1h = calculate_indicators(df_1h)
detect_signals(df_10m, df_1h)
except Exception as e:
logging.error(f"오류 발생: {e}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='업비트 트레이딩 봇')
parser.add_argument('--symbol', type=str, help='거래할 암호화폐 심볼 (예: BTC)')
args = parser.parse_args()
symbol = args.symbol
setup_logging(symbol)
while True:
main(symbol)
time.sleep(60)
4. 마무리
본 전략은 단순한 기법이지만, 다중 타임프레임을 활용함으로써 더 정교한 트레이딩이 가능합니다. 실전 적용 전에 백테스트를 수행하여 성능을 검증하는 것이 중요합니다.
추후에는 MFI나 CCI 같은 추가적인 모멘텀 지표를 활용하여 전략을 고도화하는 방법도 고려해볼 수 있습니다.
댓글
댓글 쓰기