投资组合优化实战:用Python实现智能资产配置

import numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport seaborn as snsfrom scipy.optimize import minimizeimport tushare as tsfrom datetime import datetime, timedelta# 设置样式plt.style.use('seaborn-v0_8-darkgrid')plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = False# 初始化tusharets.set_token('你的token')pro = ts.pro_api()# 获取多只股票数据（代表不同行业）def get_portfolio_data(tickers, start_date='20230101', end_date='20240601'):    """    获取投资组合数据    """    portfolio_data = {}    for ticker in tickers:        df = pro.daily(ts_code=ticker, start_date=start_date, end_date=end_date)        df['trade_date'] = pd.to_datetime(df['trade_date'])        df = df.sort_values('trade_date')        df.set_index('trade_date', inplace=True)        portfolio_data[ticker] = df['close']    # 合并数据    portfolio_df = pd.DataFrame(portfolio_data)    # 对齐数据（确保所有股票有相同的交易日期）    portfolio_df = portfolio_df.dropna()    return portfolio_df# 选择6只不同行业的代表性股票tickers = [    '600519.SH',  # 贵州茅台 - 消费    '601318.SH',  # 中国平安 - 金融    '300750.SZ',  # 宁德时代 - 新能源    '600036.SH',  # 招商银行 - 银行    '000858.SZ',  # 五粮液 - 消费    '601888.SH',  # 中国中免 - 消费]portfolio_prices = get_portfolio_data(tickers)print(f"投资组合包含{len(tickers)}只股票")print(f"数据时间范围: {portfolio_prices.index[0]} 到 {portfolio_prices.index[-1]}")print(f"数据维度: {portfolio_prices.shape}")

def calculate_portfolio_metrics(prices_df):    """    计算投资组合基础指标    """    # 计算日收益率    returns = prices_df.pct_change().dropna()    # 计算年化收益率（假设252个交易日）    annual_returns = (1 + returns.mean()) ** 252 - 1    # 计算年化波动率    annual_volatility = returns.std() * np.sqrt(252)    # 计算协方差矩阵    cov_matrix = returns.cov() * 252    # 计算相关系数矩阵    corr_matrix = returns.corr()    metrics = {        'returns': returns,        'annual_returns': annual_returns,        'annual_volatility': annual_volatility,        'cov_matrix': cov_matrix,        'corr_matrix': corr_matrix    }    return metrics# 计算投资组合指标metrics = calculate_portfolio_metrics(portfolio_prices)# 可视化相关系数矩阵plt.figure(figsize=(10, 8))sns.heatmap(metrics['corr_matrix'], annot=True, cmap='coolwarm',             center=0, fmt='.2f', square=True)plt.title('股票收益率相关系数矩阵', fontsize=15, pad=20)plt.tight_layout()plt.show()# 打印各股票表现print("各股票年化表现:")for ticker in tickers:    name = ticker[:6]    print(f"{name}: 年化收益={metrics['annual_returns'][ticker]:.2%}, "          f"年化波动={metrics['annual_volatility'][ticker]:.2%}")

def simulate_random_portfolios(num_portfolios=10000):    """    模拟随机投资组合    """    returns = metrics['returns']    cov_matrix = metrics['cov_matrix']    results = np.zeros((3, num_portfolios))    weights_record = []    for i in range(num_portfolios):        # 生成随机权重        weights = np.random.random(len(tickers))        weights /= np.sum(weights)        weights_record.append(weights)        # 计算组合收益        portfolio_return = np.sum(weights * metrics['annual_returns'])        # 计算组合波动率        portfolio_volatility = np.sqrt(np.dot(weights.T,                                              np.dot(cov_matrix, weights)))        # 计算夏普比率（假设无风险利率3%）        sharpe_ratio = (portfolio_return - 0.03) / portfolio_volatility        results[0, i] = portfolio_return        results[1, i] = portfolio_volatility        results[2, i] = sharpe_ratio    return results, weights_record# 模拟10000个随机组合results, weights_record = simulate_random_portfolios(10000)# 找到最优组合max_sharpe_idx = np.argmax(results[2])min_vol_idx = np.argmin(results[1])print(f"\n最优夏普比率组合:")print(f"年化收益: {results[0, max_sharpe_idx]:.2%}")print(f"年化波动: {results[1, max_sharpe_idx]:.2%}")print(f"夏普比率: {results[2, max_sharpe_idx]:.2f}")print(f"\n最小波动组合:")print(f"年化收益: {results[0, min_vol_idx]:.2%}")print(f"年化波动: {results[1, min_vol_idx]:.2%}")print(f"夏普比率: {results[2, min_vol_idx]:.2f}")# 可视化有效前沿plt.figure(figsize=(12, 8))plt.scatter(results[1], results[0], c=results[2],            cmap='viridis', alpha=0.6, s=10)plt.colorbar(label='夏普比率')plt.scatter(results[1, max_sharpe_idx], results[0, max_sharpe_idx],            c='red', s=200, marker='*', label='最大夏普比率组合')plt.scatter(results[1, min_vol_idx], results[0, min_vol_idx],            c='blue', s=200, marker='*', label='最小波动组合')plt.xlabel('年化波动率', fontsize=12)plt.ylabel('年化收益率', fontsize=12)plt.title('投资组合有效前沿', fontsize=15, pad=20)plt.legend()plt.grid(True, alpha=0.3)plt.tight_layout()plt.show()

def optimize_portfolio_sharpe(metrics, risk_free_rate=0.03):    """    优化投资组合以最大化夏普比率    """    returns = metrics['returns']    cov_matrix = metrics['cov_matrix']    n_assets = len(tickers)    # 约束条件：权重和为1    constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})    # 边界条件：允许做空（权重可以为负），但限制在[-0.5, 0.5]    bounds = tuple((-0.5, 0.5) for _ in range(n_assets))    # 初始权重（等权重）    init_weights = n_assets * [1.0 / n_assets]    # 目标函数：最小化负夏普比率    def negative_sharpe(weights):        portfolio_return = np.sum(weights * metrics['annual_returns'])        portfolio_volatility = np.sqrt(np.dot(weights.T,                                              np.dot(cov_matrix, weights)))        sharpe = (portfolio_return - risk_free_rate) / portfolio_volatility        return -sharpe    # 优化    optimized = minimize(negative_sharpe, init_weights,                        method='SLSQP', bounds=bounds,                        constraints=constraints)    return optimized.x# 优化权重optimal_weights = optimize_portfolio_sharpe(metrics)print("\n最优权重分配:")for ticker, weight in zip(tickers, optimal_weights):    name = ticker[:6]    print(f"{name}: {weight:.2%}")# 计算最优组合表现optimal_return = np.sum(optimal_weights * metrics['annual_returns'])optimal_volatility = np.sqrt(np.dot(optimal_weights.T,                                    np.dot(metrics['cov_matrix'], optimal_weights)))optimal_sharpe = (optimal_return - 0.03) / optimal_volatilityprint(f"\n最优组合表现:")print(f"年化收益: {optimal_return:.2%}")print(f"年化波动: {optimal_volatility:.2%}")print(f"夏普比率: {optimal_sharpe:.2f}")# 可视化权重分配plt.figure(figsize=(10, 6))colors = plt.cm.Set3(np.linspace(0, 1, len(tickers)))plt.pie(optimal_weights, labels=[t[:6] for t in tickers],         autopct='%1.1f%%', colors=colors, startangle=90)plt.title('最优投资组合权重分配', fontsize=15, pad=20)plt.tight_layout()plt.show()

def risk_parity_portfolio(cov_matrix, max_iter=1000, tolerance=1e-8):    """    风险平价模型：每个资产对组合风险的贡献相等    """    n = cov_matrix.shape[0]    weights = np.ones(n) / n  # 初始等权重    def risk_contribution(weights):        portfolio_volatility = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))        marginal_risk = np.dot(cov_matrix, weights) / portfolio_volatility        risk_contrib = weights * marginal_risk        return risk_contrib    def objective(weights):        rc = risk_contribution(weights)        # 目标：使风险贡献尽可能相等        return np.sum((rc - rc.mean()) ** 2)    # 约束条件    constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})    bounds = tuple((0, 1) for _ in range(n))  # 不允许做空    # 优化    result = minimize(objective, weights,                      method='SLSQP', bounds=bounds,                     constraints=constraints,                     options={'maxiter': max_iter})    return result.x# 计算风险平价权重rp_weights = risk_parity_portfolio(metrics['cov_matrix'])print("\n风险平价权重分配:")for ticker, weight in zip(tickers, rp_weights):    name = ticker[:6]    print(f"{name}: {weight:.2%}")# 计算风险贡献def calculate_risk_contribution(weights, cov_matrix):    """计算各资产的风险贡献"""    portfolio_volatility = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))    marginal_risk = np.dot(cov_matrix, weights) / portfolio_volatility    risk_contrib = weights * marginal_risk    risk_contrib_pct = risk_contrib / portfolio_volatility    return risk_contrib_pct# 计算并比较风险贡献print("\n风险贡献分析:")rc = calculate_risk_contribution(rp_weights, metrics['cov_matrix'])for ticker, weight, contrib in zip(tickers, rp_weights, rc):    name = ticker[:6]    print(f"{name}: 权重={weight:.2%}, 风险贡献={contrib:.2%}")# 可视化风险贡献fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 6))# 权重对比x = range(len(tickers))width = 0.35ax1.bar([i - width/2 for i in x], optimal_weights, width,         label='最大夏普', alpha=0.8)ax1.bar([i + width/2 for i in x], rp_weights, width,         label='风险平价', alpha=0.8)ax1.set_xticks(x)ax1.set_xticklabels([t[:6] for t in tickers], rotation=45)ax1.set_ylabel('权重')ax1.set_title('权重分配对比')ax1.legend()ax1.grid(True, alpha=0.3)# 风险贡献ax2.bar([t[:6] for t in tickers], rc * 100)ax2.set_ylabel('风险贡献 (%)')ax2.set_title('风险平价模型 - 各资产风险贡献')ax2.grid(True, alpha=0.3)plt.tight_layout()plt.show()

def dynamic_volatility_adjustment(prices_df, lookback=60, target_volatility=0.15):    """    基于波动率的动态资产配置    """    returns = prices_df.pct_change().dropna()    n_assets = len(tickers)    # 初始化    weights_history = []    portfolio_values = [100000]  # 初始资金10万    for i in range(lookback, len(returns)):        # 计算过去lookback天的波动率        recent_returns = returns.iloc[i-lookback:i]        recent_cov = recent_returns.cov() * 252        # 优化权重（最小化波动率）        def portfolio_volatility(weights):            return np.sqrt(np.dot(weights.T, np.dot(recent_cov, weights)))        constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})        bounds = tuple((0, 1) for _ in range(n_assets))        init_weights = n_assets * [1.0 / n_assets]        # 优化：在波动率不超过目标的前提下最大化收益        def objective(weights):            return -np.sum(weights * recent_returns.mean() * 252)        # 添加波动率约束        vol_constraint = {'type': 'ineq',                          'fun': lambda x: target_volatility - portfolio_volatility(x)}        try:            result = minimize(objective, init_weights,                            method='SLSQP', bounds=bounds,                            constraints=[constraints, vol_constraint])            current_weights = result.x        except:            # 如果优化失败，使用等权重            current_weights = init_weights        weights_history.append(current_weights)        # 计算当日组合收益        daily_return = np.sum(current_weights * returns.iloc[i])        portfolio_values.append(portfolio_values[-1] * (1 + daily_return))    return weights_history, portfolio_values# 运行动态资产配置weights_history, portfolio_values = dynamic_volatility_adjustment(portfolio_prices)# 分析结果dynamic_portfolio_returns = pd.Series(portfolio_values).pct_change().dropna()annual_return_dynamic = (1 + dynamic_portfolio_returns.mean()) ** 252 - 1annual_volatility_dynamic = dynamic_portfolio_returns.std() * np.sqrt(252)sharpe_dynamic = (annual_return_dynamic - 0.03) / annual_volatility_dynamicprint(f"\n动态资产配置表现:")print(f"年化收益: {annual_return_dynamic:.2%}")print(f"年化波动: {annual_volatility_dynamic:.2%}")print(f"夏普比率: {sharpe_dynamic:.2f}")# 可视化权重变化weights_df = pd.DataFrame(weights_history, columns=[t[:6] for t in tickers])weights_df.index = portfolio_prices.index[60:]  # 从第61天开始plt.figure(figsize=(14, 8))for column in weights_df.columns:    plt.plot(weights_df.index, weights_df[column], label=column, linewidth=2)plt.xlabel('日期')plt.ylabel('权重')plt.title('动态资产配置 - 权重变化', fontsize=15, pad=20)plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left')plt.grid(True, alpha=0.3)plt.tight_layout()plt.show()

def compare_with_benchmarks(portfolio_values, tickers):    """    与基准对比    """    # 计算等权重组合    returns = portfolio_prices.pct_change().dropna()    equal_weights = np.ones(len(tickers)) / len(tickers)    equal_portfolio_returns = returns.dot(equal_weights)    equal_portfolio_value = 100000 * (1 + equal_portfolio_returns).cumprod()    # 计算沪深300表现（作为市场基准）    # 这里简化处理，使用等权重组合的85% + 现金15%作为基准    benchmark_returns = equal_portfolio_returns * 0.85 + 0.03/252 * 0.15    benchmark_value = 100000 * (1 + benchmark_returns).cumprod()    # 创建对比DataFrame    comparison = pd.DataFrame({        '动态配置': portfolio_values,        '等权重': equal_portfolio_value.values[:len(portfolio_values)],        '市场基准': benchmark_value.values[:len(portfolio_values)]    }, index=portfolio_prices.index[60:60+len(portfolio_values)])    # 计算绩效指标    performance_metrics = pd.DataFrame()    for strategy in comparison.columns:        strategy_returns = comparison[strategy].pct_change().dropna()        ann_return = (1 + strategy_returns.mean()) ** 252 - 1        ann_vol = strategy_returns.std() * np.sqrt(252)        sharpe = (ann_return - 0.03) / ann_vol        # 最大回撤        cumulative = (1 + strategy_returns).cumprod()        peak = cumulative.expanding().max()        drawdown = (cumulative - peak) / peak        max_dd = drawdown.min()        performance_metrics[strategy] = [            f"{ann_return:.2%}",            f"{ann_vol:.2%}",            f"{sharpe:.2f}",            f"{max_dd:.2%}"        ]    performance_metrics.index = ['年化收益', '年化波动', '夏普比率', '最大回撤']    return comparison, performance_metrics# 执行对比comparison_df, metrics_df = compare_with_benchmarks(portfolio_values, tickers)# 可视化对比fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 10))# 净值曲线对比for column in comparison_df.columns:    ax1.plot(comparison_df.index, comparison_df[column] / 100000,             label=column, linewidth=2)ax1.set_ylabel('净值 (起始=1)')ax1.set_title('策略净值对比', fontsize=15, pad=20)ax1.legend()ax1.grid(True, alpha=0.3)# 回撤对比for column in comparison_df.columns:    returns = comparison_df[column].pct_change().dropna()    cumulative = (1 + returns).cumprod()    peak = cumulative.expanding().max()    drawdown = (cumulative - peak) / peak    ax2.plot(comparison_df.index[1:], drawdown, label=column, linewidth=2)ax2.set_ylabel('回撤')ax2.set_title('最大回撤对比', fontsize=15, pad=20)ax2.legend()ax2.grid(True, alpha=0.3)ax2.axhline(y=0, color='black', linestyle='-', alpha=0.3)plt.tight_layout()plt.show()print("\n绩效指标对比:")print(metrics_df)

def portfolio_construction_checklist():    """    投资组合构建检查清单    """    checklist = {        '分散化': {            '行业分散': '至少覆盖3-5个不同行业',            '相关性': '选择相关系数较低的资产',            '数量': '建议5-10只股票，避免过度分散'        },        '风险管理': {            '单只股票上限': '建议不超过20%',            '行业集中度': '单一行业不超过30%',            '止损机制': '设置动态止损线'        },        '再平衡规则': {            '时间触发': '每季度或每半年再平衡',            '阈值触发': '权重偏离目标超过5%时再平衡',            '市场触发': '市场大幅波动时检查再平衡'        }    }    return checklist# 打印检查清单checklist = portfolio_construction_checklist()for category, items in checklist.items():    print(f"\n{category}:")    for item, suggestion in items.items():        print(f"  • {item}: {suggestion}")

def common_mistakes_and_solutions():    """    常见投资组合管理误区及解决方案    """    mistakes = [        {            '误区': '过度分散，持有太多股票',            '问题': '增加管理难度，稀释优秀股票的贡献',            '解决方案': '精选5-10只核心股票，其他通过ETF配置'        },        {            '误区': '只看历史收益，忽略风险',            '问题': '高收益往往伴随高风险',            '解决方案': '使用夏普比率等风险调整后收益指标'        },        {            '误区': '频繁调仓，交易成本过高',            '问题': '侵蚀组合收益',            '解决方案': '制定明确的再平衡规则，避免情绪化交易'        },        {            '误区': '忽略相关性变化',            '问题': '危机时期相关性上升，分散效果降低',            '解决方案': '定期检查相关性矩阵，加入低相关资产'        }    ]    return pd.DataFrame(mistakes)# 显示常见误区mistakes_df = common_mistakes_and_solutions()print("\n常见误区及解决方案:")print(mistakes_df.to_string(index=False))