정량적 리스크 관리 실전 가이드: VaR·CVaR·포트폴리오 리스크 메트릭 Python 구현

VaR_alpha = -inf { x : P(L > x) <= 1 - alpha }

import numpy as np
import pandas as pd
import yfinance as yf
from typing import Tuple

def fetch_portfolio_data(
    tickers: list[str],
    start: str = "2023-01-01",
    end: str = "2026-03-01"
) -> pd.DataFrame:
    """야후 파이낸스에서 조정 종가 데이터를 다운로드한다."""
    data = yf.download(tickers, start=start, end=end, auto_adjust=True)
    prices = data["Close"][tickers]
    return prices

def calculate_portfolio_returns(
    prices: pd.DataFrame,
    weights: np.ndarray
) -> pd.Series:
    """포트폴리오 일별 수익률을 계산한다."""
    daily_returns = prices.pct_change().dropna()
    portfolio_returns = daily_returns.dot(weights)
    return portfolio_returns

def historical_var(
    returns: pd.Series,
    confidence_level: float = 0.95,
    portfolio_value: float = 1_000_000_000
) -> Tuple[float, float]:
    """
    Historical Simulation VaR를 계산한다.

    Parameters
    ----------
    returns : 포트폴리오 일별 수익률
    confidence_level : 신뢰수준 (0.95 = 95%)
    portfolio_value : 포트폴리오 총 가치 (원)

    Returns
    -------
    var_pct : VaR (수익률 기준)
    var_amount : VaR (금액 기준)
    """
    # 손실 분포의 백분위수 계산 (왼쪽 꼬리)
    var_pct = np.percentile(returns, (1 - confidence_level) * 100)
    var_amount = abs(var_pct) * portfolio_value
    return var_pct, var_amount

# 실행 예시
tickers = ["AAPL", "MSFT", "GOOGL", "AMZN"]
weights = np.array([0.25, 0.25, 0.25, 0.25])

prices = fetch_portfolio_data(tickers)
port_returns = calculate_portfolio_returns(prices, weights)

var_95_pct, var_95_amt = historical_var(port_returns, 0.95)
var_99_pct, var_99_amt = historical_var(port_returns, 0.99)

print(f"=== Historical Simulation VaR ===")
print(f"95% 1-Day VaR: {var_95_pct:.4%} (약 {var_95_amt:,.0f}원)")
print(f"99% 1-Day VaR: {var_99_pct:.4%} (약 {var_99_amt:,.0f}원)")
print(f"관측 기간: {len(port_returns)}일")
print(f"일별 수익률 평균: {port_returns.mean():.4%}")
print(f"일별 수익률 표준편차: {port_returns.std():.4%}")

VaR_alpha = -(mu + z_alpha * sigma)

from scipy import stats

def parametric_var(
    returns: pd.Series,
    confidence_level: float = 0.95,
    portfolio_value: float = 1_000_000_000
) -> dict:
    """
    Parametric (Variance-Covariance) VaR를 계산한다.
    정규분포 가정 하에서 닫힌 형태 공식을 사용한다.
    """
    mu = returns.mean()
    sigma = returns.std()

    # 정규분포 z-score
    z_score = stats.norm.ppf(1 - confidence_level)

    # VaR 계산 (음수 수익률 = 손실)
    var_pct = -(mu + z_score * sigma)
    var_amount = var_pct * portfolio_value

    return {
        "var_pct": var_pct,
        "var_amount": var_amount,
        "mu": mu,
        "sigma": sigma,
        "z_score": z_score,
    }

def multi_asset_parametric_var(
    prices: pd.DataFrame,
    weights: np.ndarray,
    confidence_level: float = 0.95,
    portfolio_value: float = 1_000_000_000
) -> dict:
    """
    다중 자산 포트폴리오의 Parametric VaR를 공분산 행렬로 계산한다.
    """
    daily_returns = prices.pct_change().dropna()

    # 공분산 행렬 (연율화하지 않고 일별 기준)
    cov_matrix = daily_returns.cov()

    # 포트폴리오 분산: w^T * Sigma * w
    port_variance = np.dot(weights.T, np.dot(cov_matrix.values, weights))
    port_sigma = np.sqrt(port_variance)
    port_mu = daily_returns.mean().dot(weights)

    z_score = stats.norm.ppf(1 - confidence_level)
    var_pct = -(port_mu + z_score * port_sigma)
    var_amount = var_pct * portfolio_value

    return {
        "var_pct": var_pct,
        "var_amount": var_amount,
        "port_sigma": port_sigma,
        "port_mu": port_mu,
        "cov_matrix": cov_matrix,
    }

# 실행 예시
result = multi_asset_parametric_var(prices, weights, 0.95)
print(f"=== Parametric VaR ===")
print(f"포트폴리오 일별 변동성: {result['port_sigma']:.4%}")
print(f"95% 1-Day VaR: {result['var_pct']:.4%} (약 {result['var_amount']:,.0f}원)")
print(f"\n공분산 행렬:")
print(result["cov_matrix"].round(6))

def monte_carlo_var(
    prices: pd.DataFrame,
    weights: np.ndarray,
    confidence_level: float = 0.95,
    portfolio_value: float = 1_000_000_000,
    n_simulations: int = 100_000,
    n_days: int = 1,
    random_seed: int = 42
) -> dict:
    """
    Monte Carlo Simulation VaR를 계산한다.
    Cholesky 분해로 자산 간 상관관계를 반영한 시뮬레이션을 수행한다.

    Parameters
    ----------
    prices : 자산별 가격 데이터
    weights : 포트폴리오 비중
    confidence_level : 신뢰수준
    portfolio_value : 포트폴리오 총 가치
    n_simulations : 시뮬레이션 횟수
    n_days : 보유 기간 (일)
    random_seed : 재현성을 위한 시드 값

    Returns
    -------
    dict : VaR, CVaR, 시뮬레이션된 수익률 분포 등
    """
    np.random.seed(random_seed)

    daily_returns = prices.pct_change().dropna()
    mean_returns = daily_returns.mean().values
    cov_matrix = daily_returns.cov().values

    # Cholesky 분해
    L = np.linalg.cholesky(cov_matrix)

    # 상관된 난수 생성
    n_assets = len(weights)
    Z = np.random.standard_normal((n_simulations, n_assets))
    correlated_returns = Z @ L.T

    # n일 보유 기간 시뮬레이션
    if n_days > 1:
        cumulative_returns = np.zeros(n_simulations)
        for _ in range(n_days):
            Z = np.random.standard_normal((n_simulations, n_assets))
            daily_sim = mean_returns + Z @ L.T
            port_daily = daily_sim @ weights
            cumulative_returns += port_daily
        portfolio_returns_sim = cumulative_returns
    else:
        simulated_returns = mean_returns + correlated_returns
        portfolio_returns_sim = simulated_returns @ weights

    # VaR 및 CVaR 계산
    var_pct = np.percentile(portfolio_returns_sim, (1 - confidence_level) * 100)
    cvar_pct = portfolio_returns_sim[
        portfolio_returns_sim <= var_pct
    ].mean()

    var_amount = abs(var_pct) * portfolio_value
    cvar_amount = abs(cvar_pct) * portfolio_value

    return {
        "var_pct": var_pct,
        "var_amount": var_amount,
        "cvar_pct": cvar_pct,
        "cvar_amount": cvar_amount,
        "simulated_returns": portfolio_returns_sim,
        "n_simulations": n_simulations,
    }

# 실행 예시
mc_result = monte_carlo_var(prices, weights, 0.95, n_simulations=100_000)
print(f"=== Monte Carlo VaR (100,000 시나리오) ===")
print(f"95% 1-Day VaR: {mc_result['var_pct']:.4%} ({mc_result['var_amount']:,.0f}원)")
print(f"95% 1-Day CVaR: {mc_result['cvar_pct']:.4%} ({mc_result['cvar_amount']:,.0f}원)")

CVaR_alpha = -E[X | X <= -VaR_alpha]
           = -(1 / (1 - alpha)) * integral_{-inf}^{-VaR_alpha} x * f(x) dx

def calculate_cvar(
    returns: pd.Series,
    confidence_level: float = 0.95,
    portfolio_value: float = 1_000_000_000
) -> dict:
    """
    Historical Simulation 기반 CVaR(Expected Shortfall)을 계산한다.

    CVaR = VaR 임계점 이하 수익률의 평균
    """
    var_threshold = np.percentile(returns, (1 - confidence_level) * 100)

    # VaR 임계점 이하의 수익률만 필터링
    tail_losses = returns[returns <= var_threshold]

    cvar_pct = tail_losses.mean()
    cvar_amount = abs(cvar_pct) * portfolio_value

    return {
        "var_pct": var_threshold,
        "var_amount": abs(var_threshold) * portfolio_value,
        "cvar_pct": cvar_pct,
        "cvar_amount": cvar_amount,
        "tail_count": len(tail_losses),
        "total_observations": len(returns),
        "tail_ratio": len(tail_losses) / len(returns),
    }

# 다양한 신뢰수준에서 VaR와 CVaR 비교
print(f"{'신뢰수준':>10} {'VaR':>12} {'CVaR':>12} {'CVaR/VaR':>10}")
print("-" * 50)
for cl in [0.90, 0.95, 0.975, 0.99]:
    result = calculate_cvar(port_returns, cl)
    ratio = result["cvar_amount"] / result["var_amount"]
    print(
        f"{cl:>10.1%} "
        f"{result['var_amount']:>12,.0f} "
        f"{result['cvar_amount']:>12,.0f} "
        f"{ratio:>10.2f}x"
    )

from scipy.optimize import minimize

def portfolio_cvar(
    weights: np.ndarray,
    returns_matrix: np.ndarray,
    confidence_level: float = 0.95
) -> float:
    """주어진 비중에서 포트폴리오의 CVaR를 계산한다."""
    portfolio_returns = returns_matrix @ weights
    var_threshold = np.percentile(
        portfolio_returns, (1 - confidence_level) * 100
    )
    cvar = portfolio_returns[portfolio_returns <= var_threshold].mean()
    return -cvar  # 최소화 문제이므로 부호 반전 (손실을 양수로)

def optimize_min_cvar_portfolio(
    prices: pd.DataFrame,
    confidence_level: float = 0.95,
    target_return: float = None,
    min_weight: float = 0.0,
    max_weight: float = 1.0
) -> dict:
    """
    CVaR를 최소화하는 최적 포트폴리오 비중을 산출한다.

    Parameters
    ----------
    prices : 자산별 가격 데이터
    confidence_level : CVaR 신뢰수준
    target_return : 목표 기대수익률 (None이면 제약 없음)
    min_weight : 최소 자산 비중
    max_weight : 최대 자산 비중
    """
    daily_returns = prices.pct_change().dropna()
    returns_matrix = daily_returns.values
    n_assets = returns_matrix.shape[1]
    asset_names = prices.columns.tolist()

    # 초기 비중: 동일 비중
    initial_weights = np.ones(n_assets) / n_assets

    # 제약조건
    constraints = [
        # 비중 합계 = 1
        {"type": "eq", "fun": lambda w: np.sum(w) - 1.0},
    ]

    # 목표 수익률 제약 (선택)
    if target_return is not None:
        mean_returns = daily_returns.mean().values
        constraints.append({
            "type": "ineq",
            "fun": lambda w: w @ mean_returns - target_return
        })

    # 비중 범위 제약
    bounds = [(min_weight, max_weight)] * n_assets

    # 최적화 실행
    result = minimize(
        portfolio_cvar,
        initial_weights,
        args=(returns_matrix, confidence_level),
        method="SLSQP",
        bounds=bounds,
        constraints=constraints,
        options={"maxiter": 1000, "ftol": 1e-10}
    )

    optimal_weights = result.x
    opt_port_returns = returns_matrix @ optimal_weights

    # 최적 포트폴리오 메트릭 계산
    opt_var = np.percentile(opt_port_returns, (1 - confidence_level) * 100)
    opt_cvar = opt_port_returns[opt_port_returns <= opt_var].mean()

    return {
        "weights": dict(zip(asset_names, optimal_weights.round(4))),
        "annual_return": float(np.mean(opt_port_returns) * 252),
        "annual_volatility": float(np.std(opt_port_returns) * np.sqrt(252)),
        "var_95": float(opt_var),
        "cvar_95": float(opt_cvar),
        "optimization_success": result.success,
        "sharpe_ratio": float(
            (np.mean(opt_port_returns) * 252)
            / (np.std(opt_port_returns) * np.sqrt(252))
        ),
    }

# 실행 예시
tickers = ["AAPL", "MSFT", "GOOGL", "AMZN", "BRK-B", "JNJ", "JPM", "V"]
prices = fetch_portfolio_data(tickers)

opt_result = optimize_min_cvar_portfolio(
    prices,
    confidence_level=0.95,
    min_weight=0.05,
    max_weight=0.30
)

print("=== CVaR 최소화 최적 포트폴리오 ===")
print(f"최적화 성공: {opt_result['optimization_success']}")
print(f"\n자산별 비중:")
for asset, weight in opt_result["weights"].items():
    bar = "█" * int(weight * 50)
    print(f"  {asset:>6}: {weight:>7.2%} {bar}")
print(f"\n연환산 기대수익률: {opt_result['annual_return']:.2%}")
print(f"연환산 변동성: {opt_result['annual_volatility']:.2%}")
print(f"샤프 비율: {opt_result['sharpe_ratio']:.3f}")
print(f"95% 1-Day VaR: {opt_result['var_95']:.4%}")
print(f"95% 1-Day CVaR: {opt_result['cvar_95']:.4%}")

def generate_cvar_efficient_frontier(
    prices: pd.DataFrame,
    n_points: int = 30,
    confidence_level: float = 0.95
) -> pd.DataFrame:
    """CVaR 기준 Efficient Frontier를 생성한다."""
    daily_returns = prices.pct_change().dropna()
    mean_returns = daily_returns.mean()

    # 목표 수익률 범위 설정
    min_ret = mean_returns.min()
    max_ret = mean_returns.max()
    target_returns = np.linspace(min_ret * 0.5, max_ret * 1.2, n_points)

    frontier_data = []
    for target_ret in target_returns:
        try:
            result = optimize_min_cvar_portfolio(
                prices,
                confidence_level=confidence_level,
                target_return=target_ret,
                min_weight=0.0,
                max_weight=0.40
            )
            if result["optimization_success"]:
                frontier_data.append({
                    "target_return": target_ret * 252,
                    "annual_return": result["annual_return"],
                    "annual_volatility": result["annual_volatility"],
                    "cvar_95": abs(result["cvar_95"]),
                    "sharpe_ratio": result["sharpe_ratio"],
                })
        except Exception:
            continue

    return pd.DataFrame(frontier_data)

frontier_df = generate_cvar_efficient_frontier(prices)
print("=== CVaR Efficient Frontier (상위 10개 포인트) ===")
print(frontier_df.sort_values("sharpe_ratio", ascending=False).head(10).to_string())

import matplotlib.pyplot as plt
import matplotlib.dates as mdates

def build_risk_dashboard(
    prices: pd.DataFrame,
    weights: np.ndarray,
    window: int = 252,
    confidence_level: float = 0.95,
    portfolio_value: float = 1_000_000_000
) -> pd.DataFrame:
    """
    롤링 VaR/CVaR 대시보드 데이터를 생성한다.
    """
    daily_returns = prices.pct_change().dropna()
    port_returns = daily_returns.dot(weights)

    dashboard = pd.DataFrame(index=port_returns.index[window:])
    dashboard["actual_return"] = port_returns.iloc[window:].values

    rolling_var = []
    rolling_cvar = []

    for i in range(window, len(port_returns)):
        hist_window = port_returns.iloc[i - window:i]
        var_val = np.percentile(
            hist_window, (1 - confidence_level) * 100
        )
        cvar_val = hist_window[hist_window <= var_val].mean()
        rolling_var.append(var_val)
        rolling_cvar.append(cvar_val)

    dashboard["var"] = rolling_var
    dashboard["cvar"] = rolling_cvar
    dashboard["breach"] = dashboard["actual_return"] < dashboard["var"]

    return dashboard

def plot_risk_dashboard(dashboard: pd.DataFrame) -> None:
    """리스크 대시보드를 시각화한다."""
    fig, axes = plt.subplots(3, 1, figsize=(14, 12), sharex=True)

    # 1. 실제 수익률 vs VaR/CVaR
    ax1 = axes[0]
    ax1.plot(
        dashboard.index, dashboard["actual_return"],
        color="steelblue", alpha=0.6, linewidth=0.8,
        label="Actual Return"
    )
    ax1.plot(
        dashboard.index, dashboard["var"],
        color="orange", linewidth=1.2, label="95% VaR"
    )
    ax1.plot(
        dashboard.index, dashboard["cvar"],
        color="red", linewidth=1.2, label="95% CVaR"
    )
    # VaR breach 표시
    breach_dates = dashboard[dashboard["breach"]].index
    breach_returns = dashboard.loc[breach_dates, "actual_return"]
    ax1.scatter(
        breach_dates, breach_returns,
        color="red", s=20, zorder=5, label="VaR Breach"
    )
    ax1.set_ylabel("Daily Return")
    ax1.set_title("Risk Dashboard: Actual Returns vs VaR/CVaR")
    ax1.legend(loc="lower left")
    ax1.axhline(y=0, color="gray", linestyle="--", linewidth=0.5)

    # 2. 롤링 VaR 추이
    ax2 = axes[1]
    ax2.fill_between(
        dashboard.index,
        dashboard["cvar"], dashboard["var"],
        alpha=0.3, color="red", label="VaR-CVaR Gap"
    )
    ax2.plot(
        dashboard.index, dashboard["var"],
        color="orange", label="95% VaR"
    )
    ax2.plot(
        dashboard.index, dashboard["cvar"],
        color="red", label="95% CVaR"
    )
    ax2.set_ylabel("Risk Level")
    ax2.set_title("Rolling VaR & CVaR Trend")
    ax2.legend()

    # 3. 누적 VaR Breach 비율
    ax3 = axes[2]
    cumulative_breach = dashboard["breach"].cumsum()
    cumulative_count = range(1, len(dashboard) + 1)
    breach_ratio = cumulative_breach / cumulative_count
    ax3.plot(
        dashboard.index, breach_ratio,
        color="red", linewidth=1.2, label="Cumulative Breach Ratio"
    )
    ax3.axhline(
        y=0.05, color="gray", linestyle="--",
        label="Expected 5% (95% CL)"
    )
    ax3.set_ylabel("Breach Ratio")
    ax3.set_title("Cumulative VaR Breach Ratio")
    ax3.legend()
    ax3.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m"))

    plt.tight_layout()
    plt.savefig("risk_dashboard.png", dpi=150, bbox_inches="tight")
    plt.show()

# 실행
tickers = ["AAPL", "MSFT", "GOOGL", "AMZN"]
weights = np.array([0.25, 0.25, 0.25, 0.25])
prices = fetch_portfolio_data(tickers)

dashboard = build_risk_dashboard(prices, weights, window=252)
plot_risk_dashboard(dashboard)

# 서머리 통계
total_days = len(dashboard)
breach_count = dashboard["breach"].sum()
breach_pct = breach_count / total_days
print(f"\n=== 리스크 대시보드 요약 ===")
print(f"분석 기간: {dashboard.index[0].strftime('%Y-%m-%d')} ~ "
      f"{dashboard.index[-1].strftime('%Y-%m-%d')}")
print(f"총 거래일: {total_days}일")
print(f"VaR Breach 횟수: {breach_count}회")
print(f"Breach 비율: {breach_pct:.2%} (이론적 기대치: 5.00%)")

from dataclasses import dataclass
from datetime import datetime

@dataclass
class RiskAlert:
    alert_type: str
    severity: str  # "WARNING", "CRITICAL"
    metric_name: str
    current_value: float
    threshold: float
    timestamp: datetime
    message: str

def check_risk_limits(
    dashboard: pd.DataFrame,
    var_limit: float = -0.03,
    cvar_limit: float = -0.05,
    max_consecutive_breaches: int = 3
) -> list[RiskAlert]:
    """리스크 한도 초과 여부를 점검하고 알림을 생성한다."""
    alerts = []
    latest = dashboard.iloc[-1]
    now = datetime.now()

    # 1. VaR 한도 초과 점검
    if latest["var"] < var_limit:
        alerts.append(RiskAlert(
            alert_type="VAR_LIMIT_BREACH",
            severity="WARNING",
            metric_name="95% VaR",
            current_value=latest["var"],
            threshold=var_limit,
            timestamp=now,
            message=f"VaR가 한도를 초과했습니다: "
                    f"현재 {latest['var']:.4%} < 한도 {var_limit:.4%}"
        ))

    # 2. CVaR 한도 초과 점검
    if latest["cvar"] < cvar_limit:
        alerts.append(RiskAlert(
            alert_type="CVAR_LIMIT_BREACH",
            severity="CRITICAL",
            metric_name="95% CVaR",
            current_value=latest["cvar"],
            threshold=cvar_limit,
            timestamp=now,
            message=f"CVaR가 한도를 초과했습니다: "
                    f"현재 {latest['cvar']:.4%} < 한도 {cvar_limit:.4%}"
        ))

    # 3. 연속 breach 점검
    recent_breaches = dashboard["breach"].tail(max_consecutive_breaches)
    if recent_breaches.all():
        alerts.append(RiskAlert(
            alert_type="CONSECUTIVE_BREACHES",
            severity="CRITICAL",
            metric_name="VaR Breach Streak",
            current_value=float(max_consecutive_breaches),
            threshold=float(max_consecutive_breaches),
            timestamp=now,
            message=f"VaR breach가 {max_consecutive_breaches}일 "
                    f"연속 발생했습니다. 즉시 점검이 필요합니다."
        ))

    return alerts

# 잘못된 방식: 매일 같은 시나리오 생성
np.random.seed(42)  # 프로덕션에서 이렇게 하면 안 됨

# 올바른 방식: 날짜 기반 시드 또는 시드 미사용
import hashlib
date_str = datetime.now().strftime("%Y%m%d")
seed = int(hashlib.sha256(date_str.encode()).hexdigest(), 16) % (2**32)
rng = np.random.default_rng(seed)

def kupiec_pof_test(
    actual_returns: pd.Series,
    var_series: pd.Series,
    confidence_level: float = 0.95
) -> dict:
    """
    Kupiec POF(Proportion of Failures) Test를 수행한다.

    귀무가설: 실제 breach 비율 = 이론적 기대 비율 (1 - confidence_level)
    """
    n = len(actual_returns)
    breaches = (actual_returns < var_series).sum()
    p_expected = 1 - confidence_level
    p_observed = breaches / n

    # 우도비(Likelihood Ratio) 통계량
    if breaches == 0 or breaches == n:
        lr_stat = 0.0
    else:
        lr_stat = -2 * (
            np.log((1 - p_expected)**(n - breaches) * p_expected**breaches)
            - np.log(
                (1 - p_observed)**(n - breaches) * p_observed**breaches
            )
        )

    # 카이제곱 분포 (자유도 1) 기반 p-value
    p_value = 1 - stats.chi2.cdf(lr_stat, df=1)

    return {
        "total_observations": n,
        "expected_breaches": int(n * p_expected),
        "actual_breaches": int(breaches),
        "expected_ratio": p_expected,
        "observed_ratio": p_observed,
        "lr_statistic": lr_stat,
        "p_value": p_value,
        "reject_h0_5pct": p_value < 0.05,
        "model_status": "PASS" if p_value >= 0.05 else "FAIL",
    }

# 백테스트 실행
backtest_result = kupiec_pof_test(
    dashboard["actual_return"],
    dashboard["var"],
    confidence_level=0.95
)

print("=== Kupiec POF Test 결과 ===")
print(f"총 관측일: {backtest_result['total_observations']}")
print(f"기대 breach: {backtest_result['expected_breaches']}회 "
      f"({backtest_result['expected_ratio']:.1%})")
print(f"실제 breach: {backtest_result['actual_breaches']}회 "
      f"({backtest_result['observed_ratio']:.2%})")
print(f"LR 통계량: {backtest_result['lr_statistic']:.4f}")
print(f"p-value: {backtest_result['p_value']:.4f}")
print(f"모형 판정 (5% 유의수준): {backtest_result['model_status']}")