Blame for webui/akquant_wf.py

# -*- coding: utf-8 -*-
"""

蒸鱼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:  # rolling
                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

            # Pipeline防泄漏: 每个窗口独立fit
            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}


# ============================================================
# Purged K-Fold + Embargo 验证器
# ============================================================
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


# ============================================================
# QuantStats HTML报告生成
# ============================================================
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,
    }


# ============================================================
# __main__ 入口
# ============================================================
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:  # quick
        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','')}")