CCI 과매도 탈출에 샀더니 685번 손절, 추세·변동성 필터를 붙인 결과

#!/usr/bin/env python3
"""CCI 역추세 진입에 추세·변동성 필터를 추가한 결과를 비교한다."""

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"
CCI_WINDOW = 20
EMA_WINDOW = 200
EMA_SLOPE_DAYS = 20
ATR_WINDOW = 14
MAX_ATR_PERCENT = 5.0
STOP_LOSS = 0.08
MAX_HOLDING_DAYS = 20
COST_RATE = 0.0005
MIN_ROWS = 252
CHART_FILE = Path("cci_filter_chart.png")
RESULT_FILE = Path("cci_filter_results.csv")
TRADE_FILE = Path("cci_filter_trades.csv")

def download_prices() -> dict[str, pd.DataFrame]:
    prices = {}
    for ticker in TICKERS:
        frame = pd.DataFrame()
        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)
            frame = retry.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:
                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
    return prices

def mean_absolute_deviation(values: np.ndarray) -> float:
    return float(np.mean(np.abs(values - values.mean())))

def prepare_price(prices: dict[str, pd.DataFrame], ticker: str) -> pd.DataFrame:
    data = prices[ticker].copy()
    typical_price = (data["high"] + data["low"] + data["close"]) / 3.0
    typical_average = typical_price.rolling(CCI_WINDOW, min_periods=CCI_WINDOW).mean()
    mean_deviation = typical_price.rolling(CCI_WINDOW, min_periods=CCI_WINDOW).apply(mean_absolute_deviation, raw=True)
    data["cci"] = (typical_price - typical_average) / (0.015 * mean_deviation)
    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["atr_percent"] = data["atr"] / data["close"] * 100.0
    data["ema_long"] = data["close"].ewm(span=EMA_WINDOW, adjust=False, min_periods=EMA_WINDOW).mean()
    data["trend_filter"] = (data["close"] > data["ema_long"]) & (data["ema_long"] > data["ema_long"].shift(EMA_SLOPE_DAYS))
    data["volatility_filter"] = data["atr_percent"] <= MAX_ATR_PERCENT
    data["cci_entry_signal"] = (data["cci"] > -100.0) & (data["cci"].shift(1) <= -100.0)
    data["filtered_entry_signal"] = data["cci_entry_signal"] & data["trend_filter"] & data["volatility_filter"]
    data["cci_exit_signal"] = (data["cci"] > 100.0) & (data["cci"].shift(1) <= 100.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, "entry_cci": pending_order["entry_cci"], "entry_atr_percent": pending_order["entry_atr_percent"], "trend_pass": pending_order["trend_pass"], "volatility_pass": pending_order["volatility_pass"], "minimum_close": entry_price}
            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"]
                maximum_adverse_excursion = current_trade["minimum_close"] / current_trade["entry_price"] - 1.0
                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, "maximum_adverse_excursion": maximum_adverse_excursion, "exit_reason": pending_order["reason"], "forced_exit": False})
                current_trade = None
            pending_order = None

        if shares > 0.0:
            current_trade["minimum_close"] = min(current_trade["minimum_close"], float(row["close"]))
            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": "손절"}
            elif bool(row["cci_exit_signal"]):
                pending_order = {"side": "sell", "signal_date": date, "reason": "CCI 청산"}
            elif holding_days >= MAX_HOLDING_DAYS:
                pending_order = {"side": "sell", "signal_date": date, "reason": "시간 청산"}
        elif bool(row[entry_column]):
            pending_order = {"side": "buy", "signal_date": date, "entry_cci": float(row["cci"]), "entry_atr_percent": float(row["atr_percent"]), "trend_pass": bool(row["trend_filter"]), "volatility_pass": bool(row["volatility_filter"])}

        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"]
        maximum_adverse_excursion = min(current_trade["minimum_close"], last_close) / current_trade["entry_price"] - 1.0
        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, "maximum_adverse_excursion": maximum_adverse_excursion, "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), "손절수": int((trades["exit_reason"] == "손절").sum()), "손절비율": float((trades["exit_reason"] == "손절").mean()), "중앙거래수익률": float(trades["trade_return"].median()), "평균MAE": float(trades["maximum_adverse_excursion"].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, all_trades: dict[str, pd.DataFrame], prepared: dict[str, pd.DataFrame]) -> None:
    colors = {"CCI 단독": "#F59E0B", "추세·변동성 필터": "#0F766E"}
    rejected = all_trades["CCI 단독"].loc[~all_trades["CCI 단독"]["trend_pass"] | ~all_trades["CCI 단독"]["volatility_pass"]]
    worst = rejected.sort_values("trade_return").iloc[0]
    example = prepared[worst["ticker"]]
    start_position = max(0, example.index.get_loc(pd.Timestamp(worst["entry_date"])) - 35)
    end_position = min(len(example), example.index.get_loc(pd.Timestamp(worst["exit_date"])) + 21)
    window = example.iloc[start_position:end_position]

    fig = plt.figure(figsize=(13, 9), facecolor="white")
    grid = fig.add_gridspec(2, 2, width_ratios=[1.55, 1.0], height_ratios=[1.45, 1.0], hspace=0.32, wspace=0.25)
    price_axis = fig.add_subplot(grid[0, 0])
    cci_axis = fig.add_subplot(grid[1, 0], sharex=price_axis)
    distribution_axis = fig.add_subplot(grid[0, 1])
    reason_axis = fig.add_subplot(grid[1, 1])

    price_axis.plot(window.index, window["close"], color="#14213D", linewidth=2.2, label="종가")
    price_axis.plot(window.index, window["ema_long"], color="#64748B", linewidth=1.8, linestyle="--", label="200일 EMA")
    price_axis.scatter(pd.Timestamp(worst["entry_date"]), worst["entry_price"], s=110, color=AMBER, edgecolor="white", linewidth=1.3, zorder=5, label="CCI 단독 진입")
    price_axis.scatter(pd.Timestamp(worst["exit_date"]), worst["exit_price"], s=110, color="#B91C1C", marker="X", edgecolor="white", linewidth=1.0, zorder=5, label=f"{worst['exit_reason']} {worst['trade_return']:+.1%}")
    price_axis.set_title(f"필터가 막은 하락 지속 사례 · {worst['ticker']}", fontsize=16, fontweight="bold")
    price_axis.set_ylabel("수정주가")
    price_axis.legend(frameon=False, fontsize=9)

    cci_axis.plot(window.index, window["cci"], color="#7C3AED", linewidth=1.8)
    cci_axis.axhline(-100.0, color=AMBER, linestyle="--", linewidth=1.4)
    cci_axis.axhline(100.0, color="#0F766E", linestyle="--", linewidth=1.2)
    cci_axis.fill_between(window.index, window["cci"], -100.0, where=window["cci"] < -100.0, color="#FDE68A", alpha=0.55)
    cci_axis.set_title("CCI 과매도 탈출 뒤에도 가격은 더 내려갈 수 있다", fontsize=13, fontweight="bold")
    cci_axis.set_ylabel("CCI(20)")

    labels = list(all_trades)
    distributions = [all_trades[label]["trade_return"].clip(-0.2, 0.2) * 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="#64748B", linewidth=1.1)
    distribution_axis.set_title("전체 거래 수익률 분포", fontsize=15, fontweight="bold")
    distribution_axis.set_ylabel("거래 수익률(%, -20~+20 표시)")

    reason_table = pd.DataFrame({label: all_trades[label]["exit_reason"].value_counts(normalize=True) for label in labels}).fillna(0.0).T
    reason_order = [reason for reason in ("CCI 청산", "시간 청산", "손절", "기간 종료") if reason in reason_table.columns]
    reason_colors = {"CCI 청산": "#0F766E", "시간 청산": "#94A3B8", "손절": "#B91C1C", "기간 종료": "#F59E0B"}
    left = np.zeros(len(labels))
    for reason in reason_order:
        values = reason_table[reason].to_numpy() * 100.0
        reason_axis.barh(labels, values, left=left, color=reason_colors[reason], label=reason)
        left += values
    reason_axis.set_title("청산 사유 구성", fontsize=15, fontweight="bold")
    reason_axis.set_xlabel("거래 비중(%)")
    reason_axis.legend(frameon=False, fontsize=9, loc="lower center", bbox_to_anchor=(0.5, -0.42), ncol=2)

    fig.suptitle("CCI 과매도 신호는 추세와 변동성을 확인해야 하락 지속 구간을 피할 수 있다", fontsize=20, fontweight="bold", y=0.99)
    for axis in (price_axis, cci_axis, distribution_axis, reason_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()
AMBER = "#F59E0B"
prices = download_prices()
prepared = {ticker: prepare_price(prices, ticker) for ticker in TICKERS}
calendar = pd.DatetimeIndex(sorted(set().union(*(data.index for data in prepared.values()))))
trade_frames = {"CCI 단독": [], "추세·변동성 필터": []}
equity_series = {"CCI 단독": [], "추세·변동성 필터": []}
exposure_series = {"CCI 단독": [], "추세·변동성 필터": []}
for ticker, data in prepared.items():
    if len(data) < MIN_ROWS or data.dropna(subset=["cci", "ema_long", "atr_percent"]).empty:
        continue
    baseline_trades, baseline_daily = run_backtest(data, "cci_entry_signal")
    filtered_trades, filtered_daily = run_backtest(data, "filtered_entry_signal")
    for label, trades, daily in (("CCI 단독", baseline_trades, baseline_daily), ("추세·변동성 필터", 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))

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)
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, all_trades, prepared)

rejected = all_trades["CCI 단독"].loc[~all_trades["CCI 단독"]["trend_pass"] | ~all_trades["CCI 단독"]["volatility_pass"]]
print(summary.to_string(formatters={"누적수익률": "{:+.2%}".format, "MDD": "{:.2%}".format, "Profit Factor": "{:.2f}".format, "승률": "{:.2%}".format, "손절비율": "{:.2%}".format, "중앙거래수익률": "{:+.2%}".format, "평균MAE": "{:.2%}".format, "노출비중": "{:.2%}".format}))
print(f"CCI 단독 거래 중 필터 조건 미충족: {len(rejected)}건, 손실 {int((rejected['trade_return'] < 0.0).sum())}건, 수익 {int((rejected['trade_return'] > 0.0).sum())}건")
print(f"미충족 거래 중 8% 손절: {int((rejected['exit_reason'] == '손절').sum())}건")
print(f"결과 저장: {RESULT_FILE}")
print(f"거래 저장: {TRADE_FILE}")
print(f"차트 저장: {CHART_FILE}")