최고 수익률만 고르면 과최적화일까? EMA 변수 49개를 검증해봤다

평탄 점수 = 주변 9개 누적수익률 중앙값 - 주변 9개 누적수익률 표준편차

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]
DOWNLOAD_START = "2012-01-01"
TRAIN_START = "2014-01-01"
TRAIN_END = "2022-01-01"
TEST_START = "2022-01-01"
TEST_END = "2026-01-01"
COST_RATE = 0.0005
MIN_ROWS = 252
PRICE_CACHE_FILE = Path("parameter_stability_prices.parquet")
CHART_FILE = Path("parameter_stability_chart.png")
RESULT_FILE = Path("parameter_stability_results.csv")
TRADE_FILE = Path("parameter_stability_trades.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=TEST_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 < TEST_END]) < MIN_ROWS:
            for attempt in range(3):
                retry = yf.download(ticker, start=DOWNLOAD_START, end=TEST_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 < TEST_END]) >= MIN_ROWS:
                    break
                time.sleep(1.0 + attempt)
        if len(frame.loc[frame.index < TEST_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_price(raw: pd.DataFrame) -> pd.DataFrame:
    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()
    return data

def run_strategy(data: pd.DataFrame, fast_window: int, slow_window: int, start: str, end: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    fast = data[f"ema_{fast_window}"]
    slow = data[f"ema_{slow_window}"]
    regime = fast > slow
    entry_signal = regime & ~regime.shift(1, fill_value=False)
    sample = data.loc[(data.index >= start) & (data.index < end)].copy()
    if len(sample) < MIN_ROWS:
        return pd.DataFrame(), pd.DataFrame()

    sample_regime = regime.reindex(sample.index).fillna(False)
    sample_entry = entry_signal.reindex(sample.index).fillna(False)
    entry_dates = sample.index[sample_entry]
    position_close = pd.Series(False, index=sample.index)
    if len(entry_dates):
        position_close.loc[entry_dates[0]:] = sample_regime.loc[entry_dates[0]:]
    position_open = position_close.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 = sample["close"].shift(1)

    factor = pd.Series(1.0, index=sample.index)
    factor.loc[entry_day] = sample.loc[entry_day, "close"] / sample.loc[entry_day, "open"] * (1.0 - COST_RATE)
    factor.loc[holding_day] = sample.loc[holding_day, "close"] / previous_close.loc[holding_day]
    factor.loc[exit_day] = sample.loc[exit_day, "open"] / previous_close.loc[exit_day] * (1.0 - COST_RATE)
    if bool(position_open.iloc[-1]):
        factor.iloc[-1] *= 1.0 - COST_RATE

    daily = pd.DataFrame({"equity": factor.cumprod(), "exposure": position_open.astype(float)})
    entries = list(sample.index[entry_day])
    exits = list(sample.index[exit_day])
    trades = []
    for entry_date in entries:
        later_exits = [exit_date for exit_date in exits if exit_date > entry_date]
        if later_exits:
            exit_date = later_exits[0]
            exit_price = float(sample.loc[exit_date, "open"])
            forced_exit = False
        else:
            exit_date = sample.index[-1]
            exit_price = float(sample.loc[exit_date, "close"])
            forced_exit = True
        entry_price = float(sample.loc[entry_date, "open"])
        holding_days = int(sample.index.get_loc(exit_date) - sample.index.get_loc(entry_date))
        trades.append({"entry_date": entry_date, "exit_date": exit_date, "entry_price": entry_price, "exit_price": exit_price, "trade_return": exit_price * (1.0 - COST_RATE) / (entry_price / (1.0 - COST_RATE)) - 1.0, "holding_days": holding_days, "forced_exit": forced_exit})
    return pd.DataFrame(trades), daily

def evaluate_combo(prepared: dict[str, pd.DataFrame], fast_window: int, slow_window: int, start: str, end: str) -> tuple[dict[str, float], pd.DataFrame, pd.DataFrame]:
    calendar = pd.DatetimeIndex(sorted(set().union(*(data.loc[(data.index >= start) & (data.index < end)].index for data in prepared.values()))))
    equity_series = []
    exposure_series = []
    trade_frames = []
    for ticker, data in prepared.items():
        trades, daily = run_strategy(data, fast_window, slow_window, start, end)
        if daily.empty:
            continue
        trades.insert(0, "ticker", ticker)
        trade_frames.append(trades)
        equity_series.append(daily["equity"].reindex(calendar).ffill().fillna(1.0).rename(ticker))
        exposure_series.append(daily["exposure"].reindex(calendar).ffill().fillna(0.0).rename(ticker))

    portfolio = pd.DataFrame({"equity": pd.concat(equity_series, axis=1).mean(axis=1), "exposure": pd.concat(exposure_series, axis=1).mean(axis=1)})
    trades = pd.concat(trade_frames, ignore_index=True)
    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"]
    summary = {"return": float(portfolio["equity"].iloc[-1] - 1.0), "mdd": float(drawdown.min()), "profit_factor": float(winners.sum() / losers.abs().sum()), "win_rate": float((trades["trade_return"] > 0.0).mean()), "trade_count": len(trades), "exposure": float(portfolio["exposure"].mean())}
    return summary, trades, portfolio

def select_candidates(results: pd.DataFrame) -> tuple[tuple[int, int], tuple[int, int], pd.DataFrame]:
    train_matrix = results.pivot(index="fast_window", columns="slow_window", values="train_return").reindex(index=FAST_WINDOWS, columns=SLOW_WINDOWS)
    peak_row = results.loc[results["train_return"].idxmax()]
    peak = (int(peak_row["fast_window"]), int(peak_row["slow_window"]))
    plateau_rows = []
    for fast_index in range(1, len(FAST_WINDOWS) - 1):
        for slow_index in range(1, len(SLOW_WINDOWS) - 1):
            neighborhood = train_matrix.iloc[fast_index - 1:fast_index + 2, slow_index - 1:slow_index + 2].to_numpy().ravel()
            median_return = float(np.nanmedian(neighborhood))
            standard_deviation = float(np.nanstd(neighborhood))
            plateau_rows.append({"fast_window": FAST_WINDOWS[fast_index], "slow_window": SLOW_WINDOWS[slow_index], "neighbor_median": median_return, "neighbor_std": standard_deviation, "neighbor_min": float(np.nanmin(neighborhood)), "plateau_score": median_return - standard_deviation})
    plateau_table = pd.DataFrame(plateau_rows)
    stable_row = plateau_table.loc[plateau_table["plateau_score"].idxmax()]
    stable = (int(stable_row["fast_window"]), int(stable_row["slow_window"]))
    return peak, stable, plateau_table

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_heatmap(axis: plt.Axes, matrix: pd.DataFrame, title: str, peak: tuple[int, int], stable: tuple[int, int]) -> None:
    values = matrix.to_numpy() * 100.0
    image = axis.imshow(values, cmap="YlGn", aspect="auto", origin="lower")
    axis.set_xticks(np.arange(len(SLOW_WINDOWS)), SLOW_WINDOWS)
    axis.set_yticks(np.arange(len(FAST_WINDOWS)), FAST_WINDOWS)
    axis.set_xlabel("느린 EMA 기간")
    axis.set_ylabel("빠른 EMA 기간")
    axis.set_title(title, fontsize=15, fontweight="bold")
    for row in range(len(FAST_WINDOWS)):
        for column in range(len(SLOW_WINDOWS)):
            axis.text(column, row, f"{values[row, column]:.0f}", ha="center", va="center", fontsize=8, color="#14213D")
    peak_position = (SLOW_WINDOWS.index(peak[1]), FAST_WINDOWS.index(peak[0]))
    stable_position = (SLOW_WINDOWS.index(stable[1]), FAST_WINDOWS.index(stable[0]))
    axis.add_patch(plt.Rectangle((peak_position[0] - 0.46, peak_position[1] - 0.46), 0.92, 0.92, fill=False, edgecolor="#DC2626", linewidth=3.0))
    axis.add_patch(plt.Rectangle((stable_position[0] - 0.40, stable_position[1] - 0.40), 0.80, 0.80, fill=False, edgecolor="#2563EB", linewidth=3.0))
    plt.colorbar(image, ax=axis, fraction=0.046, pad=0.04, label="누적수익률(%)")

def draw_chart(results: pd.DataFrame, peak: tuple[int, int], stable: tuple[int, int]) -> None:
    train_matrix = results.pivot(index="fast_window", columns="slow_window", values="train_return").reindex(index=FAST_WINDOWS, columns=SLOW_WINDOWS)
    test_matrix = results.pivot(index="fast_window", columns="slow_window", values="test_return").reindex(index=FAST_WINDOWS, columns=SLOW_WINDOWS)
    peak_row = results.loc[(results["fast_window"] == peak[0]) & (results["slow_window"] == peak[1])].iloc[0]
    stable_row = results.loc[(results["fast_window"] == stable[0]) & (results["slow_window"] == stable[1])].iloc[0]

    fig = plt.figure(figsize=(14, 10), facecolor="white")
    grid = fig.add_gridspec(2, 2, height_ratios=[1.55, 0.85], hspace=0.35, wspace=0.24)
    train_axis = fig.add_subplot(grid[0, 0])
    test_axis = fig.add_subplot(grid[0, 1])
    comparison_axis = fig.add_subplot(grid[1, :])
    draw_heatmap(train_axis, train_matrix, "변수 선택 구간", peak, stable)
    draw_heatmap(test_axis, test_matrix, "남겨둔 검증 구간", peak, stable)

    labels = [f"최고점\nEMA {peak[0]}/{peak[1]}", f"평탄 구간\nEMA {stable[0]}/{stable[1]}"]
    x = np.arange(2)
    width = 0.32
    train_values = np.array([peak_row["train_return"], stable_row["train_return"]]) * 100.0
    test_values = np.array([peak_row["test_return"], stable_row["test_return"]]) * 100.0
    comparison_axis.bar(x - width / 2, train_values, width, color="#0F766E", label="변수 선택 구간")
    comparison_axis.bar(x + width / 2, test_values, width, color="#F59E0B", label="남겨둔 검증 구간")
    comparison_axis.set_xticks(x, labels)
    comparison_axis.set_ylabel("누적수익률(%)")
    comparison_axis.set_title("최고점과 평탄 구간 후보의 성과 이동", fontsize=15, fontweight="bold")
    comparison_axis.legend(frameon=False, ncol=2)
    for container in comparison_axis.containers:
        comparison_axis.bar_label(container, labels=[f"{bar.get_height():.1f}%" for bar in container], padding=4, fontsize=10)
    comparison_axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
    comparison_axis.spines["top"].set_visible(False)
    comparison_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 = {ticker: prepare_price(frame) for ticker, frame in prices.items()}
result_rows = []
for fast_window in FAST_WINDOWS:
    for slow_window in SLOW_WINDOWS:
        train_summary, _, _ = evaluate_combo(prepared, fast_window, slow_window, TRAIN_START, TRAIN_END)
        test_summary, _, _ = evaluate_combo(prepared, fast_window, slow_window, TEST_START, TEST_END)
        result_rows.append({"fast_window": fast_window, "slow_window": slow_window, **{f"train_{key}": value for key, value in train_summary.items()}, **{f"test_{key}": value for key, value in test_summary.items()}})

results = pd.DataFrame(result_rows)
peak, stable, plateau_table = select_candidates(results)
results = results.merge(plateau_table, on=["fast_window", "slow_window"], how="left")
results["candidate"] = ""
results.loc[(results["fast_window"] == peak[0]) & (results["slow_window"] == peak[1]), "candidate"] = "최고점"
results.loc[(results["fast_window"] == stable[0]) & (results["slow_window"] == stable[1]), "candidate"] = "평탄 구간"

candidate_trades = []
for label, (fast_window, slow_window) in {"최고점": peak, "평탄 구간": stable}.items():
    for sample, start, end in (("변수 선택", TRAIN_START, TRAIN_END), ("검증", TEST_START, TEST_END)):
        _, trades, _ = evaluate_combo(prepared, fast_window, slow_window, start, end)
        candidate_trades.append(trades.assign(candidate=label, sample=sample, fast_window=fast_window, slow_window=slow_window))

tagged_trades = pd.concat(candidate_trades, ignore_index=True)
results.to_csv(RESULT_FILE, index=False, encoding="utf-8-sig")
tagged_trades.to_csv(TRADE_FILE, index=False, encoding="utf-8-sig")
draw_chart(results, peak, stable)

candidate_rows = results.loc[results["candidate"] != "", ["candidate", "fast_window", "slow_window", "train_return", "train_mdd", "test_return", "test_mdd", "neighbor_median", "neighbor_std", "neighbor_min"]]
print(candidate_rows.to_string(index=False, formatters={"train_return": "{:+.2%}".format, "train_mdd": "{:.2%}".format, "test_return": "{:+.2%}".format, "test_mdd": "{:.2%}".format, "neighbor_median": "{:+.2%}".format, "neighbor_std": "{:.2%}".format, "neighbor_min": "{:+.2%}".format}))
print(f"전체 변수 조합: {len(results)}개")
print(f"결과 저장: {RESULT_FILE}")
print(f"후보 거래 저장: {TRADE_FILE}")
print(f"차트 저장: {CHART_FILE}")