종가 신호를 바로 샀더니 82.9%, 파이썬 백테스트가 틀어진 이유

t일 장 마감
→ 종가와 거래량 확정
→ 진입 또는 청산 신호 계산
→ 주문 대기 상태로 전환
→ t+1일 시가에 주문 체결

미보유
→ 진입 신호
→ 매수 주문 대기
→ 다음 거래일 시가 체결
→ 보유
→ 청산 신호
→ 매도 주문 대기
→ 다음 거래일 시가 체결
→ 미보유

python -m pip install yfinance pandas matplotlib

from pathlib import Path

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


TICKER = "005930.KS"
START_DATE = "2018-01-01"
END_DATE = "2026-01-01"
VOLUME_MULTIPLIER = 1.2
COST_RATE = 0.0005
CHART_FILE = Path("execution_chart.png")
RESULT_FILE = Path("execution_results.csv")


def run_backtest(data: pd.DataFrame, execution_mode: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    if execution_mode not in {"same_close", "next_open"}:
        raise ValueError(f"지원하지 않는 체결 방식입니다: {execution_mode}")

    cash = 1.0
    shares = 0.0
    pending_order = None
    current_trade = None
    trades = []
    equity_records = []
    dates = list(data.index)

    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_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"]):
            if execution_mode == "same_close":
                exit_price = float(row["close"])
                cash = shares * exit_price * (1.0 - COST_RATE)
                shares = 0.0
                trades.append({**current_trade, "exit_signal_date": date, "exit_date": date, "exit_price": exit_price, "trade_return": cash / current_trade["capital_before"] - 1.0, "forced_exit": False})
                current_trade = None
            elif index + 1 < len(data):
                pending_order = {"side": "sell", "signal_date": date}
        elif shares == 0.0 and bool(row["entry_signal"]):
            if execution_mode == "same_close":
                capital_before = cash
                entry_price = float(row["close"])
                shares = cash * (1.0 - COST_RATE) / entry_price
                cash = 0.0
                current_trade = {"entry_signal_date": date, "entry_date": date, "entry_price": entry_price, "capital_before": capital_before}
            elif index + 1 < len(data):
                pending_order = {"side": "buy", "signal_date": date}

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

    if shares > 0.0:
        last_date = dates[-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})
        equity_records[-1]["equity"] = cash

    trade_frame = pd.DataFrame(trades)
    equity_frame = pd.DataFrame(equity_records).set_index("date")
    return trade_frame, equity_frame


def summarize(trades: pd.DataFrame, equity: pd.DataFrame) -> dict[str, float]:
    elapsed_years = (equity.index[-1] - equity.index[0]).days / 365.25
    total_return = float(equity["equity"].iloc[-1] - 1.0)
    cagr = float(equity["equity"].iloc[-1] ** (1.0 / elapsed_years) - 1.0)
    drawdown = equity["equity"] / equity["equity"].cummax() - 1.0
    return {"누적수익률": total_return, "CAGR": cagr, "MDD": float(drawdown.min()), "승률": float((trades["trade_return"] > 0.0).mean()), "거래수": float(len(trades))}


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

price = yf.download(TICKER, start=START_DATE, end=END_DATE, auto_adjust=True, progress=False, multi_level_index=False)
if price.empty:
    raise RuntimeError(f"가격 데이터를 받지 못했습니다: {TICKER}")

price = price.rename(columns=str.lower).dropna().copy()
price["sma_20"] = price["close"].rolling(20).mean()
price["sma_60"] = price["close"].rolling(60).mean()
price["sma_120"] = price["close"].rolling(120).mean()
price["volume_average_20"] = price["volume"].shift(1).rolling(20).mean()
price["entry_signal"] = (price["sma_60"] > price["sma_120"]) & (price["close"] > price["sma_120"]) & (price["close"] > price["sma_20"]) & (price["close"].shift(1) <= price["sma_20"].shift(1)) & (price["volume"] >= price["volume_average_20"] * VOLUME_MULTIPLIER)
price["exit_signal"] = (price["close"] < price["sma_20"]) & (price["close"].shift(1) >= price["sma_20"].shift(1))

same_close_trades, same_close_equity = run_backtest(price, "same_close")
next_open_trades, next_open_equity = run_backtest(price, "next_open")
same_close_summary = summarize(same_close_trades, same_close_equity)
next_open_summary = summarize(next_open_trades, next_open_equity)
summary = pd.DataFrame([same_close_summary, next_open_summary], index=["당일 종가 체결", "다음 날 시가 체결"])
summary["거래수"] = summary["거래수"].astype(int)

trade_comparison = same_close_trades[["entry_signal_date", "trade_return"]].rename(columns={"trade_return": "당일_종가_거래수익률"}).merge(next_open_trades[["entry_signal_date", "entry_date", "entry_price", "trade_return"]].rename(columns={"trade_return": "다음날_시가_거래수익률"}), on="entry_signal_date", how="inner")
trade_comparison["수익률차이"] = trade_comparison["당일_종가_거래수익률"] - trade_comparison["다음날_시가_거래수익률"]
trade_comparison.to_csv(RESULT_FILE, index=False, encoding="utf-8-sig")

print(summary.to_string(formatters={"누적수익률": "{:.2%}".format, "CAGR": "{:.2%}".format, "MDD": "{:.2%}".format, "승률": "{:.2%}".format}))
print(f"누적수익률 차이: {same_close_summary['누적수익률'] - next_open_summary['누적수익률']:.2%}p")
print(trade_comparison.tail(10).to_string(index=False))

fig, (equity_axis, difference_axis) = plt.subplots(2, 1, figsize=(12, 8), gridspec_kw={"height_ratios": [3, 1.4]}, facecolor="white")
equity_axis.plot(same_close_equity.index, same_close_equity["equity"], color="#F59E0B", linewidth=2.0, label=f"당일 종가 체결 {same_close_summary['누적수익률']:.1%}")
equity_axis.plot(next_open_equity.index, next_open_equity["equity"], color="#0F766E", linewidth=2.0, label=f"다음 날 시가 체결 {next_open_summary['누적수익률']:.1%}")
equity_axis.set_title("같은 종가 신호, 다른 체결 시점", fontsize=17, fontweight="bold", pad=16)
equity_axis.set_ylabel("누적 자산")
equity_axis.legend(loc="upper left", frameon=False)
equity_axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
equity_axis.yaxis.tick_right()
equity_axis.yaxis.set_label_position("right")

bar_colors = ["#F59E0B" if value >= 0.0 else "#0F766E" for value in trade_comparison["수익률차이"]]
difference_axis.bar(trade_comparison["entry_signal_date"], trade_comparison["수익률차이"] * 100.0, color=bar_colors, width=20.0)
difference_axis.axhline(0.0, color="#64748B", linewidth=1.0)
difference_axis.text(0.01, 0.92, "양수: 당일 종가 체결 수익률이 높음", transform=difference_axis.transAxes, color="#64748B", fontsize=10, va="top")
difference_axis.set_ylabel("종가-시가(%p)")
difference_axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
difference_axis.yaxis.tick_right()
difference_axis.yaxis.set_label_position("right")

for axis in (equity_axis, difference_axis):
    axis.spines["top"].set_visible(False)
    axis.spines["left"].set_visible(False)
    axis.spines["right"].set_color("#D1D5DB")
    axis.spines["bottom"].set_color("#D1D5DB")

fig.tight_layout()
fig.savefig(CHART_FILE, dpi=160, bbox_inches="tight", facecolor="white")
plt.show()

position = signal.shift(1)