상관계수를 낮췄는데 수익률도 함께 줄어든 이유

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
TARGET_VOLATILITY = 0.10
MIN_ROWS = 252
PRICE_CACHE_FILE = Path("portfolio_correlation_prices.parquet")
CHART_FILE = Path("portfolio_correlation_chart.png")
RESULT_FILE = Path("portfolio_correlation_results.csv")
CORRELATION_FILE = Path("portfolio_correlation_matrix.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),
}
ANCHOR_STRATEGY = "EMA 20·100"

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_returns(returns: pd.Series) -> dict[str, float]:
    equity = (1.0 + returns).cumprod()
    drawdown = equity / equity.cummax() - 1.0
    elapsed_years = max((equity.index[-1] - equity.index[0]).days / 365.25, 1.0 / 365.25)
    annual_volatility = float(returns.std(ddof=0) * np.sqrt(252.0))
    cagr = float(equity.iloc[-1] ** (1.0 / elapsed_years) - 1.0)
    return {"return": float(equity.iloc[-1] - 1.0), "cagr": cagr, "annual_volatility": annual_volatility, "mdd": float(drawdown.min()), "calmar": cagr / abs(float(drawdown.min())) if drawdown.min() < 0.0 else np.nan}

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(train_correlation: pd.DataFrame, test_returns: pd.DataFrame, pair_returns: pd.DataFrame, results: pd.DataFrame, high_partner: str, low_partner: str) -> None:
    colors = {"고상관 조합": "#F59E0B", "저상관 조합": "#0F766E", "EMA 20·100 단독": "#64748B"}
    fig = plt.figure(figsize=(14, 10), facecolor="white")
    grid = fig.add_gridspec(2, 2, height_ratios=[1.35, 1.0], hspace=0.34, wspace=0.28)
    equity_axis = fig.add_subplot(grid[0, :])
    matrix_axis = fig.add_subplot(grid[1, 0])
    risk_axis = fig.add_subplot(grid[1, 1])

    for label in pair_returns:
        equity = (1.0 + pair_returns[label]).cumprod()
        row = results.loc[results["portfolio"] == label].iloc[0]
        equity_axis.plot(equity.index, equity, color=colors[label], linewidth=2.5 if "조합" in label else 1.8, label=f"{label}  수익률 {row['return']:+.1%} · MDD {row['mdd']:.1%}")
    equity_axis.axhline(1.0, color="#9CA3AF", linewidth=1.0)
    equity_axis.set_title("같은 기준 변동성으로 맞춘 뒤 남겨둔 구간의 성과만 비교했다", fontsize=18, fontweight="bold", pad=16)
    equity_axis.set_ylabel("표본외 누적 자산")
    equity_axis.legend(loc="upper left", frameon=False, ncol=3)

    matrix_values = train_correlation.to_numpy()
    image = matrix_axis.imshow(matrix_values, cmap="RdYlGn_r", vmin=0.0, vmax=1.0, aspect="auto")
    matrix_axis.set_xticks(np.arange(len(train_correlation.columns)), train_correlation.columns, rotation=40, ha="right", fontsize=8)
    matrix_axis.set_yticks(np.arange(len(train_correlation.index)), train_correlation.index, fontsize=8)
    matrix_axis.set_title("변수 선택 구간 일간 수익률 상관계수", fontsize=14, fontweight="bold")
    for row in range(len(train_correlation.index)):
        for column in range(len(train_correlation.columns)):
            matrix_axis.text(column, row, f"{matrix_values[row, column]:.2f}", ha="center", va="center", fontsize=7, color="#14213D")
    plt.colorbar(image, ax=matrix_axis, fraction=0.046, pad=0.04)

    labels = ["고상관 조합", "저상관 조합"]
    x = np.arange(len(labels))
    width = 0.34
    pair_rows = results.set_index("portfolio").loc[labels]
    risk_axis.bar(x - width / 2, pair_rows["annual_volatility"] * 100.0, width, color="#94A3B8", label="연 변동성")
    risk_axis.bar(x + width / 2, pair_rows["mdd"].abs() * 100.0, width, color=[colors[label] for label in labels], label="MDD 절댓값")
    risk_axis.set_xticks(x, [f"고상관\nEMA+{high_partner}", f"저상관\nEMA+{low_partner}"])
    risk_axis.set_ylabel("위험(%)")
    risk_axis.set_title("표본외 위험 비교", fontsize=14, fontweight="bold")
    risk_axis.legend(frameon=False)
    for container in risk_axis.containers:
        risk_axis.bar_label(container, labels=[f"{bar.get_height():.1f}%" for bar in container], padding=3, fontsize=9)

    for axis in (equity_axis, risk_axis):
        axis.grid(axis="y", color="#E5E7EB", linewidth=0.8)
        axis.spines["top"].set_visible(False)
        axis.spines["right"].set_visible(False)
    fig.suptitle("수익률이 함께 움직이지 않는 시스템을 섞을수록 포트폴리오 위험이 낮아졌다", 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)
train_correlation = train_returns.corr()
eligible = [strategy for strategy in STRATEGIES if strategy != ANCHOR_STRATEGY]
high_partner = train_correlation.loc[ANCHOR_STRATEGY, eligible].idxmax()
low_partner = train_correlation.loc[ANCHOR_STRATEGY, eligible].idxmin()
train_volatility = train_returns.std(ddof=0) * np.sqrt(252.0)
volatility_scale = TARGET_VOLATILITY / train_volatility
scaled_test_returns = test_returns.mul(volatility_scale, axis=1)
pair_returns = pd.DataFrame({
    "EMA 20·100 단독": scaled_test_returns[ANCHOR_STRATEGY],
    "고상관 조합": 0.5 * scaled_test_returns[ANCHOR_STRATEGY] + 0.5 * scaled_test_returns[high_partner],
    "저상관 조합": 0.5 * scaled_test_returns[ANCHOR_STRATEGY] + 0.5 * scaled_test_returns[low_partner],
})

result_rows = []
for label, returns in pair_returns.items():
    metrics = summarize_returns(returns)
    if label == "고상관 조합":
        partner = high_partner
    elif label == "저상관 조합":
        partner = low_partner
    else:
        partner = "-"
    result_rows.append({"portfolio": label, "partner": partner, "train_correlation": 1.0 if partner == "-" else float(train_correlation.loc[ANCHOR_STRATEGY, partner]), "test_correlation": 1.0 if partner == "-" else float(test_returns[[ANCHOR_STRATEGY, partner]].corr().iloc[0, 1]), **metrics})
results = pd.DataFrame(result_rows)
train_correlation.to_csv(CORRELATION_FILE, encoding="utf-8-sig")
results.to_csv(RESULT_FILE, index=False, encoding="utf-8-sig")
draw_chart(train_correlation, test_returns, pair_returns, results, high_partner, low_partner)

print(f"고상관 파트너: {high_partner} · 선택 상관 {train_correlation.loc[ANCHOR_STRATEGY, high_partner]:.3f}")
print(f"저상관 파트너: {low_partner} · 선택 상관 {train_correlation.loc[ANCHOR_STRATEGY, low_partner]:.3f}")
print(results.to_string(index=False, formatters={"train_correlation": "{:.3f}".format, "test_correlation": "{:.3f}".format, "return": "{:+.2%}".format, "cagr": "{:+.2%}".format, "annual_volatility": "{:.2%}".format, "mdd": "{:.2%}".format, "calmar": "{:.2f}".format}))
print(f"결과 저장: {RESULT_FILE}")
print(f"상관행렬 저장: {CORRELATION_FILE}")
print(f"차트 저장: {CHART_FILE}")