EMA 변수를 매년 다시 고르면 과최적화를 피할 수 있을까?

2014~2017 선택 → 2018 검증
2014~2018 선택 → 2019 검증
2014~2019 선택 → 2020 검증

2014~2017 선택 → 2018 검증
2015~2018 선택 → 2019 검증
2016~2019 선택 → 2020 검증

평탄 점수 = 주변 9개 CAGR 중앙값 - 주변 9개 CAGR 표준편차

python -m pip install yfinance pandas numpy matplotlib pyarrow

#!/usr/bin/env python3
"""EMA 변수 선택을 확장형과 롤링형 전진분석으로 반복해 표본외 성과를 비교한다."""

from __future__ import annotations

import time
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"]
FAST_WINDOWS = [10, 15, 20, 25, 30, 35, 40]
SLOW_WINDOWS = [60, 80, 100, 120, 140, 160, 180]
TEST_YEARS = list(range(2018, 2026))
DOWNLOAD_START = "2012-01-01"
DOWNLOAD_END = "2026-01-01"
INITIAL_TRAIN_START = "2014-01-01"
ROLLING_YEARS = 4
COST_RATE = 0.0005
MIN_ROWS = 252
MIN_PERIOD_ROWS = 200
PRICE_CACHE_FILE = Path("walk_forward_prices.parquet")
CHART_FILE = Path("walk_forward_chart.png")
STEP_FILE = Path("walk_forward_steps.csv")
SUMMARY_FILE = Path("walk_forward_summary.csv")

def read_price_cache(path: Path) -> dict[str, pd.DataFrame]:
    cached = pd.read_parquet(path)
    cached["date"] = pd.to_datetime(cached["date"])
    return {ticker: group.set_index("date")[["open", "high", "low", "close", "volume"]].sort_index() for ticker, group in cached.groupby("ticker")}

def download_prices() -> dict[str, pd.DataFrame]:
    if PRICE_CACHE_FILE.is_file():
        return read_price_cache(PRICE_CACHE_FILE)

    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("가격 데이터를 받지 못했습니다.")
    prices = {}
    for ticker in TICKERS:
        frame = raw[ticker].rename(columns=str.lower).dropna(subset=["open", "high", "low", "close"]).copy()
        if len(frame.loc[frame.index < DOWNLOAD_END]) < MIN_ROWS:
            for attempt in range(3):
                retry = yf.download(ticker, start=DOWNLOAD_START, end=DOWNLOAD_END, auto_adjust=True, progress=False, threads=False)
                if isinstance(retry.columns, pd.MultiIndex):
                    retry.columns = retry.columns.get_level_values(0)
                candidate = retry.rename(columns=str.lower).dropna(subset=["open", "high", "low", "close"]).copy()
                if len(candidate) > len(frame):
                    frame = candidate
                if len(frame.loc[frame.index < DOWNLOAD_END]) >= MIN_ROWS:
                    break
                time.sleep(1.0 + attempt)
        if len(frame.loc[frame.index < DOWNLOAD_END]) < MIN_ROWS:
            raise RuntimeError(f"검증에 필요한 가격 데이터가 부족합니다: {ticker}")
        prices[ticker] = frame
    PRICE_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
    cached = pd.concat([frame.reset_index(names="date").assign(ticker=ticker) for ticker, frame in prices.items()], ignore_index=True)
    cached.to_parquet(PRICE_CACHE_FILE, index=False)
    return prices

def prepare_prices(prices: dict[str, pd.DataFrame]) -> dict[str, pd.DataFrame]:
    prepared = {}
    for ticker, raw in prices.items():
        data = raw.copy()
        for window in sorted(set(FAST_WINDOWS + SLOW_WINDOWS)):
            data[f"ema_{window}"] = data["close"].ewm(span=window, adjust=False, min_periods=window).mean()
        prepared[ticker] = data
    return prepared

def build_strategy_matrices(prepared: dict[str, pd.DataFrame], fast_window: int, slow_window: int, calendar: pd.DatetimeIndex) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    factor_columns = []
    position_columns = []
    entry_factor_columns = []
    for ticker, data in prepared.items():
        regime = data[f"ema_{fast_window}"] > data[f"ema_{slow_window}"]
        position_open = regime.shift(1, fill_value=False)
        previous_position = position_open.shift(1, fill_value=False)
        entry_day = position_open & ~previous_position
        holding_day = position_open & previous_position
        exit_day = ~position_open & previous_position
        previous_close = data["close"].shift(1)
        factor = pd.Series(1.0, index=data.index)
        factor.loc[entry_day] = data.loc[entry_day, "close"] / data.loc[entry_day, "open"] * (1.0 - COST_RATE)
        factor.loc[holding_day] = data.loc[holding_day, "close"] / previous_close.loc[holding_day]
        factor.loc[exit_day] = data.loc[exit_day, "open"] / previous_close.loc[exit_day] * (1.0 - COST_RATE)
        entry_factor = data["close"] / data["open"] * (1.0 - COST_RATE)
        factor_columns.append(factor.reindex(calendar).fillna(1.0).rename(ticker))
        position_columns.append(position_open.reindex(calendar).fillna(False).rename(ticker))
        entry_factor_columns.append(entry_factor.reindex(calendar).fillna(1.0).rename(ticker))
    return pd.concat(factor_columns, axis=1), pd.concat(position_columns, axis=1).astype(bool), pd.concat(entry_factor_columns, axis=1)

def evaluate_period(factors: pd.DataFrame, positions: pd.DataFrame, entry_factors: pd.DataFrame, start: str, end: str, include_equity: bool = False) -> tuple[dict[str, float], pd.DataFrame | None]:
    mask = (factors.index >= start) & (factors.index < end)
    sample_factors = factors.loc[mask].copy()
    sample_positions = positions.loc[mask].copy()
    sample_entry_factors = entry_factors.loc[mask]
    if len(sample_factors) < MIN_PERIOD_ROWS:
        raise RuntimeError(f"검증 구간 데이터가 부족합니다: {start}~{end}")

    first_position = sample_positions.iloc[0]
    sample_factors.iloc[0] = np.where(first_position, sample_entry_factors.iloc[0], 1.0)
    last_position = sample_positions.iloc[-1]
    sample_factors.iloc[-1] *= np.where(last_position, 1.0 - COST_RATE, 1.0)
    ticker_equity = sample_factors.cumprod()
    equity = ticker_equity.mean(axis=1)
    drawdown = equity / equity.cummax() - 1.0
    elapsed_years = max((equity.index[-1] - equity.index[0]).days / 365.25, 1.0 / 365.25)
    entries = sample_positions.astype(int).diff().eq(1).sum().sum() + int(first_position.sum())
    summary = {"return": float(equity.iloc[-1] - 1.0), "cagr": float(equity.iloc[-1] ** (1.0 / elapsed_years) - 1.0), "mdd": float(drawdown.min()), "trade_count": int(entries), "exposure": float(sample_positions.mean(axis=1).mean())}
    daily = pd.DataFrame({"equity": equity, "drawdown": drawdown}) if include_equity else None
    return summary, daily

def select_plateau(grid: pd.DataFrame) -> tuple[int, int, float, float, float]:
    matrix = grid.pivot(index="fast_window", columns="slow_window", values="train_cagr").reindex(index=FAST_WINDOWS, columns=SLOW_WINDOWS)
    candidates = []
    for fast_index in range(1, len(FAST_WINDOWS) - 1):
        for slow_index in range(1, len(SLOW_WINDOWS) - 1):
            neighborhood = matrix.iloc[fast_index - 1:fast_index + 2, slow_index - 1:slow_index + 2].to_numpy().ravel()
            median_cagr = float(np.nanmedian(neighborhood))
            standard_deviation = float(np.nanstd(neighborhood))
            candidates.append({"fast_window": FAST_WINDOWS[fast_index], "slow_window": SLOW_WINDOWS[slow_index], "neighbor_median_cagr": median_cagr, "neighbor_std_cagr": standard_deviation, "plateau_score": median_cagr - standard_deviation})
    selected = pd.DataFrame(candidates).sort_values(["plateau_score", "neighbor_median_cagr"], ascending=False).iloc[0]
    return int(selected["fast_window"]), int(selected["slow_window"]), float(selected["neighbor_median_cagr"]), float(selected["neighbor_std_cagr"]), float(selected["plateau_score"])

def chain_segments(segments: list[pd.DataFrame]) -> pd.DataFrame:
    chained = []
    capital = 1.0
    for segment in segments:
        part = segment.copy()
        part["equity"] *= capital
        capital = float(part["equity"].iloc[-1])
        chained.append(part[["equity"]])
    result = pd.concat(chained)
    result["drawdown"] = result["equity"] / result["equity"].cummax() - 1.0
    return result

def summarize_method(method: str, steps: pd.DataFrame, equity: pd.DataFrame) -> dict[str, float | int | str]:
    elapsed_years = (equity.index[-1] - equity.index[0]).days / 365.25
    parameter_pairs = list(zip(steps["fast_window"], steps["slow_window"]))
    changes = sum(left != right for left, right in zip(parameter_pairs, parameter_pairs[1:]))
    efficiency = steps["test_return"] / steps["train_cagr"].replace(0.0, np.nan)
    return {"method": method, "return": float(equity["equity"].iloc[-1] - 1.0), "cagr": float(equity["equity"].iloc[-1] ** (1.0 / elapsed_years) - 1.0), "mdd": float(equity["drawdown"].min()), "positive_years": int((steps["test_return"] > 0.0).sum()), "parameter_changes": int(changes), "unique_parameters": int(len(set(parameter_pairs))), "median_forward_efficiency": float(efficiency.median()), "average_trades": float(steps["test_trade_count"].mean()), "average_exposure": float(steps["test_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(steps: pd.DataFrame, equities: dict[str, pd.DataFrame], summary: pd.DataFrame) -> None:
    colors = {"확장형": "#0F766E", "롤링형": "#F59E0B", "고정 EMA 20·100": "#64748B"}
    fig = plt.figure(figsize=(14, 10), facecolor="white")
    grid = fig.add_gridspec(2, 2, height_ratios=[1.35, 1.0], hspace=0.34, wspace=0.24)
    equity_axis = fig.add_subplot(grid[0, :])
    return_axis = fig.add_subplot(grid[1, 0])
    parameter_axis = fig.add_subplot(grid[1, 1])

    for method, equity in equities.items():
        label_row = summary.loc[summary["method"] == method].iloc[0]
        equity_axis.plot(equity.index, equity["equity"], color=colors[method], linewidth=2.5 if method != "고정 EMA 20·100" else 1.8, label=f"{method} {label_row['return']:+.1%} · MDD {label_row['mdd']:.1%}")
    equity_axis.axhline(1.0, color="#9CA3AF", linewidth=1.0)
    equity_axis.set_title("한 해씩 변수를 다시 고른 뒤 표본외 성과만 연결했다", fontsize=18, fontweight="bold", pad=16)
    equity_axis.set_ylabel("동일 비중 누적 자산")
    equity_axis.legend(loc="upper left", frameon=False, ncol=3)

    x = np.arange(len(TEST_YEARS))
    width = 0.36
    for offset, method in ((-width / 2, "확장형"), (width / 2, "롤링형")):
        values = steps.loc[steps["method"] == method].set_index("test_year").reindex(TEST_YEARS)["test_return"] * 100.0
        return_axis.bar(x + offset, values, width, color=colors[method], label=method)
    return_axis.axhline(0.0, color="#9CA3AF", linewidth=1.0)
    return_axis.set_xticks(x, [str(year)[2:] for year in TEST_YEARS])
    return_axis.set_xlabel("검증 연도")
    return_axis.set_ylabel("표본외 수익률(%)")
    return_axis.set_title("연도별 표본외 성과", fontsize=14, fontweight="bold")
    return_axis.legend(frameon=False)

    parameter_axis.set_xlim(-0.6, len(TEST_YEARS) - 0.4)
    parameter_axis.set_ylim(-0.6, 1.6)
    parameter_axis.set_xticks(x, [str(year)[2:] for year in TEST_YEARS])
    parameter_axis.set_yticks([0, 1], ["롤링형", "확장형"])
    parameter_axis.set_xlabel("검증 연도")
    parameter_axis.set_title("매년 다시 선택된 EMA 기간", fontsize=14, fontweight="bold")
    for row, method in enumerate(("롤링형", "확장형")):
        method_steps = steps.loc[steps["method"] == method].set_index("test_year").reindex(TEST_YEARS)
        for column, (_, selected) in enumerate(method_steps.iterrows()):
            rectangle = plt.Rectangle((column - 0.45, row - 0.35), 0.9, 0.7, facecolor=colors[method], alpha=0.16, edgecolor=colors[method], linewidth=1.4)
            parameter_axis.add_patch(rectangle)
            parameter_axis.text(column, row, f"{int(selected['fast_window'])}/{int(selected['slow_window'])}", ha="center", va="center", color="#14213D", fontsize=9, fontweight="bold")
    parameter_axis.grid(False)

    for axis in (equity_axis, return_axis, parameter_axis):
        axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
        axis.spines["top"].set_visible(False)
        axis.spines["right"].set_visible(False)
    fig.suptitle("확장형과 롤링형 전진분석은 같은 전략에서도 다른 변수 경로를 만든다", fontsize=21, fontweight="bold", y=0.99)
    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()
prices = download_prices()
prepared = prepare_prices(prices)
calendar = pd.DatetimeIndex(sorted(set().union(*(data.loc[(data.index >= INITIAL_TRAIN_START) & (data.index < DOWNLOAD_END)].index for data in prepared.values()))))
periods = set()
for test_year in TEST_YEARS:
    test_start = f"{test_year}-01-01"
    test_end = f"{test_year + 1}-01-01"
    periods.add((INITIAL_TRAIN_START, test_start))
    periods.add((f"{test_year - ROLLING_YEARS}-01-01", test_start))
    periods.add((test_start, test_end))

period_results = {}
for fast_window in FAST_WINDOWS:
    for slow_window in SLOW_WINDOWS:
        factors, positions, entry_factors = build_strategy_matrices(prepared, fast_window, slow_window, calendar)
        for start, end in periods:
            summary, _ = evaluate_period(factors, positions, entry_factors, start, end)
            period_results[(start, end, fast_window, slow_window)] = summary

step_rows = []
equity_segments = {"확장형": [], "롤링형": [], "고정 EMA 20·100": []}
for method in ("확장형", "롤링형"):
    for test_year in TEST_YEARS:
        train_start = INITIAL_TRAIN_START if method == "확장형" else f"{test_year - ROLLING_YEARS}-01-01"
        train_end = f"{test_year}-01-01"
        test_start = train_end
        test_end = f"{test_year + 1}-01-01"
        grid_rows = []
        for fast_window in FAST_WINDOWS:
            for slow_window in SLOW_WINDOWS:
                train = period_results[(train_start, train_end, fast_window, slow_window)]
                grid_rows.append({"fast_window": fast_window, "slow_window": slow_window, **{f"train_{key}": value for key, value in train.items()}})
        grid_frame = pd.DataFrame(grid_rows)
        fast_window, slow_window, neighbor_median_cagr, neighbor_std_cagr, plateau_score = select_plateau(grid_frame)
        train = period_results[(train_start, train_end, fast_window, slow_window)]
        test = period_results[(test_start, test_end, fast_window, slow_window)]
        factors, positions, entry_factors = build_strategy_matrices(prepared, fast_window, slow_window, calendar)
        _, test_equity = evaluate_period(factors, positions, entry_factors, test_start, test_end, include_equity=True)
        equity_segments[method].append(test_equity)
        step_rows.append({"method": method, "test_year": test_year, "train_start": train_start, "train_end": train_end, "fast_window": fast_window, "slow_window": slow_window, "neighbor_median_cagr": neighbor_median_cagr, "neighbor_std_cagr": neighbor_std_cagr, "plateau_score": plateau_score, **{f"train_{key}": value for key, value in train.items()}, **{f"test_{key}": value for key, value in test.items()}})

fixed_factors, fixed_positions, fixed_entry_factors = build_strategy_matrices(prepared, 20, 100, calendar)
fixed_rows = []
for test_year in TEST_YEARS:
    test_start = f"{test_year}-01-01"
    test_end = f"{test_year + 1}-01-01"
    fixed_summary, fixed_equity = evaluate_period(fixed_factors, fixed_positions, fixed_entry_factors, test_start, test_end, include_equity=True)
    equity_segments["고정 EMA 20·100"].append(fixed_equity)
    fixed_rows.append({"test_year": test_year, **fixed_summary})

steps = pd.DataFrame(step_rows)
equities = {method: chain_segments(segments) for method, segments in equity_segments.items()}
summary_rows = [summarize_method(method, steps.loc[steps["method"] == method], equities[method]) for method in ("확장형", "롤링형")]
fixed_steps = pd.DataFrame(fixed_rows).rename(columns={"return": "test_return", "trade_count": "test_trade_count", "exposure": "test_exposure"})
fixed_equity = equities["고정 EMA 20·100"]
elapsed_years = (fixed_equity.index[-1] - fixed_equity.index[0]).days / 365.25
summary_rows.append({"method": "고정 EMA 20·100", "return": float(fixed_equity["equity"].iloc[-1] - 1.0), "cagr": float(fixed_equity["equity"].iloc[-1] ** (1.0 / elapsed_years) - 1.0), "mdd": float(fixed_equity["drawdown"].min()), "positive_years": int((fixed_steps["test_return"] > 0.0).sum()), "parameter_changes": 0, "unique_parameters": 1, "median_forward_efficiency": np.nan, "average_trades": float(fixed_steps["test_trade_count"].mean()), "average_exposure": float(fixed_steps["test_exposure"].mean())})
summary = pd.DataFrame(summary_rows)
steps.to_csv(STEP_FILE, index=False, encoding="utf-8-sig")
summary.to_csv(SUMMARY_FILE, index=False, encoding="utf-8-sig")
draw_chart(steps, equities, summary)

print(summary.to_string(index=False, formatters={"return": "{:+.2%}".format, "cagr": "{:+.2%}".format, "mdd": "{:.2%}".format, "median_forward_efficiency": lambda value: "-" if pd.isna(value) else f"{value:.1%}", "average_trades": "{:.1f}".format, "average_exposure": "{:.1%}".format}))
print("\n선택 변수")
print(steps.pivot(index="test_year", columns="method", values=["fast_window", "slow_window", "test_return"]).to_string())
print(f"단계 결과 저장: {STEP_FILE}")
print(f"요약 저장: {SUMMARY_FILE}")
print(f"차트 저장: {CHART_FILE}")