model/src/data/technical_indicators.py

"""
기술적 지표 계산 관련 함수 모음
"""

import numpy as np
import pandas as pd

def calculate_ema_series(series, span):
    """주어진 Series에 대해 지수이동평균(EMA)을 계산"""
    return series.ewm(span=span, adjust=False).mean()

def calculate_macd(data, short_span, long_span, signal_span):
    """MACD, Signal Line 계산"""
    ema_short = calculate_ema_series(data['Close'], short_span)
    ema_long = calculate_ema_series(data['Close'], long_span)
    macd = ema_short - ema_long
    signal = calculate_ema_series(macd, signal_span)
    return macd, signal

def calculate_cmf(data, period):
    """Chaikin Money Flow 계산"""
    high = data['High']
    low = data['Low']
    close = data['Close']
    volume = data['Volume']
    
    # Money Flow Multiplier
    mf_multiplier = ((close - low) - (high - close)) / (high - low).replace(0, 0.0001)
    
    # Money Flow Volume
    mf_volume = mf_multiplier * volume
    
    # Chaikin Money Flow
    cmf = mf_volume.rolling(window=period).sum() / volume.rolling(window=period).sum()
    
    # 결과 저장
    data[f'CMF_{period}'] = cmf
    return data

def calculate_rsi(data, period=14):
    """RSI 계산"""
    close = data['Close']
    delta = close.diff()
    
    # 상승/하락 분리
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    
    # 평균 계산
    avg_gain = gain.rolling(window=period).mean()
    avg_loss = loss.rolling(window=period).mean()
    
    # RS 계산
    rs = avg_gain / avg_loss.replace(0, 1e-10)  # 0으로 나누기 방지
    
    # RSI 계산
    rsi = 100 - (100 / (1 + rs))
    
    # 결과 저장
    data[f'RSI_{period}'] = rsi
    return data


# -----------------------------
# 기술적 지표 추가 함수
# -----------------------------
def add_technical_indicators(df, ema_params, macd_params, cmf_period, rsi_params):
    """
    주가 데이터에 최적화된 파라미터로 모든 기술적 지표를 추가하는 함수
    """
    # EMA 계산
    df['EMA_short'] = calculate_ema_series(df['Close'], ema_params['short'])
    df['EMA_long'] = calculate_ema_series(df['Close'], ema_params['long'])
    df['EMA_diff'] = df['EMA_short'] - df['EMA_long']
    
    # MACD 계산
    ema_fast = calculate_ema_series(df['Close'], macd_params['fast'])
    ema_slow = calculate_ema_series(df['Close'], macd_params['slow'])
    df['MACD'] = ema_fast - ema_slow
    df['MACD_signal'] = calculate_ema_series(df['MACD'], macd_params['signal'])
    df['MACD_diff'] = df['MACD'] - df['MACD_signal']
    
    # CMF 계산
    df = calculate_cmf(df, cmf_period)
    df['CMF'] = df[f'CMF_{cmf_period}']
    df.drop(columns=[f'CMF_{cmf_period}'], inplace=True)

    # RSI 계산
    df = calculate_rsi(df, rsi_params['period'])
    df['RSI'] = df[f'RSI_{rsi_params["period"]}']

    df.drop(columns=[f'RSI_{rsi_params["period"]}'], inplace=True)
    
    # RSI 임계값 저장
    df['RSI_upper'] = rsi_params['upper_threshold']
    df['RSI_lower'] = rsi_params['lower_threshold']
    
    # 수익률 계산
    df['Return'] = df['Close'].pct_change()
    
    return df