波动率控制仓位系列（五）源代码汇总

前面分章节为了说明每一步的任务，将代码拆分开进行详细的说明。需要读者自己拼接起来，这个事情还是比较繁琐的。另外，一些代码的细节可能在前后写文章的时候出现了不一致。因此我最后再加一章，主要是将所有的代码汇总到一起并且至少调试通过，让读者可以自己运行与验证。

"""
波动率控制仓位 —— 完整回测脚本
使用 hq.get_k_data() 获取数据
"""

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
from pytdx.hq import TdxHq_API

plt.rcParams['font.sans-serif'] = ['SimHei'] # 黑体（Windows 自带）
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示为方块的问题

# ==================== 1. 数据获取（使用 hq.get_k_data） ====================

def get_etf_daily_data(hq, code, start_date='20180101', end_date='20251231'):
"""
分批获取 ETF 日线数据，返回包含 date 和 close 的 DataFrame
hq 需具备 get_k_data(code, start_date, end_date) 方法，
返回的 DataFrame 应包含 'date' 和 'close' 列（允许有更多列）。
"""
start = datetime.strptime(start_date, '%Y%m%d')
end = datetime.strptime(end_date, '%Y%m%d')
all_data = []
batch_size = 600
current_start = start

while current_start < end:
current_end = min(current_start + timedelta(days=batch_size), end)
try:
df = hq.get_k_data(
code=code,
start_date=current_start.strftime('%Y%m%d'),
end_date=current_end.strftime('%Y%m%d')
)
if df is not None and len(df) > 0:
all_data.append(df)
except Exception as e:
print(f' 获取 {code} 时出错：{e}')
current_start = current_end + timedelta(days=1)

if not all_data:
return pd.DataFrame()

df_all = pd.concat(all_data)
df_all['date'] = pd.to_datetime(df_all['date'])
df_all = df_all[['date', 'close']].set_index('date').sort_index()
return df_all

def 波动控制验证():
# 候选池：10只行业ETF + 1只国债ETF（代码 -> 名称）
etf_info = {
'512880': '证券ETF', '512010': '医药ETF', '159995': '芯片ETF',
'515790': '光伏ETF', '515030': '新能源车ETF', '512660': '军工ETF',
'512690': '酒ETF', '159869': '游戏ETF', '512800': '银行ETF',
'512400': '有色金属ETF', '511010': '国债ETF'
}
safe_asset = '511010'
sector_etfs = [c for c in etf_info.keys() if c != safe_asset]

# ———- 请在此处初始化你的 hq 对象 ———-
# 示例（请替换为实际库）：
# from your_data_lib import DataSource
# hq = DataSource()
# ————————————————
# 如果暂无可用 hq，可以跳过数据下载，使用之前保存的 price_df 继续回测。

try:
print("连接数据源…")
TDX_SERVER_IP = "116.205.183.150"
hq = TdxHq_API()
hq.connect(TDX_SERVER_IP, 7709)

etf_data = {}
for code, name in etf_info.items():
print(f'下载 {name}({code})…')
df = get_etf_daily_data(hq, code)
if not df.empty:
df.rename(columns={'close': code}, inplace=True)
etf_data[code] = df
print(f'{name}({code}):', df)
hq.disconnect()

price_df = pd.concat(etf_data.values(), axis=1)
price_df.ffill(inplace=True)
price_df.dropna(inplace=True)
print(f'数据范围：{price_df.index[0].date()} 至 {price_df.index[-1].date()}\\n')
except Exception as e:
print(f"数据下载失败：{e}")
print("尝试从本地文件 'etf_price.csv' 读取…")
try:
price_df = pd.read_csv('etf_price.csv', index_col=0, parse_dates=True)
except FileNotFoundError:
print("本地文件不存在，请先下载数据并保存为 etf_price.csv。")
exit()

# ==================== 2. 信号生成（仅对行业ETF计算动量） ====================
momentum_20 = price_df[sector_etfs].pct_change(periods=20).dropna()
best_etf = momentum_20.idxmax(axis=1)
max_val = momentum_20.max(axis=1)

signal_df = pd.DataFrame({'best_etf': best_etf, 'max_momentum': max_val})
signal_df['hold'] = signal_df['best_etf'].shift(1) # 信号后移一天
signal_df['max_shifted'] = signal_df['max_momentum'].shift(1)
signal_df.loc[signal_df['max_shifted'] < 0, 'hold'] = safe_asset
signal_df.dropna(inplace=True)

# 周频调仓：直接使用周五的 hold 信号（已 shift 过，不再重复）
signal_df['weekday'] = signal_df.index.dayofweek
friday_signals = signal_df[signal_df['weekday'] == 4].copy()
friday_signals['trade_signal'] = friday_signals['hold']
friday_signals.loc[friday_signals['max_shifted'] < 0, 'trade_signal'] = safe_asset

daily_signal = friday_signals['trade_signal'].reindex(price_df.index)
daily_signal.ffill(inplace=True)
daily_signal.fillna(safe_asset, inplace=True)

# ==================== 3. 收益率与净值计算 ====================
daily_returns = price_df.pct_change().fillna(0)

# 3.1 等权买入持有基准
benchmark_returns = daily_returns[sector_etfs].mean(axis=1)
benchmark_nav = (1 + benchmark_returns).cumprod()
benchmark_nav.name = 'Benchmark'

# 3.2 满仓轮动
strategy_returns_full = pd.Series(0.0, index=price_df.index)
for i in range(1, len(strategy_returns_full)):
today = strategy_returns_full.index[i]
yesterday = strategy_returns_full.index[i-1]
asset = daily_signal.loc[yesterday]
if asset in daily_returns.columns:
strategy_returns_full.loc[today] = daily_returns.loc[today, asset]
strategy_nav_full = (1 + strategy_returns_full).cumprod()
strategy_nav_full.name = 'FullPosition'

# 3.3 波动率控制轮动
daily_std = daily_returns[sector_etfs].rolling(20).std()
hist_vol = daily_std * np.sqrt(252)
hist_vol_shifted = hist_vol.shift(1)
target_vol = 0.15
position_ratio = (target_vol / hist_vol_shifted).clip(upper=1.0)

strategy_returns_vol = pd.Series(0.0, index=price_df.index)
for i in range(1, len(strategy_returns_vol)):
today = strategy_returns_vol.index[i]
yesterday = strategy_returns_vol.index[i-1]
asset = daily_signal.loc[yesterday]
if asset == safe_asset or asset not in sector_etfs:
strategy_returns_vol.loc[today] = daily_returns.loc[today, asset] if asset in daily_returns.columns else 0.0
else:
ratio = position_ratio.loc[yesterday, asset] if asset in position_ratio.columns else 1.0
asset_ret = daily_returns.loc[today, asset]
bond_ret = daily_returns.loc[today, safe_asset] if safe_asset in daily_returns.columns else 0.0
strategy_returns_vol.loc[today] = ratio * asset_ret + (1 – ratio) * bond_ret

strategy_nav_vol = (1 + strategy_returns_vol).cumprod()
strategy_nav_vol.name = 'VolControl'

# ==================== 4. 绩效指标计算 ====================
def calc_metrics(nav_series, returns_series, risk_free=0.03):
total_return = nav_series.iloc[-1] / nav_series.iloc[0] – 1
days = (nav_series.index[-1] – nav_series.index[0]).days
years = days / 365.25
annual_return = (1 + total_return) ** (1 / years) – 1 if years > 0 else 0
peak = nav_series.expanding().max()
drawdown = (nav_series – peak) / peak
max_dd = drawdown.min()
annual_vol = returns_series.std() * np.sqrt(252) if len(returns_series) > 0 else 0
excess = returns_series – risk_free / 252
sharpe = (excess.mean() / excess.std()) * np.sqrt(252) if excess.std() != 0 else 0
calmar = annual_return / abs(max_dd) if max_dd != 0 else 0
return {
'累计收益率': f'{total_return:.2%}',
'年化收益率': f'{annual_return:.2%}',
'最大回撤': f'{max_dd:.2%}',
'年化波动率': f'{annual_vol:.2%}',
'夏普比率': f'{sharpe:.2f}',
'卡玛比率': f'{calmar:.2f}'
}

bench_metrics = calc_metrics(benchmark_nav, benchmark_returns)
full_metrics = calc_metrics(strategy_nav_full, strategy_returns_full)
vol_metrics = calc_metrics(strategy_nav_vol, strategy_returns_vol)

print("========== 绩效对比 ==========")
print(f"{'指标':<10} {'基准':<12} {'满仓轮动':<12} {'波动率控制':<12}")
print("-" * 46)
for key in bench_metrics.keys():
print(f"{key:<10} {bench_metrics[key]:<12} {full_metrics[key]:<12} {vol_metrics[key]:<12}")

# ==================== 5. 净值曲线可视化 ====================
plt.figure(figsize=(14, 7))
plt.plot(benchmark_nav, label='等权买入持有基准', linewidth=1.5, alpha=0.6, color='gray')
plt.plot(strategy_nav_full, label='满仓轮动（无控制）', linewidth=1.5, color='red', alpha=0.8)
plt.plot(strategy_nav_vol, label='波动率控制轮动（目标15%）', linewidth=1.8, color='blue')
plt.title('净值曲线对比：满仓轮动 vs 波动率控制', fontsize=14)
plt.xlabel('日期')
plt.ylabel('净值（初始为1）')
plt.legend(loc='upper left')
plt.grid(alpha=0.3)
plt.tight_layout()
plt.show()

if __name__ == '__main__':
波动控制验证()

代码运行结果如下：