삼중 EMA 전략: 휩쏘를 줄였는데 수익률도 낮아진 이유

휩쏘 = 진입 후 20거래일 이내에 청산했고 거래비용 반영 후 손실인 거래

python -m pip install yfinance pandas numpy matplotlib

from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import yfinance as yf
from matplotlib import font_manager


TICKERS = ["005930.KS", "000660.KS", "005380.KS", "000270.KS", "035420.KS", "035720.KS", "068270.KS", "207940.KS", "105560.KS", "055550.KS", "066570.KS", "012330.KS", "028260.KS", "017670.KS", "009150.KS", "003670.KS", "042700.KS", "000810.KS", "196170.KQ", "247540.KQ", "086520.KQ", "058470.KQ", "035900.KQ", "041510.KQ", "067160.KQ", "263750.KQ", "293490.KQ", "112040.KQ", "036930.KQ", "240810.KQ"]
DOWNLOAD_START = "2013-01-01"
TEST_START = "2014-01-01"
TEST_END = "2026-01-01"
FAST_WINDOW = 20
CONFIRM_WINDOW = 50
SLOW_WINDOW = 100
COST_RATE = 0.0005
WHIPSAW_DAYS = 20
MIN_ROWS = 252
CHART_FILE = Path("triple_ema_chart.png")
RESULT_FILE = Path("triple_ema_results.csv")
SIGNAL_FILE = Path("triple_ema_signals.csv")


def prepare_price(raw: pd.DataFrame, ticker: str) -> pd.DataFrame:
    data = raw[ticker].rename(columns=str.lower).dropna(subset=["open", "close"]).copy()
    data["ema_fast"] = data["close"].ewm(span=FAST_WINDOW, adjust=False, min_periods=FAST_WINDOW).mean()
    data["ema_confirm"] = data["close"].ewm(span=CONFIRM_WINDOW, adjust=False, min_periods=CONFIRM_WINDOW).mean()
    data["ema_slow"] = data["close"].ewm(span=SLOW_WINDOW, adjust=False, min_periods=SLOW_WINDOW).mean()
    data["two_regime"] = data["ema_fast"] > data["ema_slow"]
    data["three_regime"] = (data["ema_fast"] > data["ema_confirm"]) & (data["ema_confirm"] > data["ema_slow"])
    data["two_entry_signal"] = data["two_regime"] & ~data["two_regime"].shift(1, fill_value=False)
    data["three_entry_signal"] = data["three_regime"] & ~data["three_regime"].shift(1, fill_value=False)
    data["exit_signal"] = ~data["two_regime"] & data["two_regime"].shift(1, fill_value=False)
    return data.loc[(data.index >= TEST_START) & (data.index < TEST_END)].copy()


def run_backtest(data: pd.DataFrame, entry_column: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    cash = 1.0
    shares = 0.0
    pending_order = None
    current_trade = None
    trades = []
    daily_records = []

    for index, (date, row) in enumerate(data.iterrows()):
        if pending_order is not None:
            if pending_order["side"] == "buy" and shares == 0.0:
                capital_before = cash
                entry_price = float(row["open"])
                shares = cash * (1.0 - COST_RATE) / entry_price
                cash = 0.0
                current_trade = {"entry_signal_date": pending_order["signal_date"], "entry_date": date, "entry_index": index, "entry_price": entry_price, "capital_before": capital_before}
            elif pending_order["side"] == "sell" and shares > 0.0:
                exit_price = float(row["open"])
                cash = shares * exit_price * (1.0 - COST_RATE)
                shares = 0.0
                trade_return = cash / current_trade["capital_before"] - 1.0
                holding_days = index - current_trade["entry_index"]
                trades.append({**current_trade, "exit_signal_date": pending_order["signal_date"], "exit_date": date, "exit_price": exit_price, "trade_return": trade_return, "holding_days": holding_days, "whipsaw": trade_return < 0.0 and holding_days <= WHIPSAW_DAYS, "forced_exit": False})
                current_trade = None
            pending_order = None

        if shares > 0.0 and bool(row["exit_signal"]):
            pending_order = {"side": "sell", "signal_date": date}
        elif shares == 0.0 and bool(row[entry_column]):
            pending_order = {"side": "buy", "signal_date": date}

        daily_records.append({"date": date, "equity": cash + shares * float(row["close"]), "exposure": float(shares > 0.0)})

    if shares > 0.0:
        last_index = len(data) - 1
        last_date = data.index[-1]
        last_close = float(data.iloc[-1]["close"])
        cash = shares * last_close * (1.0 - COST_RATE)
        trade_return = cash / current_trade["capital_before"] - 1.0
        holding_days = last_index - current_trade["entry_index"]
        trades.append({**current_trade, "exit_signal_date": last_date, "exit_date": last_date, "exit_price": last_close, "trade_return": trade_return, "holding_days": holding_days, "whipsaw": trade_return < 0.0 and holding_days <= WHIPSAW_DAYS, "forced_exit": True})
        daily_records[-1]["equity"] = cash
        daily_records[-1]["exposure"] = 0.0

    return pd.DataFrame(trades), pd.DataFrame(daily_records).set_index("date")


def analyze_confirmations(data: pd.DataFrame, two_trades: pd.DataFrame, ticker: str) -> pd.DataFrame:
    rows = []
    positions = pd.Series(np.arange(len(data)), index=data.index)
    for trade in two_trades.itertuples():
        episode = data.loc[trade.entry_signal_date:trade.exit_signal_date]
        confirmations = episode.index[episode["three_entry_signal"]]
        confirmed = len(confirmations) > 0
        confirmation_date = confirmations[0] if confirmed else pd.NaT
        delay_days = int(positions.loc[confirmation_date] - positions.loc[trade.entry_signal_date]) if confirmed else np.nan
        rows.append({"ticker": ticker, "two_entry_signal_date": trade.entry_signal_date, "two_exit_signal_date": trade.exit_signal_date, "two_trade_return": trade.trade_return, "two_holding_days": trade.holding_days, "two_whipsaw": trade.whipsaw, "confirmed": confirmed, "confirmation_date": confirmation_date, "delay_days": delay_days})
    return pd.DataFrame(rows)


def summarize(trades: pd.DataFrame, portfolio: pd.DataFrame) -> dict[str, float]:
    drawdown = portfolio["equity"] / portfolio["equity"].cummax() - 1.0
    winners = trades.loc[trades["trade_return"] > 0.0, "trade_return"]
    losers = trades.loc[trades["trade_return"] < 0.0, "trade_return"]
    profit_factor = float(winners.sum() / losers.abs().sum())
    return {"누적수익률": float(portfolio["equity"].iloc[-1] - 1.0), "MDD": float(drawdown.min()), "Profit Factor": profit_factor, "승률": float((trades["trade_return"] > 0.0).mean()), "거래수": len(trades), "휩쏘수": int(trades["whipsaw"].sum()), "휩쏘비율": float(trades["whipsaw"].mean()), "평균보유일": float(trades["holding_days"].mean()), "노출비중": float(portfolio["exposure"].mean())}


def configure_font() -> None:
    available_fonts = {item.name for item in font_manager.fontManager.ttflist}
    for font_name in ("Malgun Gothic", "AppleGothic", "UnDotum", "DejaVu Sans"):
        if font_name in available_fonts:
            plt.rcParams["font.family"] = font_name
            break
    plt.rcParams["axes.unicode_minus"] = False


def draw_chart(summary: pd.DataFrame, portfolios: dict[str, pd.DataFrame], signals: pd.DataFrame) -> None:
    colors = {"2중 EMA": "#F59E0B", "3중 EMA 확인": "#0F766E"}
    fig = plt.figure(figsize=(13, 9), facecolor="white")
    grid = fig.add_gridspec(2, 2, height_ratios=[1.8, 1.2], hspace=0.35, wspace=0.24)
    equity_axis = fig.add_subplot(grid[0, :])
    count_axis = fig.add_subplot(grid[1, 0])
    delay_axis = fig.add_subplot(grid[1, 1])

    for label, portfolio in portfolios.items():
        equity_axis.plot(portfolio.index, portfolio["equity"], color=colors[label], linewidth=2.3, label=f"{label} {summary.loc[label, '누적수익률']:+.1%}")
    equity_axis.set_title("50일 EMA 확인선은 휩쏘와 함께 일부 추세도 건너뛴다", fontsize=18, fontweight="bold", pad=16)
    equity_axis.set_ylabel("30종목 동일 비중 누적 자산")
    equity_axis.legend(loc="upper left", frameon=False)

    labels = list(summary.index)
    x = np.arange(len(labels))
    width = 0.34
    count_axis.bar(x - width / 2, summary["거래수"], width, color="#94A3B8", label="전체 거래")
    count_axis.bar(x + width / 2, summary["휩쏘수"], width, color=[colors[label] for label in labels], label=f"{WHIPSAW_DAYS}일 이내 손실")
    count_axis.set_xticks(x, labels)
    count_axis.set_title("거래 수와 단기 손실 거래", fontsize=14, fontweight="bold")
    count_axis.set_ylabel("거래 수")
    count_axis.legend(frameon=False)
    for container in count_axis.containers:
        count_axis.bar_label(container, padding=3)

    confirmed_delays = signals.loc[signals["confirmed"], "delay_days"].dropna()
    delay_axis.hist(confirmed_delays, bins=np.arange(-0.5, max(confirmed_delays.max() + 2.0, 12.0), 3.0), color="#0F766E", edgecolor="white", alpha=0.88)
    delay_axis.axvline(confirmed_delays.median(), color="#F59E0B", linewidth=2.2, linestyle="--", label=f"중앙값 {confirmed_delays.median():.0f}거래일")
    delay_axis.set_title("3중 정배열 확인까지 걸린 시간", fontsize=14, fontweight="bold")
    delay_axis.set_xlabel("2중 EMA 신호 이후 거래일")
    delay_axis.set_ylabel("신호 수")
    delay_axis.legend(frameon=False)

    for axis in (equity_axis, count_axis, delay_axis):
        axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
        axis.spines["top"].set_visible(False)
        axis.spines["right"].set_visible(False)

    CHART_FILE.parent.mkdir(parents=True, exist_ok=True)
    fig.savefig(CHART_FILE, dpi=160, bbox_inches="tight", facecolor="white")
    plt.close(fig)


configure_font()
raw = yf.download(TICKERS, start=DOWNLOAD_START, end=TEST_END, auto_adjust=True, progress=False, group_by="ticker", threads=False)
if raw.empty:
    raise RuntimeError("가격 데이터를 받지 못했습니다.")

trade_frames = {"2중 EMA": [], "3중 EMA 확인": []}
equity_series = {"2중 EMA": [], "3중 EMA 확인": []}
exposure_series = {"2중 EMA": [], "3중 EMA 확인": []}
signal_frames = []
calendar = raw.loc[(raw.index >= TEST_START) & (raw.index < TEST_END)].index
for ticker in TICKERS:
    data = prepare_price(raw, ticker)
    if len(data) < MIN_ROWS or data.dropna(subset=["ema_fast", "ema_confirm", "ema_slow"]).empty:
        continue
    two_trades, two_daily = run_backtest(data, "two_entry_signal")
    three_trades, three_daily = run_backtest(data, "three_entry_signal")
    for label, trades, daily in (("2중 EMA", two_trades, two_daily), ("3중 EMA 확인", three_trades, three_daily)):
        trades.insert(0, "ticker", ticker)
        trade_frames[label].append(trades)
        equity_series[label].append(daily["equity"].reindex(calendar).ffill().fillna(1.0).rename(ticker))
        exposure_series[label].append(daily["exposure"].reindex(calendar).ffill().fillna(0.0).rename(ticker))
    signal_frames.append(analyze_confirmations(data, two_trades, ticker))

portfolios = {}
all_trades = {}
summary_rows = {}
for label in trade_frames:
    all_trades[label] = pd.concat(trade_frames[label], ignore_index=True)
    portfolios[label] = pd.DataFrame({"equity": pd.concat(equity_series[label], axis=1).mean(axis=1), "exposure": pd.concat(exposure_series[label], axis=1).mean(axis=1)})
    summary_rows[label] = summarize(all_trades[label], portfolios[label])

summary = pd.DataFrame(summary_rows).T
summary[["거래수", "휩쏘수"]] = summary[["거래수", "휩쏘수"]].astype(int)
signals = pd.concat(signal_frames, ignore_index=True)
skipped = signals.loc[~signals["confirmed"]]
confirmed = signals.loc[signals["confirmed"]]
summary.to_csv(RESULT_FILE, encoding="utf-8-sig")
signals.to_csv(SIGNAL_FILE, index=False, encoding="utf-8-sig")
draw_chart(summary, portfolios, signals)

print(summary.to_string(formatters={"누적수익률": "{:+.2%}".format, "MDD": "{:.2%}".format, "Profit Factor": "{:.2f}".format, "승률": "{:.2%}".format, "휩쏘비율": "{:.2%}".format, "평균보유일": "{:.1f}".format, "노출비중": "{:.2%}".format}))
print(f"2중 EMA 신호 중 3중 정배열 확인: {len(confirmed)}/{len(signals)} ({len(confirmed) / len(signals):.2%})")
print(f"3중 정배열 확인 지연 중앙값: {confirmed['delay_days'].median():.0f}거래일")
print(f"건너뛴 2중 EMA 거래: {len(skipped)}건, 이 중 손실 {int((skipped['two_trade_return'] < 0.0).sum())}건, 수익 {int((skipped['two_trade_return'] > 0.0).sum())}건")
print(f"결과 저장: {RESULT_FILE}")
print(f"신호 저장: {SIGNAL_FILE}")
print(f"차트 저장: {CHART_FILE}")