"""基于深度学习的ETF轮动策略实现目标：择时 + 仓位管理 + 控制回撤 + 提高夏普比率"""import numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport seaborn as snsfrom datetime import datetime, timedeltaimport warningswarnings.filterwarnings('ignore')# 深度学习相关库import torchimport torch.nn as nnimport torch.optim as optimfrom torch.utils.data import DataLoader, TensorDataset, Datasetimport torch.nn.functional as Ffrom sklearn.preprocessing import StandardScaler, MinMaxScalerfrom sklearn.model_selection import TimeSeriesSplit# 金融相关库import tushare as tsimport talib# 设置中文显示plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = False# 设置随机种子np.random.seed(42)torch.manual_seed(42)# ==================== 第一部分：数据获取与预处理 ====================class ETFDataLoader:    """ETF数据加载与预处理类"""    def __init__(self, token=None, start_date='2010-01-01', end_date=None):        """        初始化数据加载器        Args:            token: tushare token            start_date: 开始日期            end_date: 结束日期，默认为今天        """        if token:            self.pro = ts.pro_api(token)        else:            # 如果没有token，使用在线数据            self.pro = None        self.start_date = start_date        self.end_date = end_date or datetime.now().strftime('%Y-%m-%d')        # A股主要ETF列表        self.etf_list = {            '510300': '沪深300ETF',            '510500': '中证500ETF',             '159915': '创业板ETF',            '510050': '上证50ETF',            '512000': '券商ETF',            '512100': '1000ETF',            '512760': '半导体ETF',            '159919': '300ETF',            '588000': '科创50ETF',            '511010': '国债ETF'  # 防御性资产        }        self.features = {}    def fetch_etf_data(self, etf_code, save_local=True):        """        获取ETF数据        Args:            etf_code: ETF代码            save_local: 是否保存到本地        Returns:            DataFrame: ETF数据        """        print(f"正在获取{etf_code}数据...")        # 如果是tushare pro用户        if self.pro:            try:                df = self.pro.fund_daily(ts_code=f'{etf_code}.SH' if etf_code.startswith('51') else f'{etf_code}.SZ',                                         start_date=self.start_date,                                         end_date=self.end_date)                df = df.sort_values('trade_date')                df.index = pd.to_datetime(df['trade_date'])                df = df[['open', 'high', 'low', 'close', 'vol', 'amount']]                df.columns = ['open', 'high', 'low', 'close', 'volume', 'amount']            except:                # 回退到通用方法                df = self._fetch_from_web(etf_code)        else:            df = self._fetch_from_web(etf_code)        # 计算收益率        df['returns'] = df['close'].pct_change()        # 填充缺失值        df = df.fillna(method='ffill').fillna(method='bfill')        if save_local:            df.to_csv(f'etf_data_{etf_code}.csv')        return df    def _fetch_from_web(self, etf_code):        """从网络获取数据（备用方法）"""        try:            # 使用tushare通用接口            df = ts.get_k_data(etf_code, start=self.start_date, end=self.end_date)            df.index = pd.to_datetime(df['date'])            df = df[['open', 'high', 'low', 'close', 'volume']]            df.columns = ['open', 'high', 'low', 'close', 'volume']            return df        except:            # 模拟数据（用于测试）            print(f"无法获取{etf_code}真实数据，使用模拟数据...")            return self._generate_mock_data(etf_code)    def _generate_mock_data(self, etf_code):        """生成模拟数据用于测试"""        dates = pd.date_range(start=self.start_date, end=self.end_date, freq='D')        # 排除周末        dates = dates[dates.dayofweek < 5]        # 基础价格（不同ETF有不同的波动特性）        base_price = {            '510300': 5.0,  # 沪深300ETF            '510500': 6.0,  # 中证500ETF            '159915': 2.5,  # 创业板ETF            '510050': 3.0,  # 上证50ETF            '511010': 100.0, # 国债ETF        }.get(etf_code, 5.0)        n = len(dates)        np.random.seed(hash(etf_code) % 10000)        # 生成随机游走价格序列        returns = np.random.normal(0.0001, 0.02, n)  # 日均收益0.01%，波动2%        prices = base_price * np.exp(np.cumsum(returns))        # 生成OHLCV数据        df = pd.DataFrame(index=dates[:len(prices)])        df['close'] = prices        # 生成开盘价、最高价、最低价        noise = np.random.normal(0, 0.005, len(prices))        df['open'] = df['close'].shift(1) * (1 + noise)        df['high'] = df[['open', 'close']].max(axis=1) * (1 + abs(np.random.normal(0, 0.01, len(prices))))        df['low'] = df[['open', 'close']].min(axis=1) * (1 - abs(np.random.normal(0, 0.01, len(prices))))        # 成交量        df['volume'] = np.random.lognormal(10, 1, len(prices))        # 确保价格合理性        df['high'] = df[['open', 'close', 'high']].max(axis=1)        df['low'] = df[['open', 'close', 'low']].min(axis=1)        return df    def compute_technical_features(self, df, window_sizes=[5, 10, 20, 60]):        """        计算技术特征        Args:            df: 包含价格数据的DataFrame            window_sizes: 计算指标的窗口大小列表        Returns:            DataFrame: 包含特征的DataFrame        """        features_df = pd.DataFrame(index=df.index)        # 价格特征        features_df['close'] = df['close']        features_df['returns'] = df['returns']        # 波动率特征        for window in window_sizes:            # 滚动波动率            features_df[f'volatility_{window}'] = df['returns'].rolling(window).std()            # 滚动收益率            features_df[f'return_{window}'] = df['close'].pct_change(window)            # 滚动最大回撤            features_df[f'max_dd_{window}'] = -df['close'].rolling(window).apply(                lambda x: (x.max() - x[-1]) / x.max() if x.max() > 0 else 0            )        # 移动平均线        for window in [5, 10, 20, 30, 60]:            features_df[f'ma_{window}'] = df['close'].rolling(window).mean()            # 价格与均线的距离            features_df[f'price_ma_ratio_{window}'] = df['close'] / features_df[f'ma_{window}'] - 1        # 动量指标        features_df['rsi_14'] = talib.RSI(df['close'], timeperiod=14)        features_df['macd'], features_df['macd_signal'], features_df['macd_hist'] = talib.MACD(            df['close'], fastperiod=12, slowperiod=26, signalperiod=9        )        # 布林带        features_df['bb_upper'], features_df['bb_middle'], features_df['bb_lower'] = talib.BBANDS(            df['close'], timeperiod=20, nbdevup=2, nbdevdn=2, matype=0        )        features_df['bb_width'] = (features_df['bb_upper'] - features_df['bb_lower']) / features_df['bb_middle']        features_df['bb_position'] = (df['close'] - features_df['bb_lower']) / (features_df['bb_upper'] - features_df['bb_lower'])        # 成交量特征        features_df['volume'] = df['volume']        features_df['volume_ma_ratio'] = df['volume'] / df['volume'].rolling(20).mean()        features_df['volume_obv'] = talib.OBV(df['close'], df['volume'])        # 价格形态特征        features_df['atr'] = talib.ATR(df['high'], df['low'], df['close'], timeperiod=14)        features_df['adx'] = talib.ADX(df['high'], df['low'], df['close'], timeperiod=14)        # 相关性特征（需要多个ETF）        # 这里简化处理，实际应用中需要多个ETF的价格计算相关性        # 市场状态特征        features_df['day_of_week'] = df.index.dayofweek        features_df['month'] = df.index.month        features_df['is_month_start'] = (df.index.day <= 3).astype(int)        features_df['is_month_end'] = (df.index.day >= 25).astype(int)        return features_df.fillna(method='ffill').fillna(0)    def create_dataset(self, features_list, target_etf='510300', lookback=60, forward=5):        """        创建深度学习数据集        Args:            features_list: 特征DataFrame列表（每个ETF一个）            target_etf: 目标ETF代码            lookback: 回看窗口            forward: 预测未来窗口        Returns:            X, y: 特征和标签        """        # 对齐所有ETF的数据        aligned_data = []        min_len = min(len(f) for f in features_list)        for f in features_list:            aligned_data.append(f.iloc[-min_len:].values)        # 创建3D特征矩阵 [样本数, 时间步长, 特征数]        n_samples = min_len - lookback - forward        n_features = sum(f.shape[1] for f in features_list)        X = np.zeros((n_samples, lookback, n_features))        y = np.zeros((n_samples, 2))  # 二元分类：涨/跌        # 填充特征矩阵        for i in range(n_samples):            feature_idx = 0            for etf_idx, etf_features in enumerate(aligned_data):                n_etf_features = features_list[etf_idx].shape[1]                X[i, :, feature_idx:feature_idx + n_etf_features] = etf_features[i:i+lookback]                feature_idx += n_etf_features        # 创建标签：未来forward天的收益率和方向        target_idx = list(self.etf_list.keys()).index(target_etf) if target_etf in self.etf_list else 0        target_features = features_list[target_idx]        for i in range(n_samples):            future_price = target_features.iloc[i+lookback+forward-1]['close']            current_price = target_features.iloc[i+lookback-1]['close']            # 收益率            future_return = (future_price - current_price) / current_price            # 分类标签：1表示上涨，0表示下跌            y[i, 0] = future_return  # 连续值用于回归            y[i, 1] = 1 if future_return > 0 else 0  # 分类标签        return X, y    def create_multi_etf_dataset(self, etf_data_dict, lookback=60, forward=5):        """        创建多ETF轮动数据集        Args:            etf_data_dict: ETF数据字典 {代码: DataFrame}            lookback: 回看窗口            forward: 预测未来窗口        Returns:            X, y_multi: 特征和多ETF标签        """        # 获取所有ETF的特征数据        features_list = []        etf_codes = []        for code, df in etf_data_dict.items():            features = self.compute_technical_features(df)            features_list.append(features)            etf_codes.append(code)        # 对齐数据        aligned_data = []        min_len = min(len(f) for f in features_list)        for f in features_list:            aligned_data.append(f.iloc[-min_len:].values)        # 创建3D特征矩阵        n_samples = min_len - lookback - forward        n_features = sum(f.shape[1] for f in features_list)        X = np.zeros((n_samples, lookback, n_features))        # 填充特征矩阵        for i in range(n_samples):            feature_idx = 0            for etf_idx, etf_features in enumerate(aligned_data):                n_etf_features = features_list[etf_idx].shape[1]                X[i, :, feature_idx:feature_idx + n_etf_features] = etf_features[i:i+lookback]                feature_idx += n_etf_features        # 创建多ETF标签：每个ETF未来forward天的收益率        y_multi = np.zeros((n_samples, len(etf_codes)))        for etf_idx, features in enumerate(features_list):            for i in range(n_samples):                future_price = features.iloc[i+lookback+forward-1]['close']                current_price = features.iloc[i+lookback-1]['close']                y_multi[i, etf_idx] = (future_price - current_price) / current_price        return X, y_multi, etf_codes# ==================== 第二部分：深度学习模型 ====================class AttentionLayer(nn.Module):    """注意力机制层"""    def __init__(self, input_dim):        super(AttentionLayer, self).__init__()        self.attention = nn.Sequential(            nn.Linear(input_dim, 64),            nn.Tanh(),            nn.Linear(64, 1)        )    def forward(self, x):        # x形状: [batch_size, seq_len, features]        weights = self.attention(x)  # [batch_size, seq_len, 1]        weights = torch.softmax(weights, dim=1)        weighted = torch.sum(x * weights, dim=1)  # [batch_size, features]        return weighted, weightsclass LSTMModel(nn.Module):    """LSTM时序预测模型"""    def __init__(self, input_dim, hidden_dim=128, num_layers=2, output_dim=1, dropout=0.3):        super(LSTMModel, self).__init__()        self.lstm = nn.LSTM(            input_dim, hidden_dim, num_layers,             batch_first=True, dropout=dropout if num_layers > 1 else 0        )        self.attention = AttentionLayer(hidden_dim)        self.fc = nn.Sequential(            nn.Linear(hidden_dim, 64),            nn.BatchNorm1d(64),            nn.ReLU(),            nn.Dropout(dropout),            nn.Linear(64, 32),            nn.BatchNorm1d(32),            nn.ReLU(),            nn.Dropout(dropout),            nn.Linear(32, output_dim)        )    def forward(self, x):        # LSTM层        lstm_out, (h_n, c_n) = self.lstm(x)  # lstm_out: [batch, seq_len, hidden_dim]        # 注意力层        attn_out, attn_weights = self.attention(lstm_out)        # 全连接层        output = self.fc(attn_out)        return output, attn_weightsclass MultiTaskETFModel(nn.Module):    """多任务学习模型：同时预测收益率、波动率和方向"""    def __init__(self, input_dim, hidden_dim=128, num_layers=2, dropout=0.3):        super(MultiTaskETFModel, self).__init__()        # 共享的LSTM层        self.lstm = nn.LSTM(            input_dim, hidden_dim, num_layers,             batch_first=True, dropout=dropout if num_layers > 1 else 0        )        self.attention = AttentionLayer(hidden_dim)        # 共享的特征提取层        self.shared_fc = nn.Sequential(            nn.Linear(hidden_dim, 64),            nn.BatchNorm1d(64),            nn.ReLU(),            nn.Dropout(dropout)        )        # 任务特定的输出层        # 任务1: 收益率预测（回归）        self.return_head = nn.Sequential(            nn.Linear(64, 32),            nn.ReLU(),            nn.Linear(32, 1)  # 预测收益率        )        # 任务2: 波动率预测（回归）        self.volatility_head = nn.Sequential(            nn.Linear(64, 32),            nn.ReLU(),            nn.Linear(32, 1)  # 预测波动率        )        # 任务3: 方向预测（分类）        self.direction_head = nn.Sequential(            nn.Linear(64, 32),            nn.ReLU(),            nn.Linear(32, 2)  # 二分类：涨/跌        )    def forward(self, x):        # LSTM层        lstm_out, _ = self.lstm(x)        # 注意力层        attn_out, attn_weights = self.attention(lstm_out)        # 共享特征        shared_features = self.shared_fc(attn_out)        # 各个任务的输出        return_pred = self.return_head(shared_features)        volatility_pred = self.volatility_head(shared_features)        direction_pred = self.direction_head(shared_features)        return {            'return': return_pred,            'volatility': volatility_pred,            'direction': direction_pred,            'attention': attn_weights        }class TransformerETFModel(nn.Module):    """Transformer模型用于时序预测"""    def __init__(self, input_dim, hidden_dim=128, num_heads=4, num_layers=3, output_dim=1, dropout=0.1):        super(TransformerETFModel, self).__init__()        self.input_projection = nn.Linear(input_dim, hidden_dim)        # Transformer编码器        encoder_layer = nn.TransformerEncoderLayer(            d_model=hidden_dim,            nhead=num_heads,            dim_feedforward=hidden_dim*4,            dropout=dropout,            batch_first=True        )        self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)        # 注意力池化        self.attention_pool = AttentionLayer(hidden_dim)        # 输出层        self.output_layer = nn.Sequential(            nn.Linear(hidden_dim, 64),            nn.ReLU(),            nn.Dropout(dropout),            nn.Linear(64, output_dim)        )    def forward(self, x):        # 输入投影        x = self.input_projection(x)        # Transformer编码        transformer_out = self.transformer_encoder(x)        # 注意力池化        pooled, attn_weights = self.attention_pool(transformer_out)        # 输出        output = self.output_layer(pooled)        return output, attn_weights# ==================== 第三部分：仓位管理与风险控制 ====================class RiskAwarePositionManager:    """风险感知的仓位管理器"""    def __init__(self, initial_capital=1000000, max_position_ratio=0.2,                  stop_loss_pct=0.05, max_drawdown_limit=0.15, volatility_scaling=True):        """        初始化仓位管理器        Args:            initial_capital: 初始资金            max_position_ratio: 单个ETF最大仓位比例            stop_loss_pct: 止损比例            max_drawdown_limit: 最大回撤限制            volatility_scaling: 是否根据波动率调整仓位        """        self.initial_capital = initial_capital        self.current_capital = initial_capital        self.max_position_ratio = max_position_ratio        self.stop_loss_pct = stop_loss_pct        self.max_drawdown_limit = max_drawdown_limit        self.volatility_scaling = volatility_scaling        # 持仓记录        self.positions = {}        self.trade_history = []        self.equity_curve = [initial_capital]        self.max_equity = initial_capital        # 风险指标        self.current_drawdown = 0        self.peak_equity = initial_capital    def calculate_position_size(self, predictions, volatilities=None, market_state='normal'):        """        计算仓位大小        Args:            predictions: 模型预测的收益率 [n_etfs]            volatilities: 波动率预测 [n_etfs]            market_state: 市场状态 ('bull', 'bear', 'normal')        Returns:            dict: 每个ETF的权重        """        n_etfs = len(predictions)        # 基础权重：基于预测收益率        raw_weights = np.array(predictions)        # 1. 负预测值归零（不做空）        raw_weights[raw_weights < 0] = 0        # 2. 波动率调整（如果启用）        if self.volatility_scaling and volatilities is not None:            # 波动率倒数作为调整因子（波动率越高，仓位越小）            vol_adjustment = 1.0 / (volatilities + 1e-8)            raw_weights = raw_weights * vol_adjustment        # 3. 市场状态调整        market_adjustment = self._get_market_adjustment(market_state)        raw_weights = raw_weights * market_adjustment        # 4. 应用最大仓位限制        max_weight = self.max_position_ratio        if market_state == 'bear':            max_weight *= 0.5  # 熊市减半仓位限制        raw_weights = np.clip(raw_weights, 0, max_weight)        # 5. 归一化确保权重和为1        if raw_weights.sum() > 0:            weights = raw_weights / raw_weights.sum()        else:            # 如果没有正预测，全部持有现金            weights = np.zeros(n_etfs)        # 确保没有超限        weights = np.clip(weights, 0, max_weight)        return weights    def _get_market_adjustment(self, market_state):        """根据市场状态获取调整因子"""        if market_state == 'bull':            return 1.2  # 牛市增加仓位        elif market_state == 'bear':            return 0.5  # 熊市大幅降低仓位        elif market_state == 'high_volatility':            return 0.7  # 高波动市场降低仓位        else:  # 'normal'            return 1.0    def assess_market_state(self, recent_returns, recent_volatility):        """评估市场状态"""        avg_return = np.mean(recent_returns)        avg_vol = np.mean(recent_volatility)        if avg_return > 0.001 and avg_vol < 0.02:  # 正收益且低波动            return 'bull'        elif avg_return < -0.001:  # 负收益            return 'bear'        elif avg_vol > 0.025:  # 高波动            return 'high_volatility'        else:            return 'normal'    def apply_stop_loss(self, positions, current_prices, entry_prices):        """应用止损规则"""        for etf, position in positions.items():            if position['shares'] > 0:                current_value = current_prices[etf] * position['shares']                entry_value = entry_prices[etf] * position['shares']                # 计算盈亏比例                pnl_pct = (current_value - entry_value) / entry_value                # 如果亏损超过止损比例，平仓                if pnl_pct < -self.stop_loss_pct:                    print(f"止损触发: {etf}, 亏损: {pnl_pct:.2%}")                    return True, etf        return False, None    def update_drawdown(self, current_equity):        """更新回撤指标"""        if current_equity > self.peak_equity:            self.peak_equity = current_equity        self.current_drawdown = (self.peak_equity - current_equity) / self.peak_equity        # 如果回撤超过限制，强制减仓        if self.current_drawdown > self.max_drawdown_limit:            return 'reduce_position'        return 'normal'    def risk_parity_allocation(self, volatilities, correlations=None):        """        风险平价分配        Args:            volatilities: 各ETF的波动率            correlations: 相关系数矩阵        Returns:            array: 风险平价权重        """        n = len(volatilities)        # 如果没有相关矩阵，假设独立        if correlations is None:            correlations = np.eye(n)        # 计算协方差矩阵        cov_matrix = np.outer(volatilities, volatilities) * correlations        try:            # 计算风险贡献的逆            inv_vol = 1.0 / volatilities            risk_parity_weights = inv_vol / inv_vol.sum()        except:            # 如果波动率有零值，使用等权重            risk_parity_weights = np.ones(n) / n        return risk_parity_weights    def execute_rebalancing(self, target_weights, current_weights, prices,                             transaction_cost=0.001, min_trade_ratio=0.01):        """        执行再平衡        Args:            target_weights: 目标权重            current_weights: 当前权重            prices: 当前价格            transaction_cost: 交易成本比例            min_trade_ratio: 最小交易比例（避免小额交易）        Returns:            dict: 交易指令        """        trades = {}        total_value = self.current_capital        for i, etf in enumerate(prices.keys()):            target_value = total_value * target_weights[i]            current_value = total_value * current_weights[i]            value_diff = target_value - current_value            # 只有变化超过阈值时才交易            if abs(value_diff) > total_value * min_trade_ratio:                shares_to_trade = value_diff / prices[etf]                # 考虑交易成本                cost = abs(shares_to_trade * prices[etf]) * transaction_cost                trades[etf] = {                    'shares': shares_to_trade,                    'value': value_diff,                    'cost': cost                }        return trades# ==================== 第四部分：ETF轮动策略主类 ====================class DeepLearningETFRotationStrategy:    """深度学习ETF轮动策略主类"""    def __init__(self, config=None):        """        初始化策略        Args:            config: 配置字典        """        self.config = config or self._default_config()        self.data_loader = ETFDataLoader(            token=self.config.get('tushare_token'),            start_date=self.config.get('start_date', '2015-01-01'),            end_date=self.config.get('end_date')        )        # 模型        self.model = None        self.scaler = StandardScaler()        # 仓位管理        self.position_manager = RiskAwarePositionManager(            initial_capital=self.config.get('initial_capital', 1000000),            max_position_ratio=self.config.get('max_position_ratio', 0.3),            stop_loss_pct=self.config.get('stop_loss_pct', 0.05),            max_drawdown_limit=self.config.get('max_drawdown_limit', 0.15),            volatility_scaling=self.config.get('volatility_scaling', True)        )        # 数据缓存        self.etf_data = {}        self.features = {}        # 回测结果        self.backtest_results = {}    def _default_config(self):        """默认配置"""        return {            'initial_capital': 1000000,            'lookback_window': 60,            'forward_window': 5,            'train_test_split': 0.8,            'batch_size': 32,            'epochs': 50,            'learning_rate': 0.001,            'max_position_ratio': 0.3,            'stop_loss_pct': 0.05,            'max_drawdown_limit': 0.15,            'transaction_cost': 0.001,            'etf_codes': ['510300', '510500', '159915', '511010'],  # 包含防御性资产            'model_type': 'multitask',  # 'lstm', 'transformer', 'multitask'            'use_attention': True        }    def load_and_prepare_data(self):        """加载和准备数据"""        print("加载ETF数据...")        # 加载每个ETF的数据        for code in self.config['etf_codes']:            df = self.data_loader.fetch_etf_data(code, save_local=True)            self.etf_data[code] = df            # 计算特征            features = self.data_loader.compute_technical_features(df)            self.features[code] = features            print(f"{code}: {len(df)} 个交易日数据，{features.shape[1]} 个特征")        # 创建多ETF数据集        X, y_multi, etf_codes = self.data_loader.create_multi_etf_dataset(            self.etf_data,            lookback=self.config['lookback_window'],            forward=self.config['forward_window']        )        # 特征标准化        n_samples, n_timesteps, n_features = X.shape        X_reshaped = X.reshape(-1, n_features)        X_scaled = self.scaler.fit_transform(X_reshaped)        X_scaled = X_scaled.reshape(n_samples, n_timesteps, n_features)        # 划分训练集和测试集（时间序列划分）        split_idx = int(len(X_scaled) * self.config['train_test_split'])        self.X_train = X_scaled[:split_idx]        self.X_test = X_scaled[split_idx:]        self.y_train = y_multi[:split_idx]        self.y_test = y_multi[split_idx:]        self.train_dates = list(self.etf_data[self.config['etf_codes'][0]].index)[            self.config['lookback_window'] + self.config['forward_window']:split_idx +             self.config['lookback_window'] + self.config['forward_window']        ]        self.test_dates = list(self.etf_data[self.config['etf_codes'][0]].index)[            split_idx + self.config['lookback_window'] + self.config['forward_window']:        ]        print(f"训练集: {len(self.X_train)} 个样本, 测试集: {len(self.X_test)} 个样本")        return X_scaled, y_multi    def build_model(self):        """构建深度学习模型"""        n_features = self.X_train.shape[2]        n_etfs = len(self.config['etf_codes'])        if self.config['model_type'] == 'lstm':            self.model = LSTMModel(                input_dim=n_features,                hidden_dim=128,                num_layers=2,                output_dim=n_etfs,                dropout=0.3            )        elif self.config['model_type'] == 'transformer':            self.model = TransformerETFModel(                input_dim=n_features,                hidden_dim=128,                num_heads=4,                num_layers=3,                output_dim=n_etfs,                dropout=0.1            )        elif self.config['model_type'] == 'multitask':            self.model = MultiTaskETFModel(                input_dim=n_features,                hidden_dim=128,                num_layers=2,                dropout=0.3            )        else:            raise ValueError(f"未知模型类型: {self.config['model_type']}")        print(f"构建 {self.config['model_type']} 模型, 输入特征: {n_features}, 输出ETF数: {n_etfs}")        return self.model    def train_model(self):        """训练模型"""        if self.model is None:            self.build_model()        # 准备数据加载器        train_dataset = TensorDataset(            torch.FloatTensor(self.X_train),            torch.FloatTensor(self.y_train)        )        train_loader = DataLoader(            train_dataset,            batch_size=self.config['batch_size'],            shuffle=True,            drop_last=True        )        # 损失函数和优化器        if self.config['model_type'] == 'multitask':            criterion = {                'return': nn.MSELoss(),                'volatility': nn.MSELoss(),                'direction': nn.CrossEntropyLoss()            }            # 多任务损失权重            loss_weights = {                'return': 1.0,                'volatility': 0.5,                'direction': 0.8            }        else:            criterion = nn.MSELoss()        optimizer = optim.Adam(self.model.parameters(), lr=self.config['learning_rate'])        scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', patience=5, factor=0.5)        # 训练循环        self.model.train()        train_losses = []        for epoch in range(self.config['epochs']):            epoch_loss = 0            for batch_X, batch_y in train_loader:                optimizer.zero_grad()                if self.config['model_type'] == 'multitask':                    outputs = self.model(batch_X)                    # 计算多任务损失                    loss = 0                    return_loss = criterion['return'](outputs['return'], batch_y.mean(dim=1, keepdim=True))                    volatility_target = torch.abs(batch_y).mean(dim=1, keepdim=True)                    volatility_loss = criterion['volatility'](outputs['volatility'], volatility_target)                    # 方向标签（平均收益率为正为1，否则为0）                    direction_target = (batch_y.mean(dim=1) > 0).long()                    direction_loss = criterion['direction'](outputs['direction'], direction_target)                    loss = (loss_weights['return'] * return_loss +                            loss_weights['volatility'] * volatility_loss +                            loss_weights['direction'] * direction_loss)                else:                    outputs, _ = self.model(batch_X)                    loss = criterion(outputs, batch_y)                loss.backward()                torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)                optimizer.step()                epoch_loss += loss.item()            avg_loss = epoch_loss / len(train_loader)            train_losses.append(avg_loss)            scheduler.step(avg_loss)            if (epoch + 1) % 10 == 0:                print(f"Epoch [{epoch+1}/{self.config['epochs']}], Loss: {avg_loss:.6f}")        print("模型训练完成")        # 绘制训练损失曲线        plt.figure(figsize=(10, 5))        plt.plot(train_losses)        plt.title('训练损失曲线')        plt.xlabel('Epoch')        plt.ylabel('Loss')        plt.grid(True)        plt.show()        return train_losses    def predict(self, X):        """使用模型进行预测"""        self.model.eval()        with torch.no_grad():            if self.config['model_type'] == 'multitask':                outputs = self.model(torch.FloatTensor(X))                predictions = outputs['return'].numpy()                volatilities = outputs['volatility'].numpy()                direction_probs = F.softmax(outputs['direction'], dim=1).numpy()                return predictions, volatilities, direction_probs            else:                outputs, attn_weights = self.model(torch.FloatTensor(X))                return outputs.numpy(), attn_weights.numpy()    def backtest(self):        """运行回测"""        print("开始回测...")        # 初始化回测变量        dates = self.test_dates        n_dates = len(dates)        n_etfs = len(self.config['etf_codes'])        # 存储回测结果        portfolio_values = [self.position_manager.initial_capital]        positions_history = []        weights_history = []        returns_history = []        # 当前权重（开始时全部现金）        current_weights = np.zeros(n_etfs)        # 回测循环        for i in range(n_dates - self.config['forward_window']):            current_date = dates[i]            # 获取当前特征            start_idx = i + len(self.X_train)            lookback_slice = slice(start_idx, start_idx + self.config['lookback_window'])            # 准备输入数据            X_batch = self.X_test[i:i+1]  # 保持3D形状            # 模型预测            if self.config['model_type'] == 'multitask':                predictions, volatilities, direction_probs = self.predict(X_batch)                predictions = predictions.flatten()                volatilities = volatilities.flatten()            else:                predictions, _ = self.predict(X_batch)                predictions = predictions.flatten()                volatilities = None            # 获取当前价格            current_prices = {}            for j, code in enumerate(self.config['etf_codes']):                if current_date in self.etf_data[code].index:                    current_prices[code] = self.etf_data[code].loc[current_date, 'close']                else:                    # 如果当天没有数据，使用最近的价格                    current_prices[code] = self.etf_data[code].iloc[                        self.etf_data[code].index.get_loc(current_date, method='nearest')                    ]['close']            # 评估市场状态            recent_returns = []            recent_volatility = []            for code in self.config['etf_codes']:                if i >= 20:  # 至少有20天数据                    recent_data = self.etf_data[code].iloc[max(0, i-20):i]                    recent_returns.extend(recent_data['returns'].dropna().tolist())                    recent_volatility.append(recent_data['returns'].std())            if recent_returns:                market_state = self.position_manager.assess_market_state(                    recent_returns, recent_volatility                )            else:                market_state = 'normal'            # 计算目标权重            target_weights = self.position_manager.calculate_position_size(                predictions, volatilities, market_state            )            # 应用风险平价调整            if volatilities is not None:                risk_parity_weights = self.position_manager.risk_parity_allocation(                    np.abs(volatilities) + 0.01  # 避免零波动率                )                # 结合模型预测和风险平价                target_weights = 0.7 * target_weights + 0.3 * risk_parity_weights            # 执行再平衡            trades = self.position_manager.execute_rebalancing(                target_weights, current_weights, current_prices,                transaction_cost=self.config.get('transaction_cost', 0.001),                min_trade_ratio=0.005            )            # 更新当前权重            current_weights = target_weights            # 计算当日组合收益            daily_return = 0            for j, code in enumerate(self.config['etf_codes']):                if code in self.etf_data and i > 0:                    # 获取当日收益率                    if current_date in self.etf_data[code].index:                        etf_return = self.etf_data[code].loc[current_date, 'returns']                    else:                        etf_return = 0                    daily_return += current_weights[j] * etf_return            # 更新组合价值            prev_value = portfolio_values[-1]            current_value = prev_value * (1 + daily_return)            # 扣除交易成本            if trades:                total_cost = sum(trade['cost'] for trade in trades.values())                current_value -= total_cost            portfolio_values.append(current_value)            returns_history.append(daily_return)            weights_history.append(current_weights.copy())            # 更新仓位管理器状态            self.position_manager.current_capital = current_value            self.position_manager.equity_curve.append(current_value)            # 更新回撤并检查是否需要减仓            risk_status = self.position_manager.update_drawdown(current_value)            if risk_status == 'reduce_position' and np.sum(current_weights) > 0.5:                # 强制减仓到50%                reduction_factor = 0.5 / np.sum(current_weights)                current_weights = current_weights * reduction_factor            # 每30天打印一次进度            if i % 30 == 0:                print(f"日期: {current_date}, 组合价值: {current_value:.2f}, 收益率: {daily_return:.4%}")        # 存储回测结果        self.backtest_results = {            'dates': dates[:len(portfolio_values)-1],            'portfolio_values': portfolio_values[1:],            'returns': returns_history,            'weights': weights_history,            'etf_codes': self.config['etf_codes']        }        print("回测完成")        return self.backtest_results    def calculate_performance_metrics(self):        """计算绩效指标"""        if not self.backtest_results:            print("请先运行回测")            return None        returns = np.array(self.backtest_results['returns'])        portfolio_values = np.array(self.backtest_results['portfolio_values'])        # 计算累计收益        cumulative_returns = (portfolio_values[-1] / portfolio_values[0]) - 1        # 计算年化收益率        n_years = len(returns) / 252        annualized_return = (1 + cumulative_returns) ** (1 / n_years) - 1        # 计算年化波动率        annualized_volatility = np.std(returns) * np.sqrt(252)        # 计算夏普比率（假设无风险利率3%）        risk_free_rate = 0.03        excess_returns = returns - risk_free_rate/252        sharpe_ratio = np.mean(excess_returns) / np.std(returns) * np.sqrt(252)        # 计算最大回撤        peak = np.maximum.accumulate(portfolio_values)        drawdown = (portfolio_values - peak) / peak        max_drawdown = np.min(drawdown)        # 计算Sortino比率（只考虑下行风险）        downside_returns = returns[returns < 0]        downside_std = np.std(downside_returns) if len(downside_returns) > 0 else 0        sortino_ratio = (annualized_return - risk_free_rate) / (downside_std * np.sqrt(252)) if downside_std > 0 else 0        # 计算Calmar比率        calmar_ratio = annualized_return / abs(max_drawdown) if max_drawdown != 0 else 0        # 计算胜率        winning_days = np.sum(returns > 0) / len(returns)        # 计算盈亏比        avg_win = np.mean(returns[returns > 0]) if np.sum(returns > 0) > 0 else 0        avg_loss = np.mean(returns[returns < 0]) if np.sum(returns < 0) > 0 else 0        profit_factor = abs(avg_win / avg_loss) if avg_loss != 0 else float('inf')        metrics = {            '累计收益': cumulative_returns,            '年化收益率': annualized_return,            '年化波动率': annualized_volatility,            '夏普比率': sharpe_ratio,            '最大回撤': max_drawdown,            'Sortino比率': sortino_ratio,            'Calmar比率': calmar_ratio,            '胜率': winning_days,            '盈亏比': profit_factor,            '总交易日数': len(returns)        }        return metrics    def plot_results(self):        """绘制回测结果"""        if not self.backtest_results:            print("请先运行回测")            return        dates = self.backtest_results['dates']        portfolio_values = self.backtest_results['portfolio_values']        returns = self.backtest_results['returns']        weights = np.array(self.backtest_results['weights'])        etf_codes = self.backtest_results['etf_codes']        fig, axes = plt.subplots(3, 2, figsize=(15, 12))        # 1. 组合净值曲线        axes[0, 0].plot(dates, portfolio_values)        axes[0, 0].set_title('组合净值曲线')        axes[0, 0].set_xlabel('日期')        axes[0, 0].set_ylabel('组合价值')        axes[0, 0].grid(True)        # 2. 累计收益率曲线        cumulative_returns = (np.array(portfolio_values) / portfolio_values[0]) - 1        axes[0, 1].plot(dates, cumulative_returns)        axes[0, 1].set_title('累计收益率曲线')        axes[0, 1].set_xlabel('日期')        axes[0, 1].set_ylabel('累计收益率')        axes[0, 1].grid(True)        # 3. 回撤曲线        peak = np.maximum.accumulate(portfolio_values)        drawdown = (portfolio_values - peak) / peak        axes[1, 0].fill_between(dates, drawdown, 0, alpha=0.3)        axes[1, 0].set_title('回撤曲线')        axes[1, 0].set_xlabel('日期')        axes[1, 0].set_ylabel('回撤')        axes[1, 0].grid(True)        # 4. 月度收益率热力图        # 创建月度收益率数据        monthly_returns = []        monthly_dates = []        df_returns = pd.Series(returns, index=dates[:len(returns)])        monthly_series = df_returns.resample('M').apply(lambda x: (1 + x).prod() - 1)        # 重塑为矩阵形式        monthly_matrix = []        for year in sorted(set(d.year for d in monthly_series.index)):            year_data = monthly_series[monthly_series.index.year == year]            if len(year_data) == 12:                monthly_matrix.append(year_data.values)        if monthly_matrix:            monthly_matrix = np.array(monthly_matrix).T  # 转置使月份为行            im = axes[1, 1].imshow(monthly_matrix, aspect='auto', cmap='RdYlGn')            axes[1, 1].set_title('月度收益率热力图')            axes[1, 1].set_xlabel('年份')            axes[1, 1].set_ylabel('月份')            plt.colorbar(im, ax=axes[1, 1])        # 5. 仓位权重变化        for i in range(len(etf_codes)):            axes[2, 0].plot(dates[:len(weights)], weights[:, i], label=etf_codes[i])        axes[2, 0].set_title('仓位权重变化')        axes[2, 0].set_xlabel('日期')        axes[2, 0].set_ylabel('权重')        axes[2, 0].legend()        axes[2, 0].grid(True)        # 6. 收益率分布直方图        axes[2, 1].hist(returns, bins=50, alpha=0.7, edgecolor='black')        axes[2, 1].axvline(x=np.mean(returns), color='red', linestyle='--', label=f'均值: {np.mean(returns):.4%}')        axes[2, 1].set_title('日收益率分布')        axes[2, 1].set_xlabel('日收益率')        axes[2, 1].set_ylabel('频数')        axes[2, 1].legend()        axes[2, 1].grid(True)        plt.tight_layout()        plt.show()        # 打印绩效指标        metrics = self.calculate_performance_metrics()        if metrics:            print("\n" + "="*50)            print("策略绩效指标:")            print("="*50)            for key, value in metrics.items():                if isinstance(value, float):                    if '率' in key or '比' in key:                        print(f"{key}: {value:.4%}" if '回撤' not in key else f"{key}: {value:.4%}")                    elif '收益' in key:                        print(f"{key}: {value:.4%}")                    else:                        print(f"{key}: {value:.4f}")                else:                    print(f"{key}: {value}")    def run_optimization(self):        """运行参数优化（简化版）"""        print("开始参数优化...")        # 定义参数网格        param_grid = {            'lookback_window': [30, 60, 90],            'forward_window': [3, 5, 10],            'max_position_ratio': [0.2, 0.3, 0.4],            'stop_loss_pct': [0.03, 0.05, 0.08]        }        best_params = None        best_sharpe = -np.inf        # 简化优化：只测试少数组合        import itertools        keys, values = zip(*param_grid.items())        param_combinations = [dict(zip(keys, v)) for v in itertools.product(*values)]        # 只测试前几种组合（实际应用中可以更多）        for i, params in enumerate(param_combinations[:4]):            print(f"\n测试参数组合 {i+1}/{len(param_combinations[:4])}: {params}")            # 更新配置            for key, value in params.items():                self.config[key] = value            # 重新准备数据            self.load_and_prepare_data()            # 重新训练模型            self.build_model()            self.train_model()            # 运行回测            self.backtest()            # 计算夏普比率            metrics = self.calculate_performance_metrics()            sharpe = metrics['夏普比率']            print(f"夏普比率: {sharpe:.4f}")            if sharpe > best_sharpe:                best_sharpe = sharpe                best_params = params.copy()        print(f"\n最优参数: {best_params}")        print(f"最优夏普比率: {best_sharpe:.4f}")        return best_params, best_sharpe# ==================== 第五部分：主程序执行 ====================def main():    """主函数"""    print("="*60)    print("深度学习ETF轮动策略")    print("目标：择时 + 仓位管理 + 控制回撤 + 提高夏普比率")    print("="*60)    # 配置策略    config = {        'initial_capital': 1000000,        'lookback_window': 60,  # 60天历史数据        'forward_window': 5,    # 预测未来5天        'train_test_split': 0.8,        'batch_size': 32,        'epochs': 50,           # 训练轮数        'learning_rate': 0.001,        'max_position_ratio': 0.3,      # 单个ETF最大仓位30%        'stop_loss_pct': 0.05,          # 止损比例5%        'max_drawdown_limit': 0.15,     # 最大回撤限制15%        'transaction_cost': 0.001,      # 交易成本0.1%        'etf_codes': ['510300', '510500', '159915', '511010'],  # ETF列表        'model_type': 'multitask',      # 使用多任务学习模型        'volatility_scaling': True,     # 启用波动率调整        'start_date': '2016-01-01',     # 数据开始日期        'end_date': '2023-12-31'        # 数据结束日期    }    # 初始化策略    strategy = DeepLearningETFRotationStrategy(config)    # 1. 加载和准备数据    print("\n[阶段1] 加载和准备数据...")    strategy.load_and_prepare_data()    # 2. 构建和训练模型    print("\n[阶段2] 构建和训练深度学习模型...")    strategy.build_model()    strategy.train_model()    # 3. 运行回测    print("\n[阶段3] 运行策略回测...")    backtest_results = strategy.backtest()    # 4. 分析结果    print("\n[阶段4] 分析回测结果...")    strategy.plot_results()    # 5. 参数优化（可选）    print("\n[阶段5] 运行参数优化...")    optimize = input("是否运行参数优化？(y/n): ").lower() == 'y'    if optimize:        best_params, best_sharpe = strategy.run_optimization()        # 使用最优参数重新运行        print("\n使用最优参数重新运行策略...")        for key, value in best_params.items():            strategy.config[key] = value        strategy.load_and_prepare_data()        strategy.build_model()        strategy.train_model()        strategy.backtest()        strategy.plot_results()    print("\n策略执行完成！")if __name__ == "__main__":    main()