갭 상승을 따라 사면 왜 물릴까? 갭 지속·메우기 규칙 비교

갭 상승률 = 당일 시가 ÷ 전일 종가 - 1

당일 갭 메움 = 당일 저가 ≤ 전일 종가
당일 갭 미메움 = 당일 저가 > 전일 종가

종가 위치 = (종가 - 저가) ÷ (고가 - 저가)
강한 종가 = 종가 > 시가 그리고 종가 위치 ≥ 0.70

python -m pip install numpy pandas yfinance

#!/usr/bin/env python3
"""yfinance 수정 일봉으로 갭 상승의 지속·메우기 규칙을 비교한다."""

from __future__ import annotations

import numpy as np
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"
GAP_MINIMUM_RATE = 0.03
MINIMUM_AVERAGE_TRADING_VALUE = 10_000_000_000
CLOSE_LOCATION_MINIMUM = 0.70
RELATIVE_VOLUME_MINIMUM = 1.50
STOP_LOSS_RATE = 0.08
MAX_HOLDING_DAYS = 5
ONE_WAY_COST_RATE = 0.0015


def add_signals(price_data: pd.DataFrame) -> pd.DataFrame:
    signal_data = price_data.copy()
    signal_data["previous_close"] = signal_data["close"].shift(1)
    signal_data["gap_rate"] = signal_data["open"] / signal_data["previous_close"] - 1.0
    price_range = (signal_data["high"] - signal_data["low"]).replace(0.0, np.nan)
    signal_data["close_location"] = (signal_data["close"] - signal_data["low"]) / price_range
    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()
    liquid_market = signal_data["average_trading_value"] >= MINIMUM_AVERAGE_TRADING_VALUE
    signal_data["gap_entry"] = (signal_data["gap_rate"] >= GAP_MINIMUM_RATE) & liquid_market
    signal_data["gap_held_entry"] = signal_data["gap_entry"] & (signal_data["low"] > signal_data["previous_close"])
    signal_data["close_strength_entry"] = signal_data["gap_held_entry"] & (signal_data["close"] > signal_data["open"]) & (signal_data["close_location"] >= CLOSE_LOCATION_MINIMUM)
    signal_data["volume_entry"] = signal_data["close_strength_entry"] & (signal_data["relative_volume"] >= RELATIVE_VOLUME_MINIMUM)
    return signal_data.loc[TEST_START:TEST_END].copy()


def run_backtest(signal_data: pd.DataFrame, entry_column: str, same_open_entry: bool = False) -> tuple[pd.Series, pd.Series, list[dict]]:
    cash = 1.0
    position_shares = 0.0
    entry_price = 0.0
    capital_before_entry = 0.0
    entry_index = -1
    active_trade: dict = {}
    pending_order: tuple | None = None
    equity_values: list[float] = []
    exposure_values: list[float] = []
    trades: list[dict] = []

    for row_index, (date, price_row) in enumerate(signal_data.iterrows()):
        if pending_order is not None:
            action = pending_order[0]
            if action == "buy" and position_shares == 0.0:
                signal_date = pending_order[1]
                capital_before_entry = cash
                entry_price = float(price_row["open"])
                position_shares = cash * (1.0 - ONE_WAY_COST_RATE) / entry_price
                cash = 0.0
                entry_index = row_index
                active_trade = {**pending_order[2], "signal_date": signal_date, "entry_date": date, "entry_price": entry_price}
            elif action == "sell" and position_shares > 0.0:
                exit_price = float(price_row["open"])
                cash = position_shares * exit_price * (1.0 - ONE_WAY_COST_RATE)
                trades.append({**active_trade, "exit_date": date, "exit_price": exit_price, "exit_reason": pending_order[1], "holding_days": row_index - entry_index, "trade_return": cash / capital_before_entry - 1.0})
                position_shares = 0.0
                entry_price = 0.0
                entry_index = -1
                active_trade = {}
            pending_order = None

        if same_open_entry and position_shares == 0.0 and bool(price_row[entry_column]):
            capital_before_entry = cash
            entry_price = float(price_row["open"])
            position_shares = cash * (1.0 - ONE_WAY_COST_RATE) / entry_price
            cash = 0.0
            entry_index = row_index
            active_trade = {**signal_snapshot(price_row), "signal_date": date, "entry_date": date, "entry_price": entry_price}

        if position_shares > 0.0:
            close_return = float(price_row["close"]) / entry_price - 1.0
            holding_days = row_index - entry_index + 1
            if close_return <= -STOP_LOSS_RATE:
                pending_order = ("sell", "종가 기준 8% 손절")
            elif holding_days >= MAX_HOLDING_DAYS:
                pending_order = ("sell", "5거래일 시간 청산")
        elif not same_open_entry and bool(price_row[entry_column]):
            pending_order = ("buy", date, signal_snapshot(price_row))

        equity_values.append(cash + position_shares * float(price_row["close"]))
        exposure_values.append(float(position_shares > 0.0))

    if position_shares > 0.0:
        exit_price = float(signal_data.iloc[-1]["close"])
        cash = position_shares * exit_price * (1.0 - ONE_WAY_COST_RATE)
        trades.append({**active_trade, "exit_date": signal_data.index[-1], "exit_price": exit_price, "exit_reason": "검증 종료", "holding_days": len(signal_data) - entry_index, "trade_return": cash / capital_before_entry - 1.0})
        equity_values[-1] = cash
    return (
        pd.Series(equity_values, index=signal_data.index, name="equity"),
        pd.Series(exposure_values, index=signal_data.index, name="exposure"),
        trades,
    )


def signal_snapshot(price_row: pd.Series) -> dict[str, float | bool]:
    return {
        "gap_rate": float(price_row["gap_rate"]),
        "previous_close": float(price_row["previous_close"]),
        "signal_open": float(price_row["open"]),
        "signal_high": float(price_row["high"]),
        "signal_low": float(price_row["low"]),
        "signal_close": float(price_row["close"]),
        "gap_filled_same_day": bool(price_row["low"] <= price_row["previous_close"]),
        "close_location": float(price_row["close_location"]),
        "relative_volume": float(price_row["relative_volume"]),
        "average_trading_value": float(price_row["average_trading_value"]),
    }


def summarize_strategy(equity_curves: list[pd.Series], exposure_curves: list[pd.Series], trades: list[dict]) -> dict[str, float | int]:
    calendar = pd.DatetimeIndex(sorted(set().union(*(curve.index for curve in equity_curves))))
    portfolio_equity = pd.concat([curve.reindex(calendar).ffill().fillna(1.0) for curve in equity_curves], axis=1).mean(axis=1)
    portfolio_exposure = pd.concat([curve.reindex(calendar).fillna(0.0) for curve in exposure_curves], axis=1).mean(axis=1)
    drawdown = portfolio_equity / portfolio_equity.cummax() - 1.0
    trade_returns = pd.Series([trade["trade_return"] for trade in trades], dtype=float)
    return {
        "누적수익률": float(portfolio_equity.iloc[-1] - 1.0),
        "MDD": float(drawdown.min()),
        "평균노출": float(portfolio_exposure.mean()),
        "거래수": len(trades),
        "승률": float((trade_returns > 0.0).mean()),
        "평균거래수익률": float(trade_returns.mean()),
        "중앙거래수익률": float(trade_returns.median()),
    }


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


def main() -> None:
    strategies = {
        "낙관적 당일 시가": ("gap_entry", True),
        "모든 갭·다음 시가": ("gap_entry", False),
        "당일 미메움": ("gap_held_entry", False),
        "종가 강도": ("close_strength_entry", False),
        "거래량까지": ("volume_entry", False),
    }
    equity_by_strategy = {name: [] for name in strategies}
    exposure_by_strategy = {name: [] for name in strategies}
    trades_by_strategy = {name: [] for name in strategies}
    for ticker in TICKERS:
        signal_data = add_signals(download_price_data(ticker))
        for strategy_name, (entry_column, same_open_entry) in strategies.items():
            equity, exposure, trades = run_backtest(signal_data, entry_column, same_open_entry)
            equity_by_strategy[strategy_name].append(equity)
            exposure_by_strategy[strategy_name].append(exposure)
            trades_by_strategy[strategy_name].extend(trades)
    summary = pd.DataFrame(
        [
            {
                "전략": name,
                **summarize_strategy(
                    equity_by_strategy[name],
                    exposure_by_strategy[name],
                    trades_by_strategy[name],
                ),
            }
            for name in strategies
        ]
    )
    print(summary.to_string(index=False, formatters={"누적수익률": "{:+.1%}".format, "MDD": "{:.1%}".format, "평균노출": "{:.1%}".format, "승률": "{:.1%}".format, "평균거래수익률": "{:+.2%}".format, "중앙거래수익률": "{:+.2%}".format}))


if __name__ == "__main__":
    main()