변동성 돌파 전략, 시가에 샀다고 계산하면 왜 수익률이 부풀까?

돌파 가격 = 전일 종가 + 0.5 × (전일 고가 - 전일 저가)

python -m pip install pandas yfinance

"""yfinance 수정 일봉으로 변동성 돌파 전략의 체결 가정을 비교하는 본문 예제."""

import pandas as pd
import yfinance as yf


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 = "2012-01-01"
TEST_START = "2014-01-01"
TEST_END = "2026-01-01"
BREAKOUT_MULTIPLIER = 0.5
ONE_WAY_COST_RATE = 0.0015


def add_breakout_signal(price_data: pd.DataFrame) -> pd.DataFrame:
    signal_data = price_data.copy()
    signal_data["previous_close"] = signal_data["close"].shift(1)
    signal_data["previous_range"] = signal_data["high"].shift(1) - signal_data["low"].shift(1)
    signal_data["breakout_price"] = signal_data["previous_close"] + BREAKOUT_MULTIPLIER * signal_data["previous_range"]
    signal_data["breakout_signal"] = signal_data["high"] >= signal_data["breakout_price"]
    signal_data["gap_above_breakout"] = signal_data["open"] > signal_data["breakout_price"]
    return signal_data.loc[TEST_START:TEST_END].dropna(subset=["breakout_price"])


def build_trade_table(signal_data: pd.DataFrame, execution_rule: str) -> pd.DataFrame:
    trade_data = signal_data.loc[signal_data["breakout_signal"]].copy()
    if execution_rule == "retrospective_open":
        trade_data["entry_price"] = trade_data["open"]
        trade_data["exit_price"] = signal_data["open"].shift(-1).reindex(trade_data.index)
        exit_dates = pd.Series(signal_data.index, index=signal_data.index).shift(-1)
    elif execution_rule == "fixed_breakout":
        trade_data["entry_price"] = trade_data["breakout_price"]
        trade_data["exit_price"] = signal_data["open"].shift(-1).reindex(trade_data.index)
        exit_dates = pd.Series(signal_data.index, index=signal_data.index).shift(-1)
    elif execution_rule == "gap_aware_breakout":
        trade_data["entry_price"] = trade_data[["open", "breakout_price"]].max(axis=1)
        trade_data["exit_price"] = signal_data["open"].shift(-1).reindex(trade_data.index)
        exit_dates = pd.Series(signal_data.index, index=signal_data.index).shift(-1)
    elif execution_rule == "next_open_confirmation":
        trade_data["entry_price"] = signal_data["open"].shift(-1).reindex(trade_data.index)
        trade_data["exit_price"] = signal_data["open"].shift(-2).reindex(trade_data.index)
        exit_dates = pd.Series(signal_data.index, index=signal_data.index).shift(-2)
    else:
        raise ValueError(f"알 수 없는 체결 규칙입니다: {execution_rule}")

    trade_data["exit_date"] = exit_dates.reindex(trade_data.index).values
    trade_data = trade_data.dropna(subset=["entry_price", "exit_price", "exit_date"])
    trade_data["trade_return"] = trade_data["exit_price"] / trade_data["entry_price"] * (1.0 - ONE_WAY_COST_RATE) ** 2 - 1.0
    return trade_data


def build_equity_curve(signal_data: pd.DataFrame, trade_data: pd.DataFrame) -> pd.Series:
    daily_multiplier = pd.Series(1.0, index=signal_data.index)
    return_by_exit_date = trade_data.groupby("exit_date")["trade_return"].apply(lambda values: (1.0 + values).prod())
    valid_exit_dates = pd.DatetimeIndex(return_by_exit_date.index).intersection(daily_multiplier.index)
    daily_multiplier.loc[valid_exit_dates] = return_by_exit_date.reindex(valid_exit_dates).values
    return daily_multiplier.cumprod()


def summarize_strategy(equity_curves: list[pd.Series], trade_tables: list[pd.DataFrame]) -> dict[str, float | int]:
    portfolio_equity = pd.concat(equity_curves, axis=1).ffill().fillna(1.0).mean(axis=1)
    drawdown_curve = portfolio_equity / portfolio_equity.cummax() - 1.0
    all_trades = pd.concat(trade_tables, ignore_index=True)
    return {
        "누적수익률": float(portfolio_equity.iloc[-1] - 1.0),
        "MDD": float(drawdown_curve.min()),
        "거래 수": int(len(all_trades)),
        "승률": float((all_trades["trade_return"] > 0.0).mean()),
        "평균 거래수익률": float(all_trades["trade_return"].mean()),
    }


def download_price_data(ticker: str) -> pd.DataFrame:
    downloaded_prices = yf.download(
        ticker,
        start=DOWNLOAD_START,
        end=TEST_END,
        auto_adjust=True,
        progress=False,
        multi_level_index=False,
        threads=False,
    )
    price_data = downloaded_prices.rename(columns=str.lower).dropna(subset=["open", "high", "low", "close"]).copy()
    if len(price_data) < 500:
        raise RuntimeError(f"검증 데이터가 부족합니다: {ticker}")
    return price_data


def build_summary() -> tuple[pd.DataFrame, pd.DataFrame]:
    execution_rules = {
        "당일 시가 소급 체결": "retrospective_open",
        "돌파가 고정 체결": "fixed_breakout",
        "갭 반영 돌파 체결": "gap_aware_breakout",
        "종가 확인 후 다음 시가": "next_open_confirmation",
    }
    equity_by_rule = {rule_name: [] for rule_name in execution_rules}
    trades_by_rule = {rule_name: [] for rule_name in execution_rules}

    for ticker in TICKERS:
        signal_data = add_breakout_signal(download_price_data(ticker))
        for rule_name, execution_rule in execution_rules.items():
            trade_data = build_trade_table(signal_data, execution_rule)
            equity_by_rule[rule_name].append(build_equity_curve(signal_data, trade_data))
            trades_by_rule[rule_name].append(trade_data)

    summary_rows = [
        {"체결 가정": rule_name, **summarize_strategy(equity_by_rule[rule_name], trades_by_rule[rule_name])}
        for rule_name in execution_rules
    ]
    fixed_breakout_trades = pd.concat(trades_by_rule["돌파가 고정 체결"], ignore_index=True)
    gap_aware_trades = pd.concat(trades_by_rule["갭 반영 돌파 체결"], ignore_index=True)
    gap_mask = gap_aware_trades["gap_above_breakout"]
    gap_summary = pd.DataFrame(
        [
            {
                "전체 신호": len(gap_aware_trades),
                "목표가 위 갭": int(gap_mask.sum()),
                "갭 비율": float(gap_mask.mean()),
                "고정 체결 갭 평균": float(fixed_breakout_trades.loc[gap_mask, "trade_return"].mean()),
                "갭 반영 평균": float(gap_aware_trades.loc[gap_mask, "trade_return"].mean()),
            }
        ]
    )
    return pd.DataFrame(summary_rows), gap_summary


def main() -> None:
    summary, gap_summary = build_summary()
    print(
        summary.to_string(
            index=False,
            formatters={
                "누적수익률": "{:+.1%}".format,
                "MDD": "{:.1%}".format,
                "승률": "{:.1%}".format,
                "평균 거래수익률": "{:+.3%}".format,
            },
        )
    )
    print(
        gap_summary.to_string(
            index=False,
            formatters={
                "갭 비율": "{:.1%}".format,
                "고정 체결 갭 평균": "{:+.3%}".format,
                "갭 반영 평균": "{:+.3%}".format,
            },
        )
    )


if __name__ == "__main__":
    main()