リスクメトリクス計算

Value at Risk、Expected Shortfall、ドローダウン分析を含むポートフォリオ管理向けの包括的なリスク測定ツールキット。

このスキルを使用する場合

• ポートフォリオリスクの測定
• リスク制限の実装
• リスクダッシュボードの構築
• リスク調整済みリターンの計算
• ポジションサイズの設定
• 規制レポーティング

コアコンセプト

1. リスクメトリクスのカテゴリ

カテゴリ	メトリクス	使用ケース
ボラティリティ	標準偏差、ベータ	一般的なリスク
テールリスク	VaR、CVaR	極端な損失
ドローダウン	最大DD、Calmar	資本保全
リスク調整済み	シャープ、ソルティーノ	パフォーマンス

2. 時間軸

イントラデイ:   デイトレーダー向けの分単位/時間単位VaR
日次:           標準的なリスク報告
週次:           リバランシング決定
月次:           パフォーマンス属性分析
年次:           戦略配置

実装

パターン1: コアリスクメトリクス

import numpy as np
import pandas as pd
from scipy import stats
from typing import Dict, Optional, Tuple

class RiskMetrics:
    """コアリスクメトリクス計算。"""

    def __init__(self, returns: pd.Series, rf_rate: float = 0.02):
        """
        Args:
            returns: 周期的なリターンのシリーズ
            rf_rate: 年間無リスク金利
        """
        self.returns = returns
        self.rf_rate = rf_rate
        self.ann_factor = 252  # 1年間の営業日数

    # ボラティリティメトリクス
    def volatility(self, annualized: bool = True) -> float:
        """リターンの標準偏差。"""
        vol = self.returns.std()
        if annualized:
            vol *= np.sqrt(self.ann_factor)
        return vol

    def downside_deviation(self, threshold: float = 0, annualized: bool = True) -> float:
        """閾値以下のリターンの標準偏差。"""
        downside = self.returns[self.returns < threshold]
        if len(downside) == 0:
            return 0.0
        dd = downside.std()
        if annualized:
            dd *= np.sqrt(self.ann_factor)
        return dd

    def beta(self, market_returns: pd.Series) -> float:
        """市場に対するベータ。"""
        aligned = pd.concat([self.returns, market_returns], axis=1).dropna()
        if len(aligned) < 2:
            return np.nan
        cov = np.cov(aligned.iloc[:, 0], aligned.iloc[:, 1])
        return cov[0, 1] / cov[1, 1] if cov[1, 1] != 0 else 0

    # Value at Risk
    def var_historical(self, confidence: float = 0.95) -> float:
        """信頼度レベルでの過去VaR。"""
        return -np.percentile(self.returns, (1 - confidence) * 100)

    def var_parametric(self, confidence: float = 0.95) -> float:
        """正規分布を仮定したパラメトリックVaR。"""
        z_score = stats.norm.ppf(confidence)
        return self.returns.mean() - z_score * self.returns.std()

    def var_cornish_fisher(self, confidence: float = 0.95) -> float:
        """正規性から外れた分布を考慮したCornish-Fisher展開によるVaR。"""
        z = stats.norm.ppf(confidence)
        s = stats.skew(self.returns)  # 歪度
        k = stats.kurtosis(self.returns)  # 超過尖度

        # Cornish-Fisher展開
        z_cf = (z + (z**2 - 1) * s / 6 +
                (z**3 - 3*z) * k / 24 -
                (2*z**3 - 5*z) * s**2 / 36)

        return -(self.returns.mean() + z_cf * self.returns.std())

    # 条件付きVaR(Expected Shortfall)
    def cvar(self, confidence: float = 0.95) -> float:
        """Expected Shortfall / CVaR / 平均VaR。"""
        var = self.var_historical(confidence)
        return -self.returns[self.returns <= -var].mean()

    # ドローダウン分析
    def drawdowns(self) -> pd.Series:
        """ドローダウンシリーズを計算。"""
        cumulative = (1 + self.returns).cumprod()
        running_max = cumulative.cummax()
        return (cumulative - running_max) / running_max

    def max_drawdown(self) -> float:
        """最大ドローダウン。"""
        return self.drawdowns().min()

    def avg_drawdown(self) -> float:
        """平均ドローダウン。"""
        dd = self.drawdowns()
        return dd[dd < 0].mean() if (dd < 0).any() else 0

    def drawdown_duration(self) -> Dict[str, int]:
        """ドローダウン期間の統計。"""
        dd = self.drawdowns()
        in_drawdown = dd < 0

        # ドローダウン期間を検出
        drawdown_starts = in_drawdown & ~in_drawdown.shift(1).fillna(False)
        drawdown_ends = ~in_drawdown & in_drawdown.shift(1).fillna(False)

        durations = []
        current_duration = 0

        for i in range(len(dd)):
            if in_drawdown.iloc[i]:
                current_duration += 1
            elif current_duration > 0:
                durations.append(current_duration)
                current_duration = 0

        if current_duration > 0:
            durations.append(current_duration)

        return {
            "max_duration": max(durations) if durations else 0,
            "avg_duration": np.mean(durations) if durations else 0,
            "current_duration": current_duration
        }

    # リスク調整済みリターン
    def sharpe_ratio(self) -> float:
        """年率化シャープレシオ。"""
        excess_return = self.returns.mean() * self.ann_factor - self.rf_rate
        vol = self.volatility(annualized=True)
        return excess_return / vol if vol > 0 else 0

    def sortino_ratio(self) -> float:
        """ソルティーノレシオ(下方偏差を使用)。"""
        excess_return = self.returns.mean() * self.ann_factor - self.rf_rate
        dd = self.downside_deviation(threshold=0, annualized=True)
        return excess_return / dd if dd > 0 else 0

    def calmar_ratio(self) -> float:
        """Calmarレシオ(リターン/最大ドローダウン)。"""
        annual_return = (1 + self.returns).prod() ** (self.ann_factor / len(self.returns)) - 1
        max_dd = abs(self.max_drawdown())
        return annual_return / max_dd if max_dd > 0 else 0

    def omega_ratio(self, threshold: float = 0) -> float:
        """オメガレシオ。"""
        returns_above = self.returns[self.returns > threshold] - threshold
        returns_below = threshold - self.returns[self.returns <= threshold]

        if returns_below.sum() == 0:
            return np.inf

        return returns_above.sum() / returns_below.sum()

    # インフォメーションレシオ
    def information_ratio(self, benchmark_returns: pd.Series) -> float:
        """ベンチマークに対するインフォメーションレシオ。"""
        active_returns = self.returns - benchmark_returns
        tracking_error = active_returns.std() * np.sqrt(self.ann_factor)
        active_return = active_returns.mean() * self.ann_factor
        return active_return / tracking_error if tracking_error > 0 else 0

    # サマリー
    def summary(self) -> Dict[str, float]:
        """包括的なリスクサマリーを生成。"""
        dd_stats = self.drawdown_duration()

        return {
            # リターン
            "total_return": (1 + self.returns).prod() - 1,
            "annual_return": (1 + self.returns).prod() ** (self.ann_factor / len(self.returns)) - 1,

            # ボラティリティ
            "annual_volatility": self.volatility(),
            "downside_deviation": self.downside_deviation(),

            # VaR & CVaR
            "var_95_historical": self.var_historical(0.95),
            "var_99_historical": self.var_historical(0.99),
            "cvar_95": self.cvar(0.95),

            # ドローダウン
            "max_drawdown": self.max_drawdown(),
            "avg_drawdown": self.avg_drawdown(),
            "max_drawdown_duration": dd_stats["max_duration"],

            # リスク調整済み
            "sharpe_ratio": self.sharpe_ratio(),
            "sortino_ratio": self.sortino_ratio(),
            "calmar_ratio": self.calmar_ratio(),
            "omega_ratio": self.omega_ratio(),

            # 分布
            "skewness": stats.skew(self.returns),
            "kurtosis": stats.kurtosis(self.returns),
        }

パターン2: ポートフォリオリスク

class PortfolioRisk:
    """ポートフォリオレベルのリスク計算。"""

    def __init__(
        self,
        returns: pd.DataFrame,
        weights: Optional[pd.Series] = None
    ):
        """
        Args:
            returns: 資産リターンのDataFrame(列=資産)
            weights: ポートフォリオウェイト(デフォルト: 等ウェイト)
        """
        self.returns = returns
        self.weights = weights if weights is not None else \
            pd.Series(1/len(returns.columns), index=returns.columns)
        self.ann_factor = 252

    def portfolio_return(self) -> float:
        """加重ポートフォリオリターン。"""
        return (self.returns @ self.weights).mean() * self.ann_factor

    def portfolio_volatility(self) -> float:
        """ポートフォリオボラティリティ。"""
        cov_matrix = self.returns.cov() * self.ann_factor
        port_var = self.weights @ cov_matrix @ self.weights
        return np.sqrt(port_var)

    def marginal_risk_contribution(self) -> pd.Series:
        """資産ごとのリスクの限界貢献度。"""
        cov_matrix = self.returns.cov() * self.ann_factor
        port_vol = self.portfolio_volatility()

        # 限界貢献度
        mrc = (cov_matrix @ self.weights) / port_vol
        return mrc

    def component_risk(self) -> pd.Series:
        """総リスクに対する構成要素の貢献度。"""
        mrc = self.marginal_risk_contribution()
        return self.weights * mrc

    def risk_parity_weights(self, target_vol: float = None) -> pd.Series:
        """リスクパリティウェイトを計算。"""
        from scipy.optimize import minimize

        n = len(self.returns.columns)
        cov_matrix = self.returns.cov() * self.ann_factor

        def risk_budget_objective(weights):
            port_vol = np.sqrt(weights @ cov_matrix @ weights)
            mrc = (cov_matrix @ weights) / port_vol
            rc = weights * mrc
            target_rc = port_vol / n  # リスク貢献度を等しく
            return np.sum((rc - target_rc) ** 2)

        constraints = [
            {"type": "eq", "fun": lambda w: np.sum(w) - 1},  # ウェイトの合計が1
        ]
        bounds = [(0.01, 1.0) for _ in range(n)]  # 最小1%、最大100%
        x0 = np.array([1/n] * n)

        result = minimize(
            risk_budget_objective,
            x0,
            method="SLSQP",
            bounds=bounds,
            constraints=constraints
        )

        return pd.Series(result.x, index=self.returns.columns)

    def correlation_matrix(self) -> pd.DataFrame:
        """資産相関マトリックス。"""
        return self.returns.corr()

    def diversification_ratio(self) -> float:
        """多様化レシオ(高いほど多様化)。"""
        asset_vols = self.returns.std() * np.sqrt(self.ann_factor)
        weighted_vol = (self.weights * asset_vols).sum()
        port_vol = self.portfolio_volatility()
        return weighted_vol / port_vol if port_vol > 0 else 1

    def tracking_error(self, benchmark_returns: pd.Series) -> float:
        """ベンチマークに対するトラッキングエラー。"""
        port_returns = self.returns @ self.weights
        active_returns = port_returns - benchmark_returns
        return active_returns.std() * np.sqrt(self.ann_factor)

    def conditional_correlation(
        self,
        threshold_percentile: float = 10
    ) -> pd.DataFrame:
        """ストレス期間中の相関。"""
        port_returns = self.returns @ self.weights
        threshold = np.percentile(port_returns, threshold_percentile)
        stress_mask = port_returns <= threshold
        return self.returns[stress_mask].corr()

パターン3: ローリングリスクメトリクス

class RollingRiskMetrics:
    """ローリングウィンドウリスク計算。"""

    def __init__(self, returns: pd.Series, window: int = 63):
        """
        Args:
            returns: リターンシリーズ
            window: ローリングウィンドウサイズ(デフォルト: 63=約3ヶ月)
        """
        self.returns = returns
        self.window = window

    def rolling_volatility(self, annualized: bool = True) -> pd.Series:
        """ローリングボラティリティ。"""
        vol = self.returns.rolling(self.window).std()
        if annualized:
            vol *= np.sqrt(252)
        return vol

    def rolling_sharpe(self, rf_rate: float = 0.02) -> pd.Series:
        """ローリングシャープレシオ。"""
        rolling_return = self.returns.rolling(self.window).mean() * 252
        rolling_vol = self.rolling_volatility()
        return (rolling_return - rf_rate) / rolling_vol

    def rolling_var(self, confidence: float = 0.95) -> pd.Series:
        """ローリング過去VaR。"""
        return self.returns.rolling(self.window).apply(
            lambda x: -np.percentile(x, (1 - confidence) * 100),
            raw=True
        )

    def rolling_max_drawdown(self) -> pd.Series:
        """ローリング最大ドローダウン。"""
        def max_dd(returns):
            cumulative = (1 + returns).cumprod()
            running_max = cumulative.cummax()
            drawdowns = (cumulative - running_max) / running_max
            return drawdowns.min()

        return self.returns.rolling(self.window).apply(max_dd, raw=False)

    def rolling_beta(self, market_returns: pd.Series) -> pd.Series:
        """市場に対するローリングベータ。"""
        def calc_beta(window_data):
            port_ret = window_data.iloc[:, 0]
            mkt_ret = window_data.iloc[:, 1]
            cov = np.cov(port_ret, mkt_ret)
            return cov[0, 1] / cov[1, 1] if cov[1, 1] != 0 else 0

        combined = pd.concat([self.returns, market_returns], axis=1)
        return combined.rolling(self.window).apply(
            lambda x: calc_beta(x.to_frame()),
            raw=False
        ).iloc[:, 0]

    def volatility_regime(
        self,
        low_threshold: float = 0.10,
        high_threshold: float = 0.20
    ) -> pd.Series:
        """ボラティリティレジームを分類。"""
        vol = self.rolling_volatility()

        def classify(v):
            if v < low_threshold:
                return "low"
            elif v > high_threshold:
                return "high"
            else:
                return "normal"

        return vol.apply(classify)

パターン4: ストレステスト

class StressTester:
    """過去の危機期間と仮想的なストレステスト。"""

    # 過去の危機期間
    HISTORICAL_SCENARIOS = {
        "2008_financial_crisis": ("2008-09-01", "2009-03-31"),
        "2020_covid_crash": ("2020-02-19", "2020-03-23"),
        "2022_rate_hikes": ("2022-01-01", "2022-10-31"),
        "dot_com_bust": ("2000-03-01", "2002-10-01"),
        "flash_crash_2010": ("2010-05-06", "2010-05-06"),
    }

    def __init__(self, returns: pd.Series, weights: pd.Series = None):
        self.returns = returns
        self.weights = weights

    def historical_stress_test(
        self,
        scenario_name: str,
        historical_data: pd.DataFrame
    ) -> Dict[str, float]:
        """過去の危機期間に対してポートフォリオをテスト。"""
        if scenario_name not in self.HISTORICAL_SCENARIOS:
            raise ValueError(f"Unknown scenario: {scenario_name}")

        start, end = self.HISTORICAL_SCENARIOS[scenario_name]

        # 危機期間中のリターンを取得
        crisis_returns = historical_data.loc[start:end]

        if self.weights is not None:
            port_returns = (crisis_returns @ self.weights)
        else:
            port_returns = crisis_returns

        total_return = (1 + port_returns).prod() - 1
        max_dd = self._calculate_max_dd(port_returns)
        worst_day = port_returns.min()

        return {
            "scenario": scenario_name,
            "period": f"{start} to {end}",
            "total_return": total_return,
            "max_drawdown": max_dd,
            "worst_day": worst_day,
            "volatility": port_returns.std() * np.sqrt(252)
        }

    def hypothetical_stress_test(
        self,
        shocks: Dict[str, float]
    ) -> float:
        """
        仮想的なショックに対してポートフォリオをテスト。

        Args:
            shocks: {資産: ショックリターン} の辞書
        """
        if self.weights is None:
            raise ValueError("Weights required for hypothetical stress test")

        total_impact = 0
        for asset, shock in shocks.items():
            if asset in self.weights.index:
                total_impact += self.weights[asset] * shock

        return total_impact

    def monte_carlo_stress(
        self,
        n_simulations: int = 10000,
        horizon_days: int = 21,
        vol_multiplier: float = 2.0
    ) -> Dict[str, float]:
        """上昇したボラティリティでのモンテカルロストレステスト。"""
        mean = self.returns.mean()
        vol = self.returns.std() * vol_multiplier

        simulations = np.random.normal(
            mean,
            vol,
            (n_simulations, horizon_days)
        )

        total_returns = (1 + simulations).prod(axis=1) - 1

        return {
            "expected_loss": -total_returns.mean(),
            "var_95": -np.percentile(total_returns, 5),
            "var_99": -np.percentile(total_returns, 1),
            "worst_case": -total_returns.min(),
            "prob_10pct_loss": (total_returns < -0.10).mean()
        }

    def _calculate_max_dd(self, returns: pd.Series) -> float:
        cumulative = (1 + returns).cumprod()
        running_max = cumulative.cummax()
        drawdowns = (cumulative - running_max) / running_max
        return drawdowns.min()

クイックリファレンス

# 日次利用
metrics = RiskMetrics(returns)
print(f"Sharpe: {metrics.sharpe_ratio():.2f}")
print(f"Max DD: {metrics.max_drawdown():.2%}")
print(f"VaR 95%: {metrics.var_historical(0.95):.2%}")

# 完全なサマリー
summary = metrics.summary()
for metric, value in summary.items():
    print(f"{metric}: {value:.4f}")

ベストプラクティス

すべきこと

• 複数のメトリクスを使用する - 単一のメトリクスではすべてのリスクをキャプチャできない
• テールリスクを検討する - VaRだけでは不十分、CVaRも使用する
• ローリング分析 - リスクは時間とともに変わる
• ストレステストを実施 - 過去と仮想的なテスト
• 前提条件を記録 - 分布、ルックバック期間など

すべきではないこと

• VaR単独に依存しない - テールリスクを過小評価
• 正規分布を仮定しない - リターンは太いテール
• 相関を無視しない - ストレス期間中に増加
• 短いルックバックを使用しない - レジーム変化を見落とす
• 取引コストを忘れない - 実現リスクに影響