손절 없는 전략은 승률이 높은데 왜 위험할까? 초기·추적·시간 손절 비교

python -m pip install pandas yfinance

"""yfinance 배당 조정 제외 OHLC로 초기·추적·시간 손절을 비교하는 본문 예제."""

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-08-29"
BREAKOUT_WINDOW = 20
TREND_WINDOW = 120
TREND_SLOPE_WINDOW = 20
MAXIMUM_HOLDING_DAYS = 60
INITIAL_STOP_RATE = 0.10
TRAILING_STOP_RATE = 0.10
TIME_STOP_DAYS = 20
ONE_WAY_COST_RATE = 0.0015


def add_entry_signal(price_data: pd.DataFrame) -> pd.DataFrame:
    signal_data = price_data.copy()
    signal_data["entry_channel"] = signal_data["high"].shift(1).rolling(BREAKOUT_WINDOW).max()
    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))
    rising_trend = (signal_data["close"] > signal_data["sma120"]) & (signal_data["sma120"] > signal_data["sma120"].shift(TREND_SLOPE_WINDOW))
    signal_data["entry_signal"] = price_breakout & rising_trend
    return signal_data.loc[TEST_START:TEST_END].dropna(subset=["entry_channel", "sma120"])


def run_backtest(signal_data: pd.DataFrame, stop_rule: str) -> tuple[pd.Series, list[dict[str, object]]]:
    cash, position_shares = 1.0, 0.0
    active_trade: dict[str, object] | None = None
    pending_order: dict[str, object] | None = None
    equity_values: list[float] = []
    trades: list[dict[str, object]] = []
    for row_index, (trade_day, price_row) in enumerate(signal_data.iterrows()):
        if pending_order is not None:
            if pending_order["side"] == "buy" and position_shares == 0.0:
                entry_price = float(price_row["open"])
                capital_before_entry = cash
                position_shares = cash * (1.0 - ONE_WAY_COST_RATE) / entry_price
                cash = 0.0
                active_trade = {"signal_date": pending_order["signal_date"], "entry_date": trade_day, "entry_index": row_index, "entry_price": entry_price, "capital": capital_before_entry, "highest_close": float(price_row["close"])}
            elif pending_order["side"] == "sell" and position_shares > 0.0 and active_trade is not None:
                exit_price = float(price_row["open"])
                cash = position_shares * exit_price * (1.0 - ONE_WAY_COST_RATE)
                position_shares = 0.0
                trades.append({**active_trade, "exit_date": trade_day, "exit_price": exit_price, "trade_return": cash / float(active_trade["capital"]) - 1.0, "holding_days": row_index - int(active_trade["entry_index"]), "exit_reason": pending_order["reason"]})
                active_trade = None
            pending_order = None
        if position_shares > 0.0 and active_trade is not None:
            active_trade["highest_close"] = max(float(active_trade["highest_close"]), float(price_row["close"]))
            close_return = float(price_row["close"]) / float(active_trade["entry_price"]) - 1.0
            trailing_return = float(price_row["close"]) / float(active_trade["highest_close"]) - 1.0
            holding_days = row_index - int(active_trade["entry_index"])
            exit_reason = ""
            if stop_rule == "initial" and close_return <= -INITIAL_STOP_RATE:
                exit_reason = "진입가 대비 10% 손절"
            elif stop_rule == "trailing" and trailing_return <= -TRAILING_STOP_RATE:
                exit_reason = "고점 대비 10% 추적 손절"
            elif stop_rule == "time" and holding_days >= TIME_STOP_DAYS and close_return <= 0.0:
                exit_reason = "20거래일 무수익 청산"
            elif holding_days >= MAXIMUM_HOLDING_DAYS:
                exit_reason = "60거래일 경과"
            if exit_reason:
                pending_order = {"side": "sell", "reason": exit_reason}
        elif bool(price_row["entry_signal"]):
            pending_order = {"side": "buy", "signal_date": trade_day}
        equity_values.append(cash + position_shares * float(price_row["close"]))
    if position_shares > 0.0 and active_trade is not None:
        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, "trade_return": cash / float(active_trade["capital"]) - 1.0, "holding_days": len(signal_data) - 1 - int(active_trade["entry_index"]), "exit_reason": "검증 기간 종료"})
        equity_values[-1] = cash
    return pd.Series(equity_values, index=signal_data.index), trades


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


def summarize_strategy(equity_curves: list[pd.Series], trades: list[dict[str, object]]) -> 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
    trade_returns = pd.Series([float(trade["trade_return"]) for trade in trades])
    return {"누적수익률": float(portfolio_equity.iloc[-1] - 1.0), "MDD": float(drawdown.min()), "최장 회복 기간": longest_recovery_days(portfolio_equity), "거래 수": len(trades), "승률": float((trade_returns > 0.0).mean()), "최악 거래": float(trade_returns.min())}


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


def main() -> None:
    stop_rules = {"손실 제한 없음": "none", "초기 10% 손절": "initial", "추적 10% 손절": "trailing", "20일 무수익 청산": "time"}
    equity_by_rule = {name: [] for name in stop_rules}
    trades_by_rule = {name: [] for name in stop_rules}
    for ticker in TICKERS:
        signal_data = add_entry_signal(download_price_data(ticker))
        for rule_name, stop_rule in stop_rules.items():
            equity, trades = run_backtest(signal_data, stop_rule)
            equity_by_rule[rule_name].append(equity)
            trades_by_rule[rule_name].extend(trades)
    summary = pd.DataFrame([{"청산 규칙": name, **summarize_strategy(equity_by_rule[name], trades_by_rule[name])} for name in stop_rules])
    print(summary.to_string(index=False, formatters={"누적수익률": "{:+.1%}".format, "MDD": "{:.1%}".format, "승률": "{:.1%}".format, "최악 거래": "{:.1%}".format}))


if __name__ == "__main__":
    main()

오늘 종가가 손절선 아래인지 확인
→ 다음 거래일 시가에 매도
→ 밤사이 악재나 하한가·갭 하락이 있으면 -10%보다 낮은 가격에 체결

허용 손실 금액 = 계좌 평가액 × 거래당 위험 비율
매수 수량 = 허용 손실 금액 ÷ (진입가 - 손절가)