눌림목 매수는 왜 계속 물릴까? 추세·거래량 조건을 검증해봤다

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"
PULLBACK_HIGH_WINDOW = 60
TREND_WINDOW = 120
TREND_SLOPE_WINDOW = 20
VOLUME_SHORT_WINDOW = 3
VOLUME_LONG_WINDOW = 20
MAX_PULLBACK_RATE = 0.15
MIN_PULLBACK_RATE = 0.03
VOLUME_CONTRACTION_MAXIMUM = 0.85
STOP_LOSS_RATE = 0.08
MAX_HOLDING_DAYS = 40
ONE_WAY_COST_RATE = 0.0015


def add_signals(price_data: pd.DataFrame) -> pd.DataFrame:
    signal_data = price_data.copy()
    signal_data["ema20"] = signal_data["close"].ewm(span=20, adjust=False, min_periods=20).mean()
    signal_data["sma120"] = signal_data["close"].rolling(TREND_WINDOW).mean()
    signal_data["previous_high"] = signal_data["high"].shift(1).rolling(PULLBACK_HIGH_WINDOW).max()
    signal_data["pullback_rate"] = signal_data["close"] / signal_data["previous_high"] - 1.0
    signal_data["volume_ratio"] = signal_data["volume"].shift(1).rolling(VOLUME_SHORT_WINDOW).mean() / signal_data["volume"].shift(1).rolling(VOLUME_LONG_WINDOW).mean()

    recovery_signal = (signal_data["close"] > signal_data["ema20"]) & (signal_data["close"].shift(1) <= signal_data["ema20"].shift(1))
    pullback_range = signal_data["pullback_rate"].between(-MAX_PULLBACK_RATE, -MIN_PULLBACK_RATE)
    rising_trend = (signal_data["close"] > signal_data["sma120"]) & (signal_data["sma120"] > signal_data["sma120"].shift(TREND_SLOPE_WINDOW))
    contracting_volume = signal_data["volume_ratio"] < VOLUME_CONTRACTION_MAXIMUM

    signal_data["recovery_entry"] = recovery_signal & pullback_range
    signal_data["trend_entry"] = signal_data["recovery_entry"] & rising_trend
    signal_data["trend_volume_entry"] = signal_data["trend_entry"] & contracting_volume
    return signal_data.loc[TEST_START:TEST_END].dropna(subset=["ema20", "sma120", "previous_high"])


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["ema20"]) 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 run_buy_and_hold(signal_data: pd.DataFrame) -> pd.Series:
    entry_price = float(signal_data.iloc[0]["open"])
    position_shares = (1.0 - ONE_WAY_COST_RATE) / entry_price
    equity_curve = position_shares * signal_data["close"]
    equity_curve.iloc[-1] *= 1.0 - ONE_WAY_COST_RATE
    return equity_curve


def summarize_strategy(equity_curves: list[pd.Series], trade_returns: list[float] | None) -> dict[str, float | int | None]:
    portfolio_equity = pd.concat(equity_curves, axis=1).ffill().fillna(1.0).mean(axis=1)
    drawdown_curve = portfolio_equity / portfolio_equity.cummax() - 1.0
    summary: dict[str, float | int | None] = {
        "누적수익률": float(portfolio_equity.iloc[-1] - 1.0),
        "MDD": float(drawdown_curve.min()),
    }
    if trade_returns is not None:
        summary["거래 수"] = len(trade_returns)
        summary["손실 거래 비율"] = float(sum(trade_return < 0.0 for trade_return in trade_returns) / len(trade_returns))
    return summary


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 = {
        "반등만 확인": "recovery_entry",
        "상승 추세 확인": "trend_entry",
        "상승 추세·거래량 수축 확인": "trend_volume_entry",
    }
    strategy_equity = {strategy_name: [] for strategy_name in strategy_columns}
    strategy_returns = {strategy_name: [] for strategy_name in strategy_columns}
    buy_and_hold_equity = []

    for ticker in TICKERS:
        signal_data = add_signals(download_price_data(ticker))
        buy_and_hold_equity.append(run_buy_and_hold(signal_data))
        for strategy_name, entry_column in strategy_columns.items():
            equity_curve, trade_returns = run_backtest(signal_data, entry_column)
            strategy_equity[strategy_name].append(equity_curve)
            strategy_returns[strategy_name].extend(trade_returns)

    summary_rows = [
        {"전략": strategy_name, **summarize_strategy(strategy_equity[strategy_name], strategy_returns[strategy_name])}
        for strategy_name in strategy_columns
    ]
    summary_rows.append({"전략": "동일 종목 매수 후 보유", **summarize_strategy(buy_and_hold_equity, None)})
    return pd.DataFrame(summary_rows)


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


if __name__ == "__main__":
    main()