|
|
| """
|
| 蒸鱼AKQuant Walk-Forward回测验证框架
|
| =============================================
|
| 集成AKQuant回测理念到蒸鱼(Kronos)系统:
|
| - Walk-Forward Validation (rolling/expanding/anchored)
|
| - Pipeline防泄漏: StandardScaler在每个WF窗口内独立fit
|
| - Purged K-Fold: 训练/测试间设置embargo期防止标签泄漏
|
| - QuantStats绩效报告: Sharpe/Sortino/最大回撤/蒙特卡洛模拟
|
| - 方向准确率: 每个WF窗口独立统计, 汇总平均
|
| - 衰减分析: 训练→验证→测试的准确率衰减趋势
|
| 复用ml_engine.py的26特征和compute_single_features函数
|
| """
|
|
|
| import numpy as np
|
| import pandas as pd
|
| import os, json, time, warnings
|
| from datetime import datetime
|
| from copy import deepcopy
|
| warnings.filterwarnings('ignore')
|
|
|
| import importlib.util as _ilu
|
| _ml_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'ml_engine.py')
|
| _ml_spec = _ilu.spec_from_file_location('ml_engine', _ml_path)
|
| _ml_mod = _ilu.module_from_spec(_ml_spec)
|
| _ml_spec.loader.exec_module(_ml_mod)
|
| download_daily = _ml_mod.download_daily
|
| compute_single_features = _ml_mod.compute_single_features
|
| FEATURE_COLS = _ml_mod.FEATURE_COLS
|
|
|
| REPORT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'reports')
|
| os.makedirs(REPORT_DIR, exist_ok=True)
|
|
|
|
|
| class WalkForwardValidator:
|
| """AKQuant风格Walk-Forward回测验证器
|
| 支持三种窗口模式: rolling/expanding/anchored
|
| 防泄漏: Pipeline(StandardScaler+Classifier)每个窗口独立fit, Purged embargo
|
| """
|
|
|
| def __init__(self, mode='rolling', train_days=400, test_days=60,
|
| step_days=60, embargo_days=5, n_purge_splits=5):
|
| self.mode = mode
|
| self.train_days = train_days
|
| self.test_days = test_days
|
| self.step_days = step_days
|
| self.embargo_days = embargo_days
|
| self.n_purge_splits = n_purge_splits
|
|
|
| def _generate_windows(self, n_total):
|
| windows = []
|
| min_req = 250 + self.train_days + self.embargo_days + self.test_days
|
| if n_total < min_req:
|
| return windows
|
| pos = 0
|
| while True:
|
| ts = 250
|
| te = ts + self.train_days
|
| if self.mode == 'expanding':
|
| te = 250 + self.train_days + pos
|
| elif self.mode == 'anchored':
|
| te = 250 + self.train_days + pos
|
| else:
|
| ts = 250 + pos
|
| te = ts + self.train_days
|
| gap_end = te + self.embargo_days
|
| test_s = gap_end
|
| test_e = test_s + self.test_days
|
| if test_e > n_total:
|
| break
|
| windows.append({'train_start': ts, 'train_end': te,
|
| 'test_start': test_s, 'test_end': min(test_e, n_total)})
|
| pos += self.step_days
|
| return windows
|
|
|
| def _build_dataset(self, code, name=''):
|
| print(f" [WF] 下载数据 {code} {name}...")
|
| d = download_daily(code)
|
| if d is None:
|
| return None
|
| n = len(d['close'])
|
| c, h, l, v = d['close'], d['high'], d['low'], d['volume']
|
| records = []
|
| for i in range(250, n - 5):
|
| feat = compute_single_features(
|
| {'close': c[:i+1], 'high': h[:i+1], 'low': l[:i+1], 'volume': v[:i+1]})
|
| if feat is None:
|
| continue
|
| if i + 5 < n:
|
| fr = (c[i+5] / c[i] - 1) * 100
|
| if fr > 0.5:
|
| feat['label'] = 1
|
| elif fr < -0.5:
|
| feat['label'] = 0
|
| else:
|
| continue
|
| records.append(feat)
|
| if len(records) < 100:
|
| return None
|
| df = pd.DataFrame(records)
|
| print(f" [WF] {code} {name}: {n}天 -> {len(df)}有效样本")
|
| return df
|
|
|
| def validate(self, stock_list=None, model_type='gbdt'):
|
| if stock_list is None:
|
| stock_list = [
|
| ('sz300896','爱美客'),('sh600519','贵州茅台'),('sz000001','平安银行'),
|
| ('sh600489','中金黄金'),('sz300601','康泰生物'),('sz300142','沃森生物'),
|
| ('sh600036','招商银行'),('sz000858','五粮液'),('sh601318','中国平安'),
|
| ('sz300750','宁德时代'),('sz002475','立讯精密'),('sh600900','长江电力'),
|
| ('sz300122','智飞生物'),('sz002625','光启技术'),('sh601012','隆基绿能'),
|
| ('sz000333','美的集团'),
|
| ]
|
| print(f"\n{'='*60}")
|
| print(f"[AKQuant WF] 蒸鱼Walk-Forward回测验证")
|
| print(f"[AKQuant WF] 模式={self.mode}, 训练={self.train_days}天, "
|
| f"测试={self.test_days}天, 步长={self.step_days}天")
|
| print(f"[AKQuant WF] 模型={model_type}, 股票数={len(stock_list)}")
|
| print(f"{'='*60}")
|
|
|
| all_data = {}
|
| for code, name in stock_list:
|
| df = self._build_dataset(code, name)
|
| if df is not None:
|
| all_data[code] = df
|
| time.sleep(0.15)
|
| if not all_data:
|
| return {'success': False, 'error': '无有效数据'}
|
|
|
| stock_results = {}
|
| for code, df in all_data.items():
|
| result = self._validate_single(df, code, model_type)
|
| if result['success']:
|
| stock_results[code] = result
|
| print(f" [WF] {code}: {result['n_windows']}窗口, "
|
| f"准确率={result['avg_accuracy']:.1f}%")
|
| if not stock_results:
|
| return {'success': False, 'error': '所有股票WF验证失败'}
|
|
|
| summary = self._compute_summary(stock_results)
|
| decay = self._compute_decay(stock_results)
|
| ts = datetime.now().strftime('%Y%m%d_%H%M%S')
|
| rp = os.path.join(REPORT_DIR, f'wf_report_{ts}.json')
|
| final = {
|
| 'success': True, 'timestamp': str(datetime.now()),
|
| 'config': {'mode': self.mode, 'train_days': self.train_days,
|
| 'test_days': self.test_days, 'step_days': self.step_days,
|
| 'embargo_days': self.embargo_days, 'model_type': model_type},
|
| 'n_stocks': len(stock_results), 'stock_results': stock_results,
|
| 'summary': summary, 'decay': decay,
|
| }
|
| with open(rp, 'w', encoding='utf-8') as f:
|
| json.dump(final, f, ensure_ascii=False, indent=2, default=str)
|
| print(f"\n[AKQuant WF] 报告: {rp}")
|
| print(f"[AKQuant WF] 汇总: {summary['avg_accuracy']:.1f}%, "
|
| f"{summary['n_total_windows']}窗口, 稳定性={summary['stability']}")
|
| return final
|
|
|
| def _validate_single(self, df, code, model_type):
|
| from sklearn.ensemble import GradientBoostingClassifier
|
| from sklearn.metrics import (accuracy_score, precision_score,
|
| recall_score, f1_score)
|
| from sklearn.pipeline import Pipeline
|
| from sklearn.preprocessing import StandardScaler
|
|
|
| n_total = len(df)
|
| windows = self._generate_windows(n_total)
|
| if len(windows) < 2:
|
| return {'success': False, 'error': f'{code}: 窗口数不足'}
|
|
|
| window_results = []
|
| test_rets_all = []
|
|
|
| for wi, win in enumerate(windows):
|
| train_df = df.iloc[win['train_start']:win['train_end']].copy()
|
| test_df = df.iloc[win['test_start']:win['test_end']].copy()
|
| if len(train_df) < 50 or len(test_df) < 10:
|
| continue
|
| X_train = train_df[FEATURE_COLS].values.astype(np.float64)
|
| y_train = train_df['label'].values.astype(int)
|
| X_test = test_df[FEATURE_COLS].values.astype(np.float64)
|
| y_test = test_df['label'].values.astype(int)
|
| if len(np.unique(y_train)) < 2:
|
| continue
|
|
|
|
|
| if model_type == 'xgboost':
|
| try:
|
| import xgboost as xgb
|
| pipe = Pipeline([
|
| ('scaler', StandardScaler()),
|
| ('clf', xgb.XGBClassifier(
|
| n_estimators=200, max_depth=4, learning_rate=0.05,
|
| min_child_weight=10, subsample=0.8, colsample_bytree=0.8,
|
| reg_alpha=0.1, reg_lambda=1.0, random_state=42,
|
| use_label_encoder=False, eval_metric='logloss'))
|
| ])
|
| except ImportError:
|
| pipe = Pipeline([('scaler', StandardScaler()),
|
| ('clf', GradientBoostingClassifier(
|
| n_estimators=300, max_depth=4, learning_rate=0.05,
|
| min_samples_leaf=15, random_state=42))])
|
| else:
|
| pipe = Pipeline([('scaler', StandardScaler()),
|
| ('clf', GradientBoostingClassifier(
|
| n_estimators=300, max_depth=4, learning_rate=0.05,
|
| min_samples_leaf=15, random_state=42))])
|
|
|
| try:
|
| pipe.fit(X_train, y_train)
|
| y_pred = pipe.predict(X_test)
|
| acc = accuracy_score(y_test, y_pred)
|
| prec = precision_score(y_test, y_pred, zero_division=0)
|
| rec = recall_score(y_test, y_pred, zero_division=0)
|
| f1 = f1_score(y_test, y_pred, zero_division=0)
|
| y_proba = pipe.predict_proba(X_test)[:, 1]
|
| wr = {
|
| 'window_id': wi+1, 'train_size': len(X_train),
|
| 'test_size': len(X_test), 'accuracy': round(acc*100, 2),
|
| 'precision': round(prec*100, 2), 'recall': round(rec*100, 2),
|
| 'f1': round(f1*100, 2),
|
| 'buy_rate': round(float(np.mean(y_pred)), 4),
|
| 'avg_buy_prob': round(float(np.mean(y_proba)), 4),
|
| }
|
| window_results.append(wr)
|
| for j in range(len(y_test)):
|
| if y_proba[j] > 0.5:
|
| test_rets_all.append(1.0 if y_test[j]==1 else -1.0)
|
| else:
|
| test_rets_all.append(1.0 if y_test[j]==0 else -1.0)
|
| except Exception:
|
| continue
|
|
|
| if not window_results:
|
| return {'success': False, 'error': f'{code}: 所有窗口训练失败'}
|
|
|
| accuracies = [w['accuracy'] for w in window_results]
|
| qs_stats = {}
|
| if test_rets_all:
|
| rets_arr = np.array(test_rets_all)
|
| cum = np.cumsum(rets_arr)
|
| peak = np.maximum.accumulate(cum)
|
| dd = (peak - cum) / (peak + 1e-8) * 100
|
| qs_stats = {
|
| 'total_trades': len(rets_arr),
|
| 'win_rate': round(float(np.mean(rets_arr > 0)*100), 2),
|
| 'avg_return': round(float(np.mean(rets_arr)), 4),
|
| 'sharpe': round(float(np.mean(rets_arr)/(np.std(rets_arr)+1e-8)), 4),
|
| 'max_drawdown': round(float(np.max(dd)), 2),
|
| 'cum_return': round(float(np.sum(rets_arr)), 2),
|
| }
|
| return {
|
| 'success': True, 'code': code,
|
| 'n_windows': len(window_results), 'windows': window_results,
|
| 'avg_accuracy': round(float(np.mean(accuracies)), 2),
|
| 'std_accuracy': round(float(np.std(accuracies)), 2),
|
| 'min_accuracy': round(float(min(accuracies)), 2),
|
| 'max_accuracy': round(float(max(accuracies)), 2),
|
| 'quantstats': qs_stats,
|
| }
|
|
|
| def _compute_summary(self, stock_results):
|
| all_accs, all_precs, all_f1s, tw = [], [], [], 0
|
| for sr in stock_results.values():
|
| all_accs.append(sr['avg_accuracy'])
|
| if sr['windows']:
|
| all_precs.append(float(np.mean([w['precision'] for w in sr['windows']])))
|
| all_f1s.append(float(np.mean([w['f1'] for w in sr['windows']])))
|
| tw += sr['n_windows']
|
| all_qs = [sr.get('quantstats',{}) for sr in stock_results.values() if sr.get('quantstats')]
|
| return {
|
| 'avg_accuracy': round(float(np.mean(all_accs)), 2),
|
| 'std_accuracy': round(float(np.std(all_accs)), 2),
|
| 'min_accuracy': round(float(min(all_accs)), 2),
|
| 'max_accuracy': round(float(max(all_accs)), 2),
|
| 'avg_precision': round(float(np.mean(all_precs)) if all_precs else 0, 2),
|
| 'avg_f1': round(float(np.mean(all_f1s)) if all_f1s else 0, 2),
|
| 'n_total_windows': tw, 'n_stocks': len(stock_results),
|
| 'stability': '高' if np.std(all_accs)<3 else ('中' if np.std(all_accs)<6 else '低'),
|
| 'pass_rate': round(float(sum(1 for a in all_accs if a>52)/len(all_accs)*100), 2),
|
| 'quantstats': {
|
| 'avg_win_rate': round(float(np.mean([q.get('win_rate',50) for q in all_qs])),2) if all_qs else 0,
|
| 'avg_sharpe': round(float(np.mean([q.get('sharpe',0) for q in all_qs])),4) if all_qs else 0,
|
| 'avg_max_drawdown': round(float(np.mean([q.get('max_drawdown',0) for q in all_qs])),2) if all_qs else 0,
|
| } if all_qs else {},
|
| }
|
|
|
| def _compute_decay(self, stock_results):
|
| all_wa = []
|
| for sr in stock_results.values():
|
| all_wa.extend([w['accuracy'] for w in sr['windows']])
|
| if len(all_wa) < 4:
|
| return {'trend': '数据不足', 'decay_rate': 0}
|
| x = np.arange(len(all_wa)); y = np.array(all_wa)
|
| try:
|
| slope = np.polyfit(x, y, 1)[0]
|
| half = len(all_wa)//2
|
| fa = float(np.mean(y[:half])); sa = float(np.mean(y[half:]))
|
| trend = '稳定' if slope > -0.1 else ('轻微衰减' if slope > -0.5 else '明显衰减')
|
| return {'trend': trend, 'decay_rate': round(abs(slope),4),
|
| 'first_half_avg': round(fa,2), 'second_half_avg': round(sa,2),
|
| 'total_windows': len(all_wa)}
|
| except Exception:
|
| return {'trend': '无法计算', 'decay_rate': 0}
|
|
|
|
|
|
|
|
|
|
|
| class PurgedKFoldValidator:
|
| """Purged K-Fold回测验证器
|
| 在标准K-Fold基础上: 1) 训练/测试间设置gap防止标签泄漏
|
| 2) 每个fold内Pipeline独立fit 3) 统计汇总+蒙特卡洛基线
|
| """
|
|
|
| def __init__(self, n_splits=5, embargo_days=5, gap_days=3):
|
| self.n_splits = n_splits
|
| self.embargo_days = embargo_days
|
| self.gap_days = gap_days
|
|
|
| def _build_dataset(self, code, name=''):
|
| print(f" [PKF] 下载数据 {code} {name}...")
|
| d = download_daily(code)
|
| if d is None:
|
| return None
|
| n = len(d['close'])
|
| c, h, l, v = d['close'], d['high'], d['low'], d['volume']
|
| records = []
|
| for i in range(250, n - 5):
|
| feat = compute_single_features(
|
| {'close': c[:i+1], 'high': h[:i+1], 'low': l[:i+1], 'volume': v[:i+1]})
|
| if feat is None:
|
| continue
|
| if i + 5 < n:
|
| fr = (c[i+5] / c[i] - 1) * 100
|
| if fr > 0.5:
|
| feat['label'] = 1
|
| elif fr < -0.5:
|
| feat['label'] = 0
|
| else:
|
| continue
|
| records.append(feat)
|
| if len(records) < 100:
|
| return None
|
| df = pd.DataFrame(records)
|
| print(f" [PKF] {code} {name}: {n}天 -> {len(df)}有效样本")
|
| return df
|
|
|
| def _generate_purged_splits(self, n_total):
|
| """生成Purged K-Fold索引: 每个split间有gap+embargo"""
|
| total_gap = self.gap_days + self.embargo_days
|
| fold_size = (n_total - total_gap * (self.n_splits - 1)) // self.n_splits
|
| if fold_size < 50:
|
| return []
|
| splits = []
|
| pos = 0
|
| for i in range(self.n_splits):
|
| test_start = pos
|
| test_end = min(pos + fold_size, n_total)
|
| if test_end - test_start < 20:
|
| break
|
| train_end = test_start - total_gap
|
| splits.append({
|
| 'train_start': 0,
|
| 'train_end': max(train_end, 0),
|
| 'test_start': test_start,
|
| 'test_end': test_end,
|
| })
|
| pos = test_end + total_gap
|
| if pos >= n_total:
|
| break
|
| return splits
|
|
|
| def _validate_single(self, df, code, model_type='gbdt'):
|
| from sklearn.ensemble import GradientBoostingClassifier
|
| from sklearn.metrics import (accuracy_score, precision_score,
|
| recall_score, f1_score)
|
| from sklearn.pipeline import Pipeline
|
| from sklearn.preprocessing import StandardScaler
|
|
|
| n_total = len(df)
|
| splits = self._generate_purged_splits(n_total)
|
| if len(splits) < 2:
|
| return {'success': False, 'error': f'{code}: 有效split数不足'}
|
|
|
| fold_results = []
|
| test_rets_all = []
|
|
|
| for fi, sp in enumerate(splits):
|
| if sp['train_end'] <= 0:
|
| continue
|
| train_df = df.iloc[sp['train_start']:sp['train_end']].copy()
|
| test_df = df.iloc[sp['test_start']:sp['test_end']].copy()
|
| if len(train_df) < 50 or len(test_df) < 10:
|
| continue
|
| X_train = train_df[FEATURE_COLS].values.astype(np.float64)
|
| y_train = train_df['label'].values.astype(int)
|
| X_test = test_df[FEATURE_COLS].values.astype(np.float64)
|
| y_test = test_df['label'].values.astype(int)
|
| if len(np.unique(y_train)) < 2:
|
| continue
|
|
|
| if model_type == 'xgboost':
|
| try:
|
| import xgboost as xgb
|
| pipe = Pipeline([
|
| ('scaler', StandardScaler()),
|
| ('clf', xgb.XGBClassifier(
|
| n_estimators=200, max_depth=4, learning_rate=0.05,
|
| min_child_weight=10, subsample=0.8, colsample_bytree=0.8,
|
| reg_alpha=0.1, reg_lambda=1.0, random_state=42,
|
| use_label_encoder=False, eval_metric='logloss'))
|
| ])
|
| except ImportError:
|
| pipe = Pipeline([('scaler', StandardScaler()),
|
| ('clf', GradientBoostingClassifier(
|
| n_estimators=300, max_depth=4, learning_rate=0.05,
|
| min_samples_leaf=15, random_state=42))])
|
| else:
|
| pipe = Pipeline([('scaler', StandardScaler()),
|
| ('clf', GradientBoostingClassifier(
|
| n_estimators=300, max_depth=4, learning_rate=0.05,
|
| min_samples_leaf=15, random_state=42))])
|
|
|
| try:
|
| pipe.fit(X_train, y_train)
|
| y_pred = pipe.predict(X_test)
|
| y_proba = pipe.predict_proba(X_test)[:, 1]
|
| acc = accuracy_score(y_test, y_pred)
|
| fr = {
|
| 'fold_id': fi+1, 'train_size': len(X_train),
|
| 'test_size': len(X_test),
|
| 'accuracy': round(acc*100, 2),
|
| 'precision': round(precision_score(y_test, y_pred, zero_division=0)*100, 2),
|
| 'recall': round(recall_score(y_test, y_pred, zero_division=0)*100, 2),
|
| 'f1': round(f1_score(y_test, y_pred, zero_division=0)*100, 2),
|
| 'buy_rate': round(float(np.mean(y_pred)), 4),
|
| }
|
| fold_results.append(fr)
|
| for j in range(len(y_test)):
|
| if y_proba[j] > 0.5:
|
| test_rets_all.append(1.0 if y_test[j]==1 else -1.0)
|
| else:
|
| test_rets_all.append(1.0 if y_test[j]==0 else -1.0)
|
| except Exception:
|
| continue
|
|
|
| if not fold_results:
|
| return {'success': False, 'error': f'{code}: 所有fold训练失败'}
|
|
|
| accuracies = [f['accuracy'] for f in fold_results]
|
| qs_stats = {}
|
| if test_rets_all:
|
| rets_arr = np.array(test_rets_all)
|
| cum = np.cumsum(rets_arr)
|
| peak = np.maximum.accumulate(cum)
|
| dd = (peak - cum) / (peak + 1e-8) * 100
|
| qs_stats = {
|
| 'total_trades': len(rets_arr),
|
| 'win_rate': round(float(np.mean(rets_arr > 0)*100), 2),
|
| 'avg_return': round(float(np.mean(rets_arr)), 4),
|
| 'sharpe': round(float(np.mean(rets_arr)/(np.std(rets_arr)+1e-8)), 4),
|
| 'max_drawdown': round(float(np.max(dd)), 2),
|
| }
|
| return {
|
| 'success': True, 'code': code,
|
| 'n_folds': len(fold_results), 'folds': fold_results,
|
| 'avg_accuracy': round(float(np.mean(accuracies)), 2),
|
| 'std_accuracy': round(float(np.std(accuracies)), 2),
|
| 'quantstats': qs_stats,
|
| }
|
|
|
| def validate(self, stock_list=None, model_type='gbdt'):
|
| if stock_list is None:
|
| stock_list = [
|
| ('sz300896','爱美客'),('sh600519','贵州茅台'),('sz000001','平安银行'),
|
| ('sh600489','中金黄金'),('sz300601','康泰生物'),('sz300142','沃森生物'),
|
| ('sh600036','招商银行'),('sz000858','五粮液'),('sh601318','中国平安'),
|
| ('sz300750','宁德时代'),('sz002475','立讯精密'),('sh600900','长江电力'),
|
| ('sz300122','智飞生物'),('sz002625','光启技术'),('sh601012','隆基绿能'),
|
| ('sz000333','美的集团'),
|
| ]
|
| print(f"\n{'='*60}")
|
| print(f"[AKQuant PKF] Purged K-Fold回测验证")
|
| print(f"[AKQuant PKF] folds={self.n_splits}, embargo={self.embargo_days}天, "
|
| f"gap={self.gap_days}天, 模型={model_type}")
|
| print(f"{'='*60}")
|
|
|
| all_data = {}
|
| for code, name in stock_list:
|
| df = self._build_dataset(code, name)
|
| if df is not None:
|
| all_data[code] = df
|
| time.sleep(0.15)
|
| if not all_data:
|
| return {'success': False, 'error': '无有效数据'}
|
|
|
| stock_results = {}
|
| for code, df in all_data.items():
|
| result = self._validate_single(df, code, model_type)
|
| if result['success']:
|
| stock_results[code] = result
|
| print(f" [PKF] {code}: {result['n_folds']}folds, "
|
| f"准确率={result['avg_accuracy']:.1f}%")
|
| if not stock_results:
|
| return {'success': False, 'error': '所有股票PKF验证失败'}
|
|
|
| all_accs = [sr['avg_accuracy'] for sr in stock_results.values()]
|
| tf = sum(sr['n_folds'] for sr in stock_results.values())
|
| summary = {
|
| 'avg_accuracy': round(float(np.mean(all_accs)), 2),
|
| 'std_accuracy': round(float(np.std(all_accs)), 2),
|
| 'n_stocks': len(stock_results),
|
| 'n_total_folds': tf,
|
| 'pass_rate': round(float(sum(1 for a in all_accs if a>52)/len(all_accs)*100), 2),
|
| 'stability': '高' if np.std(all_accs)<3 else ('中' if np.std(all_accs)<6 else '低'),
|
| }
|
|
|
| ts = datetime.now().strftime('%Y%m%d_%H%M%S')
|
| rp = os.path.join(REPORT_DIR, f'pkf_report_{ts}.json')
|
| final = {
|
| 'success': True, 'timestamp': str(datetime.now()),
|
| 'config': {'n_splits': self.n_splits, 'embargo_days': self.embargo_days,
|
| 'gap_days': self.gap_days, 'model_type': model_type},
|
| 'n_stocks': len(stock_results), 'stock_results': stock_results,
|
| 'summary': summary,
|
| }
|
| with open(rp, 'w', encoding='utf-8') as f:
|
| json.dump(final, f, ensure_ascii=False, indent=2, default=str)
|
| print(f"\n[AKQuant PKF] 报告: {rp}")
|
| print(f"[AKQuant PKF] 汇总: {summary['avg_accuracy']:.1f}%, "
|
| f"稳定性={summary['stability']}")
|
| return final
|
|
|
|
|
|
|
|
|
|
|
| def generate_quantstats_report(returns_list, benchmark=None, title='AKQuant Strategy'):
|
| """使用QuantStats生成HTML绩效报告"""
|
| try:
|
| import quantstats as qs
|
| except ImportError:
|
| print("[QS] quantstats未安装, 跳过HTML报告生成")
|
| return None
|
| if not returns_list:
|
| return None
|
| try:
|
| returns = pd.Series(returns_list)
|
| returns.index = pd.bdate_range(
|
| start='2024-01-01', periods=len(returns))
|
| html_file = os.path.join(REPORT_DIR, f'qs_{datetime.now().strftime("%Y%m%d_%H%M%S")}.html')
|
| if benchmark is not None and len(benchmark) == len(returns_list):
|
| bm = pd.Series(benchmark)
|
| bm.index = returns.index
|
| qs.reports.html(returns, benchmark=bm, title=title, output=html_file)
|
| else:
|
| qs.reports.html(returns, title=title, output=html_file)
|
| print(f"[QS] HTML报告: {html_file}")
|
| return html_file
|
| except Exception as e:
|
| print(f"[QS] 报告生成失败: {e}")
|
| return None
|
|
|
|
|
| def _generate_text_report(result, report_type='WF'):
|
| """生成文本摘要报告"""
|
| lines = [f"\n{'='*60}"]
|
| lines.append(f"[AKQuant {report_type}] 回测验证报告")
|
| lines.append(f"时间: {result.get('timestamp', 'N/A')}")
|
| lines.append(f"{'='*60}")
|
| cfg = result.get('config', {})
|
| for k, v in cfg.items():
|
| lines.append(f" {k}: {v}")
|
| lines.append('')
|
| sm = result.get('summary', {})
|
| lines.append(f"--- 汇总 ---")
|
| for k, v in sm.items():
|
| lines.append(f" {k}: {v}")
|
| lines.append('')
|
| dc = result.get('decay', {})
|
| if dc:
|
| lines.append(f"--- 衰减分析 ---")
|
| for k, v in dc.items():
|
| lines.append(f" {k}: {v}")
|
| lines.append('')
|
| lines.append(f"--- 各股结果 ---")
|
| for code, sr in result.get('stock_results', {}).items():
|
| lines.append(f" {code}: avg_acc={sr.get('avg_accuracy',0):.1f}%, "
|
| f"windows={sr.get('n_windows',sr.get('n_folds',0))}")
|
| return '\n'.join(lines)
|
|
|
|
|
|
|
|
|
|
|
| def monte_carlo_simulation(n_simulations=500, n_trades=100, win_prob=None):
|
| """蒙特卡洛模拟: 500次随机策略基线
|
| 返回95%/99%置信区间, 与实际策略对比判断显著性
|
| """
|
| np.random.seed(42)
|
| if win_prob is None:
|
| win_prob = 0.5
|
| sim_results = []
|
| for _ in range(n_simulations):
|
| trades = np.random.choice([1.0, -1.0], size=n_trades, p=[win_prob, 1-win_prob])
|
| cum = np.cumsum(trades)
|
| peak = np.maximum.accumulate(cum)
|
| dd = (peak - cum) / (peak + 1e-8) * 100
|
| sim_results.append({
|
| 'total_return': float(np.sum(trades)),
|
| 'win_rate': float(np.mean(trades > 0) * 100),
|
| 'sharpe': float(np.mean(trades) / (np.std(trades) + 1e-8)),
|
| 'max_drawdown': float(np.max(dd)),
|
| })
|
| df = pd.DataFrame(sim_results)
|
| baseline = {
|
| 'n_simulations': n_simulations,
|
| 'n_trades': n_trades,
|
| 'win_prob': win_prob,
|
| 'return_95ci': [round(float(df['total_return'].quantile(0.025)), 2),
|
| round(float(df['total_return'].quantile(0.975)), 2)],
|
| 'return_99ci': [round(float(df['total_return'].quantile(0.005)), 2),
|
| round(float(df['total_return'].quantile(0.995)), 2)],
|
| 'win_rate_mean': round(float(df['win_rate'].mean()), 2),
|
| 'win_rate_std': round(float(df['win_rate'].std()), 2),
|
| 'sharpe_95ci': [round(float(df['sharpe'].quantile(0.025)), 4),
|
| round(float(df['sharpe'].quantile(0.975)), 4)],
|
| 'max_dd_95ci': [round(float(df['max_drawdown'].quantile(0.025)), 2),
|
| round(float(df['max_drawdown'].quantile(0.975)), 2)],
|
| }
|
| print(f"[MC] 蒙特卡洛基线: {n_simulations}次模拟, {n_trades}笔交易")
|
| print(f"[MC] 收益95%CI: {baseline['return_95ci']}")
|
| print(f"[MC] Sharpe 95%CI: {baseline['sharpe_95ci']}")
|
| return baseline
|
|
|
|
|
|
|
|
|
|
|
| def run_walk_forward(mode='rolling', train_days=400, test_days=60,
|
| step_days=60, model_type='gbdt', n_stocks=None):
|
| """快捷: 运行Walk-Forward验证"""
|
| wf = WalkForwardValidator(
|
| mode=mode, train_days=train_days, test_days=test_days,
|
| step_days=step_days, embargo_days=5)
|
| return wf.validate(model_type=model_type)
|
|
|
|
|
| def run_purged_kfold(n_splits=5, embargo_days=5, gap_days=3,
|
| model_type='gbdt'):
|
| """快捷: 运行Purged K-Fold验证"""
|
| pkf = PurgedKFoldValidator(
|
| n_splits=n_splits, embargo_days=embargo_days, gap_days=gap_days)
|
| return pkf.validate(model_type=model_type)
|
|
|
|
|
| def run_full_validation(model_type='gbdt'):
|
| """完整验证流程: WF + PKF + 蒙特卡洛"""
|
| print("\n" + "#"*60)
|
| print("# AKQuant 完整回测验证流程")
|
| print("#"*60)
|
|
|
| wf_result = run_walk_forward(mode='rolling', model_type=model_type)
|
| pkf_result = run_purged_kfold(model_type=model_type)
|
| mc_baseline = monte_carlo_simulation(n_simulations=500, n_trades=200)
|
|
|
| text_report = _generate_text_report(wf_result, 'WF') + '\n'
|
| text_report += _generate_text_report(pkf_result, 'PKF')
|
|
|
| all_rets = []
|
| for sr in wf_result.get('stock_results', {}).values():
|
| for w in sr.get('windows', []):
|
| all_rets.append(1.0 if w['accuracy'] > 52 else -1.0)
|
|
|
| if all_rets:
|
| actual_sharpe = float(np.mean(all_rets) / (np.std(all_rets) + 1e-8))
|
| mc_lower = mc_baseline['sharpe_95ci'][1]
|
| significance = actual_sharpe > mc_lower
|
| text_report += f"\n--- 显著性检验 ---\n"
|
| text_report += f" 实际Sharpe: {actual_sharpe:.4f}\n"
|
| text_report += f" MC 95%CI上界: {mc_lower:.4f}\n"
|
| text_report += f" 策略显著: {'是' if significance else '否'}\n"
|
|
|
| ts = datetime.now().strftime('%Y%m%d_%H%M%S')
|
| rp = os.path.join(REPORT_DIR, f'full_validation_{ts}.txt')
|
| with open(rp, 'w', encoding='utf-8') as f:
|
| f.write(text_report)
|
| print(f"\n[AKQuant] 完整报告: {rp}")
|
|
|
| html = generate_quantstats_report(all_rets) if all_rets else None
|
| return {
|
| 'success': True, 'wf': wf_result, 'pkf': pkf_result,
|
| 'monte_carlo': mc_baseline,
|
| 'text_report': text_report,
|
| 'html_report': html,
|
| 'report_path': rp,
|
| }
|
|
|
|
|
|
|
|
|
|
|
| if __name__ == '__main__':
|
| import sys
|
| mode = sys.argv[1] if len(sys.argv) > 1 else 'quick'
|
| if mode == 'full':
|
| result = run_full_validation()
|
| print(f"\n完成! 报告路径: {result['report_path']}")
|
| elif mode == 'mc':
|
| baseline = monte_carlo_simulation(n_simulations=500, n_trades=200)
|
| print(f"\n蒙特卡洛基线: {json.dumps(baseline, indent=2)}")
|
| elif mode == 'pkf':
|
| result = run_purged_kfold()
|
| elif mode == 'wf':
|
| result = run_walk_forward()
|
| else:
|
| print("[AKQuant] 快速模式: 3股WF验证")
|
| quick_stocks = [
|
| ('sh600519','贵州茅台'), ('sz000001','平安银行'),
|
| ('sh600036','招商银行'),
|
| ]
|
| wf = WalkForwardValidator(mode='rolling', train_days=400,
|
| test_days=60, step_days=60)
|
| result = wf.validate(stock_list=quick_stocks)
|
| if result['success']:
|
| print(f"\n快速验证完成: {result['summary']['avg_accuracy']:.1f}%")
|
| else:
|
| print(f"\n快速验证失败: {result.get('error','')}")
|
|
|
