In [1]:
Copied!
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
np.random.seed(42)
print('Libraries loaded')
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
np.random.seed(42)
print('Libraries loaded')
Libraries loaded
In [2]:
Copied!
import matplotlib.pyplot as plt
# 1. 设置系统自带的中文字体(这里使用黑体 SimHei)
plt.rcParams['font.sans-serif'] = ['SimHei'] # 如果你想用微软雅黑,可以改成 ['Microsoft YaHei']
# 2. 解决更换字体后,负号(-)显示为方块的问题
plt.rcParams['axes.unicode_minus'] = False
import matplotlib.pyplot as plt
# 1. 设置系统自带的中文字体(这里使用黑体 SimHei)
plt.rcParams['font.sans-serif'] = ['SimHei'] # 如果你想用微软雅黑,可以改成 ['Microsoft YaHei']
# 2. 解决更换字体后,负号(-)显示为方块的问题
plt.rcParams['axes.unicode_minus'] = False
1. 为什么要识别市场状态?¶
许多量化策略在牛市和熊市中完全不同:
- 动量策略在牛市高效,熊市崩溃时动量策略大幅亏损
- 均值回归在震荡市有效,趋势市中持续亏损
- 最优解:识别市场状态,切换匹配的策略或资产权重
In [3]:
Copied!
# 模拟一段带有状态切换的市场数据
np.random.seed(42)
n = 252 * 5
# 真实状态序列:0=熊市, 1=震荡, 2=牛市
true_states = np.zeros(n, dtype=int)
true_states[0:400] = 0 # 熊市
true_states[400:700] = 1 # 震荡
true_states[700:900] = 2 # 牛市
true_states[900:1000] = 0 # 熊市
true_states[1000:] = 2 # 牛市
state_params = {0:(-0.0008, 0.022), 1:(0.0002, 0.012), 2:(0.0010, 0.008)}
market_returns = np.array([np.random.normal(*state_params[s]) for s in true_states])
cum_market = (1 + pd.Series(market_returns)).cumprod()
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(13, 7), sharex=True)
ax1.plot(cum_market, color='steelblue', lw=1.2)
for start, end, lbl, color in [(0,400,'熊市','red'),(400,700,'震荡','yellow'),(700,900,'牛市','green'),(900,1000,'熊市','red'),(1000,n,'牛市','green')]:
ax1.axvspan(start, end, alpha=0.2, color=color)
ax1.set_title('模拟市场价格(不同颜色代表不同状态)'); ax1.set_ylabel('累积净值')
volatility = pd.Series(market_returns).rolling(20).std() * np.sqrt(252)
volatility.plot(ax=ax2, color='darkorange', lw=1.2)
ax2.set_title('20日滚动年化波动率'); ax2.set_ylabel('年化波动率')
plt.tight_layout(); plt.show()
# 模拟一段带有状态切换的市场数据
np.random.seed(42)
n = 252 * 5
# 真实状态序列:0=熊市, 1=震荡, 2=牛市
true_states = np.zeros(n, dtype=int)
true_states[0:400] = 0 # 熊市
true_states[400:700] = 1 # 震荡
true_states[700:900] = 2 # 牛市
true_states[900:1000] = 0 # 熊市
true_states[1000:] = 2 # 牛市
state_params = {0:(-0.0008, 0.022), 1:(0.0002, 0.012), 2:(0.0010, 0.008)}
market_returns = np.array([np.random.normal(*state_params[s]) for s in true_states])
cum_market = (1 + pd.Series(market_returns)).cumprod()
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(13, 7), sharex=True)
ax1.plot(cum_market, color='steelblue', lw=1.2)
for start, end, lbl, color in [(0,400,'熊市','red'),(400,700,'震荡','yellow'),(700,900,'牛市','green'),(900,1000,'熊市','red'),(1000,n,'牛市','green')]:
ax1.axvspan(start, end, alpha=0.2, color=color)
ax1.set_title('模拟市场价格(不同颜色代表不同状态)'); ax1.set_ylabel('累积净值')
volatility = pd.Series(market_returns).rolling(20).std() * np.sqrt(252)
volatility.plot(ax=ax2, color='darkorange', lw=1.2)
ax2.set_title('20日滚动年化波动率'); ax2.set_ylabel('年化波动率')
plt.tight_layout(); plt.show()
In [4]:
Copied!
vol_20d = pd.Series(market_returns).rolling(20).std() * np.sqrt(252)
identified_state = pd.Series('unknown', index=range(n))
identified_state[vol_20d > 0.20] = 'risk_off' # 高波动 → 防御
identified_state[vol_20d < 0.12] = 'risk_on' # 低波动 → 进攻
# 动态配置策略
# risk_on: 80% 股票 + 20% 债券
# risk_off: 20% 股票 + 80% 债券
bond_returns = np.random.normal(0.0003, 0.003, n)
stock_returns = market_returns
weights_stock = pd.Series(0.5, index=range(n)) # 默认 50/50
weights_stock[identified_state == 'risk_on'] = 0.8
weights_stock[identified_state == 'risk_off'] = 0.2
ret_dynamic = weights_stock * stock_returns + (1-weights_stock) * bond_returns
ret_passive = 0.6 * stock_returns + 0.4 * bond_returns # 60/40 被动
cum_dyn = (1 + pd.Series(ret_dynamic)).cumprod()
cum_pass = (1 + pd.Series(ret_passive)).cumprod()
plt.figure(figsize=(12, 5))
plt.plot(cum_dyn, 'blue', lw=2, label='波动率状态识别动态配置')
plt.plot(cum_pass, 'gray', lw=2, linestyle='--', label='60/40 被动配置')
plt.title('市场状态识别动态配置 vs 60/40 被动')
plt.ylabel('累积净值'); plt.legend(); plt.grid(alpha=0.3); plt.show()
def sr(r): return pd.Series(r).mean()/pd.Series(r).std()*np.sqrt(252)
print(f'动态配置 Sharpe: {sr(ret_dynamic):.3f}')
vol_20d = pd.Series(market_returns).rolling(20).std() * np.sqrt(252)
identified_state = pd.Series('unknown', index=range(n))
identified_state[vol_20d > 0.20] = 'risk_off' # 高波动 → 防御
identified_state[vol_20d < 0.12] = 'risk_on' # 低波动 → 进攻
# 动态配置策略
# risk_on: 80% 股票 + 20% 债券
# risk_off: 20% 股票 + 80% 债券
bond_returns = np.random.normal(0.0003, 0.003, n)
stock_returns = market_returns
weights_stock = pd.Series(0.5, index=range(n)) # 默认 50/50
weights_stock[identified_state == 'risk_on'] = 0.8
weights_stock[identified_state == 'risk_off'] = 0.2
ret_dynamic = weights_stock * stock_returns + (1-weights_stock) * bond_returns
ret_passive = 0.6 * stock_returns + 0.4 * bond_returns # 60/40 被动
cum_dyn = (1 + pd.Series(ret_dynamic)).cumprod()
cum_pass = (1 + pd.Series(ret_passive)).cumprod()
plt.figure(figsize=(12, 5))
plt.plot(cum_dyn, 'blue', lw=2, label='波动率状态识别动态配置')
plt.plot(cum_pass, 'gray', lw=2, linestyle='--', label='60/40 被动配置')
plt.title('市场状态识别动态配置 vs 60/40 被动')
plt.ylabel('累积净值'); plt.legend(); plt.grid(alpha=0.3); plt.show()
def sr(r): return pd.Series(r).mean()/pd.Series(r).std()*np.sqrt(252)
print(f'动态配置 Sharpe: {sr(ret_dynamic):.3f}')
动态配置 Sharpe: 1.881
🎯 练习¶
- 用指数移动平均(EWM)替代简单移动平均计算波动率,对比状态识别的稳定性。
- 引入宏观信号(如收益率曲线斜率代理变量),结合波动率构建双信号状态识别器。
- 研究 hmmlearn 库的 GaussianHMM,用真实 SPY 数据拟合 2-状态 HMM,标注历史牛熊区间。
下一节 → ../04_backtesting/10_performance_attribution.ipynb
In [ ]:
Copied!