이동평균 크로스오버란?
이동평균 크로스오버(Moving Average Crossover)는 퀀트 트레이딩에서 가장 오래되고 널리 쓰이는 추세 추종 전략입니다. 단기 이동평균이 장기 이동평균을 상향 돌파하면 매수(골든 크로스), 하향 돌파하면 매도(데드 크로스)하는 단순한 규칙이지만, 올바르게 구현하면 놀라운 효과를 발휘합니다.
이 글에서는 파이썬으로 SMA, EMA, DEMA 크로스오버를 구현하고, 파라미터 최적화와 필터링 기법까지 다룹니다. 단순 구현을 넘어 실전에서 수익을 내는 크로스오버 시스템을 만드는 방법을 단계별로 설명합니다.
SMA 크로스오버 기본 구현
가장 기본적인 단순 이동평균(SMA) 크로스오버부터 구현합니다:
import numpy as np
import pandas as pd
import yfinance as yf
def sma_crossover(data: pd.DataFrame, fast: int = 20,
slow: int = 60) -> pd.DataFrame:
"""SMA 크로스오버 시그널 생성"""
df = data.copy()
df["sma_fast"] = df["Close"].rolling(fast).mean()
df["sma_slow"] = df["Close"].rolling(slow).mean()
# 크로스오버 시그널
df["signal"] = 0
df.loc[df["sma_fast"] > df["sma_slow"], "signal"] = 1 # 골든 크로스
df.loc[df["sma_fast"] <= df["sma_slow"], "signal"] = -1 # 데드 크로스
# 실제 포지션 (다음 날 진입)
df["position"] = df["signal"].shift(1)
df["returns"] = df["Close"].pct_change()
df["strategy_returns"] = df["position"] * df["returns"]
return df.dropna()
# SPY 데이터로 백테스트
data = yf.download("SPY", start="2015-01-01", end="2025-12-31")
result = sma_crossover(data, fast=20, slow=60)
# 성과 요약
cumulative = (1 + result["strategy_returns"]).cumprod()
buy_hold = (1 + result["returns"]).cumprod()
print(f"전략 누적수익: {cumulative.iloc[-1] - 1:.1%}")
print(f"Buy&Hold 누적수익: {buy_hold.iloc[-1] - 1:.1%}")
print(f"거래 횟수: {(result['signal'].diff() != 0).sum()}")기본 SMA 크로스오버의 가장 큰 문제는 횡보장에서의 잦은 whipsaw(헛신호)입니다. 추세가 없는 시장에서 이동평균이 반복적으로 교차하며 손실을 누적합니다.
EMA 크로스오버: 최신 가격에 더 민감하게
지수 이동평균(EMA)은 최근 가격에 더 높은 가중치를 부여해 SMA보다 빠르게 반응합니다:
def ema_crossover(data: pd.DataFrame, fast: int = 12,
slow: int = 26) -> pd.DataFrame:
"""EMA 크로스오버 (MACD 기본 파라미터 활용)"""
df = data.copy()
df["ema_fast"] = df["Close"].ewm(span=fast, adjust=False).mean()
df["ema_slow"] = df["Close"].ewm(span=slow, adjust=False).mean()
# 크로스오버 강도 측정
df["spread"] = df["ema_fast"] - df["ema_slow"]
df["spread_pct"] = df["spread"] / df["Close"] * 100
df["signal"] = 0
df.loc[df["spread"] > 0, "signal"] = 1
df.loc[df["spread"] <= 0, "signal"] = -1
df["position"] = df["signal"].shift(1)
df["returns"] = df["Close"].pct_change()
df["strategy_returns"] = df["position"] * df["returns"]
return df.dropna()
def dema_crossover(data: pd.DataFrame, fast: int = 12,
slow: int = 26) -> pd.DataFrame:
"""DEMA(Double EMA) 크로스오버 - 지연 최소화"""
df = data.copy()
# DEMA = 2 * EMA(n) - EMA(EMA(n))
ema_fast = df["Close"].ewm(span=fast, adjust=False).mean()
df["dema_fast"] = 2 * ema_fast - ema_fast.ewm(
span=fast, adjust=False
).mean()
ema_slow = df["Close"].ewm(span=slow, adjust=False).mean()
df["dema_slow"] = 2 * ema_slow - ema_slow.ewm(
span=slow, adjust=False
).mean()
df["signal"] = 0
df.loc[df["dema_fast"] > df["dema_slow"], "signal"] = 1
df.loc[df["dema_fast"] <= df["dema_slow"], "signal"] = -1
df["position"] = df["signal"].shift(1)
df["returns"] = df["Close"].pct_change()
df["strategy_returns"] = df["position"] * df["returns"]
return df.dropna()
# 세 가지 비교
for name, func, params in [
("SMA 20/60", sma_crossover, {"fast": 20, "slow": 60}),
("EMA 12/26", ema_crossover, {"fast": 12, "slow": 26}),
("DEMA 12/26", dema_crossover, {"fast": 12, "slow": 26}),
]:
r = func(data, **params)
cum = (1 + r["strategy_returns"]).cumprod().iloc[-1] - 1
trades = (r["signal"].diff() != 0).sum()
print(f"{name}: 수익={cum:.1%}, 거래={trades}회")EMA는 SMA보다 빠르게 반응하지만 노이즈에도 민감합니다. DEMA는 EMA의 지연을 더욱 줄이면서도 SMA보다 부드러운 곡선을 유지합니다. 각 이동평균의 특성을 이해하고 시장에 맞게 선택하는 것이 중요합니다.
Whipsaw 필터: 횡보장 손실 방지
크로스오버 전략의 최대 적은 횡보장입니다. 세 가지 필터로 whipsaw를 줄입니다:
class FilteredCrossover:
"""필터링된 크로스오버 전략"""
def __init__(self, fast: int = 20, slow: int = 60):
self.fast = fast
self.slow = slow
def threshold_filter(self, data: pd.DataFrame,
min_spread_pct: float = 0.5) -> pd.DataFrame:
"""임계값 필터: 스프레드가 일정 이상이어야 진입"""
df = data.copy()
df["sma_fast"] = df["Close"].rolling(self.fast).mean()
df["sma_slow"] = df["Close"].rolling(self.slow).mean()
spread_pct = abs(
df["sma_fast"] - df["sma_slow"]
) / df["Close"] * 100
df["signal"] = 0
bullish = (df["sma_fast"] > df["sma_slow"]) & (
spread_pct > min_spread_pct
)
bearish = (df["sma_fast"] < df["sma_slow"]) & (
spread_pct > min_spread_pct
)
df.loc[bullish, "signal"] = 1
df.loc[bearish, "signal"] = -1
# 시그널 없으면 이전 포지션 유지
df["signal"] = df["signal"].replace(0, np.nan).ffill().fillna(0)
return df
def confirmation_filter(self, data: pd.DataFrame,
confirm_days: int = 3) -> pd.DataFrame:
"""확인 필터: N일 연속 크로스 유지 시에만 진입"""
df = data.copy()
df["sma_fast"] = df["Close"].rolling(self.fast).mean()
df["sma_slow"] = df["Close"].rolling(self.slow).mean()
raw_signal = (df["sma_fast"] > df["sma_slow"]).astype(int)
# N일 연속 같은 시그널이면 확정
confirmed = raw_signal.rolling(confirm_days).sum()
df["signal"] = 0
df.loc[confirmed == confirm_days, "signal"] = 1
df.loc[confirmed == 0, "signal"] = -1
df["signal"] = df["signal"].replace(0, np.nan).ffill().fillna(0)
return df
def volatility_filter(self, data: pd.DataFrame,
vol_lookback: int = 20,
vol_threshold: float = 0.15) -> pd.DataFrame:
"""변동성 필터: 저변동성(횡보) 구간 거래 중단"""
df = data.copy()
df["sma_fast"] = df["Close"].rolling(self.fast).mean()
df["sma_slow"] = df["Close"].rolling(self.slow).mean()
df["volatility"] = (
df["Close"].pct_change().rolling(vol_lookback).std()
* np.sqrt(252)
)
df["signal"] = 0
is_trending = df["volatility"] > vol_threshold
df.loc[
(df["sma_fast"] > df["sma_slow"]) & is_trending, "signal"
] = 1
df.loc[
(df["sma_fast"] < df["sma_slow"]) & is_trending, "signal"
] = -1
return df
# 필터 효과 비교
fc = FilteredCrossover(20, 60)
for name, method in [
("임계값 필터", fc.threshold_filter),
("확인 필터", fc.confirmation_filter),
("변동성 필터", fc.volatility_filter),
]:
df = method(data).dropna()
df["position"] = df["signal"].shift(1)
df["returns"] = df["Close"].pct_change()
df["strategy_returns"] = df["position"] * df["returns"]
cum = (1 + df["strategy_returns"].dropna()).cumprod().iloc[-1] - 1
trades = (df["signal"].diff() != 0).sum()
print(f"{name}: 수익={cum:.1%}, 거래={trades}회")세 가지 필터를 조합하면 whipsaw를 70~80%까지 줄일 수 있습니다. 핵심은 "확실한 추세에서만 진입"하는 것입니다.
파라미터 최적화: 워크포워드 방식
최적 이동평균 기간을 찾되, 과적합을 방지하는 워크포워드 최적화를 적용합니다:
def walk_forward_optimize(data: pd.DataFrame,
train_years: int = 3,
test_months: int = 6) -> pd.DataFrame:
"""워크포워드 최적화"""
results = []
dates = data.index
train_days = train_years * 252
test_days = test_months * 21
i = train_days
while i + test_days <= len(data):
# 학습 구간
train_data = data.iloc[i - train_days:i]
# 학습 구간에서 최적 파라미터 탐색
best_sharpe = -np.inf
best_params = (20, 60)
for fast in range(10, 51, 5):
for slow in range(40, 201, 10):
if fast >= slow:
continue
r = sma_crossover(train_data, fast, slow)
ret = r["strategy_returns"]
if ret.std() == 0:
continue
sr = ret.mean() / ret.std() * np.sqrt(252)
if sr > best_sharpe:
best_sharpe = sr
best_params = (fast, slow)
# 테스트 구간에 적용
test_data = data.iloc[i:i + test_days]
r = sma_crossover(test_data, *best_params)
results.append({
"period_start": dates[i],
"period_end": dates[min(i + test_days - 1, len(data) - 1)],
"fast": best_params[0],
"slow": best_params[1],
"train_sharpe": round(best_sharpe, 2),
"test_return": r["strategy_returns"].sum(),
})
i += test_days
results_df = pd.DataFrame(results)
total_ret = (1 + results_df["test_return"]).prod() - 1
print(f"워크포워드 총 수익: {total_ret:.1%}")
print(f"평균 최적 파라미터: "
f"fast={results_df['fast'].mean():.0f}, "
f"slow={results_df['slow'].mean():.0f}")
return results_df
wf_results = walk_forward_optimize(data)워크포워드 최적화는 "미래 데이터를 절대 보지 않는" 방식으로 파라미터를 선택합니다. 고정 파라미터보다 성과가 낮을 수 있지만, 실전에서의 성과를 더 정확하게 예측합니다.
다중 시간프레임 크로스오버
단일 시간프레임보다 다중 시간프레임 확인이 신뢰도를 높입니다:
def multi_timeframe_crossover(data: pd.DataFrame) -> pd.DataFrame:
"""다중 시간프레임 크로스오버 전략"""
df = data.copy()
# 단기 크로스오버 (5/20)
df["fast_5"] = df["Close"].rolling(5).mean()
df["slow_20"] = df["Close"].rolling(20).mean()
df["short_signal"] = (df["fast_5"] > df["slow_20"]).astype(int)
# 중기 크로스오버 (20/60)
df["mid_20"] = df["Close"].rolling(20).mean()
df["mid_60"] = df["Close"].rolling(60).mean()
df["mid_signal"] = (df["mid_20"] > df["mid_60"]).astype(int)
# 장기 추세 (60/200)
df["long_60"] = df["Close"].rolling(60).mean()
df["long_200"] = df["Close"].rolling(200).mean()
df["long_signal"] = (df["long_60"] > df["long_200"]).astype(int)
# 종합 시그널: 3개 중 2개 이상 일치
df["total_score"] = (
df["short_signal"] + df["mid_signal"] + df["long_signal"]
)
df["signal"] = 0
df.loc[df["total_score"] >= 2, "signal"] = 1 # 2/3 이상 매수
df.loc[df["total_score"] <= 0, "signal"] = -1 # 전부 매도
df["signal"] = df["signal"].replace(0, np.nan).ffill().fillna(0)
df["position"] = df["signal"].shift(1)
df["returns"] = df["Close"].pct_change()
df["strategy_returns"] = df["position"] * df["returns"]
return df.dropna()
mtf = multi_timeframe_crossover(data)
cum = (1 + mtf["strategy_returns"]).cumprod().iloc[-1] - 1
print(f"다중 시간프레임 수익: {cum:.1%}")다중 시간프레임의 핵심 원칙: 장기 추세 방향으로만 거래합니다. 장기가 상승 추세일 때 단기 골든 크로스에서 매수하고, 장기가 하락일 때는 단기 시그널을 무시합니다.
성과 분석과 리스크 관리
크로스오버 전략의 성과를 종합 성과 지표로 평가합니다:
def evaluate_strategy(returns: pd.Series, name: str) -> dict:
"""전략 성과 종합 평가"""
cumulative = (1 + returns).cumprod()
max_dd = (cumulative / cumulative.cummax() - 1).min()
annual_ret = returns.mean() * 252
annual_vol = returns.std() * np.sqrt(252)
sharpe = annual_ret / annual_vol if annual_vol > 0 else 0
# 승률과 손익비
wins = returns[returns > 0]
losses = returns[returns < 0]
win_rate = len(wins) / (len(wins) + len(losses)) if len(losses) > 0 else 0
profit_factor = (
wins.sum() / abs(losses.sum()) if losses.sum() != 0 else float("inf")
)
return {
"name": name,
"annual_return": f"{annual_ret:.1%}",
"annual_volatility": f"{annual_vol:.1%}",
"sharpe": round(sharpe, 2),
"max_drawdown": f"{max_dd:.1%}",
"win_rate": f"{win_rate:.1%}",
"profit_factor": round(profit_factor, 2),
}
# 전략별 비교
strategies = [
("SMA 20/60", sma_crossover(data, 20, 60)),
("EMA 12/26", ema_crossover(data, 12, 26)),
("다중 시간프레임", multi_timeframe_crossover(data)),
]
for name, r in strategies:
eval_result = evaluate_strategy(r["strategy_returns"], name)
print(f"n{eval_result['name']}:")
for k, v in eval_result.items():
if k != "name":
print(f" {k}: {v}")실전 적용 체크리스트
- 추세장 전용: 크로스오버는 추세 추종 전략. 시장 레짐 탐지로 횡보장을 필터링
- 필터 필수: 임계값, 확인, 변동성 필터 중 최소 하나 적용
- 다중 시간프레임: 장기 추세 방향과 일치하는 시그널만 실행
- 슬리피지 반영: 크로스오버 시점은 거래량이 몰리므로 슬리피지 5~10bps 반영
- 워크포워드 검증: 고정 파라미터 대신 롤링 최적화 적용
- 손절 결합: 크로스오버 시그널과 별도로 ATR 기반 손절 설정
결론
이동평균 크로스오버는 "너무 단순하다"고 과소평가되지만, 필터링, 다중 시간프레임, 워크포워드 최적화를 결합하면 여전히 강력한 전략입니다. 복잡한 머신러닝 모델보다 해석이 쉽고, 과적합 위험이 낮으며, 구현과 운영이 간단합니다. 퀀트 자동매매의 첫 번째 전략으로 크로스오버를 마스터하면, 이후 모든 추세 추종 전략의 기초가 됩니다.