|
|
@@ -0,0 +1,546 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+"""
|
|
|
+创业板50指数点位量化交易策略回测框架
|
|
|
+训练集:2017-2023 | 验证集:2024-2025
|
|
|
+"""
|
|
|
+
|
|
|
+import pandas as pd
|
|
|
+import numpy as np
|
|
|
+import matplotlib.pyplot as plt
|
|
|
+from datetime import datetime, timedelta
|
|
|
+import warnings
|
|
|
+warnings.filterwarnings('ignore')
|
|
|
+
|
|
|
+# 设置中文显示
|
|
|
+plt.rcParams['font.sans-serif'] = ['DejaVu Sans']
|
|
|
+plt.rcParams['axes.unicode_minus'] = False
|
|
|
+
|
|
|
+# ==================== 1. 数据生成/加载 ====================
|
|
|
+
|
|
|
+def generate_mock_data(start='2017-01-01', end='2025-12-31', seed=42):
|
|
|
+ """
|
|
|
+ 生成模拟的创业板50指数数据(用于演示框架)
|
|
|
+ 实际使用时应替换为真实数据
|
|
|
+ """
|
|
|
+ np.random.seed(seed)
|
|
|
+ dates = pd.date_range(start=start, end=end, freq='D')
|
|
|
+ # 只保留交易日(简化:去掉周末)
|
|
|
+ dates = dates[dates.dayofweek < 5]
|
|
|
+
|
|
|
+ n = len(dates)
|
|
|
+
|
|
|
+ # 生成带有趋势和波动的价格序列
|
|
|
+ # 创业板50基准约2000点
|
|
|
+ returns = np.random.normal(0.0003, 0.016, n) # 日均收益0.03%,波动1.6%
|
|
|
+
|
|
|
+ # 添加一些趋势性(牛市、熊市、震荡)
|
|
|
+ # 2019-2021牛市
|
|
|
+ bull_mask = (dates >= '2019-01-01') & (dates <= '2021-12-31')
|
|
|
+ returns[bull_mask] += 0.001
|
|
|
+
|
|
|
+ # 2022熊市
|
|
|
+ bear_mask = (dates >= '2022-01-01') & (dates <= '2022-12-31')
|
|
|
+ returns[bear_mask] -= 0.001
|
|
|
+
|
|
|
+ # 2024-2025震荡
|
|
|
+ osc_mask = dates >= '2024-01-01'
|
|
|
+ returns[osc_mask] = np.random.normal(0, 0.012, sum(osc_mask))
|
|
|
+
|
|
|
+ # 计算价格
|
|
|
+ price = 2000 * np.exp(np.cumsum(returns))
|
|
|
+
|
|
|
+ # 生成OHLC数据
|
|
|
+ df = pd.DataFrame(index=dates)
|
|
|
+ df['close'] = price
|
|
|
+ df['open'] = price * (1 + np.random.normal(0, 0.005, n))
|
|
|
+ df['high'] = np.maximum(df[['open', 'close']].max(axis=1) * (1 + np.abs(np.random.normal(0, 0.008, n))),
|
|
|
+ df[['open', 'close']].max(axis=1))
|
|
|
+ df['low'] = np.minimum(df[['open', 'close']].min(axis=1) * (1 - np.abs(np.random.normal(0, 0.008, n))),
|
|
|
+ df[['open', 'close']].min(axis=1))
|
|
|
+
|
|
|
+ return df
|
|
|
+
|
|
|
+# ==================== 2. 技术指标计算 ====================
|
|
|
+
|
|
|
+def calculate_atr(high, low, close, period=20):
|
|
|
+ """计算ATR(平均真实波幅)"""
|
|
|
+ tr1 = high - low
|
|
|
+ tr2 = abs(high - close.shift(1))
|
|
|
+ tr3 = abs(low - close.shift(1))
|
|
|
+ tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
|
|
|
+ atr = tr.rolling(window=period).mean()
|
|
|
+ return atr
|
|
|
+
|
|
|
+def calculate_rsrs(high, low, n=20, m=250):
|
|
|
+ """
|
|
|
+ 计算RSRS指标(阻力支撑相对强度)- 优化版
|
|
|
+ 返回: (rsrs_score, r_squared)
|
|
|
+ """
|
|
|
+ # 使用rolling计算斜率和R²
|
|
|
+ def rolling_beta(x):
|
|
|
+ if len(x) < n or np.std(x[:n//2]) == 0:
|
|
|
+ return np.nan
|
|
|
+ low_vals = x[:n//2]
|
|
|
+ high_vals = x[n//2:]
|
|
|
+ if np.std(low_vals) == 0:
|
|
|
+ return 0
|
|
|
+ beta = np.corrcoef(low_vals, high_vals)[0,1] * np.std(high_vals) / np.std(low_vals)
|
|
|
+ return beta
|
|
|
+
|
|
|
+ # 简化计算:使用 rolling.apply
|
|
|
+ slopes = pd.Series(index=high.index, dtype=float)
|
|
|
+ r2s = pd.Series(index=high.index, dtype=float)
|
|
|
+
|
|
|
+ for i in range(n-1, len(high)):
|
|
|
+ low_window = low.iloc[i-n+1:i+1].values
|
|
|
+ high_window = high.iloc[i-n+1:i+1].values
|
|
|
+
|
|
|
+ if np.std(low_window) > 0:
|
|
|
+ beta = np.corrcoef(low_window, high_window)[0,1] * np.std(high_window) / np.std(low_window)
|
|
|
+ # R²
|
|
|
+ y_pred = np.mean(high_window) + beta * (low_window - np.mean(low_window))
|
|
|
+ ss_res = np.sum((high_window - y_pred) ** 2)
|
|
|
+ ss_tot = np.sum((high_window - np.mean(high_window)) ** 2)
|
|
|
+ r2 = 1 - ss_res / ss_tot if ss_tot > 0 else 0
|
|
|
+
|
|
|
+ slopes.iloc[i] = beta
|
|
|
+ r2s.iloc[i] = r2
|
|
|
+
|
|
|
+ # 计算标准分(滚动M日)
|
|
|
+ rsrs = pd.Series(index=high.index, dtype=float)
|
|
|
+ for i in range(m+n-2, len(slopes)):
|
|
|
+ slope_window = slopes.iloc[i-m+1:i+1]
|
|
|
+ if slope_window.std() > 0:
|
|
|
+ zscore = (slopes.iloc[i] - slope_window.mean()) / slope_window.std()
|
|
|
+ rsrs.iloc[i] = zscore * r2s.iloc[i]
|
|
|
+
|
|
|
+ return rsrs, r2s
|
|
|
+
|
|
|
+def calculate_rsi(close, period=14):
|
|
|
+ """计算RSI指标"""
|
|
|
+ delta = close.diff()
|
|
|
+ gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
|
|
|
+ loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
|
|
|
+ rs = gain / loss
|
|
|
+ rsi = 100 - (100 / (1 + rs))
|
|
|
+ return rsi
|
|
|
+
|
|
|
+# ==================== 3. 市场状态判断 ====================
|
|
|
+
|
|
|
+def classify_state(close, ma20, ma60, ma120, atr_percent, rsrs, close_series):
|
|
|
+ """
|
|
|
+ 判断市场状态:BULL(牛市)/ BEAR(熊市)/ OSCILLATE(震荡)
|
|
|
+ close_series: 用于计算涨跌幅的收盘价序列
|
|
|
+ """
|
|
|
+ # 趋势得分
|
|
|
+ trend_score = 0
|
|
|
+ if close > ma20: trend_score += 1
|
|
|
+ if ma20 > ma60: trend_score += 1
|
|
|
+ if ma60 > ma120: trend_score += 1
|
|
|
+
|
|
|
+ # 波动率
|
|
|
+ high_volatility = atr_percent > 5
|
|
|
+
|
|
|
+ # 熔断检测(极端情况)
|
|
|
+ daily_return = close_series.pct_change().iloc[-1] if len(close_series) > 1 else 0
|
|
|
+ crash = daily_return < -0.07
|
|
|
+
|
|
|
+ if crash or (trend_score <= 1 and high_volatility and close < ma60):
|
|
|
+ return "BEAR"
|
|
|
+ elif trend_score >= 2 and not high_volatility:
|
|
|
+ return "BULL"
|
|
|
+ else:
|
|
|
+ return "OSCILLATE"
|
|
|
+
|
|
|
+# ==================== 4. 策略核心 ====================
|
|
|
+
|
|
|
+class CYB50Strategy:
|
|
|
+ """创业板50指数交易策略"""
|
|
|
+
|
|
|
+ def __init__(self, params=None):
|
|
|
+ # 默认参数
|
|
|
+ self.params = params or {
|
|
|
+ 'rsrs_n': 20,
|
|
|
+ 'rsrs_m': 250,
|
|
|
+ 'bull_buy': 0.5,
|
|
|
+ 'bull_sell': -0.7,
|
|
|
+ 'bear_buy': 1.5,
|
|
|
+ 'bull_max': 1.0,
|
|
|
+ 'osc_max': 0.6,
|
|
|
+ 'stop_loss': 0.10,
|
|
|
+ 'min_change': 0.20
|
|
|
+ }
|
|
|
+ self.current_position = 0
|
|
|
+ self.state_history = []
|
|
|
+ self.entry_price = None
|
|
|
+
|
|
|
+ def generate_signal(self, data):
|
|
|
+ """生成交易信号"""
|
|
|
+ close = data['close']
|
|
|
+ high = data['high']
|
|
|
+ low = data['low']
|
|
|
+
|
|
|
+ # 计算指标
|
|
|
+ rsrs, r2 = calculate_rsrs(high, low,
|
|
|
+ self.params['rsrs_n'],
|
|
|
+ self.params['rsrs_m'])
|
|
|
+
|
|
|
+ ma20 = close.rolling(20).mean()
|
|
|
+ ma60 = close.rolling(60).mean()
|
|
|
+ ma120 = close.rolling(120).mean()
|
|
|
+ atr = calculate_atr(high, low, close, 20)
|
|
|
+ atr_percent = atr / close * 100
|
|
|
+
|
|
|
+ # 获取当前值
|
|
|
+ curr_rsrs = rsrs.iloc[-1]
|
|
|
+ curr_close = close.iloc[-1]
|
|
|
+ curr_ma20 = ma20.iloc[-1]
|
|
|
+ curr_ma60 = ma60.iloc[-1]
|
|
|
+ curr_ma120 = ma120.iloc[-1]
|
|
|
+ curr_atr_pct = atr_percent.iloc[-1]
|
|
|
+
|
|
|
+ # 检查是否有足够数据
|
|
|
+ if pd.isna(curr_rsrs):
|
|
|
+ return 0, "INIT"
|
|
|
+
|
|
|
+ # 判断市场状态
|
|
|
+ state = classify_state(curr_close, curr_ma20, curr_ma60,
|
|
|
+ curr_ma120, curr_atr_pct, curr_rsrs, close)
|
|
|
+
|
|
|
+ # 状态防抖(连续3日确认,极端情况除外)
|
|
|
+ self.state_history.append(state)
|
|
|
+ if len(self.state_history) >= 3:
|
|
|
+ # 熔断检测:单日大跌立即转熊
|
|
|
+ daily_return = close.pct_change().iloc[-1]
|
|
|
+ if daily_return < -0.07:
|
|
|
+ state = "BEAR"
|
|
|
+ elif len(self.state_history) >= 3:
|
|
|
+ # 正常防抖
|
|
|
+ recent_states = self.state_history[-3:]
|
|
|
+ if len(set(recent_states)) > 1:
|
|
|
+ state = self.state_history[-2] if len(self.state_history) >= 2 else state
|
|
|
+
|
|
|
+ # 根据状态确定仓位
|
|
|
+ target_pos = self._calculate_position(state, curr_rsrs, curr_atr_pct)
|
|
|
+
|
|
|
+ # 止损检查
|
|
|
+ if self.entry_price is not None and self.current_position > 0:
|
|
|
+ current_drawdown = (curr_close - self.entry_price) / self.entry_price
|
|
|
+ if current_drawdown < -self.params['stop_loss']:
|
|
|
+ target_pos = 0
|
|
|
+ self.entry_price = None
|
|
|
+
|
|
|
+ # 最小调仓幅度过滤
|
|
|
+ if abs(target_pos - self.current_position) < self.params['min_change']:
|
|
|
+ target_pos = self.current_position
|
|
|
+
|
|
|
+ # 更新入场价
|
|
|
+ if target_pos > 0 and self.current_position == 0:
|
|
|
+ self.entry_price = curr_close
|
|
|
+ elif target_pos == 0:
|
|
|
+ self.entry_price = None
|
|
|
+
|
|
|
+ self.current_position = target_pos
|
|
|
+ return target_pos, state
|
|
|
+
|
|
|
+ def _calculate_position(self, state, rsrs, atr_percent):
|
|
|
+ """根据状态和指标计算目标仓位"""
|
|
|
+ p = self.params
|
|
|
+
|
|
|
+ if state == "BULL":
|
|
|
+ if rsrs > p['bull_buy']:
|
|
|
+ pos = p['bull_max']
|
|
|
+ elif rsrs < p['bull_sell']:
|
|
|
+ pos = 0
|
|
|
+ else:
|
|
|
+ pos = p['bull_max'] * 0.5
|
|
|
+
|
|
|
+ elif state == "BEAR":
|
|
|
+ # 熊市:空仓为主
|
|
|
+ if rsrs < -p['bear_buy']:
|
|
|
+ pos = 0.1 # 极端超卖,10%仓位博反弹
|
|
|
+ else:
|
|
|
+ pos = 0
|
|
|
+
|
|
|
+ else: # OSCILLATE
|
|
|
+ if rsrs > 0.7:
|
|
|
+ pos = p['osc_max']
|
|
|
+ elif rsrs < -0.7:
|
|
|
+ pos = 0
|
|
|
+ else:
|
|
|
+ pos = p['osc_max'] * 0.5
|
|
|
+
|
|
|
+ # 波动率调整
|
|
|
+ if atr_percent > 5:
|
|
|
+ pos *= 0.6
|
|
|
+
|
|
|
+ return np.clip(pos, 0, 1)
|
|
|
+
|
|
|
+# ==================== 5. 回测引擎 ====================
|
|
|
+
|
|
|
+def backtest(data, strategy, initial_capital=1000000, start_date=None, end_date=None):
|
|
|
+ """
|
|
|
+ 回测引擎
|
|
|
+ """
|
|
|
+ # 数据切片
|
|
|
+ if start_date:
|
|
|
+ data = data[data.index >= start_date]
|
|
|
+ if end_date:
|
|
|
+ data = data[data.index <= end_date]
|
|
|
+
|
|
|
+ dates = []
|
|
|
+ positions = []
|
|
|
+ navs = []
|
|
|
+ states = []
|
|
|
+
|
|
|
+ capital = initial_capital
|
|
|
+ current_nav = 1.0
|
|
|
+
|
|
|
+ # 跳过前250日(warm-up)
|
|
|
+ start_idx = 250
|
|
|
+
|
|
|
+ for i in range(start_idx, len(data)):
|
|
|
+ curr_data = data.iloc[:i+1]
|
|
|
+ curr_date = data.index[i]
|
|
|
+
|
|
|
+ # 获取信号
|
|
|
+ position, state = strategy.generate_signal(curr_data)
|
|
|
+
|
|
|
+ # 计算当日收益(使用前一日仓位)
|
|
|
+ if i > start_idx:
|
|
|
+ daily_return = data['close'].iloc[i] / data['close'].iloc[i-1] - 1
|
|
|
+ prev_position = positions[-1] if positions else 0
|
|
|
+ strategy_return = daily_return * prev_position
|
|
|
+ current_nav *= (1 + strategy_return)
|
|
|
+
|
|
|
+ dates.append(curr_date)
|
|
|
+ positions.append(position)
|
|
|
+ navs.append(current_nav)
|
|
|
+ states.append(state)
|
|
|
+
|
|
|
+ # 构建结果DataFrame
|
|
|
+ results = pd.DataFrame({
|
|
|
+ 'date': dates,
|
|
|
+ 'position': positions,
|
|
|
+ 'nav': navs,
|
|
|
+ 'state': states
|
|
|
+ }).set_index('date')
|
|
|
+
|
|
|
+ # 计算指数基准
|
|
|
+ index_data = data.iloc[start_idx:].copy()
|
|
|
+ results['index_close'] = index_data['close']
|
|
|
+ results['index_nav'] = results['index_close'] / results['index_close'].iloc[0]
|
|
|
+
|
|
|
+ # 计算指标
|
|
|
+ metrics = calculate_metrics(results['nav'], results['index_nav'])
|
|
|
+
|
|
|
+ return results, metrics
|
|
|
+
|
|
|
+def calculate_metrics(strategy_nav, index_nav):
|
|
|
+ """计算回测指标"""
|
|
|
+ # 收益率
|
|
|
+ total_return = strategy_nav.iloc[-1] - 1
|
|
|
+ days = len(strategy_nav)
|
|
|
+ annual_return = (1 + total_return) ** (252 / days) - 1
|
|
|
+
|
|
|
+ # 指数收益
|
|
|
+ index_return = index_nav.iloc[-1] - 1
|
|
|
+ index_annual = (1 + index_return) ** (252 / days) - 1
|
|
|
+
|
|
|
+ # 最大回撤
|
|
|
+ running_max = strategy_nav.expanding().max()
|
|
|
+ drawdown = (strategy_nav - running_max) / running_max
|
|
|
+ max_drawdown = drawdown.min()
|
|
|
+
|
|
|
+ # 波动率
|
|
|
+ daily_returns = strategy_nav.pct_change().dropna()
|
|
|
+ volatility = daily_returns.std() * np.sqrt(252)
|
|
|
+
|
|
|
+ # 夏普比率(假设无风险利率3%)
|
|
|
+ excess_return = annual_return - 0.03
|
|
|
+ sharpe = excess_return / volatility if volatility > 0 else 0
|
|
|
+
|
|
|
+ # 卡玛比率
|
|
|
+ calmar = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0
|
|
|
+
|
|
|
+ # 胜率
|
|
|
+ positive_days = (daily_returns > 0).sum()
|
|
|
+ total_days = len(daily_returns)
|
|
|
+ win_rate = positive_days / total_days
|
|
|
+
|
|
|
+ # Beta
|
|
|
+ index_returns = index_nav.pct_change().dropna()
|
|
|
+ covariance = daily_returns.cov(index_returns)
|
|
|
+ variance = index_returns.var()
|
|
|
+ beta = covariance / variance if variance > 0 else 1
|
|
|
+
|
|
|
+ # 年化超额收益
|
|
|
+ excess_annual = annual_return - index_annual
|
|
|
+
|
|
|
+ return {
|
|
|
+ 'total_return': total_return,
|
|
|
+ 'annual_return': annual_return,
|
|
|
+ 'index_return': index_return,
|
|
|
+ 'index_annual': index_annual,
|
|
|
+ 'excess_annual': excess_annual,
|
|
|
+ 'max_drawdown': max_drawdown,
|
|
|
+ 'volatility': volatility,
|
|
|
+ 'sharpe': sharpe,
|
|
|
+ 'calmar': calmar,
|
|
|
+ 'win_rate': win_rate,
|
|
|
+ 'beta': beta,
|
|
|
+ 'trading_days': days
|
|
|
+ }
|
|
|
+
|
|
|
+# ==================== 6. 参数优化 ====================
|
|
|
+
|
|
|
+def grid_search(data, param_grid):
|
|
|
+ """网格搜索最优参数 - 简化版"""
|
|
|
+ best_score = -999
|
|
|
+ best_params = None
|
|
|
+ best_metrics = None
|
|
|
+
|
|
|
+ # 只测试1组参数(加速演示)
|
|
|
+ test_params = [
|
|
|
+ {'rsrs_n': 20, 'rsrs_m': 250, 'bull_buy': 0.5, 'bull_sell': -0.7,
|
|
|
+ 'bear_buy': 1.5, 'bull_max': 1.0, 'osc_max': 0.6, 'stop_loss': 0.10, 'min_change': 0.20},
|
|
|
+ ]
|
|
|
+
|
|
|
+ for params in test_params:
|
|
|
+ print(f"测试参数: {params}")
|
|
|
+ strategy = CYB50Strategy(params)
|
|
|
+ results, metrics = backtest(data, strategy, start_date='2018-02-01', end_date='2023-12-31')
|
|
|
+
|
|
|
+ # 综合评分
|
|
|
+ score = metrics['sharpe'] * 0.4 + metrics['calmar'] * 0.4 + metrics['excess_annual'] * 2
|
|
|
+
|
|
|
+ print(f" 年化: {metrics['annual_return']*100:.1f}%, 回撤: {metrics['max_drawdown']*100:.1f}%, 夏普: {metrics['sharpe']:.2f}, 评分: {score:.2f}")
|
|
|
+
|
|
|
+ if score > best_score and metrics['max_drawdown'] > -0.40:
|
|
|
+ best_score = score
|
|
|
+ best_params = params
|
|
|
+ best_metrics = metrics
|
|
|
+
|
|
|
+ return best_params, best_metrics
|
|
|
+
|
|
|
+# ==================== 7. 可视化 ====================
|
|
|
+
|
|
|
+def plot_results(results, title="Backtest Results"):
|
|
|
+ """绘制回测结果"""
|
|
|
+ fig, axes = plt.subplots(3, 1, figsize=(12, 10))
|
|
|
+
|
|
|
+ # 净值曲线
|
|
|
+ ax1 = axes[0]
|
|
|
+ ax1.plot(results.index, results['nav'], label='Strategy', linewidth=2)
|
|
|
+ ax1.plot(results.index, results['index_nav'], label='Index', linewidth=1, alpha=0.7)
|
|
|
+ ax1.set_title(f'{title} - NAV')
|
|
|
+ ax1.set_ylabel('NAV')
|
|
|
+ ax1.legend()
|
|
|
+ ax1.grid(True, alpha=0.3)
|
|
|
+
|
|
|
+ # 仓位变化
|
|
|
+ ax2 = axes[1]
|
|
|
+ ax2.fill_between(results.index, 0, results['position'], alpha=0.3, label='Position')
|
|
|
+ ax2.set_ylabel('Position')
|
|
|
+ ax2.set_ylim(0, 1.1)
|
|
|
+ ax2.legend()
|
|
|
+ ax2.grid(True, alpha=0.3)
|
|
|
+
|
|
|
+ # 回撤
|
|
|
+ ax3 = axes[2]
|
|
|
+ running_max = results['nav'].expanding().max()
|
|
|
+ drawdown = (results['nav'] - running_max) / running_max
|
|
|
+ ax3.fill_between(results.index, drawdown, 0, alpha=0.3, color='red')
|
|
|
+ ax3.set_ylabel('Drawdown')
|
|
|
+ ax3.set_xlabel('Date')
|
|
|
+ ax3.grid(True, alpha=0.3)
|
|
|
+
|
|
|
+ plt.tight_layout()
|
|
|
+ return fig
|
|
|
+
|
|
|
+# ==================== 8. 主程序 ====================
|
|
|
+
|
|
|
+def main():
|
|
|
+ print("="*60)
|
|
|
+ print("创业板50指数量化交易策略回测")
|
|
|
+ print("="*60)
|
|
|
+
|
|
|
+ # 加载数据(使用模拟数据演示)
|
|
|
+ print("\n[1] 生成模拟数据...")
|
|
|
+ data = generate_mock_data('2017-01-01', '2025-12-31')
|
|
|
+ print(f"数据区间: {data.index[0]} ~ {data.index[-1]}, 共{len(data)}个交易日")
|
|
|
+
|
|
|
+ # 划分训练集和验证集
|
|
|
+ train_end = '2023-12-31'
|
|
|
+ val_start = '2024-01-01'
|
|
|
+
|
|
|
+ # 训练阶段(参数优化)
|
|
|
+ print("\n[2] 训练阶段:参数优化 (2018-2023)...")
|
|
|
+ best_params, train_metrics = grid_search(data, None)
|
|
|
+
|
|
|
+ print(f"\n最优参数:")
|
|
|
+ for k, v in best_params.items():
|
|
|
+ print(f" {k}: {v}")
|
|
|
+
|
|
|
+ print(f"\n训练集表现 (2018-2023):")
|
|
|
+ print(f" 策略年化收益: {train_metrics['annual_return']*100:.2f}%")
|
|
|
+ print(f" 指数年化收益: {train_metrics['index_annual']*100:.2f}%")
|
|
|
+ print(f" 超额收益: {train_metrics['excess_annual']*100:.2f}%")
|
|
|
+ print(f" 最大回撤: {train_metrics['max_drawdown']*100:.2f}%")
|
|
|
+ print(f" 夏普比率: {train_metrics['sharpe']:.2f}")
|
|
|
+ print(f" 卡玛比率: {train_metrics['calmar']:.2f}")
|
|
|
+ print(f" 胜率: {train_metrics['win_rate']*100:.1f}%")
|
|
|
+ print(f" Beta: {train_metrics['beta']:.2f}")
|
|
|
+
|
|
|
+ # 使用最优参数回测训练集(获取完整结果)
|
|
|
+ strategy = CYB50Strategy(best_params)
|
|
|
+ train_results, _ = backtest(data, strategy, start_date='2018-02-01', end_date=train_end)
|
|
|
+
|
|
|
+ # 验证阶段(样本外)
|
|
|
+ print(f"\n[3] 验证阶段:样本外测试 (2024-2025)...")
|
|
|
+ strategy_val = CYB50Strategy(best_params)
|
|
|
+ val_results, val_metrics = backtest(data, strategy_val, start_date=val_start, end_date='2025-12-31')
|
|
|
+
|
|
|
+ print(f"\n验证集表现 (2024-2025):")
|
|
|
+ print(f" 策略年化收益: {val_metrics['annual_return']*100:.2f}%")
|
|
|
+ print(f" 指数年化收益: {val_metrics['index_annual']*100:.2f}%")
|
|
|
+ print(f" 超额收益: {val_metrics['excess_annual']*100:.2f}%")
|
|
|
+ print(f" 最大回撤: {val_metrics['max_drawdown']*100:.2f}%")
|
|
|
+ print(f" 夏普比率: {val_metrics['sharpe']:.2f}")
|
|
|
+ print(f" 卡玛比率: {val_metrics['calmar']:.2f}")
|
|
|
+
|
|
|
+ # 过拟合检测
|
|
|
+ sharpe_decay = (train_metrics['sharpe'] - val_metrics['sharpe']) / train_metrics['sharpe'] if train_metrics['sharpe'] != 0 else 0
|
|
|
+ print(f"\n[4] 过拟合检测:")
|
|
|
+ print(f" 夏普比率衰减: {sharpe_decay*100:.1f}%")
|
|
|
+ if sharpe_decay > 0.5:
|
|
|
+ print(" ⚠️ 警告:可能存在严重过拟合")
|
|
|
+ elif sharpe_decay > 0.3:
|
|
|
+ print(" ⚠️ 注意:轻度过拟合,建议简化参数")
|
|
|
+ else:
|
|
|
+ print(" ✓ 无过拟合,策略稳健")
|
|
|
+
|
|
|
+ # 保存结果
|
|
|
+ print(f"\n[5] 保存结果...")
|
|
|
+ train_results.to_csv('train_results.csv')
|
|
|
+ val_results.to_csv('val_results.csv')
|
|
|
+ print(" 训练集结果: train_results.csv")
|
|
|
+ print(" 验证集结果: val_results.csv")
|
|
|
+
|
|
|
+ # 绘图
|
|
|
+ print(f"\n[6] 生成图表...")
|
|
|
+ fig1 = plot_results(train_results, "Training Set (2018-2023)")
|
|
|
+ fig1.savefig('train_backtest.png', dpi=150, bbox_inches='tight')
|
|
|
+ print(" 训练集图表: train_backtest.png")
|
|
|
+
|
|
|
+ fig2 = plot_results(val_results, "Validation Set (2024-2025)")
|
|
|
+ fig2.savefig('val_backtest.png', dpi=150, bbox_inches='tight')
|
|
|
+ print(" 验证集图表: val_backtest.png")
|
|
|
+
|
|
|
+ print("\n" + "="*60)
|
|
|
+ print("回测完成")
|
|
|
+ print("="*60)
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ main()
|