本项目旨在帮助学习者构建一个完整的投资组合优化与风险管理工具集，涵盖从资产配置到风险监控的全流程。学习者将在提供的框架基础上，实现各种投资组合构建方法，开发风险度量指标，并设计动态管理策略。该项目模拟实际投资管理过程中面临的挑战，帮助学习者将量化金融理论应用于实践。
完成本项目后，学习者应能够：
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import scipy.optimize as sco
class PortfolioFramework:
def __init__(self, returns_data):
"""
初始化投资组合框架
参数:
returns_data: DataFrame, 资产收益率数据
"""
self.returns = returns_data
self.mean_returns = returns_data.mean()
self.cov_matrix = returns_data.cov()
self.num_assets = len(returns_data.columns)
self.asset_names = returns_data.columns
def portfolio_performance(self, weights):
"""
计算给定权重的投资组合表现
参数:
weights: array, 资产权重
返回:
returns, volatility, sharpe_ratio
"""
returns = np.sum(self.mean_returns * weights)
volatility = np.sqrt(np.dot(weights.T, np.dot(self.cov_matrix, weights)))
sharpe_ratio = returns / volatility
return returns, volatility, sharpe_ratio
# 待实现的其他方法
def min_variance_portfolio(self):
"""最小方差投资组合实现"""
pass
def max_sharpe_portfolio(self):
"""最大夏普比率投资组合实现"""
pass
def efficient_frontier(self, points=100):
"""有效前沿计算与可视化"""
pass
class RiskMetrics:
def __init__(self, returns_data, weights=None):
"""
初始化风险指标计算器
参数:
returns_data: DataFrame, 资产收益率数据
weights: array, 可选, 资产权重
"""
self.returns = returns_data
self.weights = weights
if weights is not None:
self.portfolio_returns = np.sum(returns_data * weights, axis=1)
def historical_var(self, confidence_level=0.95):
"""
计算历史模拟法VaR
参数:
confidence_level: float, 置信水平
返回:
VaR值
"""
if self.weights is None:
raise ValueError("需要设置投资组合权重")
alpha = 1 - confidence_level
return -np.percentile(self.portfolio_returns, alpha * 100)
# 待实现的风险指标方法
def historical_cvar(self, confidence_level=0.95):
"""条件风险价值(CVaR)计算"""
pass
def parametric_var(self, confidence_level=0.95):
"""参数法VaR计算"""
pass
def stress_test(self, scenario):
"""压力测试实现"""
pass
class PortfolioOptimizer:
def __init__(self, returns_data, risk_free_rate=0.0):
"""
初始化投资组合优化器
参数:
returns_data: DataFrame, 资产收益率数据
risk_free_rate: float, 无风险利率
"""
self.returns = returns_data
self.mean_returns = returns_data.mean()
self.cov_matrix = returns_data.cov()
self.num_assets = len(returns_data.columns)
self.risk_free_rate = risk_free_rate
def optimize(self, objective_function, constraints, bounds):
"""
通用优化方法
参数:
objective_function: callable, 目标函数
constraints: list, 约束条件
bounds: tuple, 边界条件
返回:
最优权重
"""
initial_guess = np.array([1.0 / self.num_assets] * self.num_assets)
result = sco.minimize(
objective_function,
initial_guess,
method='SLSQP',
bounds=bounds,
constraints=constraints
)
return result['x']
# 待实现的特定优化目标
def min_volatility_objective(self):
"""最小波动率目标函数与约束定义"""
pass
def max_sharpe_objective(self):
"""最大夏普比率目标函数与约束定义"""
pass
def target_return_objective(self, target_return):
"""特定目标收益率下的最小风险组合"""
pass
class PortfolioVisualizer:
def __init__(self, portfolio_framework):
"""
初始化投资组合可视化工具
参数:
portfolio_framework: PortfolioFramework实例
"""
self.framework = portfolio_framework
def plot_efficient_frontier(self, min_return=None, max_return=None, points=100):
"""
绘制有效前沿
参数:
min_return: float, 可选, 最小收益率
max_return: float, 可选, 最大收益率
points: int, 计算点数
"""
# 基本绘图代码框架
plt.figure(figsize=(12, 8))
plt.title('投资组合优化: 有效前沿')
plt.xlabel('波动率 (%)')
plt.ylabel('预期收益率 (%)')
# 待实现有效前沿计算与绘制代码
def plot_asset_weights(self, weights, title='投资组合权重分配'):
"""
绘制资产权重分配
参数:
weights: array, 资产权重
title: str, 图表标题
"""
plt.figure(figsize=(12, 8))
plt.title(title)
assets = self.framework.asset_names
plt.pie(weights, labels=assets, autopct='%1.1f%%')
plt.show()
# 待实现的其他可视化方法
def plot_risk_contribution(self, weights):
"""风险贡献可视化"""
pass
def plot_performance_metrics(self, portfolio_returns):
"""绩效指标可视化"""
pass
min_variance_portfolio方法max_sharpe_portfolio方法risk_parity_portfolio方法black_litterman_portfolio方法import numpy as np
import pandas as pd
import scipy.optimize as sco
import matplotlib.pyplot as plt
class PortfolioOptimizer:
"""
投资组合优化器，实现多种投资组合构建方法
"""
def __init__(self, returns, cov_matrix=None, risk_free_rate=0.0):
"""
初始化投资组合优化器
参数:
returns: 资产收益率数据框或矩阵
cov_matrix: 协方差矩阵，如不提供则从returns计算
risk_free_rate: 无风险利率，默认为0
"""
self.returns = returns
self.mean_returns = returns.mean()
self.cov_matrix = returns.cov() if cov_matrix is None else cov_matrix
self.risk_free_rate = risk_free_rate
self.num_assets = len(self.mean_returns)
def portfolio_performance(self, weights):
"""
计算给定权重的投资组合表现
参数:
weights: 各资产权重数组
返回:
returns: 投资组合预期收益率
volatility: 投资组合波动率
sharpe_ratio: 夏普比率
"""
returns = np.sum(self.mean_returns * weights)
volatility = np.sqrt(np.dot(weights.T, np.dot(self.cov_matrix, weights)))
sharpe_ratio = (returns - self.risk_free_rate) / volatility
return returns, volatility, sharpe_ratio
def min_variance_portfolio(self, constraint_set=None):
"""
构建最小方差投资组合
参数:
constraint_set: 约束条件集合，可以是None, 'long_only', 'weight_limit'等
返回:
weights: 最优权重
returns: 投资组合预期收益率
volatility: 投资组合波动率
"""
# 实现最小方差投资组合的逻辑
# TODO: 学习者需要实现这一部分
# 提示: 使用scipy.optimize库进行优化
# 1. 定义目标函数: 最小化投资组合方差
# 2. 设置约束条件: 权重和为1，以及其他约束
# 3. 设置初始值并优化
# 4. 返回最优权重和投资组合表现
pass # 学习者移除此行并实现方法
def max_sharpe_portfolio(self, constraint_set=None):
"""
构建最大夏普比率投资组合
参数:
constraint_set: 约束条件集合，可以是None, 'long_only', 'weight_limit'等
返回:
weights: 最优权重
returns: 投资组合预期收益率
volatility: 投资组合波动率
sharpe_ratio: 夏普比率
"""
# 实现最大夏普比率投资组合的逻辑
# TODO: 学习者需要实现这一部分
# 提示: 使用scipy.optimize库进行优化
# 1. 定义目标函数: 最大化夏普比率 (或最小化负夏普比率)
# 2. 设置约束条件: 权重和为1，以及其他约束
# 3. 设置初始值并优化
# 4. 返回最优权重和投资组合表现
pass # 学习者移除此行并实现方法
def risk_parity_portfolio(self, risk_target=None, constraint_set=None):
"""
构建风险平价投资组合
参数:
risk_target: 目标风险水平，如不提供则自动确定
constraint_set: 约束条件集合
返回:
weights: 最优权重
returns: 投资组合预期收益率
volatility: 投资组合波动率
risk_contribution: 各资产的风险贡献
"""
# 实现风险平价投资组合的逻辑
# TODO: 学习者需要实现这一部分
# 提示:
# 1. 计算各资产对总风险的贡献
# 2. 设置目标函数，使各资产风险贡献尽可能相等
# 3. 设置约束条件并优化
# 4. 返回最优权重和投资组合表现
pass # 学习者移除此行并实现方法
def black_litterman_portfolio(self, market_weights, tau=0.05, views=None, view_confidences=None, constraint_set=None):
"""
使用Black-Litterman模型构建投资组合
参数:
market_weights: 市场投资组合权重
tau: 不确定性参数
views: 投资者观点，字典形式 {资产索引: 预期收益率}
view_confidences: 观点的确信度
constraint_set: 约束条件集合
返回:
weights: 最优权重
returns: 投资组合预期收益率
volatility: 投资组合波动率
"""
# 实现Black-Litterman模型的逻辑
# TODO: 学习者需要实现这一部分
# 提示:
# 1. 计算市场隐含收益率
# 2. 构建投资者观点矩阵和确信度矩阵
# 3. 结合市场隐含收益率和投资者观点，计算后验收益率
# 4. 利用后验收益率和协方差矩阵，构建最优投资组合
pass # 学习者移除此行并实现方法
def plot_efficient_frontier(self, min_return=None, max_return=None, points=100):
"""
绘制有效前沿
参数:
min_return: 最小收益率，默认为最小方差组合收益率
max_return: 最大收益率，默认为最大收益率资产的收益率
points: 有效前沿上的点数
返回:
fig: matplotlib图形对象
"""
# 计算有效前沿
if min_return is None:
min_var_weights, min_var_return, min_var_volatility = self.min_variance_portfolio()
min_return = min_var_return
if max_return is None:
max_return = np.max(self.mean_returns)
target_returns = np.linspace(min_return, max_return, points)
efficient_frontier = []
# 对每个目标收益率求解最小方差投资组合
for target_return in target_returns:
# TODO: 学习者需要实现这部分 - 求解给定目标收益率下的最小方差
# 将结果添加到efficient_frontier列表
pass
# 绘制有效前沿
# TODO: 学习者需要实现这部分 - 绘制有效前沿曲线
# 并标记最小方差点和最大夏普比率点
pass # 学习者移除此行并实现方法
# 示例使用方法
def example_usage():
"""示例使用代码，展示如何使用PortfolioOptimizer类"""
# 加载示例数据
# 这里假设我们有一个包含多只股票日收益率的CSV文件
# returns_data = pd.read_csv('stock_returns.csv', index_col=0, parse_dates=True)
# 为了测试，生成一些随机数据
np.random.seed(42)
num_assets = 5
num_observations = 1000
returns_data = pd.DataFrame(
np.random.normal(0.001, 0.02, (num_observations, num_assets)),
columns=[f'Asset_{i}' for i in range(1, num_assets + 1)]
)
# 创建投资组合优化器实例
optimizer = PortfolioOptimizer(returns_data, risk_free_rate=0.0001)
# 1. 最小方差投资组合
min_var_weights, min_var_return, min_var_volatility = optimizer.min_variance_portfolio()
print("最小方差投资组合:")
print(f"权重: {min_var_weights}")
print(f"预期收益率: {min_var_return:.4f}")
print(f"波动率: {min_var_volatility:.4f}")
# 2. 最大夏普比率投资组合
max_sharpe_weights, max_sharpe_return, max_sharpe_volatility, max_sharpe = optimizer.max_sharpe_portfolio()
print("\n最大夏普比率投资组合:")
print(f"权重: {max_sharpe_weights}")
print(f"预期收益率: {max_sharpe_return:.4f}")
print(f"波动率: {max_sharpe_volatility:.4f}")
print(f"夏普比率: {max_sharpe:.4f}")
# 3. 风险平价投资组合
rp_weights, rp_return, rp_volatility, rp_risk_contrib = optimizer.risk_parity_portfolio()
print("\n风险平价投资组合:")
print(f"权重: {rp_weights}")
print(f"预期收益率: {rp_return:.4f}")
print(f"波动率: {rp_volatility:.4f}")
print(f"风险贡献: {rp_risk_contrib}")
# 4. Black-Litterman模型
# 假设市场权重平均分配
market_weights = np.ones(num_assets) / num_assets
# 设置一些观点
views = {0: 0.003, 2: -0.001} # Asset_1预期收益率为0.3%，Asset_3预期收益率为-0.1%
view_confidences = {0: 0.6, 2: 0.7} # 对这些观点的信心
bl_weights, bl_return, bl_volatility = optimizer.black_litterman_portfolio(
market_weights, views=views, view_confidences=view_confidences
)
print("\nBlack-Litterman投资组合:")
print(f"权重: {bl_weights}")
print(f"预期收益率: {bl_return:.4f}")
print(f"波动率: {bl_volatility:.4f}")
# 绘制有效前沿
optimizer.plot_efficient_frontier()
plt.show()
if __name__ == "__main__":
example_usage()parametric_var方法historical_cvar方法stress_test方法risk_decomposition方法import numpy as np
import pandas as pd
import scipy.stats as stats
from typing import List, Dict, Tuple, Optional, Union
import matplotlib.pyplot as plt
class RiskMetrics:
"""
风险度量指标类
包含VaR、CVaR、压力测试和风险分解方法
"""
def __init__(self, returns: pd.DataFrame, weights: np.ndarray, confidence_level: float = 0.95):
"""
初始化风险度量类
参数:
returns: 资产收益率时间序列数据，DataFrame格式
weights: 资产权重，np.ndarray格式
confidence_level: 置信水平，默认为0.95
"""
self.returns = returns
self.weights = weights
self.confidence_level = confidence_level
self.portfolio_returns = self._calculate_portfolio_returns()
def _calculate_portfolio_returns(self) -> pd.Series:
"""
计算投资组合的历史收益率
返回:
投资组合历史收益率序列
"""
return self.returns.dot(self.weights)
def parametric_var(self, time_horizon: int = 1) -> float:
"""
使用参数法(正态分布假设)计算VaR
参数:
time_horizon: 风险持有期，默认为1天
返回:
在给定置信水平下的VaR值
实现要点:
1. 计算投资组合的均值和标准差
2. 根据正态分布假设，使用z统计量计算VaR
3. 根据时间跨度调整VaR
"""
# TODO: 实现参数法VaR计算
# 提示: 使用scipy.stats.norm.ppf获取正态分布的分位数
pass
def historical_var(self, time_horizon: int = 1) -> float:
"""
使用历史模拟法计算VaR
参数:
time_horizon: 风险持有期，默认为1天
返回:
在给定置信水平下的VaR值
实现要点:
1. 对历史收益率进行排序
2. 选择对应置信水平的分位数
3. 根据时间跨度调整VaR
"""
# TODO: 实现历史模拟法VaR计算
pass
def monte_carlo_var(self, num_simulations: int = 10000, time_horizon: int = 1) -> float:
"""
使用蒙特卡洛模拟法计算VaR
参数:
num_simulations: 模拟次数
time_horizon: 风险持有期，默认为1天
返回:
在给定置信水平下的VaR值
实现要点:
1. 估计收益率的均值和协方差矩阵
2. 生成多元正态分布的随机样本
3. 计算模拟的投资组合收益率
4. 确定VaR值
"""
# TODO: 实现蒙特卡洛模拟法VaR计算
pass
def compare_var_methods(self) -> Dict[str, float]:
"""
比较三种VaR计算方法的结果
返回:
包含三种方法VaR值的字典
实现要点:
1. 调用三种VaR计算方法
2. 比较结果并分析差异
3. 可视化展示三种方法的VaR
"""
# TODO: 实现VaR方法比较
pass
def historical_cvar(self, time_horizon: int = 1) -> float:
"""
使用历史模拟法计算CVaR (Conditional Value at Risk)
参数:
time_horizon: 风险持有期，默认为1天
返回:
在给定置信水平下的CVaR值
实现要点:
1. 计算VaR阈值
2. 提取低于VaR阈值的收益率
3. 计算这些收益率的平均值
"""
# TODO: 实现历史CVaR计算
pass
def parametric_cvar(self, time_horizon: int = 1) -> float:
"""
使用参数法计算CVaR
参数:
time_horizon: 风险持有期，默认为1天
返回:
在给定置信水平下的CVaR值
实现要点:
1. 使用正态分布的性质计算CVaR
2. CVaR是正态分布中，小于VaR的部分的期望值
"""
# TODO: 实现参数法CVaR计算
pass
def compare_var_cvar(self) -> Dict[str, float]:
"""
比较VaR和CVaR的结果，分析二者区别
返回:
包含VaR和CVaR值的字典
实现要点:
1. 计算VaR和CVaR
2. 分析二者的区别和适用场景
3. 可视化展示比较结果
"""
# TODO: 实现VaR和CVaR的比较分析
pass
def stress_test(self, scenarios: Dict[str, pd.DataFrame] = None) -> Dict[str, float]:
"""
执行压力测试，评估投资组合在极端情况下的潜在损失
参数:
scenarios: 情景数据字典，每个键为情景名称，值为该情景下的资产收益率
如果为None，则使用内置的历史和假设情景
返回:
每个情景下的投资组合损失
实现要点:
1. 定义历史极端事件情景(如2008金融危机、2020新冠危机等)
2. 定义假设情景(如利率上升300bp、股市下跌20%等)
3. 在每种情景下计算投资组合的预期损失
"""
# TODO: 实现压力测试框架
# 如果没有提供情景，创建默认情景
if scenarios is None:
scenarios = self._create_default_scenarios()
# 计算每个情景下的投资组合损失
pass
def _create_default_scenarios(self) -> Dict[str, pd.DataFrame]:
"""
创建默认压力测试情景
返回:
情景数据字典
实现要点:
1. 包含历史极端事件(如2008金融危机、2020疫情等)
2. 包含假设情景(如市场下跌20%、利率上升300bp等)
"""
# TODO: 创建默认压力测试情景
pass
def risk_decomposition(self, method: str = 'contribution') -> pd.Series:
"""
计算投资组合风险分解
参数:
method: 风险分解方法，可选值为'contribution'(风险贡献)或'pct_contribution'(风险贡献百分比)
返回:
各资产的风险贡献或风险贡献百分比
实现要点:
1. 计算投资组合波动率
2. 计算资产对投资组合风险的边际贡献(Marginal Contribution to Risk)
3. 计算每个资产的风险贡献(Component Contribution to Risk)
4. 可选计算风险贡献百分比
"""
# TODO: 实现风险分解
pass
def factor_risk_decomposition(self, factor_exposures: pd.DataFrame, factor_covariance: pd.DataFrame) -> pd.Series:
"""
基于因子模型的风险分解
参数:
factor_exposures: 资产对因子的暴露度，DataFrame格式
factor_covariance: 因子协方差矩阵，DataFrame格式
返回:
各因子的风险贡献
实现要点:
1. 计算投资组合对各因子的暴露
2. 计算因子对总风险的贡献
3. 分析哪些因子是主要风险来源
"""
# TODO: 实现因子风险分解
pass
def plot_var_distribution(self, num_bins: int = 50) -> None:
"""
绘制收益率分布和VaR值
参数:
num_bins: 直方图的柱数
实现要点:
1. 绘制收益率直方图
2. 在图上标记VaR位置
3. 可选添加正态分布拟合曲线
"""
# TODO: 实现VaR分布可视化
pass
def plot_risk_decomposition(self) -> None:
"""
可视化风险分解结果
实现要点:
1. 使用饼图或条形图展示风险贡献
2. 清晰标记各部分的贡献大小
"""
# TODO: 实现风险分解可视化
pass
# 使用示例
if __name__ == "__main__":
# 模拟数据
np.random.seed(42)
n_assets = 4
n_days = 1000
# 创建随机收益率数据
returns_data = pd.DataFrame(
np.random.normal(0.0005, 0.01, (n_days, n_assets)),
columns=[f'Asset_{i+1}' for i in range(n_assets)]
)
# 设置资产权重
weights = np.array([0.25, 0.25, 0.3, 0.2])
# 初始化风险度量类
risk_metrics = RiskMetrics(returns_data, weights, confidence_level=0.95)
# 测试VaR计算
print("Parametric VaR:", risk_metrics.parametric_var())
print("Historical VaR:", risk_metrics.historical_var())
print("Monte Carlo VaR:", risk_metrics.monte_carlo_var())
# 测试CVaR计算
print("Historical CVaR:", risk_metrics.historical_cvar())
# 比较VaR和CVaR
var_cvar_comparison = risk_metrics.compare_var_cvar()
print("VaR vs CVaR:", var_cvar_comparison)
# 执行压力测试
stress_results = risk_metrics.stress_test()
print("Stress Test Results:", stress_results)
# 风险分解
risk_contrib = risk_metrics.risk_decomposition()
print("Risk Contribution:", risk_contrib)
# 可视化
risk_metrics.plot_var_distribution()
risk_metrics.plot_risk_decomposition()periodic_rebalance方法threshold_rebalance方法dynamic_target_weights方法import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from enum import Enum
class RebalanceFrequency(Enum):
"""再平衡频率枚举类"""
DAILY = "D"
WEEKLY = "W"
MONTHLY = "M"
QUARTERLY = "Q"
YEARLY = "Y"
class PortfolioRebalancer:
"""投资组合再平衡策略类"""
def __init__(self, initial_weights, asset_names, price_data, trading_cost=0.001):
"""
初始化再平衡策略类
参数:
initial_weights: 初始资产权重
asset_names: 资产名称列表
price_data: 资产价格数据，DataFrame格式，索引为日期，列为资产
trading_cost: 交易成本(单边)，默认为0.1%
"""
self.initial_weights = initial_weights
self.asset_names = asset_names
self.price_data = price_data[asset_names]
self.trading_cost = trading_cost
# 计算收益率数据
self.returns_data = self.price_data.pct_change().dropna()
def periodic_rebalance(self, frequency=RebalanceFrequency.MONTHLY, target_weights=None, initial_capital=10000.0):
"""
执行周期性再平衡策略
参数:
frequency: 再平衡频率，默认为每月
target_weights: 目标权重，如果为None则使用初始权重
initial_capital: 初始资金，默认为10000
返回:
回测结果DataFrame
"""
# 设置目标权重，如果未提供则使用初始权重
if target_weights is None:
target_weights = self.initial_weights
# 获取价格数据日期范围
dates = self.price_data.index
# 初始化结果数据结构
results = pd.DataFrame(index=dates)
results['portfolio_value'] = 0.0
# 为每个资产创建权重和价值列
for asset in self.asset_names:
results[f'{asset}_weight'] = 0.0
results[f'{asset}_value'] = 0.0
# 添加交易成本列
results['trade_cost'] = 0.0
results['cum_trade_cost'] = 0.0
# 确定再平衡日期 (学生实现)
# rebalance_dates = ...
# 模拟投资组合的每日价值 (学生实现)
# current_weights = ...
# current_capital = initial_capital
# 实现每日模拟逻辑 (学生实现)
# for date in dates:
# # 检查是否是再平衡日期
# # 更新资产价值和权重
# # 更新结果数据
return results
def threshold_rebalance(self, threshold=0.05, target_weights=None, initial_capital=10000.0):
"""
执行阈值触发再平衡策略
参数:
threshold: 权重偏离阈值，当任一资产权重偏离目标权重超过此值时触发再平衡
target_weights: 目标权重，如果为None则使用初始权重
initial_capital: 初始资金，默认为10000
返回:
回测结果DataFrame
"""
# 设置目标权重，如果未提供则使用初始权重
if target_weights is None:
target_weights = self.initial_weights
# 获取价格数据日期范围
dates = self.price_data.index
# 初始化结果数据结构
results = pd.DataFrame(index=dates)
results['portfolio_value'] = 0.0
# 为每个资产创建权重和价值列
for asset in self.asset_names:
results[f'{asset}_weight'] = 0.0
results[f'{asset}_value'] = 0.0
# 添加交易成本列
results['trade_cost'] = 0.0
results['cum_trade_cost'] = 0.0
# 模拟投资组合的每日价值 (学生实现)
# current_weights = target_weights.copy()
# current_capital = initial_capital
# 实现每日模拟逻辑 (学生实现)
# for date in dates:
# # 检查当前权重是否偏离目标权重超过阈值
# # 更新资产价值和权重
# # 更新结果数据
return results
def dynamic_target_weights(self, weight_rules, condition_func, initial_capital=10000.0):
"""
执行动态目标权重调整策略
参数:
weight_rules: 不同市场状态对应的目标权重字典
condition_func: 市场状态判断函数，接收日期和价格数据，返回市场状态标识
initial_capital: 初始资金，默认为10000
返回:
回测结果DataFrame
"""
# 获取价格数据日期范围
dates = self.price_data.index
# 初始化结果数据结构
results = pd.DataFrame(index=dates)
results['portfolio_value'] = 0.0
results['market_state'] = ''
# 为每个资产创建权重和价值列
for asset in self.asset_names:
results[f'{asset}_weight'] = 0.0
results[f'{asset}_value'] = 0.0
# 添加交易成本列
results['trade_cost'] = 0.0
results['cum_trade_cost'] = 0.0
# 模拟投资组合的每日价值 (学生实现)
# current_market_state = list(weight_rules.keys())[0]
# current_weights = weight_rules[current_market_state].copy()
# current_capital = initial_capital
# 实现每日模拟逻辑 (学生实现)
# for date in dates:
# # 确定当前市场状态
# # 根据市场状态选择目标权重
# # 判断是否需要再平衡
# # 更新资产价值和权重
# # 更新结果数据
return results
def analyze_strategy(self, results):
"""
分析再平衡策略的表现
参数:
results: 回测结果
返回:
包含各项表现指标的字典
"""
# 提取投资组合价值序列
portfolio_values = results['portfolio_value']
# 计算投资组合收益率
portfolio_returns = portfolio_values.pct_change().dropna()
# 计算表现指标 (学生实现)
# total_return = ...
# annual_return = ...
# volatility = ...
# max_drawdown = ...
# 计算交易成本 (学生实现)
# total_cost = ...
# avg_cost_per_trade = ...
# 返回分析结果
metrics = {
'total_return': 0, # 学生替换为实际计算结果
'annual_return': 0,
'volatility': 0,
'max_drawdown': 0,
'sharpe_ratio': 0,
'total_cost': 0,
'avg_cost_per_trade': 0
}
return metrics
def compare_strategies(self, strategies_results):
"""
比较多种再平衡策略的表现
参数:
strategies_results: 多个策略的回测结果字典
返回:
包含各策略表现指标的DataFrame
"""
# 初始化结果DataFrame
comparison = pd.DataFrame()
# 分析每个策略并添加到比较结果中
for strategy_name, results in strategies_results.items():
metrics = self.analyze_strategy(results)
comparison[strategy_name] = pd.Series(metrics)
return comparison
def plot_results(self, strategies_results, figsize=(12, 8)):
"""
绘制策略结果对比图
参数:
strategies_results: 多个策略的回测结果字典
figsize: 图表大小
"""
plt.figure(figsize=figsize)
# 绘制每个策略的投资组合价值
for strategy_name, results in strategies_results.items():
# 标准化为初始资本=1
normalized_value = results['portfolio_value'] / results['portfolio_value'].iloc[0]
plt.plot(results.index, normalized_value, label=strategy_name)
plt.title('投资组合价值比较')
plt.xlabel('日期')
plt.ylabel('标准化价值')
plt.legend()
plt.grid(True)
plt.show()
# 示例使用
if __name__ == "__main__":
# 创建模拟数据
np.random.seed(42)
n_assets = 3
n_days = 252 * 2 # 2年交易日
# 创建日期索引
start_date = pd.Timestamp('2020-01-01')
date_range = pd.date_range(start=start_date, periods=n_days, freq='B')
# 创建随机价格数据
price_data = pd.DataFrame(index=date_range)
initial_prices = np.random.uniform(50, 200, n_assets)
asset_names = [f'资产_{i+1}' for i in range(n_assets)]
for i, asset in enumerate(asset_names):
# 模拟资产价格的随机游走
returns = np.random.normal(0.0005, 0.01, n_days)
prices = initial_prices[i] * (1 + returns).cumprod()
price_data[asset] = prices
# 设置初始资产权重
initial_weights = np.array([0.33, 0.33, 0.34])
# 初始化再平衡策略类
rebalancer = PortfolioRebalancer(
initial_weights=initial_weights,
asset_names=asset_names,
price_data=price_data,
trading_cost=0.001
)
# 测试周期性再平衡策略
monthly_results = rebalancer.periodic_rebalance(
frequency=RebalanceFrequency.MONTHLY,
initial_capital=10000.0
)
# 测试阈值触发再平衡策略
threshold_results = rebalancer.threshold_rebalance(
threshold=0.05,
initial_capital=10000.0
)
# 定义简单的市场状态判断函数 - 基于30日移动平均线
def market_condition(date, price_data):
if date < price_data.index[30]: # 前30天没有足够的数据
return 'neutral'
asset1_price = price_data.loc[date, '资产_1']
asset1_ma30 = price_data['资产_1'].loc[:date].tail(30).mean()
if asset1_price > asset1_ma30:
return 'bullish' # 牛市
else:
return 'bearish' # 熊市
# 定义不同市场状态的权重
weight_rules = {
'bullish': np.array([0.4, 0.4, 0.2]), # 牛市偏向风险资产
'bearish': np.array([0.2, 0.2, 0.6]), # 熊市偏向保守资产
'neutral': np.array([0.33, 0.33, 0.34]) # 中性市场平均分配
}
# 测试动态目标权重调整策略
dynamic_results = rebalancer.dynamic_target_weights(
weight_rules=weight_rules,
condition_func=market_condition,
initial_capital=10000.0
)
# 比较策略表现
strategies = {
"月度再平衡": monthly_results,
"阈值再平衡": threshold_results,
"动态权重": dynamic_results
}
comparison = rebalancer.compare_strategies(strategies)
print(comparison)
# 绘制策略结果对比
rebalancer.plot_results(strategies)risk_monitor类extreme_event_detection方法risk_trend_analysis方法import numpy as np
import pandas as pd
from datetime import datetime
import matplotlib.pyplot as plt
from typing import Dict, List, Tuple, Optional, Union, Callable
class RiskMonitor:
"""风险监控类，用于监控和检测金融风险指标"""
def __init__(self, thresholds: Dict[str, Dict[str, float]]):
"""
初始化风险监控器
参数:
thresholds: 风险指标阈值字典，格式为 {'指标名': {'upper': 上限阈值, 'lower': 下限阈值}}
"""
# TODO: 初始化风险监控器，存储阈值设置和其他必要属性
self.thresholds = thresholds
self.alerts_history = []
# 添加其他必要的属性初始化
def check_threshold_breach(self, indicator_name: str, value: float) -> Optional[Dict]:
"""
检查指标值是否突破阈值
参数:
indicator_name: 指标名称
value: 当前指标值
返回:
如果突破阈值，返回警报信息字典；否则返回None
"""
# TODO: 实现阈值突破检测逻辑
pass
def trigger_alert(self, alert_info: Dict) -> None:
"""
触发警报机制
参数:
alert_info: 警报信息字典
"""
# TODO: 实现警报机制，可以是日志记录、消息通知等
pass
def monitor_indicators(self, current_data: Dict[str, float]) -> List[Dict]:
"""
监控多个风险指标并触发警报
参数:
current_data: 当前各指标数据，格式为 {'指标名': 指标值}
返回:
触发的警报列表
"""
# TODO: 实现多指标监控逻辑
pass
def extreme_event_detection(self, time_series_data: pd.DataFrame,
window_size: int = 20,
std_threshold: float = 3.0) -> List[Dict]:
"""
检测极端事件
参数:
time_series_data: 时间序列数据，DataFrame格式，包含日期和各指标列
window_size: 滑动窗口大小，用于计算统计异常
std_threshold: 标准差倍数阈值，超过均值多少倍标准差视为异常
返回:
检测到的极端事件列表
"""
# TODO: 实现基于统计方法的极端事件检测算法
pass
def risk_mitigation_plan(self, event_type: str, severity: float) -> Dict:
"""
根据极端事件类型和严重程度，生成风险缓解措施
参数:
event_type: 极端事件类型
severity: 事件严重程度
返回:
风险缓解措施方案
"""
# TODO: 实现风险缓解措施生成逻辑
pass
def risk_trend_analysis(self, time_series_data: pd.DataFrame,
indicator: str,
window_size: int = 30) -> Dict:
"""
分析风险指标的时间序列趋势
参数:
time_series_data: 时间序列数据，DataFrame格式，包含日期和各指标列
indicator: 要分析的指标名称
window_size: 分析窗口大小
返回:
趋势分析结果
"""
# TODO: 实现风险指标时间序列趋势分析
pass
def detect_trend_change(self, time_series: pd.Series,
change_threshold: float = 0.1) -> List[Dict]:
"""
检测趋势突变点
参数:
time_series: 单个指标的时间序列
change_threshold: 变化阈值，超过此阈值认为是趋势突变
返回:
检测到的趋势突变点列表
"""
# TODO: 实现趋势突变检测算法
pass
def visualize_risk_indicators(self, time_series_data: pd.DataFrame,
indicators: List[str],
alerts: List[Dict] = None) -> None:
"""
可视化风险指标和警报
参数:
time_series_data: 时间序列数据
indicators: 要可视化的指标列表
alerts: 要在图上标记的警报列表
"""
# TODO: 实现风险指标可视化功能
pass
# 使用示例
if __name__ == "__main__":
# 示例阈值设置
example_thresholds = {
'volatility': {'upper': 0.03, 'lower': None},
'drawdown': {'upper': 0.05, 'lower': None},
'var': {'upper': 0.02, 'lower': None},
'liquidity_ratio': {'upper': None, 'lower': 1.5}
}
# 创建风险监控器实例
risk_monitor = RiskMonitor(example_thresholds)
# TODO: 添加测试数据和调用各方法的示例代码
# 示例：监控当前风险指标
current_indicators = {
'volatility': 0.035, # 超过阈值
'drawdown': 0.04, # 未超过阈值
'var': 0.025, # 超过阈值
'liquidity_ratio': 1.7 # 未超过阈值
}
alerts = risk_monitor.monitor_indicators(current_indicators)
print(f"监测到 {len(alerts)} 个风险警报")
# 添加其他测试代码...historical_scenario方法monte_carlo_simulation方法sensitivity_analysis方法import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import stats
from typing import Dict, List, Tuple, Optional, Union, Callable
import datetime as dt
from dataclasses import dataclass
@dataclass
class Portfolio:
"""投资组合数据结构"""
assets: Dict[str, float] # 资产及其权重
positions: Dict[str, float] # 资产及其持仓量
historical_returns: pd.DataFrame # 历史收益率数据
class ScenarioAnalysis:
"""情景分析工具类，用于评估投资组合在不同情景下的表现"""
def __init__(self, portfolio: Portfolio):
"""
初始化情景分析工具
参数:
portfolio: 待分析的投资组合
"""
self.portfolio = portfolio
self.historical_data = None # 用于存储历史情景数据
self.simulation_results = {} # 用于存储模拟结果
def load_historical_data(self, data_source: Union[str, pd.DataFrame]) -> None:
"""
加载历史市场数据
参数:
data_source: 数据源，可以是文件路径或DataFrame
"""
# TODO: 实现历史数据加载逻辑
pass
def define_historical_scenarios(self) -> Dict[str, Dict]:
"""
定义典型历史市场事件
返回:
历史情景定义字典，格式为 {'情景名': {'start_date': 开始日期, 'end_date': 结束日期, 'description': 描述}}
"""
# TODO: 定义典型历史市场事件
# 示例：金融危机、科技泡沫、疫情冲击等
scenarios = {
'financial_crisis_2008': {
'start_date': dt.datetime(2008, 9, 1),
'end_date': dt.datetime(2009, 3, 31),
'description': '2008年金融危机期间市场剧烈下跌'
},
'covid_crash_2020': {
'start_date': dt.datetime(2020, 2, 19),
'end_date': dt.datetime(2020, 3, 23),
'description': '2020年新冠疫情导致的市场崩盘'
}
# TODO: 添加更多历史情景
}
return scenarios
def historical_scenario(self, scenario_name: str) -> Dict:
"""
在指定的历史情景下分析投资组合表现
参数:
scenario_name: 情景名称，必须是define_historical_scenarios()中定义的情景之一
返回:
分析结果字典，包含在该情景下的关键指标
"""
# TODO: 实现历史情景回放分析
pass
def calculate_scenario_metrics(self, returns: pd.Series) -> Dict:
"""
计算情景分析的关键指标
参数:
returns: 情景期间的投资组合收益率序列
返回:
指标字典，包含关键风险收益指标
"""
# TODO: 实现情景分析指标计算
pass
def monte_carlo_simulation(self,
model_type: str = 'gbm',
num_simulations: int = 1000,
time_horizon: int = 252,
confidence_level: float = 0.95) -> Dict:
"""
执行蒙特卡洛模拟分析
参数:
model_type: 随机过程模型类型 ('gbm', 'jump_diffusion', 'regime_switching')
num_simulations: 模拟次数
time_horizon: 模拟时间长度（交易日）
confidence_level: 置信水平
返回:
模拟结果字典
"""
# TODO: 实现蒙特卡洛模拟
pass
def geometric_brownian_motion(self,
initial_price: float,
mu: float,
sigma: float,
time_horizon: int,
num_simulations: int) -> np.ndarray:
"""
几何布朗运动模型
参数:
initial_price: 初始价格
mu: 预期收益率（年化）
sigma: 波动率（年化）
time_horizon: 模拟时间步数
num_simulations: 模拟路径数量
返回:
模拟路径数组，形状为(num_simulations, time_horizon)
"""
# TODO: 实现几何布朗运动模型
pass
def jump_diffusion_model(self,
initial_price: float,
mu: float,
sigma: float,
jump_intensity: float,
jump_mean: float,
jump_std: float,
time_horizon: int,
num_simulations: int) -> np.ndarray:
"""
跳跃扩散模型
参数:
initial_price: 初始价格
mu: 预期收益率（年化）
sigma: 波动率（年化）
jump_intensity: 跳跃强度（年化）
jump_mean: 跳跃幅度均值
jump_std: 跳跃幅度标准差
time_horizon: 模拟时间步数
num_simulations: 模拟路径数量
返回:
模拟路径数组，形状为(num_simulations, time_horizon)
"""
# TODO: 实现跳跃扩散模型
pass
def regime_switching_model(self,
initial_price: float,
mu_regimes: List[float],
sigma_regimes: List[float],
transition_matrix: np.ndarray,
time_horizon: int,
num_simulations: int) -> np.ndarray:
"""
区制转换模型
参数:
initial_price: 初始价格
mu_regimes: 各区制下的预期收益率列表
sigma_regimes: 各区制下的波动率列表
transition_matrix: 区制转换概率矩阵
time_horizon: 模拟时间步数
num_simulations: 模拟路径数量
返回:
模拟路径数组，形状为(num_simulations, time_horizon)
"""
# TODO: 实现区制转换模型
pass
def analyze_simulation_results(self,
simulation_paths: np.ndarray,
initial_value: float,
confidence_level: float = 0.95) -> Dict:
"""
分析蒙特卡洛模拟结果
参数:
simulation_paths: 模拟路径数组
initial_value: 初始投资组合价值
confidence_level: 置信水平
返回:
分析结果字典
"""
# TODO: 实现模拟结果分析
pass
def visualize_simulation(self,
simulation_paths: np.ndarray,
title: str = "Monte Carlo Simulation",
num_paths_to_plot: int = 100) -> None:
"""
可视化蒙特卡洛模拟结果
参数:
simulation_paths: 模拟路径数组
title: 图表标题
num_paths_to_plot: 要绘制的路径数量
"""
# TODO: 实现模拟结果可视化
pass
def sensitivity_analysis(self,
parameters: Dict[str, List[float]],
target_metric: str = 'expected_return') -> pd.DataFrame:
"""
执行敏感性分析
参数:
parameters: 要分析的参数及其范围，格式为 {'参数名': [参数值列表]}
target_metric: 目标指标名称
返回:
敏感性分析结果DataFrame
"""
# TODO: 实现敏感性分析
pass
def one_way_sensitivity(self,
parameter: str,
values: List[float],
target_metric: str) -> pd.Series:
"""
单因素敏感性分析
参数:
parameter: 参数名称
values: 参数值列表
target_metric: 目标指标名称
返回:
结果Series，索引为参数值，值为目标指标
"""
# TODO: 实现单因素敏感性分析
pass
def identify_vulnerabilities(self, sensitivity_results: pd.DataFrame) -> Dict:
"""
基于敏感性分析结果识别投资组合脆弱点
参数:
sensitivity_results: 敏感性分析结果
返回:
脆弱点分析结果字典
"""
# TODO: 实现脆弱点识别逻辑
pass
def visualize_sensitivity(self, sensitivity_results: pd.DataFrame, title: str = "Sensitivity Analysis") -> None:
"""
可视化敏感性分析结果
参数:
sensitivity_results: 敏感性分析结果
title: 图表标题
"""
# TODO: 实现敏感性分析可视化
pass
# 使用示例
if __name__ == "__main__":
# 创建示例投资组合
example_portfolio = Portfolio(
assets={'AAPL': 0.3, 'MSFT': 0.3, 'GOOG': 0.2, 'AMZN': 0.2},
positions={'AAPL': 100, 'MSFT': 200, 'GOOG': 50, 'AMZN': 40},
historical_returns=pd.DataFrame() # 这里应该加载实际数据
)
# 创建情景分析工具实例
scenario_tool = ScenarioAnalysis(example_portfolio)
# TODO: 添加测试代码，展示各种功能的使用
# 示例：历史情景分析
# financial_crisis_results = scenario_tool.historical_scenario('financial_crisis_2008')
# print("2008金融危机情景下的表现:", financial_crisis_results)
# 示例：蒙特卡洛模拟
# mc_results = scenario_tool.monte_carlo_simulation(model_type='gbm', num_simulations=1000)
# print("蒙特卡洛模拟结果:", mc_results)
# 示例：敏感性分析
# sensitivity_params = {
# 'volatility': [0.1, 0.15, 0.2, 0.25, 0.3],
# 'correlation': [-0.2, 0, 0.2, 0.4, 0.6]
# }
# sensitivity_results = scenario_tool.sensitivity_analysis(sensitivity_params)
# print("敏感性分析结果:\n", sensitivity_results)完成项目后，请提交：
为便于项目实施，可使用以下数据：
完成基本要求后，可考虑以下扩展方向：
现代投资组合理论指南
优化算法比较文档
风险管理最佳实践