[알고리즘 트레이딩] MFI, CCI 조합 최적 매개변수 탐색 프로그램

현재 MFI, CCI 조합으로 트레이딩 연습중입니다.

MFI의 매개변수 기본값은 14일, CCI의 매개변수 기본값은 20일입니다. 

MFI, CCI 조합으로 트레이딩 시 기본적인 매매 전략은 아래와 같습니다.

매수 시

- MFI: 20 이하 (과매도 상태), CCI: -100 이하 (과매도 영역)

매도 시

-  MFI: 80 이상 (과매수 상태), CCI: 100 이상 (과매수 영역)

위 매도 전략을 사용할 때 MFI, CCI 기본 매개변수 값들과 잘 들어맞지 않는 경우가 있습니다. 그래서 파이썬을 활용하여 최적의 MFI, CCI 매개변수 값을 찾아주는 프로그램을 작성하였습니다. 60초 단위로 최적의 MFI, CCI 매개변수 값이 계산되어 출력됩니다.

import ccxt
import pandas as pd
import numpy as np
import time
import logging
import argparse
from datetime import datetime
from colorama import init, Fore
import winsound
import matplotlib.pyplot as plt

# 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,
})

# colorama 초기화
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='10m', 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

# MFI 계산
def calculate_mfi(df, period):
    tp = (df['high'] + df['low'] + df['close']) / 3
    mf = tp * df['volume']
    positive_mf = mf.where(tp > tp.shift(1), 0).rolling(window=period).sum()
    negative_mf = mf.where(tp < tp.shift(1), 0).rolling(window=period).sum()
    mfi = 100 - (100 / (1 + positive_mf / negative_mf))
    return mfi

# CCI 계산
def calculate_cci(df, period):
    tp = (df['high'] + df['low'] + df['close']) / 3
    ma = tp.rolling(window=period).mean()
    md = tp.rolling(window=period).apply(lambda x: abs(x - x.mean()).mean())
    cci = (tp - ma) / (0.015 * md)
    return cci

# 신호 계산
def calculate_signals(df, mfi_period, cci_period):
    # 지표 계산
    df['mfi'] = calculate_mfi(df, mfi_period)
    df['cci'] = calculate_cci(df, cci_period)

    # 매수/매도 조건 설정
    buy_signal = (df['mfi'] < 20) & (df['cci'] < -100)
    sell_signal = (df['mfi'] > 80) & (df['cci'] > 100)

    return buy_signal.iloc[-1], sell_signal.iloc[-1]

# 성능 평가
def evaluate_strategy(df, mfi_period, cci_period):
    df['mfi'] = calculate_mfi(df, mfi_period)
    df['cci'] = calculate_cci(df, cci_period)
    df['signal'] = ((df['mfi'] < 20) & (df['cci'] < -100)).astype(int) - ((df['mfi'] > 80) & (df['cci'] > 100)).astype(int)
    df['returns'] = df['close'].pct_change() * df['signal'].shift(1)
    cumulative_returns = (1 + df['returns']).cumprod()
    return cumulative_returns.iloc[-1]

# 파라미터 최적화
def optimize_parameters(symbol, timeframe):
    df = fetch_data(symbol, timeframe)
    mfi_range = range(5, 30, 1)
    cci_range = range(5, 30, 1)
    results = []

    for mfi_period in mfi_range:
        for cci_period in cci_range:
            try:
                performance = evaluate_strategy(df, mfi_period, cci_period)
                results.append((mfi_period, cci_period, performance))
            except Exception as e:
                logging.error(f"Error in evaluation: {e}")

    best_params = max(results, key=lambda x: x[2])
    logging.info(f"최적의 MFI 기간: {best_params[0]}, CCI 기간: {best_params[1]}, 성능: {best_params[2]}")
    return best_params

# 메인 함수
def main(symbol, timeframe):
    full_symbol = f"{symbol}/KRW"
    setup_logging(symbol)
    
    while True:
        try:
            best_params = optimize_parameters(full_symbol, timeframe)
            mfi_period, cci_period = best_params[0], best_params[1]

            df = fetch_data(full_symbol, timeframe)
            buy_signal, sell_signal = calculate_signals(df, mfi_period, cci_period)

            current_mfi = df['mfi'].iloc[-1]
            current_cci = df['cci'].iloc[-1]
            logging.info(f"[{symbol}] 현재 MFI: {current_mfi:.2f}, CCI: {current_cci:.2f}")

            # 신호 처리
            if buy_signal:
                logging.info(f"{Fore.BLUE}BUY SIGNAL 발생!")
                winsound.Beep(1000, 500)
            if sell_signal:
                logging.info(f"{Fore.RED}SELL SIGNAL 발생!")
                winsound.Beep(1000, 1000)

            time.sleep(60)

        except Exception as e:
            logging.error(f"오류 발생: {e}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='업비트 트레이딩 봇')
    parser.add_argument('--symbol', type=str, help='거래할 암호화폐 심볼 (예: BTC)')
    parser.add_argument('--timeframe', type=str, default='10m', help='타임프레임 (기본값: 10m)')
    args = parser.parse_args()

    main(args.symbol, args.timeframe)

CMD 창에서 아래와 같이 입력하면 1분봉 단위에서 MFI, CCI 최적값을 찾게 됩니다.

> python upbit_mfi_cci_best_notify.py --symbol XRP --timeframe 1m

위 코드는 업비트 API를 활용한 알고리즘 트레이닝 입문하시는 저와 같은 초보자를 위한 참고용 코드입니다.

끝.