仿站怎么做员工培训

动态Beta模拟、t分布数据生成、结构性断点、滚动Beta计算、异方差处理、Fama-French三因子扩展、统计检验和可视化

import numpy as np
import pandas as pd
import statsmodels.api as sm
import matplotlib.pyplot as plt
from scipy.stats import ttest_1samp# ======================
# 参数设置
# ======================
np.random.seed(42)
NUM_MONTHS = 60
MARKET_VOLATILITY = 0.0433
ALPHA_TRUE = 0.00# ======================
# 数据生成模块
# ======================
def generate_returns():# 基础市场收益（t分布，自由度为5）market_returns = np.random.standard_t(df=5, size=NUM_MONTHS) * MARKET_VOLATILITY# 生成三种不同的Beta场景beta_scenarios = {# 场景1：时变Beta（每两年+0.1）"time_varying": [1.5 + 0.1 * (i // 24) for i in range(NUM_MONTHS)],# 场景2：结构性断点（第36个月后Beta突变）"structural_break": np.concatenate([np.full(36, 1.2), np.full(24, 1.8)]),# 场景3：恒定Beta（作为对照组）"constant": np.full(NUM_MONTHS, 1.5)}# 生成资产收益dfs = {}for scenario, beta_values in beta_scenarios.items():epsilon = np.random.normal(loc=0, scale=0.05, size=NUM_MONTHS)asset_returns = ALPHA_TRUE + np.array(beta_values) * market_returns + epsilondates = pd.date_range(start='2014-01-01', periods=NUM_MONTHS, freq='M')dfs[scenario] = pd.DataFrame({'market': market_returns,'asset': asset_returns,'true_beta': beta_values}, index=dates)return dfs# ======================
# 分析模块
# ======================
def analyze_data(df, scenario_name):# 滚动窗口Beta计算（24个月窗口）df['rolling_beta'] = df['asset'].rolling(24).cov(df['market']) / df['market'].rolling(24).var()# OLS回归X = sm.add_constant(df['market'])y = df['asset']model_ols = sm.OLS(y, X).fit()# 处理异方差性（WLS）weights = 1 / np.abs(df['market'])  # 简单权重方案model_wls = sm.WLS(y, X, weights=weights).fit()# Fama-French三因子扩展df['SMB'] = np.random.normal(0, 0.03, NUM_MONTHS)  # 模拟小市值因子df['HML'] = np.random.normal(0, 0.03, NUM_MONTHS)  # 模拟价值因子X_ff3 = sm.add_constant(df[['market', 'SMB', 'HML']])model_ff3 = sm.OLS(y, X_ff3).fit()# ======================# 可视化模块# ======================plt.figure(figsize=(15, 10))# 动态Beta可视化plt.subplot(2, 2, 1)df['true_beta'].plot(label='True Beta', lw=2)df['rolling_beta'].plot(label='24M Rolling Beta', ls='--')if 'structural_break' in scenario_name:plt.axvline(df.index[35], color='r', linestyle=':', label='Break Point')plt.title(f'Beta Dynamics: {scenario_name}')plt.legend()# 残差诊断图plt.subplot(2, 2, 2)plt.scatter(model_ols.predict(X), model_ols.resid, alpha=0.6)plt.axhline(0, color='red')plt.title('Residuals vs Fitted Values')plt.xlabel('Predicted')plt.ylabel('Residuals')# 累计收益对比plt.subplot(2, 2, 3)(1 + df[['market', 'asset']]).cumprod().plot()plt.title('Cumulative Returns Comparison')# 回归结果可视化plt.subplot(2, 2, 4)plt.scatter(df['market'], df['asset'], alpha=0.6)plt.plot(df['market'], model_ols.predict(X), color='red', label='OLS')plt.plot(df['market'], model_wls.predict(X), color='green', ls='--', label='WLS')plt.title('Regression Lines Comparison')plt.legend()plt.tight_layout()plt.show()# ======================# 统计检验输出# ======================print(f"\n=== {scenario_name.upper()} SCENARIO ANALYSIS ===")print("\n1. OLS Regression Results:")print(model_ols.summary())print("\n2. WLS Regression Results (处理异方差性):")print(model_wls.summary())print("\n3. Fama-French三因子模型:")print(model_ff3.summary())print("\n4. Beta假设检验:")t_stat, p_value = ttest_1samp(df['rolling_beta'].dropna(), popmean=1.5)print(f"H0: Beta=1.5 | t-stat={t_stat:.2f} | p-value={p_value:.3f}")# ======================
# 执行主程序
# ======================
dfs = generate_returns()
for scenario_name, df in dfs.items():analyze_data(df, scenario_name)