거래량이 20배 터졌는데 왜 실패할까? 거래량 돌파 전략 검증

상대 거래량 = 오늘 거래량 ÷ 직전 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"
ENTRY_LOOKBACK_DAYS = 20
EXIT_LOOKBACK_DAYS = 10
RELATIVE_VOLUME_MINIMUM = 2.0
MINIMUM_AVERAGE_TRADING_VALUE = 10_000_000_000
TREND_WINDOW = 120
TREND_SLOPE_WINDOW = 20
STOP_LOSS_RATE = 0.10
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["entry_channel"] = signal_data["high"].shift(1).rolling(ENTRY_LOOKBACK_DAYS).max()
    signal_data["exit_channel"] = signal_data["low"].shift(1).rolling(EXIT_LOOKBACK_DAYS).min()
    signal_data["relative_volume"] = signal_data["volume"] / signal_data["volume"].shift(1).rolling(20).mean()
    signal_data["average_trading_value"] = (signal_data["close"] * signal_data["volume"]).shift(1).rolling(20).mean()
    signal_data["sma120"] = signal_data["close"].rolling(TREND_WINDOW).mean()
    price_breakout = (signal_data["close"] > signal_data["entry_channel"]) & (signal_data["close"].shift(1) <= signal_data["entry_channel"].shift(1))
    volume_breakout = price_breakout & (signal_data["relative_volume"] >= RELATIVE_VOLUME_MINIMUM)
    liquid_market = signal_data["average_trading_value"] >= MINIMUM_AVERAGE_TRADING_VALUE
    rising_trend = (signal_data["close"] > signal_data["sma120"]) & (signal_data["sma120"] > signal_data["sma120"].shift(TREND_SLOPE_WINDOW))
    signal_data["price_entry"] = price_breakout
    signal_data["volume_entry"] = volume_breakout
    signal_data["liquid_entry"] = volume_breakout & liquid_market
    signal_data["trend_entry"] = volume_breakout & liquid_market & rising_trend
    signal_data["exit_signal"] = signal_data["close"] < signal_data["exit_channel"]
    return signal_data.loc[TEST_START:TEST_END].dropna(subset=["entry_channel", "exit_channel", "sma120"])


def run_backtest(signal_data: pd.DataFrame, entry_column: str) -> tuple[pd.Series, list[float]]:
    cash, position_shares, entry_price, capital_before_entry = 1.0, 0.0, 0.0, 0.0
    entry_index, pending_order = 0, ""
    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, entry_price, entry_index = cash, float(price_row["open"]), 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:
            cash = position_shares * float(price_row["open"]) * (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
            if close_return <= -STOP_LOSS_RATE or bool(price_row["exit_signal"]) or row_index - entry_index >= 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 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 = portfolio_equity / portfolio_equity.cummax() - 1.0
    return {"누적수익률": float(portfolio_equity.iloc[-1] - 1.0), "MDD": float(drawdown.min()), "거래 수": len(trade_returns), "승률": float(sum(value > 0.0 for value in trade_returns) / len(trade_returns))}


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


def main() -> None:
    strategy_columns = {"가격 돌파": "price_entry", "가격·거래량 돌파": "volume_entry", "거래대금 확인": "liquid_entry", "거래대금·상승 추세 확인": "trend_entry"}
    equity_by_strategy = {name: [] for name in strategy_columns}
    returns_by_strategy = {name: [] for 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, returns = run_backtest(signal_data, entry_column)
            equity_by_strategy[strategy_name].append(equity)
            returns_by_strategy[strategy_name].extend(returns)
    summary = pd.DataFrame([{"전략": name, **summarize_strategy(equity_by_strategy[name], returns_by_strategy[name])} for name in strategy_columns])
    print(summary.to_string(index=False, formatters={"누적수익률": "{:+.1%}".format, "MDD": "{:.1%}".format, "승률": "{:.1%}".format}))


if __name__ == "__main__":
    main()