자동매매를 언제 멈추고 다시 켜야 할까?

RUNNING: 실제 주문과 가상 자산곡선을 함께 기록
PAUSED: 실제 주문은 중지하고 가상 자산곡선만 기록
RESUME: 가상 자산곡선이 중지 전 고점을 회복하면 다음 거래일부터 재개

python -m pip install yfinance pandas numpy matplotlib pyarrow

#!/usr/bin/env python3
"""연속 손실과 과거 MDD 기준의 시스템 중지·재개 정책을 표본외 구간에서 비교한다."""

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"
TRAIN_START = "2014-01-01"
TRAIN_END = "2022-01-01"
TEST_START = "2022-01-01"
TEST_END = "2026-01-01"
COST_RATE = 0.0005
MDD_MULTIPLIER = 1.5
NEGATIVE_MONTH_LIMIT = 3
MIN_ROWS = 252
PRICE_CACHE_FILE = Path("system_stop_resume_prices.parquet")
CHART_FILE = Path("system_stop_resume_chart.png")
RESULT_FILE = Path("system_stop_resume_results.csv")
EVENT_FILE = Path("system_stop_resume_events.csv")

STRATEGIES = {
    "EMA 20·100": ("ema_20_100_entry", "ema_20_100_exit", None, None),
    "EMA 35·120": ("ema_35_120_entry", "ema_35_120_exit", None, None),
    "돈치안 돌파": ("donchian_entry", "channel_exit", 0.10, 120),
    "ATR 돌파": ("atr_entry", "channel_exit", 0.10, 120),
    "ROC 추세": ("roc_entry", "roc_exit", None, None),
    "CCI 역추세": ("cci_entry", "cci_exit", 0.08, 20),
}

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_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_END]) >= MIN_ROWS:
                    break
                time.sleep(1.0 + attempt)
        if len(frame.loc[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 mean_absolute_deviation(values: np.ndarray) -> float:
    return float(np.mean(np.abs(values - values.mean())))

def prepare_price(raw: pd.DataFrame) -> pd.DataFrame:
    data = raw.copy()
    for fast_window, slow_window in ((20, 100), (35, 120)):
        fast = data["close"].ewm(span=fast_window, adjust=False, min_periods=fast_window).mean()
        slow = data["close"].ewm(span=slow_window, adjust=False, min_periods=slow_window).mean()
        data[f"ema_{fast_window}_{slow_window}_entry"] = (fast > slow) & (fast.shift(1) <= slow.shift(1))
        data[f"ema_{fast_window}_{slow_window}_exit"] = (fast < slow) & (fast.shift(1) >= slow.shift(1))

    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)
    atr = true_range.ewm(alpha=1.0 / 14, adjust=False, min_periods=14).mean()
    atr_center = data["close"].ewm(span=20, adjust=False, min_periods=20).mean()
    atr_upper = atr_center + 2.0 * atr
    donchian_upper = data["high"].shift(1).rolling(20, min_periods=20).max()
    common_exit = data["low"].shift(1).rolling(10, min_periods=10).min()
    data["donchian_entry"] = (data["close"] > donchian_upper) & (data["close"].shift(1) <= donchian_upper.shift(1))
    data["atr_entry"] = (data["close"] > atr_upper) & (data["close"].shift(1) <= atr_upper.shift(1))
    data["channel_exit"] = (data["close"] < common_exit) & (data["close"].shift(1) >= common_exit.shift(1))

    roc = data["close"].pct_change(20) * 100.0
    positive_episode = (roc <= 0.0).cumsum()
    roc_candidate = roc > 2.0
    previous_candidate = roc_candidate.groupby(positive_episode).cummax().groupby(positive_episode).shift(1, fill_value=False)
    data["roc_entry"] = roc_candidate & ~previous_candidate
    data["roc_exit"] = (roc < 0.0) & (roc.shift(1) >= 0.0)

    typical_price = (data["high"] + data["low"] + data["close"]) / 3.0
    typical_average = typical_price.rolling(20, min_periods=20).mean()
    mean_deviation = typical_price.rolling(20, min_periods=20).apply(mean_absolute_deviation, raw=True)
    cci = (typical_price - typical_average) / (0.015 * mean_deviation)
    ema_long = data["close"].ewm(span=200, adjust=False, min_periods=200).mean()
    trend_filter = (data["close"] > ema_long) & (ema_long > ema_long.shift(20))
    volatility_filter = atr / data["close"] * 100.0 <= 5.0
    data["cci_entry"] = (cci > -100.0) & (cci.shift(1) <= -100.0) & trend_filter & volatility_filter
    data["cci_exit"] = (cci > 100.0) & (cci.shift(1) <= 100.0)
    return data

def run_backtest(data: pd.DataFrame, entry_column: str, exit_column: str, stop_loss: float | None, maximum_holding_days: int | None) -> pd.DataFrame:
    cash = 1.0
    shares = 0.0
    pending_order = None
    entry_price = np.nan
    entry_index = -1
    daily_records = []
    for index, (date, row) in enumerate(data.iterrows()):
        if pending_order == "buy" and shares == 0.0:
            entry_price = float(row["open"])
            shares = cash * (1.0 - COST_RATE) / entry_price
            cash = 0.0
            entry_index = index
        elif pending_order == "sell" and shares > 0.0:
            cash = shares * float(row["open"]) * (1.0 - COST_RATE)
            shares = 0.0
            entry_price = np.nan
            entry_index = -1
        pending_order = None

        if shares > 0.0:
            holding_days = index - entry_index
            close_return = float(row["close"]) / entry_price - 1.0
            if stop_loss is not None and close_return <= -stop_loss:
                pending_order = "sell"
            elif bool(row[exit_column]):
                pending_order = "sell"
            elif maximum_holding_days is not None and holding_days >= maximum_holding_days:
                pending_order = "sell"
        elif bool(row[entry_column]):
            pending_order = "buy"
        daily_records.append({"date": date, "equity": cash + shares * float(row["close"])})

    if shares > 0.0:
        cash = shares * float(data.iloc[-1]["close"]) * (1.0 - COST_RATE)
        daily_records[-1]["equity"] = cash
    return pd.DataFrame(daily_records).set_index("date")

def build_strategy_returns(prepared: dict[str, pd.DataFrame], start: str, end: str) -> pd.DataFrame:
    calendar = pd.DatetimeIndex(sorted(set().union(*(data.loc[(data.index >= start) & (data.index < end)].index for data in prepared.values()))))
    strategy_returns = {}
    for label, (entry_column, exit_column, stop_loss, maximum_holding_days) in STRATEGIES.items():
        equity_columns = []
        for ticker, full_data in prepared.items():
            data = full_data.loc[(full_data.index >= start) & (full_data.index < end)].copy()
            daily = run_backtest(data, entry_column, exit_column, stop_loss, maximum_holding_days)
            equity_columns.append(daily["equity"].reindex(calendar).ffill().fillna(1.0).rename(ticker))
        portfolio_equity = pd.concat(equity_columns, axis=1).mean(axis=1)
        strategy_returns[label] = portfolio_equity.pct_change().fillna(0.0)
    return pd.DataFrame(strategy_returns)

def summarize_equity(equity: pd.Series) -> dict[str, float]:
    drawdown = equity / equity.cummax() - 1.0
    return {"return": float(equity.iloc[-1] - 1.0), "mdd": float(drawdown.min())}

def apply_monitoring_policy(returns: pd.Series, policy: str, drawdown_limit: float | None = None) -> tuple[pd.DataFrame, pd.DataFrame]:
    shadow_equity = (1.0 + returns).cumprod()
    shadow_peak = shadow_equity.cummax()
    shadow_drawdown = shadow_equity / shadow_peak - 1.0
    monthly_returns = (1.0 + returns).groupby(returns.index.to_period("M")).prod() - 1.0
    running = True
    resume_level = np.nan
    negative_months = 0
    active_equity = 1.0
    records = []
    events = []

    for index, (date, daily_return) in enumerate(returns.items()):
        state_at_open = "RUNNING" if running else "PAUSED"
        if running:
            active_equity *= 1.0 + daily_return
        records.append({"date": date, "policy": policy, "shadow_equity": shadow_equity.iloc[index], "active_equity": active_equity, "shadow_drawdown": shadow_drawdown.iloc[index], "state": state_at_open})

        month_end = index == len(returns) - 1 or returns.index[index + 1].to_period("M") != date.to_period("M")
        if not running:
            if shadow_equity.iloc[index] >= resume_level:
                running = True
                negative_months = 0
                events.append({"date": date, "policy": policy, "event": "RESUME", "reason": "가상 자산곡선 전고점 회복", "shadow_equity": shadow_equity.iloc[index], "shadow_drawdown": shadow_drawdown.iloc[index]})
            continue

        should_pause = False
        reason = ""
        if policy == "과거 MDD 1.5배" and drawdown_limit is not None and shadow_drawdown.iloc[index] <= -drawdown_limit:
            should_pause = True
            reason = f"낙폭 {shadow_drawdown.iloc[index]:.2%}가 중지선 {-drawdown_limit:.2%} 하회"
        elif policy == "3개월 연속 손실" and month_end:
            negative_months = negative_months + 1 if monthly_returns.loc[date.to_period("M")] < 0.0 else 0
            if negative_months >= NEGATIVE_MONTH_LIMIT:
                should_pause = True
                reason = "월간 수익률 3개월 연속 음수"

        if should_pause:
            running = False
            negative_months = 0
            resume_level = shadow_peak.iloc[index]
            events.append({"date": date, "policy": policy, "event": "PAUSE", "reason": reason, "shadow_equity": shadow_equity.iloc[index], "shadow_drawdown": shadow_drawdown.iloc[index]})

    return pd.DataFrame(records).set_index("date"), pd.DataFrame(events)

def build_results(train_returns: pd.DataFrame, test_returns: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    result_rows = []
    event_frames = []
    daily_frames = []
    for strategy in STRATEGIES:
        train_equity = (1.0 + train_returns[strategy]).cumprod()
        train_mdd = abs(summarize_equity(train_equity)["mdd"])
        drawdown_limit = train_mdd * MDD_MULTIPLIER
        always_equity = (1.0 + test_returns[strategy]).cumprod()
        always_metrics = summarize_equity(always_equity)
        result_rows.append({"strategy": strategy, "policy": "계속 운용", "train_mdd": -train_mdd, "stop_line": -drawdown_limit, "return": always_metrics["return"], "mdd": always_metrics["mdd"], "pause_count": 0, "paused_days": 0})

        for policy in ("3개월 연속 손실", "과거 MDD 1.5배"):
            daily, events = apply_monitoring_policy(test_returns[strategy], policy, drawdown_limit)
            metrics = summarize_equity(daily["active_equity"])
            pause_count = int((events["event"] == "PAUSE").sum()) if not events.empty else 0
            paused_days = int((daily["state"] == "PAUSED").sum())
            result_rows.append({"strategy": strategy, "policy": policy, "train_mdd": -train_mdd, "stop_line": -drawdown_limit, "return": metrics["return"], "mdd": metrics["mdd"], "pause_count": pause_count, "paused_days": paused_days})
            daily_frames.append(daily.reset_index().assign(strategy=strategy))
            if not events.empty:
                event_frames.append(events.assign(strategy=strategy))

    results = pd.DataFrame(result_rows)
    events = pd.concat(event_frames, ignore_index=True) if event_frames else pd.DataFrame()
    daily_records = pd.concat(daily_frames, ignore_index=True)
    return results, events, daily_records

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(results: pd.DataFrame, events: pd.DataFrame, daily_records: pd.DataFrame) -> None:
    colors = {"계속 운용": "#64748B", "3개월 연속 손실": "#F59E0B", "과거 MDD 1.5배": "#0F766E"}
    fig = plt.figure(figsize=(14, 11), facecolor="white")
    grid = fig.add_gridspec(2, 2, height_ratios=[1.35, 1.0], hspace=0.34, wspace=0.27)
    equity_axis = fig.add_subplot(grid[0, :])
    ratio_axis = fig.add_subplot(grid[1, 0])
    return_axis = fig.add_subplot(grid[1, 1])

    strategy = "ROC 추세"
    policies = ["계속 운용", "3개월 연속 손실", "과거 MDD 1.5배"]
    roc_results = results.loc[results["strategy"] == strategy].set_index("policy").loc[policies]
    always = daily_records.loc[(daily_records["strategy"] == strategy) & (daily_records["policy"] == "과거 MDD 1.5배")].set_index("date")
    loss_policy = daily_records.loc[(daily_records["strategy"] == strategy) & (daily_records["policy"] == "3개월 연속 손실")].set_index("date")
    equity_axis.plot(always.index, always["shadow_equity"], color=colors["계속 운용"], linewidth=2.1, label=f"계속 운용 {roc_results.loc['계속 운용', 'return']:+.1%}")
    equity_axis.plot(loss_policy.index, loss_policy["active_equity"], color=colors["3개월 연속 손실"], linewidth=2.0, label=f"3개월 연속 손실 {roc_results.loc['3개월 연속 손실', 'return']:+.1%}")
    equity_axis.plot(always.index, always["active_equity"], color=colors["과거 MDD 1.5배"], linewidth=2.5, label=f"과거 MDD 1.5배 {roc_results.loc['과거 MDD 1.5배', 'return']:+.1%}")
    paused = always["state"].eq("PAUSED")
    equity_axis.fill_between(always.index, 0.85, 1.55, where=paused, color="#CCFBF1", alpha=0.8, step="post", label="MDD 기준 주문 중지")
    roc_events = events.loc[(events["strategy"] == strategy) & (events["policy"] == "과거 MDD 1.5배")]
    for _, event in roc_events.iterrows():
        marker = "v" if event["event"] == "PAUSE" else "^"
        color = "#B91C1C" if event["event"] == "PAUSE" else "#047857"
        label = "중지" if event["event"] == "PAUSE" else "재개"
        equity_axis.scatter(event["date"], event["shadow_equity"], marker=marker, s=90, color=color, zorder=5)
        equity_axis.annotate(label, (event["date"], event["shadow_equity"]), xytext=(0, 12 if marker == "^" else -20), textcoords="offset points", ha="center", color=color, fontweight="bold")
    equity_axis.set_ylim(0.85, 1.55)
    equity_axis.set_ylabel("표본외 누적 자산")
    equity_axis.set_title("중지 기간에도 가상 자산곡선을 계산해야 재개 시점을 정할 수 있다", fontsize=17, fontweight="bold", pad=14)
    equity_axis.legend(loc="upper left", frameon=False, ncol=2)

    always_rows = results.loc[results["policy"] == "계속 운용"].set_index("strategy")
    mdd_rows = results.loc[results["policy"] == "과거 MDD 1.5배"].set_index("strategy")
    ratios = always_rows["mdd"].abs() / always_rows["train_mdd"].abs()
    bar_colors = ["#0F766E" if mdd_rows.loc[label, "pause_count"] else "#CBD5E1" for label in ratios.index]
    ratio_axis.barh(ratios.index, ratios.values, color=bar_colors)
    ratio_axis.axvline(MDD_MULTIPLIER, color="#B91C1C", linestyle="--", linewidth=1.8, label="중지선 1.5배")
    ratio_axis.set_xlabel("표본외 MDD ÷ 개발 구간 MDD")
    ratio_axis.set_title("1.5배 중지선은 6개 중 2개만 작동", fontsize=14, fontweight="bold")
    ratio_axis.legend(frameon=False, loc="lower right")
    for index, value in enumerate(ratios.values):
        ratio_axis.text(value + 0.03, index, f"{value:.2f}배", va="center", fontsize=9)

    bars = return_axis.bar(np.arange(len(policies)), roc_results["return"] * 100.0, color=[colors[policy] for policy in policies])
    return_axis.set_xticks(np.arange(len(policies)), ["계속 운용", "3개월 손실", "MDD 1.5배"])
    return_axis.set_ylabel("누적수익률(%)")
    return_axis.set_title("ROC 추세의 중지 비용", fontsize=14, fontweight="bold")
    return_axis.bar_label(bars, labels=[f"{value:.1f}%" for value in roc_results["return"] * 100.0], padding=4, fontsize=10, fontweight="bold")
    return_axis.set_ylim(0.0, max(50.0, float((roc_results["return"] * 100.0).max() * 1.22)))

    for axis in (equity_axis, ratio_axis, return_axis):
        axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
        axis.spines["top"].set_visible(False)
        axis.spines["right"].set_visible(False)
    fig.suptitle("연속 손실만으로 끄면 과잉 중단, MDD 기준도 회복 수익을 포기했다", fontsize=21, fontweight="bold", y=0.99)
    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()}
train_returns = build_strategy_returns(prepared, TRAIN_START, TRAIN_END)
test_returns = build_strategy_returns(prepared, TEST_START, TEST_END)
results, events, daily_records = build_results(train_returns, test_returns)
RESULT_FILE.parent.mkdir(parents=True, exist_ok=True)
results.to_csv(RESULT_FILE, index=False, encoding="utf-8-sig")
events.to_csv(EVENT_FILE, index=False, encoding="utf-8-sig")
draw_chart(results, events, daily_records)

summary = results.pivot(index="strategy", columns="policy", values=["return", "mdd", "pause_count", "paused_days"])
print(summary.to_string(float_format=lambda value: f"{value:.4f}"))
print(events.to_string(index=False, formatters={"shadow_equity": "{:.4f}".format, "shadow_drawdown": "{:.2%}".format}))
print(f"결과 저장: {RESULT_FILE}")
print(f"이벤트 로그 저장: {EVENT_FILE}")
print(f"차트 저장: {CHART_FILE}")