量化风险管理实战：VaR与CVaR计算及压力测试框架

import numpy as np
import pandas as pd

def historical_var(returns, confidence=0.95):
    """
    历史模拟法计算VaR
    returns: 收益率序列
    confidence: 置信水平 (默认95%)
    """
    var = np.percentile(returns, (1 - confidence) * 100)
    return var

# 示例
returns = pd.read_csv('portfolio_returns.csv')['return']
var_95 = historical_var(returns, 0.95)
var_99 = historical_var(returns, 0.99)

print(f"95% VaR: {var_95:.2%}")
print(f"99% VaR: {var_99:.2%}")

def monte_carlo_var(returns, confidence=0.95, num_simulations=10000, horizon=1):
    """
    蒙特卡洛模拟法计算VaR
    """
    mu = returns.mean()
    sigma = returns.std()
    
    # 模拟未来收益率
    simulated_returns = np.random.normal(mu, sigma, (num_simulations, horizon))
    cumulative_returns = np.sum(simulated_returns, axis=1)
    
    # 计算VaR
    var = np.percentile(cumulative_returns, (1 - confidence) * 100)
    return var

var_mc = monte_carlo_var(returns, 0.95, num_simulations=10000)
print(f"蒙特卡洛95% VaR: {var_mc:.2%}")

def portfolio_var(returns_df, weights, confidence=0.95):
    """
    计算组合VaR
    returns_df: 各资产收益率DataFrame
    weights: 权重向量
    """
    # 计算组合收益率
    portfolio_returns = (returns_df * weights).sum(axis=1)
    
    # 计算组合VaR
    var = np.percentile(portfolio_returns, (1 - confidence) * 100)
    return var

# 示例：3只股票的组合
weights = np.array([0.4, 0.3, 0.3])
portfolio_var_value = portfolio_var(returns_df, weights, 0.95)

def risk_budget_allocation(returns_df, target_risk_contribution):
    """
    风险预算配置
    target_risk_contribution: 目标风险贡献比例
    """
    # 使用优化算法求解权重
    # 使各资产的风险贡献等于目标比例
    from scipy.optimize import minimize
    
    def objective(weights):
        portfolio_var = np.percentile(
            (returns_df * weights).sum(axis=1), 5
        )
        risk_contribution = calculate_risk_contribution(returns_df, weights)
        return np.sum((risk_contribution - target_risk_contribution)**2)
    
    # 约束：权重之和为1
    constraints = ({'type': 'eq', 'fun': lambda w: np.sum(w) - 1})
    bounds = tuple((0, 1) for _ in range(returns_df.shape[1]))
    
    result = minimize(objective, x0=np.ones(returns_df.shape[1]) / returns_df.shape[1],
                     method='SLSQP', bounds=bounds, constraints=constraints)
    
    return result.x

import numpy as np
import pandas as pd
from scipy import stats
import matplotlib.pyplot as plt

class RiskManager:
    """
    量化风险管理系统
    """
    def __init__(self, returns_df, confidence=0.95):
        self.returns = returns_df
        self.confidence = confidence
    
    def calculate_var(self, method='historical', weights=None):
        """
        计算VaR
        method: 'historical', 'parametric', 'monte_carlo'
        """
        if weights is not None:
            portfolio_returns = (self.returns * weights).sum(axis=1)
        else:
            portfolio_returns = self.returns
        
        if method == 'historical':
            var = np.percentile(portfolio_returns, (1 - self.confidence) * 100)
        
        elif method == 'parametric':
            mu = portfolio_returns.mean()
            sigma = portfolio_returns.std()
            z_score = stats.norm.ppf(1 - self.confidence)
            var = mu - z_score * sigma
        
        elif method == 'monte_carlo':
            var = self._monte_carlo_var(portfolio_returns)
        
        return var
    
    def calculate_cvar(self, method='historical', weights=None):
        """
        计算CVaR
        """
        if weights is not None:
            portfolio_returns = (self.returns * weights).sum(axis=1)
        else:
            portfolio_returns = self.returns
        
        var = self.calculate_var(method, weights)
        cvar = portfolio_returns[portfolio_returns <= var].mean()
        
        return cvar
    
    def _monte_carlo_var(self, returns, num_simulations=10000):
        """
        蒙特卡洛模拟VaR
        """
        mu = returns.mean()
        sigma = returns.std()
        simulated_returns = np.random.normal(mu, sigma, num_simulations)
        var = np.percentile(simulated_returns, (1 - self.confidence) * 100)
        return var
    
    def stress_test(self, scenario_shocks, weights=None):
        """
        压力测试
        scenario_shocks: 各资产的冲击幅度
        """
        if weights is None:
            weights = np.ones(self.returns.shape[1]) / self.returns.shape[1]
        
        stress_loss = (weights * scenario_shocks).sum()
        return stress_loss
    
    def plot_risk_dashboard(self, weights=None):
        """
        绘制风险仪表盘
        """
        if weights is not None:
            portfolio_returns = (self.returns * weights).sum(axis=1)
        else:
            portfolio_returns = self.returns
        
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        
        # 1. 收益率分布直方图
        axes[0, 0].hist(portfolio_returns, bins=50, edgecolor='black', alpha=0.7)
        axes[0, 0].axvline(self.calculate_var('historical'), color='red', 
                          linestyle='--', label=f'VaR {self.confidence*100}%')
        axes[0, 0].set_title('Returns Distribution with VaR')
        axes[0, 0].legend()
        
        # 2. VaR时间序列
        rolling_var = portfolio_returns.rolling(window=20).apply(
            lambda x: np.percentile(x, (1 - self.confidence) * 100)
        )
        axes[0, 1].plot(rolling_var, color='red', label='Rolling VaR')
        axes[0, 1].set_title('Rolling VaR (20-day)')
        axes[0, 1].legend()
        
        # 3. 回撤曲线
        cumulative_returns = (1 + portfolio_returns).cumprod()
        running_max = cumulative_returns.cummax()
        drawdown = (cumulative_returns - running_max) / running_max
        axes[1, 0].fill_between(drawdown.index, drawdown, 0, color='red', alpha=0.3)
        axes[1, 0].set_title('Drawdown Curve')
        
        # 4. QQ图（检验正态分布假设）
        stats.probplot(portfolio_returns, dist="norm", plot=axes[1, 1])
        axes[1, 1].set_title('Q-Q Plot (Normality Test)')
        
        plt.tight_layout()
        plt.savefig('risk_dashboard.png', dpi=150, bbox_inches='tight')
        plt.close()

# 使用示例
returns_df = pd.read_csv('portfolio_returns.csv', index_col=0, parse_dates=True)
risk_mgr = RiskManager(returns_df, confidence=0.95)

# 计算VaR和CVaR
var_historical = risk_mgr.calculate_var(method='historical')
cvar_historical = risk_mgr.calculate_cvar(method='historical')

print(f"Historical VaR (95%): {var_historical:.2%}")
print(f"Historical CVaR (95%): {cvar_historical:.2%}")

# 压力测试
scenario_shocks = returns_df.quantile(0.01)  # 使用1%分位数作为压力场景
stress_loss = risk_mgr.stress_test(scenario_shocks)
print(f"Stress Test Loss: {stress_loss:.2%}")

# 绘制风险仪表盘
risk_mgr.plot_risk_dashboard()