같은 돌파 전략도 청산 규칙에 따라 결과가 달라지는 이유

진입 신호 = 종가가 계산일 이전 20거래일 고가의 최댓값을 상향 돌파
체결 가격 = 신호 다음 거래일 시가

python -m pip install yfinance pandas numpy matplotlib pyarrow

#!/usr/bin/env python3
"""같은 돈치안 돌파 진입에 네 가지 청산 규칙을 적용해 성과를 비교한다."""

from __future__ import annotations

import time
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import yfinance as yf
from matplotlib import font_manager

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_WINDOW = 20
ATR_WINDOW = 14
STOP_LOSS = 0.10
PROFIT_TARGET = 0.20
TIME_EXIT_DAYS = 40
ATR_MULTIPLIER = 3.0
MAX_HOLDING_DAYS = 120
COST_RATE = 0.0005
MIN_ROWS = 504
PRICE_CACHE_FILE = Path("exit_rules_prices.parquet")
CHART_FILE = Path("exit_rules_chart.png")
RESULT_FILE = Path("exit_rules_results.csv")
TRADE_FILE = Path("exit_rules_trades.csv")

def read_price_cache(path: Path) -> dict[str, pd.DataFrame]:
    cached = pd.read_parquet(path)
    cached["date"] = pd.to_datetime(cached["date"])
    return {ticker: group.set_index("date")[["open", "high", "low", "close", "volume"]].sort_index() for ticker, group in cached.groupby("ticker")}

def download_prices() -> dict[str, pd.DataFrame]:
    if PRICE_CACHE_FILE.is_file():
        return read_price_cache(PRICE_CACHE_FILE)

    raw = yf.download(TICKERS, start=DOWNLOAD_START, end=TEST_END, auto_adjust=True, progress=False, group_by="ticker", threads=False)
    if raw.empty:
        raise RuntimeError("가격 데이터를 받지 못했습니다.")
    prices = {}
    for ticker in TICKERS:
        frame = raw[ticker].rename(columns=str.lower).dropna(subset=["open", "high", "low", "close"]).copy()
        if len(frame.loc[(frame.index >= TEST_START) & (frame.index < TEST_END)]) < MIN_ROWS:
            for attempt in range(3):
                retry = yf.download(ticker, start=DOWNLOAD_START, end=TEST_END, auto_adjust=True, progress=False, threads=False)
                if isinstance(retry.columns, pd.MultiIndex):
                    retry.columns = retry.columns.get_level_values(0)
                candidate = retry.rename(columns=str.lower).dropna(subset=["open", "high", "low", "close"]).copy()
                if len(candidate) > len(frame):
                    frame = candidate
                if len(frame.loc[(frame.index >= TEST_START) & (frame.index < TEST_END)]) >= MIN_ROWS:
                    break
                time.sleep(1.0 + attempt)
        if len(frame.loc[(frame.index >= TEST_START) & (frame.index < TEST_END)]) < MIN_ROWS:
            raise RuntimeError(f"검증에 필요한 가격 데이터가 부족합니다: {ticker}")
        prices[ticker] = frame
    PRICE_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
    cached = pd.concat([frame.reset_index(names="date").assign(ticker=ticker) for ticker, frame in prices.items()], ignore_index=True)
    cached.to_parquet(PRICE_CACHE_FILE, index=False)
    return prices

def prepare_price(raw: pd.DataFrame) -> pd.DataFrame:
    data = raw.copy()
    previous_close = data["close"].shift(1)
    true_range = pd.concat([(data["high"] - data["low"]), (data["high"] - previous_close).abs(), (data["low"] - previous_close).abs()], axis=1).max(axis=1)
    data["atr"] = true_range.ewm(alpha=1.0 / ATR_WINDOW, adjust=False, min_periods=ATR_WINDOW).mean()
    data["entry_channel"] = data["high"].shift(1).rolling(ENTRY_WINDOW, min_periods=ENTRY_WINDOW).max()
    data["entry_signal"] = (data["close"] > data["entry_channel"]) & (data["close"].shift(1) <= data["entry_channel"].shift(1))
    return data.loc[(data.index >= TEST_START) & (data.index < TEST_END)].copy()

def run_backtest(data: pd.DataFrame, exit_rule: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    cash = 1.0
    shares = 0.0
    pending_order = None
    current_trade = None
    trades = []
    daily_records = []

    for index, (date, row) in enumerate(data.iterrows()):
        if pending_order is not None:
            if pending_order["side"] == "buy" and shares == 0.0:
                capital_before = cash
                entry_price = float(row["open"])
                shares = cash * (1.0 - COST_RATE) / entry_price
                cash = 0.0
                initial_atr = float(row["atr"])
                current_trade = {"entry_signal_date": pending_order["signal_date"], "entry_date": date, "entry_index": index, "entry_price": entry_price, "capital_before": capital_before, "highest_close": float(row["close"]), "trailing_stop": entry_price - ATR_MULTIPLIER * initial_atr}
            elif pending_order["side"] == "sell" and shares > 0.0:
                exit_price = float(row["open"])
                cash = shares * exit_price * (1.0 - COST_RATE)
                shares = 0.0
                trades.append({**current_trade, "exit_signal_date": pending_order["signal_date"], "exit_date": date, "exit_price": exit_price, "trade_return": cash / current_trade["capital_before"] - 1.0, "holding_days": index - current_trade["entry_index"], "exit_reason": pending_order["reason"], "forced_exit": False})
                current_trade = None
            pending_order = None

        if shares > 0.0:
            current_trade["highest_close"] = max(current_trade["highest_close"], float(row["close"]))
            if pd.notna(row["atr"]):
                candidate_stop = current_trade["highest_close"] - ATR_MULTIPLIER * float(row["atr"])
                current_trade["trailing_stop"] = max(current_trade["trailing_stop"], candidate_stop)
            holding_days = index - current_trade["entry_index"]
            close_return = float(row["close"]) / current_trade["entry_price"] - 1.0
            if close_return <= -STOP_LOSS:
                pending_order = {"side": "sell", "signal_date": date, "reason": "10% 손실"}
            elif exit_rule == "profit_target" and close_return >= PROFIT_TARGET:
                pending_order = {"side": "sell", "signal_date": date, "reason": "20% 목표"}
            elif exit_rule == "time_exit" and holding_days >= TIME_EXIT_DAYS:
                pending_order = {"side": "sell", "signal_date": date, "reason": "40일 보유"}
            elif exit_rule == "atr_trailing" and float(row["close"]) <= current_trade["trailing_stop"]:
                pending_order = {"side": "sell", "signal_date": date, "reason": "3ATR 추적"}
            elif holding_days >= MAX_HOLDING_DAYS:
                pending_order = {"side": "sell", "signal_date": date, "reason": "120일 상한"}
        elif bool(row["entry_signal"]):
            pending_order = {"side": "buy", "signal_date": date}

        daily_records.append({"date": date, "equity": cash + shares * float(row["close"]), "exposure": float(shares > 0.0)})

    if shares > 0.0:
        last_date = data.index[-1]
        last_close = float(data.iloc[-1]["close"])
        cash = shares * last_close * (1.0 - COST_RATE)
        trades.append({**current_trade, "exit_signal_date": last_date, "exit_date": last_date, "exit_price": last_close, "trade_return": cash / current_trade["capital_before"] - 1.0, "holding_days": len(data) - 1 - current_trade["entry_index"], "exit_reason": "기간 종료", "forced_exit": True})
        daily_records[-1]["equity"] = cash
        daily_records[-1]["exposure"] = 0.0

    return pd.DataFrame(trades), pd.DataFrame(daily_records).set_index("date")

def summarize(trades: pd.DataFrame, portfolio: pd.DataFrame) -> dict[str, float]:
    drawdown = portfolio["equity"] / portfolio["equity"].cummax() - 1.0
    winners = trades.loc[trades["trade_return"] > 0.0, "trade_return"]
    losers = trades.loc[trades["trade_return"] < 0.0, "trade_return"]
    return {"누적수익률": float(portfolio["equity"].iloc[-1] - 1.0), "MDD": float(drawdown.min()), "Profit Factor": float(winners.sum() / losers.abs().sum()), "승률": float((trades["trade_return"] > 0.0).mean()), "거래수": len(trades), "평균이익": float(winners.mean()), "평균손실": float(losers.mean()), "평균손익비": float(winners.mean() / losers.abs().mean()), "중앙보유일": float(trades["holding_days"].median()), "노출비중": float(portfolio["exposure"].mean()), "손실청산수": int((trades["exit_reason"] == "10% 손실").sum())}

def configure_font() -> None:
    available_fonts = {item.name for item in font_manager.fontManager.ttflist}
    for font_name in ("Malgun Gothic", "AppleGothic", "UnDotum", "DejaVu Sans"):
        if font_name in available_fonts:
            plt.rcParams["font.family"] = font_name
            break
    plt.rcParams["axes.unicode_minus"] = False

def draw_chart(summary: pd.DataFrame, portfolios: dict[str, pd.DataFrame], all_trades: dict[str, pd.DataFrame]) -> None:
    colors = {"고정 손절": "#64748B", "+20% 목표": "#F59E0B", "40일 보유": "#7C3AED", "3ATR 추적": "#0F766E"}
    fig = plt.figure(figsize=(13, 9), facecolor="white")
    grid = fig.add_gridspec(2, 2, height_ratios=[1.65, 1.0], hspace=0.35, wspace=0.28)
    equity_axis = fig.add_subplot(grid[0, :])
    risk_axis = fig.add_subplot(grid[1, 0])
    distribution_axis = fig.add_subplot(grid[1, 1])

    for label, portfolio in portfolios.items():
        equity_axis.plot(portfolio.index, portfolio["equity"], color=colors[label], linewidth=2.1, label=f"{label} {summary.loc[label, '누적수익률']:+.1%}")
    equity_axis.axhline(1.0, color="#CBD5E1", linewidth=1.0)
    equity_axis.set_title("같은 돌파 진입도 청산 규칙에 따라 다른 자산 경로를 만든다", fontsize=18, fontweight="bold", pad=16)
    equity_axis.set_ylabel("30종목 동일 비중 누적 자산")
    equity_axis.legend(frameon=False, ncol=2, loc="upper left")

    for label in summary.index:
        risk_axis.scatter(abs(summary.loc[label, "MDD"]) * 100.0, summary.loc[label, "누적수익률"] * 100.0, s=70 + summary.loc[label, "거래수"] / 8.0, color=colors[label], edgecolor="white", linewidth=1.2, label=label)
        risk_axis.annotate(label, (abs(summary.loc[label, "MDD"]) * 100.0, summary.loc[label, "누적수익률"] * 100.0), xytext=(5, 5), textcoords="offset points", fontsize=9)
    risk_axis.set_title("낙폭과 누적수익률", fontsize=14, fontweight="bold")
    risk_axis.set_xlabel("MDD 절댓값(%)")
    risk_axis.set_ylabel("누적수익률(%)")

    labels = list(all_trades)
    distributions = [all_trades[label]["trade_return"].clip(-0.2, 0.4) * 100.0 for label in labels]
    box = distribution_axis.boxplot(distributions, tick_labels=labels, patch_artist=True, showfliers=False, widths=0.55)
    for patch, label in zip(box["boxes"], labels):
        patch.set_facecolor(colors[label])
        patch.set_alpha(0.72)
    distribution_axis.axhline(0.0, color="#94A3B8", linewidth=1.0)
    distribution_axis.set_title("거래 수익률 분포", fontsize=14, fontweight="bold")
    distribution_axis.set_ylabel("거래 수익률(%, -20~+40 표시)")
    distribution_axis.tick_params(axis="x", rotation=12)

    fig.suptitle("청산은 수익을 끝내는 규칙이 아니라 전략의 손익 구조를 정하는 규칙이다", fontsize=20, fontweight="bold", y=0.99)
    for axis in (equity_axis, risk_axis, distribution_axis):
        axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
        axis.spines["top"].set_visible(False)
        axis.spines["right"].set_visible(False)
    CHART_FILE.parent.mkdir(parents=True, exist_ok=True)
    fig.savefig(CHART_FILE, dpi=160, bbox_inches="tight", facecolor="white")
    plt.close(fig)

configure_font()
prices = download_prices()
prepared = {ticker: prepare_price(frame) for ticker, frame in prices.items()}
calendar = pd.DatetimeIndex(sorted(set().union(*(data.index for data in prepared.values()))))
strategies = {"고정 손절": "fixed_stop", "+20% 목표": "profit_target", "40일 보유": "time_exit", "3ATR 추적": "atr_trailing"}
trade_frames = {label: [] for label in strategies}
equity_series = {label: [] for label in strategies}
exposure_series = {label: [] for label in strategies}

for ticker, data in prepared.items():
    for label, exit_rule in strategies.items():
        trades, daily = run_backtest(data, exit_rule)
        trades.insert(0, "ticker", ticker)
        trade_frames[label].append(trades)
        equity_series[label].append(daily["equity"].reindex(calendar).ffill().fillna(1.0).rename(ticker))
        exposure_series[label].append(daily["exposure"].reindex(calendar).ffill().fillna(0.0).rename(ticker))

all_trades = {}
portfolios = {}
summary_rows = {}
for label in strategies:
    all_trades[label] = pd.concat(trade_frames[label], ignore_index=True)
    portfolios[label] = pd.DataFrame({"equity": pd.concat(equity_series[label], axis=1).mean(axis=1), "exposure": pd.concat(exposure_series[label], axis=1).mean(axis=1)})
    summary_rows[label] = summarize(all_trades[label], portfolios[label])

summary = pd.DataFrame(summary_rows).T
summary[["거래수", "손실청산수"]] = summary[["거래수", "손실청산수"]].astype(int)
tagged_trades = pd.concat([trades.assign(strategy=label) for label, trades in all_trades.items()], ignore_index=True)
summary.to_csv(RESULT_FILE, encoding="utf-8-sig")
tagged_trades.to_csv(TRADE_FILE, index=False, encoding="utf-8-sig")
draw_chart(summary, portfolios, all_trades)

print(summary.to_string(formatters={"누적수익률": "{:+.2%}".format, "MDD": "{:.2%}".format, "Profit Factor": "{:.2f}".format, "승률": "{:.2%}".format, "평균이익": "{:+.2%}".format, "평균손실": "{:.2%}".format, "평균손익비": "{:.2f}".format, "중앙보유일": "{:.0f}".format, "노출비중": "{:.2%}".format}))
for label, trades in all_trades.items():
    print(f"{label} 청산 사유: {trades['exit_reason'].value_counts().to_dict()}")
print(f"결과 저장: {RESULT_FILE}")
print(f"거래 저장: {TRADE_FILE}")
print(f"차트 저장: {CHART_FILE}")