평균회귀 전략은 왜 하락 추세에서 무너질까? 과매도 필터를 검증해봤다

Z-스코어 = (종가 - 20일 평균) ÷ 20일 표준편차

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"
ZSCORE_WINDOW = 20
ZSCORE_ENTRY_LEVEL = -2.0
TREND_WINDOW = 120
TREND_SLOPE_WINDOW = 20
STOP_LOSS_RATE = 0.10
MAX_HOLDING_DAYS = 10
ONE_WAY_COST_RATE = 0.0015


def add_signals(price_data: pd.DataFrame) -> pd.DataFrame:
    signal_data = price_data.copy()
    signal_data["sma20"] = signal_data["close"].rolling(ZSCORE_WINDOW).mean()
    signal_data["std20"] = signal_data["close"].rolling(ZSCORE_WINDOW).std(ddof=0)
    signal_data["zscore"] = (signal_data["close"] - signal_data["sma20"]) / signal_data["std20"]
    signal_data["sma120"] = signal_data["close"].rolling(TREND_WINDOW).mean()
    oversold_entry = (signal_data["zscore"] <= ZSCORE_ENTRY_LEVEL) & (signal_data["zscore"].shift(1) > ZSCORE_ENTRY_LEVEL)
    rising_trend = (signal_data["close"] > signal_data["sma120"]) & (signal_data["sma120"] > signal_data["sma120"].shift(TREND_SLOPE_WINDOW))
    signal_data["oversold_entry"] = oversold_entry
    signal_data["trend_entry"] = oversold_entry & rising_trend
    return signal_data.loc[TEST_START:TEST_END].dropna(subset=["zscore", "sma120"])


def run_backtest(signal_data: pd.DataFrame, entry_column: str) -> tuple[pd.Series, list[float]]:
    cash = 1.0
    position_shares = 0.0
    entry_price = 0.0
    entry_index = 0
    capital_before_entry = 0.0
    pending_order = ""
    trade_returns: list[float] = []
    equity_values: list[float] = []

    for row_index, (_, price_row) in enumerate(signal_data.iterrows()):
        if pending_order == "buy" and position_shares == 0.0:
            capital_before_entry = cash
            entry_price = float(price_row["open"])
            entry_index = row_index
            position_shares = cash * (1.0 - ONE_WAY_COST_RATE) / entry_price
            cash = 0.0
        elif pending_order == "sell" and position_shares > 0.0:
            exit_price = float(price_row["open"])
            cash = position_shares * exit_price * (1.0 - ONE_WAY_COST_RATE)
            trade_returns.append(cash / capital_before_entry - 1.0)
            position_shares = 0.0
        pending_order = ""

        if position_shares > 0.0:
            close_return = float(price_row["close"]) / entry_price - 1.0
            holding_days = row_index - entry_index
            if close_return <= -STOP_LOSS_RATE or float(price_row["close"]) >= float(price_row["sma20"]) or holding_days >= MAX_HOLDING_DAYS:
                pending_order = "sell"
        elif bool(price_row[entry_column]):
            pending_order = "buy"
        equity_values.append(cash + position_shares * float(price_row["close"]))

    if position_shares > 0.0:
        cash = position_shares * float(signal_data.iloc[-1]["close"]) * (1.0 - ONE_WAY_COST_RATE)
        trade_returns.append(cash / capital_before_entry - 1.0)
        equity_values[-1] = cash
    return pd.Series(equity_values, index=signal_data.index), trade_returns


def calculate_max_recovery_days(equity_curve: pd.Series) -> int:
    underwater = equity_curve < equity_curve.cummax()
    longest_days = 0
    underwater_start = None
    for trade_day, is_underwater in underwater.items():
        if is_underwater and underwater_start is None:
            underwater_start = trade_day
        elif not is_underwater and underwater_start is not None:
            longest_days = max(longest_days, (trade_day - underwater_start).days)
            underwater_start = None
    if underwater_start is not None:
        longest_days = max(longest_days, (equity_curve.index[-1] - underwater_start).days)
    return longest_days


def summarize_strategy(equity_curves: list[pd.Series], trade_returns: list[float]) -> 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
    return {
        "누적수익률": float(portfolio_equity.iloc[-1] - 1.0),
        "MDD": float(drawdown_curve.min()),
        "최장 회복일": calculate_max_recovery_days(portfolio_equity),
        "거래 수": len(trade_returns),
        "승률": float(sum(trade_return > 0.0 for trade_return in trade_returns) / len(trade_returns)),
    }


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() -> pd.DataFrame:
    strategy_columns = {"과매도만 확인": "oversold_entry", "상승 추세 확인": "trend_entry"}
    equity_by_strategy = {strategy_name: [] for strategy_name in strategy_columns}
    returns_by_strategy = {strategy_name: [] for strategy_name in strategy_columns}
    for ticker in TICKERS:
        signal_data = add_signals(download_price_data(ticker))
        for strategy_name, entry_column in strategy_columns.items():
            equity_curve, trade_returns = run_backtest(signal_data, entry_column)
            equity_by_strategy[strategy_name].append(equity_curve)
            returns_by_strategy[strategy_name].extend(trade_returns)
    return pd.DataFrame(
        [
            {"전략": strategy_name, **summarize_strategy(equity_by_strategy[strategy_name], returns_by_strategy[strategy_name])}
            for strategy_name in strategy_columns
        ]
    )


def main() -> None:
    summary = build_summary()
    print(summary.to_string(index=False, formatters={"누적수익률": "{:+.1%}".format, "MDD": "{:.1%}".format, "승률": "{:.1%}".format}))


if __name__ == "__main__":
    main()