터틀트레이딩 — MA 필터와 조합 비교

 

MA 시리즈를 마무리할 때, "이동평균의 진짜 가치는 크로스 전략보다 추세 필터에 있을 수 있다"고 했습니다. 이번 글에서 그 가설을 검증합니다.

 

지금까지 만든 터틀트레이딩(피라미딩 포함)에 이동평균 필터를 얹어봅니다. 규칙은 간단합니다. 가격이 MA 위에 있을 때만 진입을 허용한다. 이미 포지션이 있으면 MA 아래로 내려가도 청산하지 않고, 기존 손절/채널 청산 규칙을 그대로 따릅니다.

---

아이디어

돈치안 채널 브레이크아웃은 신고가를 돌파하면 무조건 진입합니다. 하지만 하락 추세 속에서 일시적으로 반등하면서 신고가를 찍는 경우도 있습니다. 이런 가짜 돌파에 진입하면 바로 손절로 이어집니다.

 

MA 필터는 이걸 걸러주는 역할입니다. 가격이 장기 MA 아래에 있다면 전체적으로 하락 추세라는 뜻이니, 그 상태에서의 돌파 신호는 무시합니다. 큰 추세가 상승일 때만 진입을 허용해서, 불필요한 진입과 손절을 줄이겠다는 논리입니다.

---

테스트 조합

터틀 설정은 이전 글과 동일하고, MA 필터만 바꿔가며 10가지 조합을 비교합니다.

필터	종류
없음	기존 터틀 + 피라미딩
SMA 50 / 100 / 200	단순이동평균
EMA 50 / 100 / 200	지수이동평균
HMA 50 / 100 / 200	헐이동평균

 

기간은 단기(50), 중기(100), 장기(200) 세 가지. MA 종류는 크로스 전략에서 상위권이었던  SMA, EMA, HMA를 선택했습니다.

---

코드 구현

기존 터틀 + 피라미딩 코드를 함수화하고, MA 필터 조건을 진입부에 추가했습니다. 필터 조합을 리스트로 관리해서 한 번에 비교합니다.

import pandas as pd
import numpy as np

# ===== 설정 =====
DATA_FILE = "BTCUSDT_1d.csv"
INITIAL_CAPITAL = 10000

# 터틀 기본 설정
DC_ENTRY = 20
DC_EXIT = 10
ATR_PERIOD = 14
ATR_STOP_MULT = 2
RISK_PER_TRADE = 0.02
PYRAMID_STEP = 0.5
MAX_UNITS = 4

# MA 필터 조합 리스트: (이름, 종류, 기간)
FILTER_LIST = [
    ("None",     None,  0),
    ("SMA 50",   "sma", 50),
    ("SMA 100",  "sma", 100),
    ("SMA 200",  "sma", 200),
    ("EMA 50",   "ema", 50),
    ("EMA 100",  "ema", 100),
    ("EMA 200",  "ema", 200),
    ("HMA 50",   "hma", 50),
    ("HMA 100",  "hma", 100),
    ("HMA 200",  "hma", 200),
]
# ================

# 데이터 로드
df = pd.read_csv(DATA_FILE, index_col="timestamp", parse_dates=True)

# 돈치안 채널
df["dc_entry_upper"] = df["high"].rolling(window=DC_ENTRY).max().shift(1)
df["dc_exit_lower"] = df["low"].rolling(window=DC_EXIT).min().shift(1)

# ATR
df["h_l"] = df["high"] - df["low"]
df["h_pc"] = abs(df["high"] - df["close"].shift(1))
df["l_pc"] = abs(df["low"] - df["close"].shift(1))
df["tr"] = df[["h_l", "h_pc", "l_pc"]].max(axis=1)
df["atr"] = df["tr"].ewm(alpha=1/ATR_PERIOD, adjust=False).mean()

# MA 계산 함수
def calc_ma(series, ma_type, period):
    if ma_type == "sma":
        return series.rolling(window=period).mean()
    elif ma_type == "ema":
        return series.ewm(span=period, adjust=False).mean()
    elif ma_type == "hma":
        half = int(period / 2)
        sqrt_period = int(np.sqrt(period))
        weights_half = np.arange(1, half + 1, dtype=float)
        wma_half = series.rolling(window=half).apply(lambda x: np.dot(x, weights_half) / weights_half.sum(), raw=True)
        weights_full = np.arange(1, period + 1, dtype=float)
        wma_full = series.rolling(window=period).apply(lambda x: np.dot(x, weights_full) / weights_full.sum(), raw=True)
        diff = 2 * wma_half - wma_full
        weights_sqrt = np.arange(1, sqrt_period + 1, dtype=float)
        return diff.rolling(window=sqrt_period).apply(lambda x: np.dot(x, weights_sqrt) / weights_sqrt.sum(), raw=True)
    return None

# 백테스트 함수
def run_turtle_pyramid(df, ma_series, initial_capital):
    capital = initial_capital
    in_position = False
    units = []
    stop_loss = 0
    last_add_price = 0
    entry_atr = 0
    trades = []
    equity_curve = []

    start_idx = max(DC_ENTRY, ATR_PERIOD)

    for i in range(start_idx, len(df)):
        high = df["high"].iloc[i]
        low = df["low"].iloc[i]
        close = df["close"].iloc[i]
        entry_upper = df["dc_entry_upper"].iloc[i]
        exit_lower = df["dc_exit_lower"].iloc[i]
        atr = df["atr"].iloc[i]

        if ma_series is not None:
            ma_val = ma_series.iloc[i]
            ma_ok = not pd.isna(ma_val) and close > ma_val
        else:
            ma_ok = True

        if not in_position:
            if high > entry_upper and atr > 0 and ma_ok:
                entry_atr = atr
                entry_price = close

                risk_amount = capital * RISK_PER_TRADE
                risk_per_unit = ATR_STOP_MULT * entry_atr
                qty = risk_amount / risk_per_unit

                max_qty = capital / entry_price
                qty = min(qty, max_qty)

                capital -= qty * entry_price
                units = [{"qty": qty, "entry": entry_price}]
                stop_loss = entry_price - ATR_STOP_MULT * entry_atr
                last_add_price = entry_price
                in_position = True
        else:
            if low < stop_loss:
                total_qty = sum(u["qty"] for u in units)
                capital += total_qty * stop_loss
                avg_entry = sum(u["qty"] * u["entry"] for u in units) / total_qty
                pnl = (stop_loss - avg_entry) / avg_entry * 100
                trades.append({"pnl_pct": round(pnl, 2), "units": len(units), "exit_type": "stop"})
                units = []
                in_position = False

            elif len(units) < MAX_UNITS and high > last_add_price + PYRAMID_STEP * entry_atr:
                add_price = last_add_price + PYRAMID_STEP * entry_atr
                current_equity = capital + sum(u["qty"] for u in units) * close
                risk_amount = current_equity * RISK_PER_TRADE
                risk_per_unit = ATR_STOP_MULT * entry_atr
                qty = risk_amount / risk_per_unit

                if capital > add_price * qty:
                    capital -= qty * add_price
                    units.append({"qty": qty, "entry": add_price})
                    last_add_price = add_price
                    stop_loss = add_price - ATR_STOP_MULT * entry_atr

            if in_position and low < exit_lower:
                total_qty = sum(u["qty"] for u in units)
                capital += total_qty * close
                avg_entry = sum(u["qty"] * u["entry"] for u in units) / total_qty
                pnl = (close - avg_entry) / avg_entry * 100
                trades.append({"pnl_pct": round(pnl, 2), "units": len(units), "exit_type": "exit"})
                units = []
                in_position = False

        total_qty = sum(u["qty"] for u in units) if units else 0
        equity = capital + total_qty * close
        equity_curve.append(equity)

    if in_position:
        last_price = df["close"].iloc[-1]
        total_qty = sum(u["qty"] for u in units)
        capital += total_qty * last_price
        avg_entry = sum(u["qty"] * u["entry"] for u in units) / total_qty
        pnl = (last_price - avg_entry) / avg_entry * 100
        trades.append({"pnl_pct": round(pnl, 2), "units": len(units), "exit_type": "hold"})

    eq = pd.Series(equity_curve)
    peak = eq.cummax()
    mdd = ((eq - peak) / peak * 100).min()

    return capital, trades, mdd

# 바이앤홀드
first_price = df["close"].dropna().iloc[0]
last_price = df["close"].dropna().iloc[-1]
bnh_return = (last_price - first_price) / first_price * 100
bnh_capital = INITIAL_CAPITAL * (1 + bnh_return / 100)

# 실행 및 결과 수집
results = []

for name, ma_type, period in FILTER_LIST:
    if ma_type is not None:
        ma_series = calc_ma(df["close"], ma_type, period)
    else:
        ma_series = None

    capital, trades, mdd = run_turtle_pyramid(df, ma_series, INITIAL_CAPITAL)
    total_return = (capital - INITIAL_CAPITAL) / INITIAL_CAPITAL * 100
    wins = len([t for t in trades if t["pnl_pct"] > 0])
    losses = len([t for t in trades if t["pnl_pct"] <= 0])
    win_rate = wins / len(trades) * 100 if trades else 0

    results.append({
        "name": name,
        "capital": capital,
        "return": total_return,
        "mdd": mdd,
        "trades": len(trades),
        "wins": wins,
        "losses": losses,
        "win_rate": win_rate,
    })

# 출력
print("=" * 80)
print(f"  Turtle + Pyramid + MA Filter")
print(f"  DC {DC_ENTRY}/{DC_EXIT}, ATR {ATR_PERIOD}, Stop {ATR_STOP_MULT}x, Risk {RISK_PER_TRADE*100:.0f}%")
print(f"  Pyramid {PYRAMID_STEP}x ATR, Max {MAX_UNITS} units")
print(f"  Data: {df.index[0].strftime('%Y-%m-%d')} ~ {df.index[-1].strftime('%Y-%m-%d')}")
print("=" * 80)
print(f"  {'Filter':<12} {'Capital':>10} {'Return':>10} {'MDD':>8} {'Trades':>7} {'Win':>5} {'Lose':>5} {'WinRate':>8}")
print("-" * 80)

for r in results:
    print(f"  {r['name']:<12} ${r['capital']:>9,.0f} {r['return']:>+9.2f}% {r['mdd']:>+7.2f}% {r['trades']:>6}  {r['wins']:>5} {r['losses']:>5} {r['win_rate']:>7.1f}%")

print("-" * 80)
print(f"  {'B&H':<12} ${bnh_capital:>9,.0f} {bnh_return:>+9.2f}%")
print("=" * 80)

---

사용법

FILTER_LIST에 원하는 MA 조합을 추가/제거하면 됩니다. MA 종류는 "sma", "ema", "hma"를 지원합니다.

---

결과

---

분석

필터가 효과가 있는가?

대부분의 MA 필터가 거래 횟수를 줄여줬습니다. 필터 없이 58회였던 거래가 SMA 200에서는 39회로 줄었습니다. 하락 추세에서의 불필요한 진입을 걸러낸 결과입니다.

 

그런데 거래가 줄어든다고 꼭 좋은 건 아닙니다. SMA 100은 거래 47회에 +831%, SMA 50은 49회에 +1388%. 거래 횟수는 비슷한데 수익률 차이가 큽니다. 어떤 거래를 걸러냈느냐가 중요한 거죠.

필터가 오히려 역효과인 경우

모든 필터가 성과를 개선한 것은 아닙니다. SMA 100(+831%), SMA 200(+825%), EMA 200(+754%)은 필터 없음(+1173%)보다 수익률이 크게 낮습니다. 필터가 불필요한 진입을 막아주기도 하지만, 동시에 수익이 날 진입까지 막아버릴 수 있다는 뜻입니다.

 

특히 장기 MA 필터는 큰 하락 후 반등 초기에 문제가 됩니다. 가격이 바닥을 찍고 올라오기 시작해도, 200일 MA는 아직 한참 위에 있습니다. 가격이 MA를 넘을 때쯤이면 초기 상승분을 이미 놓친 뒤입니다. 2018년 말이나 2022년 말처럼 급락 후 급반등하는 구간에서 이런 손실이 발생했을 가능성이 높습니다.

 

필터는 "나쁜 진입을 줄여주는 도구"이지, "무조건 성과를 개선해주는 장치"가 아닙니다.

 

수익률: SMA/EMA는 50기간이 최고, HMA는 다른 패턴

기간	SMA	EMA	HMA
50	+1388%	+1276%	+1193%
100	+831%	+970%	+1289%
200	+825%	+754%	+1262%

 

SMA와 EMA는 50기간이 압도적으로 좋습니다. 필터가 너무 느리면(100, 200) 상승 초기에 "아직 MA 아래"라서 진입을 놓치는 경우가 생깁니다. 50기간은 추세 전환을 적당히 빨리 감지하면서도 노이즈는 걸러주는 균형점이었습니다.

 

HMA는 다른 패턴을 보입니다. 100기간이 +1289%로 가장 높고, 200기간도 +1262%로 큰 차이가 없습니다. HMA는 같은 기간이라도 반응이 빠르기 때문에, 100이나 200으로 설정해도 SMA/EMA의 50과 비슷한 속도로 추세를 감지합니다. 대신 그만큼 필터 효과가 약해져서 MDD 개선은 미미합니다.

 

SMA 50이 전체 1위(+1388%)이고, 필터 없음(+1173%) 대비 수익률이 약 18% 개선되었습니다. 진입 로직은 똑같은데, "언제 진입하지 않을 것인가"를 추가한 것만으로 이 차이가 납니다.

 

MDD: 느린 MA일수록 안정적

기간	SMA	EMA	HMA
50	-30.88%	-30.88%	-36.57%
100	-29.35%	-29.25%	-36.57%
200	-26.80%	-28.21%	-34.01%

 

MDD에서 MA 종류별 차이가 명확합니다. SMA와 EMA는 모든 기간에서 MDD를 줄여줬지만, HMA 50과 HMA 100은 MDD가 -36.57%로 필터 없음과 동일합니다. 필터가 없는 것이나 마찬가지라는 뜻입니다.

 

이유는 HMA의 빠른 반응에 있습니다. 하락 중에 반등이 오면 HMA는 빠르게 가격 위로 올라가서 "상승 추세"라고 판단합니다. 그래서 하락장에서의 가짜 돌파를 걸러내지 못합니다. 크로스 전략에서 HMA의 빠른 반응이 장점이었지만, 필터에서는 오히려 약점이 됩니다. 필터는 느릴수록 효과적입니다.

 

SMA 200이 -26.80%로 가장 안정적이지만, 수익률은 +825%로 가장 낮습니다. 수익률과 MDD는 여전히 트레이드오프 관계입니다.

---

시리즈 전체 비교

돈치안 채널부터 여기까지 4단계를 거쳤습니다.

단계	수익률	MDD
돈치안 (전액 투입)	+1117%	(전액 투입)
터틀 (기본)	+220%	-14.41%
터틀 + 피라미딩	+1173%	-36.57%
터틀 + 피라미딩 + SMA 50	+1388%	-30.88%

 

단순 브레이크아웃에서 출발해서 리스크 관리를 추가하고, 피라미딩으로 수익을 회복하고, MA 필터로 불필요한 진입을 줄이는 과정입니다. 최종 버전은 전액 투입 돈치안(+1117%)을 넘어서면서, MDD는 관리 가능한 수준(-30.88%)으로 유지되었습니다.

---

정리

이동평균을 매수/매도 신호로 직접 사용했을 때는 크로스 전략의 한계가 보였습니다. 하지만 다른 전략의 필터로 사용했을 때는 확실한 개선 효과가 있었습니다. 진입 로직을 바꾸지 않고, "진입하지 않을 조건"을 추가하는 것만으로 수익률과 MDD 모두 개선된 것이 그 증거입니다.

 

또한 크로스 전략과 필터에서 요구하는 MA 특성이 정반대라는 점도 확인했습니다. 크로스에서는 SMA처럼 느린 MA가, 필터에서도 SMA처럼 느린 MA가 좋았습니다. 반면 HMA처럼 빠른 MA는 크로스에서는 선전했지만 필터로서는 효과가 약했습니다. 도구의 특성을 이해하고 용도에 맞게 쓰는 것이 중요합니다.

 

다만 이번 결과도 2018~2025 BTC 일봉 데이터에 한정된 것입니다. 이전에 다뤘듯, 특정 데이터에서 가장 좋은 조합이 항상 최선은 아닙니다. 중요한 건 "SMA 50이 최고"라는 결론이 아니라, "필터를 추가하면 불필요한 진입을 줄일 수 있다"는 구조적 교훈입니다.

 

여러 기간의 MA의 값들을 비교하여 필터로 사용하는 것 또한 가능합니다. 이번 포스트에서는 가격과 MA 라인을 비교했지만, 가격, 단기 MA 라인, 장기 MA 라인의 정배열이 이루어졌을 때를 필터로 사용하는 것 또한 가능합니다. 떠오르는 아이디어를 여러가지 테스트 해보시길 권장드립니다.

---

본 포스팅은 지표의 원리와 백테스트 과정을 정리한 기술 블로그이며, 특정 자산의 매매를 권유하지 않습니다.
백테스트 결과는 과거 데이터 기반이며 실제 수익을 보장하지 않습니다. 모든 투자의 책임은 본인에게 있습니다.