돌파매매는 언제 실패할까? 돈치안 채널을 규칙으로 바꾸는 법

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
ADX_WINDOW_DAYS = 14
ADX_MINIMUM = 25.0
RELATIVE_VOLUME_MINIMUM = 1.2
STOP_LOSS_RATE = 0.10
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["breakout_signal"] = (signal_data["close"] > signal_data["entry_channel"]) & (signal_data["close"].shift(1) <= signal_data["entry_channel"].shift(1))
    signal_data["exit_signal"] = (signal_data["close"] < signal_data["exit_channel"]) & (signal_data["close"].shift(1) >= signal_data["exit_channel"].shift(1))
    previous_close = signal_data["close"].shift(1)
    true_range = pd.concat([(signal_data["high"] - signal_data["low"]), (signal_data["high"] - previous_close).abs(), (signal_data["low"] - previous_close).abs()], axis=1).max(axis=1)
    upward_move = signal_data["high"].diff()
    downward_move = -signal_data["low"].diff()
    positive_directional_move = upward_move.where((upward_move > downward_move) & (upward_move > 0.0), 0.0)
    negative_directional_move = downward_move.where((downward_move > upward_move) & (downward_move > 0.0), 0.0)
    average_true_range = true_range.ewm(alpha=1.0 / ADX_WINDOW_DAYS, adjust=False, min_periods=ADX_WINDOW_DAYS).mean()
    positive_directional_index = 100.0 * positive_directional_move.ewm(alpha=1.0 / ADX_WINDOW_DAYS, adjust=False, min_periods=ADX_WINDOW_DAYS).mean() / average_true_range
    negative_directional_index = 100.0 * negative_directional_move.ewm(alpha=1.0 / ADX_WINDOW_DAYS, adjust=False, min_periods=ADX_WINDOW_DAYS).mean() / average_true_range
    directional_index = 100.0 * (positive_directional_index - negative_directional_index).abs() / (positive_directional_index + negative_directional_index)
    signal_data["adx"] = directional_index.ewm(alpha=1.0 / ADX_WINDOW_DAYS, adjust=False, min_periods=ADX_WINDOW_DAYS).mean()
    signal_data["relative_volume"] = signal_data["volume"] / signal_data["volume"].shift(1).rolling(20).mean()
    signal_data["baseline_entry"] = signal_data["breakout_signal"]
    signal_data["relative_volume_entry"] = signal_data["breakout_signal"] & (signal_data["relative_volume"] >= RELATIVE_VOLUME_MINIMUM)
    signal_data["adx_entry"] = signal_data["breakout_signal"] & (signal_data["adx"] >= ADX_MINIMUM)
    signal_data["combined_entry"] = signal_data["relative_volume_entry"] & (signal_data["adx"] >= ADX_MINIMUM)
    return signal_data.loc[TEST_START:].dropna(subset=["entry_channel", "exit_channel"])


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
    capital_before_entry = 0.0
    pending_order = ""
    trade_returns: list[float] = []
    equity_values: list[float] = []

    for _, price_row in signal_data.iterrows():
        if pending_order == "buy" and position_shares == 0.0:
            capital_before_entry = cash
            entry_price = float(price_row["open"])
            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
            if close_return <= -STOP_LOSS_RATE or bool(price_row["exit_signal"]):
                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 = {
        "누적수익률": 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


strategy_columns = {
    "기준 돌파": "baseline_entry",
    "거래량 확인": "relative_volume_entry",
    "ADX 추세 확인": "adx_entry",
    "거래량·ADX 확인": "combined_entry",
}


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 main() -> None:
    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)})
    print(pd.DataFrame(summary_rows).to_string(index=False, formatters={"누적수익률": "{:+.1%}".format, "MDD": "{:.1%}".format, "승률": "{:.1%}".format}))


if __name__ == "__main__":
    main()