cyb50-quant/market-regime-identifier/train_and_validate.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
市场环境识别器 - 训练与验证脚本
使用2017-2023年数据训练，2024-2025年数据验证
"""

import numpy as np
import pandas as pd
import sys
from pathlib import Path

PROJECT_DIR = Path(__file__).resolve().parent
if str(PROJECT_DIR) not in sys.path:
    sys.path.insert(0, str(PROJECT_DIR))

from market_regime_hmm import (
    MarketRegimeHMM, 
    StrategySelector, 
    extract_features, 
    evaluate_model,
    calculate_hurst,
    calculate_rsi
)
from hmmlearn.hmm import GaussianHMM
import warnings
warnings.filterwarnings('ignore')

# 尝试导入数据获取库
try:
    import akshare as ak
    HAS_AKSHARE = True
except:
    HAS_AKSHARE = False
    print("警告: akshare未安装，将使用示例数据")


def fetch_index_data(index_code="sz399673", start_date="20170101", end_date="20251231"):
    """获取指数数据"""
    if HAS_AKSHARE:
        try:
            df = ak.index_zh_a_hist(symbol=index_code, period="daily", 
                                   start_date=start_date, end_date=end_date)
            df['date'] = pd.to_datetime(df['日期'])
            df = df.set_index('date').sort_index()
            df = df.rename(columns={
                '开盘': 'open',
                '收盘': 'close',
                '最高': 'high',
                '最低': 'low',
                '成交量': 'volume'
            })
            return df[['open', 'high', 'low', 'close', 'volume']]
        except Exception as e:
            print(f"数据获取失败: {e}")
            return None
    return None


def generate_synthetic_data(n_days=2000, seed=42):
    """
    生成合成数据用于演示
    模拟三种市场状态：趋势、震荡、反转
    """
    np.random.seed(seed)
    dates = pd.date_range('2017-01-01', periods=n_days, freq='B')
    
    price = 1000
    prices = []
    true_states = []  # 记录真实状态用于验证
    
    for i in range(n_days):
        # 模拟三种状态切换
        cycle_idx = (i // 200) % 3
        day_in_segment = i % 200

        if cycle_idx == 0:  # 趋势上涨
            price *= (1 + np.random.normal(0.001, 0.012))
            true_states.append(1)
        elif cycle_idx == 1:  # 震荡
            price *= (1 + np.random.normal(0, 0.015))
            true_states.append(0)
        else:  # 反转：前半段单边，后半段反向，形成真正的拐点
            if day_in_segment < 100:
                ret = np.random.normal(-0.0016, 0.016)
            else:
                ret = np.random.normal(0.0016, 0.016)
            price *= (1 + ret)
            true_states.append(2)

        prices.append(price)
    
    df = pd.DataFrame({
        'open': np.array(prices) + np.random.normal(0, 2, n_days),
        'high': np.array(prices) + np.abs(np.random.normal(5, 2, n_days)),
        'low': np.array(prices) - np.abs(np.random.normal(5, 2, n_days)),
        'close': prices,
        'volume': np.random.randint(1000000, 5000000, n_days),
        'true_state': true_states
    }, index=dates)
    
    return df


def train_and_validate():
    """训练与验证主程序"""
    print("="*70)
    print("市场环境识别器 - 训练与验证")
    print("="*70)
    
    # 获取数据
    print("\n[1/5] 获取数据...")
    df = fetch_index_data()
    
    if df is None:
        print("使用合成数据演示...")
        df = generate_synthetic_data(n_days=2000)
        df['true_state'] = None  # 移除真实状态标记
        using_synthetic = True
    else:
        using_synthetic = False
        print(f"获取到真实数据: {len(df)}条")
    
    # 划分训练集和验证集
    # 训练集: 2017-2023年 (约1500天)
    # 验证集: 2024-2025年 (约500天)
    split_date = '2024-01-01'
    
    if using_synthetic:
        # 合成数据前75%训练，后25%验证
        split_idx = int(len(df) * 0.75)
        train_df = df.iloc[:split_idx].copy()
        test_df = df.iloc[split_idx:].copy()
    else:
        train_df = df[df.index < split_date].copy()
        test_df = df[df.index >= split_date].copy()
    
    print(f"训练集: {len(train_df)}天 ({train_df.index[0].date()} ~ {train_df.index[-1].date()})")
    print(f"验证集: {len(test_df)}天 ({test_df.index[0].date()} ~ {test_df.index[-1].date()})")
    
    # 特征提取
    print("\n[2/5] 特征提取...")
    train_features = extract_features(train_df)
    test_features = extract_features(test_df)
    
    # 选择核心特征
    feature_cols = ['ret_std_5', 'momentum_10', 'vol_ratio', 'volume_change', 'intraday_trend']
    X_train = train_features[feature_cols].dropna()
    X_test = test_features[feature_cols].dropna()
    
    print(f"训练特征: {X_train.shape}")
    print(f"验证特征: {X_test.shape}")
    
    # 训练HMM模型
    print("\n[3/5] 训练HMM模型...")
    hmm = MarketRegimeHMM(n_components=3, n_iter=200)
    hmm.fit(X_train)
    
    # 验证模型
    print("\n[4/5] 模型评估...")
    print("\n--- 训练集评估 ---")
    train_results = evaluate_model(hmm, X_train)
    
    print("\n--- 验证集评估 ---")
    test_results = evaluate_model(hmm, X_test)
    
    # 验证准确率（如果有真实状态标签）
    if not using_synthetic and 'true_state' in df.columns:
        print("\n[5/5] 准确率验证...")
        # 这里可以添加与人工标注或基准的对比
        pass
    else:
        print("\n[5/5] 状态合理性检查...")
        
        # 检查状态与价格行为的对应关系
        test_states = test_results['states']
        test_df_aligned = test_df.iloc[-len(test_states):].copy()
        test_df_aligned['state'] = test_states
        
        # 计算各状态下的平均收益率
        for state_id, state_name in hmm.STATE_NAMES.items():
            mask = test_states == state_id
            if mask.any():
                state_returns = test_df_aligned[mask]['close'].pct_change().mean() * 100
                state_volatility = test_df_aligned[mask]['close'].pct_change().std() * 100
                print(f"\n{state_name}状态:")
                print(f"  平均日收益率: {state_returns:.3f}%")
                print(f"  波动率: {state_volatility:.3f}%")
                print(f"  出现天数: {mask.sum()}")
        
        # 验证逻辑:
        # 1. 趋势状态应该有较高的绝对收益率
        # 2. 震荡状态应该有较低的波动率变化
        # 3. 反转状态应该在高RSI后出现负收益
        
        print("\n" + "="*70)
        print("验证结果分析")
        print("="*70)
        
        # 计算各状态识别质量指标
        trend_returns = []
        range_returns = []
        reversal_returns = []
        
        for i in range(len(test_states)):
            if i > 0:
                ret = test_df_aligned['close'].iloc[i] / test_df_aligned['close'].iloc[i-1] - 1
                if test_states[i] == 1:  # 趋势
                    trend_returns.append(abs(ret))
                elif test_states[i] == 0:  # 震荡
                    range_returns.append(abs(ret))
                elif test_states[i] == 2:  # 反转
                    reversal_returns.append(abs(ret))
        
        if trend_returns and range_returns and reversal_returns:
            print(f"趋势状态平均绝对收益: {np.mean(trend_returns)*100:.3f}%")
            print(f"震荡状态平均绝对收益: {np.mean(range_returns)*100:.3f}%")
            print(f"反转状态平均绝对收益: {np.mean(reversal_returns)*100:.3f}%")
            
            # 简单的合理性检查
            checks_passed = 0
            checks_total = 2
            
            if np.mean(trend_returns) > np.mean(range_returns):
                print("[OK] 趋势状态收益 > 震荡状态收益")
                checks_passed += 1
            else:
                print("[ERR] 趋势状态收益应 > 震荡状态收益")
            
            if len([s for s in test_states if s == 1]) > len(test_states) * 0.1:
                print("[OK] 趋势状态出现频率合理 (>10%)")
                checks_passed += 1
            else:
                print("[ERR] 趋势状态出现频率过低")
            
            accuracy = (checks_passed / checks_total) * 100
            print(f"\n状态识别合理性: {accuracy:.0f}% ({checks_passed}/{checks_total})")
            
            if accuracy >= 50:  # 实际使用时要求72%
                print("[OK] 通过基本验证")
            else:
                print("[ERR] 需要重新训练")
    
    # 当前状态
    print("\n" + "="*70)
    print("当前市场状态")
    print("="*70)
    current_regime = hmm.get_current_regime(X_test)
    print(f"状态: {current_regime['state_name']}")
    print(f"置信度: {current_regime['confidence']:.2%}")
    
    strategy = StrategySelector.get_strategy(current_regime['state'])
    print(f"\n推荐策略: {strategy['name']}")
    print(f"仓位建议: {strategy['position_size']*100:.0f}%")
    
    # 保存模型
    print("\n[保存模型...]")
    import pickle
    model_path = PROJECT_DIR / 'hmm_model.pkl'
    with open(model_path, 'wb') as f:
        pickle.dump(hmm, f)
    print(f"模型已保存: {model_path}")
    
    # 保存特征统计
    feature_stats = {
        'feature_cols': feature_cols,
        'train_mean': X_train.mean().to_dict(),
        'train_std': X_train.std().to_dict()
    }
    stats_path = PROJECT_DIR / 'feature_stats.pkl'
    with open(stats_path, 'wb') as f:
        pickle.dump(feature_stats, f)
    print(f"特征统计已保存: {stats_path}")
    
    print("\n" + "="*70)
    print("训练完成!")
    print("="*70)


if __name__ == "__main__":
    train_and_validate()