ROC·ADX 전략: 거래를 줄여도 가짜 신호가 남는 이유

#!/usr/bin/env python3
"""ROC 진입 신호에 ADX·DMI 필터를 추가했을 때의 손익을 비교한다."""

from __future__ import annotations

from pathlib import Path
import time

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 = "2013-01-01"
TEST_START = "2014-01-01"
TEST_END = "2026-01-01"
ROC_WINDOW = 20
ROC_ENTRY_LEVEL = 2.0
ADX_WINDOW = 14
COST_RATE = 0.0005
FALSE_SIGNAL_DAYS = 20
MIN_ROWS = 252
CHART_FILE = Path("roc_adx_chart.png")
RESULT_FILE = Path("roc_adx_results.csv")
SIGNAL_FILE = Path("roc_adx_signals.csv")

def wilder_average(values: pd.Series, window: int) -> pd.Series:
    return values.ewm(alpha=1.0 / window, adjust=False, min_periods=window).mean()

def download_prices() -> dict[str, pd.DataFrame]:
    prices = {}
    for ticker in TICKERS:
        frame = pd.DataFrame()
        for attempt in range(3):
            if not frame.empty and len(frame.loc[(frame.index >= TEST_START) & (frame.index < TEST_END)]) >= MIN_ROWS:
                break
            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)
            frame = retry.rename(columns=str.lower).dropna(subset=["open", "high", "low", "close"]).copy()
            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
    return prices

def prepare_price(prices: dict[str, pd.DataFrame], ticker: str) -> pd.DataFrame:
    data = prices[ticker].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)
    upward_move = data["high"].diff()
    downward_move = -data["low"].diff()
    plus_dm = upward_move.where((upward_move > downward_move) & (upward_move > 0.0), 0.0)
    minus_dm = downward_move.where((downward_move > upward_move) & (downward_move > 0.0), 0.0)
    average_true_range = wilder_average(true_range, ADX_WINDOW)
    data["plus_di"] = 100.0 * wilder_average(plus_dm, ADX_WINDOW) / average_true_range
    data["minus_di"] = 100.0 * wilder_average(minus_dm, ADX_WINDOW) / average_true_range
    dx = 100.0 * (data["plus_di"] - data["minus_di"]).abs() / (data["plus_di"] + data["minus_di"])
    data["adx"] = wilder_average(dx.replace([np.inf, -np.inf], np.nan), ADX_WINDOW)
    data["roc"] = data["close"].pct_change(ROC_WINDOW) * 100.0
    positive_episode = (data["roc"] <= 0.0).cumsum()
    entry_candidate = data["roc"] > ROC_ENTRY_LEVEL
    previous_candidate = entry_candidate.groupby(positive_episode).cummax().groupby(positive_episode).shift(1, fill_value=False)
    data["roc_entry_signal"] = entry_candidate & ~previous_candidate
    data["adx_filter"] = (data["adx"] > data["adx"].shift(2)) & (data["plus_di"] > data["minus_di"])
    data["filtered_entry_signal"] = data["roc_entry_signal"] & data["adx_filter"]
    data["exit_signal"] = (data["roc"] < 0.0) & (data["roc"].shift(1) >= 0.0)
    return data.loc[(data.index >= TEST_START) & (data.index < TEST_END)].copy()

def run_backtest(data: pd.DataFrame, entry_column: 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
                current_trade = {"entry_signal_date": pending_order["signal_date"], "entry_date": date, "entry_index": index, "entry_price": entry_price, "capital_before": capital_before}
            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
                trade_return = cash / current_trade["capital_before"] - 1.0
                holding_days = index - current_trade["entry_index"]
                trades.append({**current_trade, "exit_signal_date": pending_order["signal_date"], "exit_date": date, "exit_price": exit_price, "trade_return": trade_return, "holding_days": holding_days, "false_signal": trade_return < 0.0 and holding_days <= FALSE_SIGNAL_DAYS, "forced_exit": False})
                current_trade = None
            pending_order = None

        if shares > 0.0 and bool(row["exit_signal"]):
            pending_order = {"side": "sell", "signal_date": date}
        elif shares == 0.0 and bool(row[entry_column]):
            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_index = len(data) - 1
        last_date = data.index[-1]
        last_close = float(data.iloc[-1]["close"])
        cash = shares * last_close * (1.0 - COST_RATE)
        trade_return = cash / current_trade["capital_before"] - 1.0
        holding_days = last_index - current_trade["entry_index"]
        trades.append({**current_trade, "exit_signal_date": last_date, "exit_date": last_date, "exit_price": last_close, "trade_return": trade_return, "holding_days": holding_days, "false_signal": trade_return < 0.0 and holding_days <= FALSE_SIGNAL_DAYS, "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 analyze_roc_signals(data: pd.DataFrame, baseline_trades: pd.DataFrame, ticker: str) -> pd.DataFrame:
    rows = []
    for trade in baseline_trades.itertuples():
        signal = data.loc[trade.entry_signal_date]
        rows.append({"ticker": ticker, "entry_signal_date": trade.entry_signal_date, "exit_signal_date": trade.exit_signal_date, "trade_return": trade.trade_return, "holding_days": trade.holding_days, "false_signal": trade.false_signal, "accepted": bool(signal["adx_filter"]), "roc": signal["roc"], "adx": signal["adx"], "plus_di": signal["plus_di"], "minus_di": signal["minus_di"]})
    return pd.DataFrame(rows)

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), "가짜신호수": int(trades["false_signal"].sum()), "가짜신호비율": float(trades["false_signal"].mean()), "평균보유일": float(trades["holding_days"].mean()), "노출비중": float(portfolio["exposure"].mean())}

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], signals: pd.DataFrame) -> None:
    colors = {"ROC 단독": "#F59E0B", "ROC + ADX·DMI": "#0F766E"}
    fig = plt.figure(figsize=(13, 9), facecolor="white")
    grid = fig.add_gridspec(2, 2, height_ratios=[1.8, 1.2], hspace=0.35, wspace=0.27)
    equity_axis = fig.add_subplot(grid[0, :])
    count_axis = fig.add_subplot(grid[1, 0])
    rejection_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.3, label=f"{label} {summary.loc[label, '누적수익률']:+.1%}")
    equity_axis.set_title("ADX·DMI 필터는 단기 손실과 좋은 ROC 신호를 함께 거른다", fontsize=18, fontweight="bold", pad=16)
    equity_axis.set_ylabel("30종목 동일 비중 누적 자산")
    equity_axis.legend(loc="upper left", frameon=False)

    labels = list(summary.index)
    x = np.arange(len(labels))
    width = 0.34
    count_axis.bar(x - width / 2, summary["거래수"], width, color="#94A3B8", label="전체 거래")
    count_axis.bar(x + width / 2, summary["가짜신호수"], width, color=[colors[label] for label in labels], label=f"{FALSE_SIGNAL_DAYS}일 이내 손실")
    count_axis.set_xticks(x, labels)
    count_axis.set_title("거래 수와 가짜 신호", fontsize=14, fontweight="bold")
    count_axis.set_ylabel("거래 수")
    count_axis.legend(frameon=False)
    for container in count_axis.containers:
        count_axis.bar_label(container, padding=3)

    rejected = signals.loc[~signals["accepted"]]
    rejection_labels = ["손실", "수익"]
    rejection_values = [int((rejected["trade_return"] < 0.0).sum()), int((rejected["trade_return"] > 0.0).sum())]
    rejection_axis.bar(rejection_labels, rejection_values, color=["#94A3B8", "#F59E0B"], width=0.55)
    rejection_axis.set_title("필터가 거른 ROC 단독 거래", fontsize=14, fontweight="bold")
    rejection_axis.set_ylabel("거래 수")
    for container in rejection_axis.containers:
        rejection_axis.bar_label(container, padding=3)

    for axis in (equity_axis, count_axis, rejection_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()

trade_frames = {"ROC 단독": [], "ROC + ADX·DMI": []}
equity_series = {"ROC 단독": [], "ROC + ADX·DMI": []}
exposure_series = {"ROC 단독": [], "ROC + ADX·DMI": []}
signal_frames = []
calendar = pd.DatetimeIndex(sorted(set().union(*(frame.loc[(frame.index >= TEST_START) & (frame.index < TEST_END)].index for frame in prices.values()))))
for ticker in TICKERS:
    data = prepare_price(prices, ticker)
    if len(data) < MIN_ROWS or data.dropna(subset=["roc", "adx", "plus_di", "minus_di"]).empty:
        continue
    baseline_trades, baseline_daily = run_backtest(data, "roc_entry_signal")
    filtered_trades, filtered_daily = run_backtest(data, "filtered_entry_signal")
    for label, trades, daily in (("ROC 단독", baseline_trades, baseline_daily), ("ROC + ADX·DMI", filtered_trades, filtered_daily)):
        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))
    signal_frames.append(analyze_roc_signals(data, baseline_trades, ticker))

portfolios = {}
all_trades = {}
summary_rows = {}
for label in trade_frames:
    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)
signals = pd.concat(signal_frames, ignore_index=True)
accepted = signals.loc[signals["accepted"]]
rejected = signals.loc[~signals["accepted"]]
summary.to_csv(RESULT_FILE, encoding="utf-8-sig")
signals.to_csv(SIGNAL_FILE, index=False, encoding="utf-8-sig")
draw_chart(summary, portfolios, signals)

print(summary.to_string(formatters={"누적수익률": "{:+.2%}".format, "MDD": "{:.2%}".format, "Profit Factor": "{:.2f}".format, "승률": "{:.2%}".format, "가짜신호비율": "{:.2%}".format, "평균보유일": "{:.1f}".format, "노출비중": "{:.2%}".format}))
print(f"ROC 단독 거래 중 필터 통과: {len(accepted)}/{len(signals)} ({len(accepted) / len(signals):.2%})")
print(f"필터가 거른 ROC 단독 거래: {len(rejected)}건, 손실 {int((rejected['trade_return'] < 0.0).sum())}건, 수익 {int((rejected['trade_return'] > 0.0).sum())}건")
print(f"거른 거래의 합산 단순수익률: {rejected['trade_return'].sum():+.2%}")
print(f"결과 저장: {RESULT_FILE}")
print(f"신호 저장: {SIGNAL_FILE}")
print(f"차트 저장: {CHART_FILE}")