Python으로 구현하는 모멘텀 트레이딩 전략

# pyproject.toml
[project.dependencies]
ta = ">=0.11"       # 기술 분석 라이브러리
pandas = ">=2.0"
numpy = ">=1.24"

import pandas as pd
import numpy as np
from ta.trend import EMAIndicator, MACD as MACDIndicator
from ta.momentum import RSIIndicator

def ma_crossover_signals(
    df: pd.DataFrame,
    fast: int = 20,
    slow: int = 60,
) -> pd.DataFrame:
    """이동평균 크로스오버 신호 생성.
 
    Args:
        df: OHLCV DataFrame (columns: open, high, low, close, volume)
        fast: 단기 이동평균 기간
        slow: 장기 이동평균 기간
 
    Returns:
        signal 컬럼이 추가된 DataFrame (1=매수, -1=매도, 0=중립)
    """
    df = df.copy()
 
    # EMA 계산
    df["ema_fast"] = EMAIndicator(
        close=df["close"], window=fast
    ).ema_indicator()
    df["ema_slow"] = EMAIndicator(
        close=df["close"], window=slow
    ).ema_indicator()
 
    # 크로스오버 감지
    df["ma_cross"] = 0
 
    # 골든크로스: fast가 slow 위로 교차
    df.loc[
        (df["ema_fast"] > df["ema_slow"])
        & (df["ema_fast"].shift(1) <= df["ema_slow"].shift(1)),
        "ma_cross",
    ] = 1
 
    # 데드크로스: fast가 slow 아래로 교차
    df.loc[
        (df["ema_fast"] < df["ema_slow"])
        & (df["ema_fast"].shift(1) >= df["ema_slow"].shift(1)),
        "ma_cross",
    ] = -1
 
    return df

def rsi_signals(
    df: pd.DataFrame,
    window: int = 14,
    overbought: float = 70,
    oversold: float = 30,
) -> pd.DataFrame:
    """RSI 기반 매매 신호 생성.
 
    Args:
        df: OHLCV DataFrame
        window: RSI 계산 기간
        overbought: 과매수 기준
        oversold: 과매도 기준
 
    Returns:
        rsi, rsi_signal 컬럼이 추가된 DataFrame
    """
    df = df.copy()
    df["rsi"] = RSIIndicator(
        close=df["close"], window=window
    ).rsi()
 
    df["rsi_signal"] = 0
 
    # 과매도 탈출 → 매수 (RSI가 oversold 아래에서 위로)
    df.loc[
        (df["rsi"] > oversold)
        & (df["rsi"].shift(1) <= oversold),
        "rsi_signal",
    ] = 1
 
    # 과매수 이탈 → 매도 (RSI가 overbought 위에서 아래로)
    df.loc[
        (df["rsi"] < overbought)
        & (df["rsi"].shift(1) >= overbought),
        "rsi_signal",
    ] = -1
 
    return df

def macd_signals(
    df: pd.DataFrame,
    fast: int = 12,
    slow: int = 26,
    signal: int = 9,
) -> pd.DataFrame:
    """MACD 크로스오버 신호 생성.
 
    Args:
        df: OHLCV DataFrame
        fast: 단기 EMA 기간
        slow: 장기 EMA 기간
        signal: 시그널 EMA 기간
 
    Returns:
        macd, macd_signal_line, macd_hist, macd_signal 컬럼 추가
    """
    df = df.copy()
    macd = MACDIndicator(
        close=df["close"],
        window_fast=fast,
        window_slow=slow,
        window_sign=signal,
    )
    df["macd"] = macd.macd()
    df["macd_signal_line"] = macd.macd_signal()
    df["macd_hist"] = macd.macd_diff()
 
    df["macd_signal"] = 0
 
    # MACD 골든크로스
    df.loc[
        (df["macd"] > df["macd_signal_line"])
        & (df["macd"].shift(1) <= df["macd_signal_line"].shift(1)),
        "macd_signal",
    ] = 1
 
    # MACD 데드크로스
    df.loc[
        (df["macd"] < df["macd_signal_line"])
        & (df["macd"].shift(1) >= df["macd_signal_line"].shift(1)),
        "macd_signal",
    ] = -1
 
    return df

def combined_momentum_signal(
    df: pd.DataFrame,
    ma_fast: int = 20,
    ma_slow: int = 60,
    rsi_window: int = 14,
    rsi_overbought: float = 70,
    rsi_oversold: float = 30,
    macd_fast: int = 12,
    macd_slow: int = 26,
    macd_signal: int = 9,
    min_confirmations: int = 2,
) -> pd.DataFrame:
    """복합 모멘텀 신호 생성.
 
    min_confirmations개 이상의 지표가 동일 방향일 때만 신호 발생.
    """
    df = ma_crossover_signals(df, ma_fast, ma_slow)
    df = rsi_signals(df, rsi_window, rsi_overbought, rsi_oversold)
    df = macd_signals(df, macd_fast, macd_slow, macd_signal)
 
    # 각 지표의 현재 방향성
    df["ma_direction"] = np.where(
        df["ema_fast"] > df["ema_slow"], 1, -1
    )
    df["rsi_direction"] = np.where(
        df["rsi"] < 50, -1, 1
    )  # RSI 50 기준
    df["macd_direction"] = np.where(
        df["macd_hist"] > 0, 1, -1
    )
 
    # 방향성 합산
    df["momentum_score"] = (
        df["ma_direction"]
        + df["rsi_direction"]
        + df["macd_direction"]
    )
 
    # 최종 신호
    df["signal"] = 0
    df.loc[
        df["momentum_score"] >= min_confirmations, "signal"
    ] = 1   # 매수
    df.loc[
        df["momentum_score"] <= -min_confirmations, "signal"
    ] = -1  # 매도
 
    return df

"kospi_standard": {
    "ma_fast": 20, "ma_slow": 60,
    "rsi_window": 14, "rsi_overbought": 70, "rsi_oversold": 30,
    "macd_fast": 12, "macd_slow": 26, "macd_signal": 9,
    "min_confirmations": 2,
}

"kosdaq_aggressive": {
    "ma_fast": 10, "ma_slow": 30,
    "rsi_window": 9, "rsi_overbought": 75, "rsi_oversold": 25,
    "macd_fast": 8, "macd_slow": 21, "macd_signal": 5,
    "min_confirmations": 2,
}

"conservative": {
    "ma_fast": 50, "ma_slow": 200,
    "rsi_window": 14, "rsi_overbought": 80, "rsi_oversold": 20,
    "macd_fast": 12, "macd_slow": 26, "macd_signal": 9,
    "min_confirmations": 3,  # 3개 모두 일치 시만
}

MOMENTUM_PARAMS = {
    "kospi_standard": {
        "ma_fast": 20, "ma_slow": 60,
        "rsi_window": 14, "rsi_overbought": 70, "rsi_oversold": 30,
        "macd_fast": 12, "macd_slow": 26, "macd_signal": 9,
        "min_confirmations": 2,
    },
    "kosdaq_aggressive": {
        "ma_fast": 10, "ma_slow": 30,
        "rsi_window": 9, "rsi_overbought": 75, "rsi_oversold": 25,
        "macd_fast": 8, "macd_slow": 21, "macd_signal": 5,
        "min_confirmations": 2,
    },
    "conservative": {
        "ma_fast": 50, "ma_slow": 200,
        "rsi_window": 14, "rsi_overbought": 80, "rsi_oversold": 20,
        "macd_fast": 12, "macd_slow": 26, "macd_signal": 9,
        "min_confirmations": 3,
    },
}

params = MOMENTUM_PARAMS["kospi_standard"]
result = combined_momentum_signal(df, **params)

def calculate_position_size(
    budget: float,
    price: float,
    stop_loss_pct: float,
    risk_per_trade_pct: float = 1.0,
) -> int:
    """고정비율 포지션 사이징.
 
    Args:
        budget: 총 투자 가능 금액
        price: 현재 주가
        stop_loss_pct: 손절 비율 (예: -3.0)
        risk_per_trade_pct: 1회 거래 최대 손실 비율 (예산 대비, 기본 1%)
 
    Returns:
        매수 수량 (정수)
    """
    max_loss_amount = budget * (risk_per_trade_pct / 100)
    loss_per_share = price * abs(stop_loss_pct / 100)
 
    if loss_per_share == 0:
        return 0
 
    shares = int(max_loss_amount / loss_per_share)
 
    # 종목당 최대 비중 10% 제한
    max_shares = int(budget * 0.10 / price)
    return min(shares, max_shares)

RISK_PARAMS = {
    "stop_loss_pct": -3.0,      # 손절: -3%
    "take_profit_pct": 6.0,     # 익절: +6% (R:R = 1:2)
    "trailing_stop_pct": -2.0,  # 트레일링 스톱: 고점 대비 -2%
    "max_holding_days": 20,     # 최대 보유 기간: 20 영업일
    "max_position_pct": 10.0,   # 종목당 최대 비중: 10%
}