EMA 교차 백테스트: 한 구간의 성공을 믿으면 안 되는 이유

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"]
PERIODS = {"2014~2017": ("2014-01-01", "2018-01-01"), "2018~2021": ("2018-01-01", "2022-01-01"), "2022~2025": ("2022-01-01", "2026-01-01")}
DOWNLOAD_START = "2013-01-01"
DOWNLOAD_END = "2026-01-01"
FAST_WINDOW = 20
SLOW_WINDOW = 100
COST_RATE = 0.0005
MIN_ROWS = 252
CHART_FILE = Path("ema_robustness_chart.png")
RESULT_FILE = Path("ema_robustness_results.csv")
SUMMARY_FILE = Path("ema_robustness_summary.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_slow"] = data["close"].ewm(span=SLOW_WINDOW, adjust=False, min_periods=SLOW_WINDOW).mean()
    data["entry_signal"] = (data["ema_fast"] > data["ema_slow"]) & (data["ema_fast"].shift(1) <= data["ema_slow"].shift(1))
    data["exit_signal"] = (data["ema_fast"] < data["ema_slow"]) & (data["ema_fast"].shift(1) >= data["ema_slow"].shift(1))
    return data


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

    for date, row in 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_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
                trades.append({**current_trade, "exit_signal_date": pending_order["signal_date"], "exit_date": date, "exit_price": exit_price, "trade_return": cash / current_trade["capital_before"] - 1.0, "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_signal"]):
            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_date = data.index[-1]
        last_close = float(data.iloc[-1]["close"])
        cash = shares * last_close * (1.0 - COST_RATE)
        trades.append({**current_trade, "exit_signal_date": last_date, "exit_date": last_date, "exit_price": last_close, "trade_return": cash / current_trade["capital_before"] - 1.0, "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 summarize_sample(ticker: str, period: str, data: pd.DataFrame, trades: pd.DataFrame, daily: pd.DataFrame) -> dict[str, float | str]:
    drawdown = daily["equity"] / daily["equity"].cummax() - 1.0
    buy_hold = float(data["close"].iloc[-1] * (1.0 - COST_RATE) / (data["open"].iloc[0] / (1.0 - COST_RATE)) - 1.0)
    return {"ticker": ticker, "period": period, "rows": len(data), "strategy_return": float(daily["equity"].iloc[-1] - 1.0), "buy_hold_return": buy_hold, "excess_return": float(daily["equity"].iloc[-1] - 1.0 - buy_hold), "mdd": float(drawdown.min()), "trade_count": len(trades), "win_rate": float((trades["trade_return"] > 0.0).mean()) if not trades.empty else np.nan, "exposure": float(daily["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(results: pd.DataFrame, summaries: pd.DataFrame, portfolios: dict[str, pd.DataFrame]) -> None:
    colors = {"2014~2017": "#64748B", "2018~2021": "#F59E0B", "2022~2025": "#0F766E"}
    fig = plt.figure(figsize=(13, 10), facecolor="white")
    grid = fig.add_gridspec(2, 2, height_ratios=[1.8, 1.3], hspace=0.34, wspace=0.24)
    equity_axis = fig.add_subplot(grid[0, :])
    scatter_axis = fig.add_subplot(grid[1, 0])
    ratio_axis = fig.add_subplot(grid[1, 1])

    for period, portfolio in portfolios.items():
        elapsed = np.arange(len(portfolio)) / 252.0
        equity_axis.plot(elapsed, portfolio["equity"], color=colors[period], linewidth=2.4, label=f"{period} {portfolio['equity'].iloc[-1] - 1.0:+.1%}")
    equity_axis.axhline(1.0, color="#9CA3AF", linewidth=1.0)
    equity_axis.set_title("같은 EMA 규칙도 검증 구간이 바뀌면 다른 경로를 만든다", fontsize=18, fontweight="bold", pad=16)
    equity_axis.set_xlabel("구간 시작 후 경과 연수")
    equity_axis.set_ylabel("유효 종목 동일 비중 누적 자산")
    equity_axis.legend(loc="upper left", frameon=False, ncol=3)

    for period, group in results.groupby("period"):
        scatter_axis.scatter(group["buy_hold_return"] * 100.0, group["strategy_return"] * 100.0, s=48, alpha=0.72, color=colors[period], label=period)
    bounds = results[["buy_hold_return", "strategy_return"]].to_numpy() * 100.0
    low = float(np.nanmin(bounds)) - 5.0
    high = float(np.nanmax(bounds)) + 5.0
    scatter_axis.plot([low, high], [low, high], color="#9CA3AF", linewidth=1.0, linestyle="--")
    scatter_axis.axhline(0.0, color="#D1D5DB", linewidth=0.8)
    scatter_axis.axvline(0.0, color="#D1D5DB", linewidth=0.8)
    scatter_axis.set_xlim(low, high)
    scatter_axis.set_ylim(low, high)
    scatter_axis.set_title("종목·구간별 EMA 전략과 매수보유", fontsize=14, fontweight="bold")
    scatter_axis.set_xlabel("매수보유 수익률(%)")
    scatter_axis.set_ylabel("EMA 교차 수익률(%)")
    scatter_axis.legend(loc="upper left", frameon=False, fontsize=8)

    x = np.arange(len(summaries))
    width = 0.34
    ratio_axis.bar(x - width / 2, summaries["profitable_ratio"] * 100.0, width, color="#0F766E", label="수익 종목")
    ratio_axis.bar(x + width / 2, summaries["beat_buy_hold_ratio"] * 100.0, width, color="#F59E0B", label="매수보유 초과")
    ratio_axis.set_xticks(x, summaries.index)
    ratio_axis.set_ylim(0, 100)
    ratio_axis.set_ylabel("종목 비율(%)")
    ratio_axis.set_title("수익 종목과 매수보유 초과 종목은 달랐다", fontsize=14, fontweight="bold")
    ratio_axis.legend(frameon=False)
    for container in ratio_axis.containers:
        ratio_axis.bar_label(container, labels=[f"{value.get_height():.0f}%" for value in container], padding=3, fontsize=9)

    for axis in (equity_axis, scatter_axis, ratio_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=DOWNLOAD_END, auto_adjust=True, progress=False, group_by="ticker", threads=False)
if raw.empty:
    raise RuntimeError("가격 데이터를 받지 못했습니다.")

price_by_ticker = {ticker: prepare_price(raw, ticker) for ticker in TICKERS}
result_rows = []
portfolio_results = {}
for period, (start, end) in PERIODS.items():
    equity_series = []
    for ticker, full_data in price_by_ticker.items():
        data = full_data.loc[(full_data.index >= start) & (full_data.index < end)].copy()
        if len(data) < MIN_ROWS or data.dropna(subset=["ema_fast", "ema_slow"]).empty:
            continue
        trades, daily = run_backtest(data)
        result_rows.append(summarize_sample(ticker, period, data, trades, daily))
        equity_series.append(daily["equity"].rename(ticker))
    aligned = pd.concat(equity_series, axis=1).ffill().fillna(1.0)
    portfolio = pd.DataFrame({"equity": aligned.mean(axis=1)})
    portfolio["drawdown"] = portfolio["equity"] / portfolio["equity"].cummax() - 1.0
    portfolio_results[period] = portfolio

results = pd.DataFrame(result_rows)
summary_rows = []
for period, group in results.groupby("period", sort=False):
    portfolio = portfolio_results[period]
    summary_rows.append({"period": period, "ticker_count": len(group), "portfolio_return": float(portfolio["equity"].iloc[-1] - 1.0), "portfolio_mdd": float(portfolio["drawdown"].min()), "median_return": float(group["strategy_return"].median()), "profitable_ratio": float((group["strategy_return"] > 0.0).mean()), "beat_buy_hold_ratio": float((group["excess_return"] > 0.0).mean()), "average_trades": float(group["trade_count"].mean()), "average_exposure": float(group["exposure"].mean())})
summaries = pd.DataFrame(summary_rows).set_index("period")
results.to_csv(RESULT_FILE, index=False, encoding="utf-8-sig")
summaries.to_csv(SUMMARY_FILE, encoding="utf-8-sig")
draw_chart(results, summaries, portfolio_results)

common = results.pivot(index="ticker", columns="period", values="strategy_return").dropna()
all_positive_ratio = float((common > 0.0).all(axis=1).mean())
print(summaries.to_string(formatters={"portfolio_return": "{:+.2%}".format, "portfolio_mdd": "{:.2%}".format, "median_return": "{:+.2%}".format, "profitable_ratio": "{:.2%}".format, "beat_buy_hold_ratio": "{:.2%}".format, "average_trades": "{:.1f}".format, "average_exposure": "{:.2%}".format}))
print(f"전체 표본: {len(results)}개")
print(f"세 구간 모두 수익인 공통 종목 비율: {all_positive_ratio:.2%} ({int((common > 0.0).all(axis=1).sum())}/{len(common)})")
print(f"결과 저장: {RESULT_FILE}")
print(f"요약 저장: {SUMMARY_FILE}")
print(f"차트 저장: {CHART_FILE}")