feat(data-source): 支持指数-ETF双轨数据获取及因子计算

- 新增使用Tushare获取A股ETF价格及净值数据的私有方法
- fetch_all方法支持接收完整代码配置，区分指数与ETF及市场类别
- 指数数据和ETF数据分别下载，ETF净值数据用于溢价率计算
- 采用A股交易日为主交易日历，非A股数据前向填充对齐
- 调整因子计算，支持指数价格计算因子，ETF价格计算收益率
- run_rotation脚本和RotationStrategy引擎适配指数-ETF配置格式
- 代码结构优化，增强多市场及加密货币处理能力

core/datasource/hybrid_source.py

@@ -209,6 +209,116 @@ class HybridDataSource:
                 if value is not None:
                     os.environ[key] = value
 
+    def _fetch_etf(self, code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]:
+        """使用 Tushare 获取A股ETF数据（fund_daily接口）"""
+        import os
+
+        # 临时清除代理环境变量
+        original_proxy = {}
+        for key in ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"]:
+            original_proxy[key] = os.environ.pop(key, None)
+
+        try:
+            import tushare as ts
+
+            pro = ts.pro_api(self._get_tushare_token())
+
+            # 转换代码格式 (510300.SH -> 510300.SH)
+            ts_code = code.replace(".SS", ".SH")
+
+            # 获取ETF日线数据
+            df = pro.fund_daily(
+                ts_code=ts_code,
+                start_date=start_date.replace("-", ""),
+                end_date=end_date.replace("-", "")
+            )
+
+            if df is None or len(df) == 0:
+                return None
+
+            # 标准化列名
+            df = df.rename(columns={
+                "trade_date": "date",
+                "open": "open",
+                "high": "high",
+                "low": "low",
+                "close": "close",
+                "vol": "volume",
+            })
+
+            # 转换日期格式
+            df["date"] = pd.to_datetime(df["date"])
+            df = df.set_index("date")
+            df = df.sort_index()
+
+            # 添加代码列
+            df["code"] = code
+
+            return df
+
+        except Exception as e:
+            print(f"Tushare 下载ETF {code} 失败: {e}")
+            return None
+
+        finally:
+            # 恢复代理环境变量
+            for key, value in original_proxy.items():
+                if value is not None:
+                    os.environ[key] = value
+
+    def _fetch_etf_nav(self, code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]:
+        """使用 Tushare 获取ETF净值数据（fund_nav接口）"""
+        import os
+
+        # 临时清除代理环境变量
+        original_proxy = {}
+        for key in ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"]:
+            original_proxy[key] = os.environ.pop(key, None)
+
+        try:
+            import tushare as ts
+
+            pro = ts.pro_api(self._get_tushare_token())
+
+            # 转换代码格式
+            ts_code = code.replace(".SS", ".SH")
+
+            # 获取ETF净值数据
+            df = pro.fund_nav(
+                ts_code=ts_code,
+                start_date=start_date.replace("-", ""),
+                end_date=end_date.replace("-", "")
+            )
+
+            if df is None or len(df) == 0:
+                return None
+
+            # 标准化列名
+            df = df.rename(columns={
+                "nav_date": "date",
+                "unit_nav": "nav",
+            })
+
+            # 转换日期格式
+            df["date"] = pd.to_datetime(df["date"])
+            df = df.set_index("date")
+            df = df.sort_index()
+
+            # 添加代码列
+            df["code"] = code
+
+            return df
+
+        except Exception as e:
+            print(f"Tushare 下载ETF净值 {code} 失败: {e}")
+            return None
+
+        finally:
+            # 恢复代理环境变量
+            for key, value in original_proxy.items():
+                if value is not None:
+                    os.environ[key] = value
+
     def _fetch_yfinance(self, code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]:
         """使用 YFinance 获取数据"""
         import time
@@ -303,39 +413,50 @@ class HybridDataSource:
 
     def fetch_all(
         self,
-        code_list,  # list[代码] 或 dict{代码: 名称}
+        code_config: dict,  # {代码: {name, etf, market}}
         benchmark_code: str,
         start_date: str,
         end_date: str,
-    ) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame], list]:
+    ) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame], Optional[pd.DataFrame], Optional[pd.DataFrame], list]:
         """
-        批量获取数据
-        注意：由于 Tushare(中国A股) 和 YFinance(美股/加密货币) 的交易日历不同，
-        这里返回的是长格式数据，由调用方分别处理各市场的数据
-
+        批量获取数据（支持指数-ETF映射）
+        
+        Args:
+            code_config: 配置字典，格式为 {index_code: {name, etf, market}}
+            benchmark_code: 基准指数代码
+            start_date: 开始日期
+            end_date: 结束日期
+            
         Returns:
-            (etf_data, benchmark_data, valid_codes)
-            etf_data: DataFrame with columns [code, close, source], index=date
+            (index_data, etf_data, etf_nav_data, benchmark_data, valid_codes)
+            - index_data: 指数数据（用于因子计算）
+            - etf_data: ETF价格数据（用于收益计算）
+            - etf_nav_data: ETF净值数据（用于溢价率计算）
+            - benchmark_data: 基准数据
+            - valid_codes: 有效代码列表
         """
-        all_data = []
+        index_data_list = []
+        etf_data_list = []
         valid_codes = []
-
-        # 兼容列表和字典格式
-        if isinstance(code_list, dict):
-            codes = list(code_list.keys())
-            code_name_map = code_list
-        else:
-            codes = code_list
-            code_name_map = {c: c for c in codes}
-
-        print(f"开始下载 {len(codes)} 只标的的数据...")
-        china_codes = [c for c in codes if self._is_china_index(c)]
-        global_codes = [c for c in codes if not self._is_china_index(c)]
+        
+        # 提取指数代码和ETF代码
+        index_codes = list(code_config.keys())
+        etf_codes = {}
+        for idx_code, cfg in code_config.items():
+            if cfg.get('etf'):
+                etf_codes[idx_code] = cfg['etf']
+        
+        print(f"开始下载 {len(index_codes)} 只标的的数据...")
+        print(f"  指数代码: {len(index_codes)} 只")
+        print(f"  ETF映射: {len(etf_codes)} 只")
+        
+        china_codes = [c for c in index_codes if self._is_china_index(c)]
+        global_codes = [c for c in index_codes if not self._is_china_index(c)]
         print(f"  中国A股指数: {len(china_codes)} 只")
         print(f"  港股/美股/加密货币: {len(global_codes)} 只")
 
         # 检查是否需要启动 socks2http 代理（用于加密货币）
-        crypto_codes = [c for c in codes if self._is_crypto(c)]
+        crypto_codes = [c for c in index_codes if self._is_crypto(c)]
         http_proxy = None
         socks2http_proc = None
         
@@ -358,8 +479,9 @@ class HybridDataSource:
             except Exception as e:
                 print(f"  ✗ 启动代理失败: {e}")
         
-        # 分别下载数据
-        for code in codes:
+        # 下载指数数据
+        print("\n  [1/2] 下载指数数据（用于因子计算）...")
+        for code in index_codes:
             if self._is_china_index(code):
                 source = "Tushare"
             elif self._is_crypto(code):
@@ -367,8 +489,8 @@ class HybridDataSource:
             else:
                 source = "YFinance"
             
-            name = code_name_map.get(code, code)
-            print(f"  下载 {code} ({name}) - {source}...", end=" ")
+            name = code_config[code].get('name', code)
+            print(f"    下载 {code} ({name}) - {source}...", end=" ")
 
             # 加密货币使用 HTTP 代理
             proxy = http_proxy if self._is_crypto(code) else None
@@ -378,95 +500,163 @@ class HybridDataSource:
                 # 标准化数据格式
                 data = data.copy()
                 data['source'] = source
+                data['code'] = code  # 确保code列正确
                 # 确保索引是日期格式且无时区，只保留日期部分（去掉时间）
                 data.index = pd.to_datetime(data.index, utc=True).tz_localize(None).normalize()
-                all_data.append(data[['code', 'close', 'source']])
+                index_data_list.append(data[['code', 'close', 'source']])
                 valid_codes.append(code)
                 print(f"✓ {len(data)} 条")
             else:
                 print("✗ 无数据")
         
+        # 下载ETF数据（价格+净值，用于溢价率计算）
+        etf_nav_data_list = []  # ETF净值数据
+        
+        if etf_codes:
+            print("\n  [2/2] 下载ETF数据（价格+净值，用于溢价率计算）...")
+            
+            for idx_code, etf_code in etf_codes.items():
+                name = code_config[idx_code].get('name', idx_code)
+                market = code_config[idx_code].get('market', 'A')
+                
+                # 加密货币跳过ETF下载
+                if market == 'CRYPTO':
+                    continue
+                    
+                print(f"    下载 ETF {etf_code} (对应指数 {idx_code})...", end=" ")
+                
+                # 获取ETF价格数据
+                price_data = self._fetch_etf(etf_code, start_date, end_date)
+                # 获取ETF净值数据
+                nav_data = self._fetch_etf_nav(etf_code, start_date, end_date)
+                
+                if price_data is not None and len(price_data) > 0:
+                    # 使用指数代码作为列名，保持与指数数据一致
+                    price_data = price_data.copy()
+                    price_data['source'] = 'Tushare-ETF'
+                    price_data['code'] = idx_code
+                    price_data.index = pd.to_datetime(price_data.index, utc=True).tz_localize(None).normalize()
+                    etf_data_list.append(price_data[['code', 'close', 'source']])
+                    
+                    # 处理净值数据
+                    if nav_data is not None and len(nav_data) > 0:
+                        nav_data = nav_data.copy()
+                        nav_data['code'] = idx_code
+                        nav_data.index = pd.to_datetime(nav_data.index, utc=True).tz_localize(None).normalize()
+                        etf_nav_data_list.append(nav_data[['code', 'nav']])
+                        print(f"✓ 价格{len(price_data)}条 净值{len(nav_data)}条")
+                    else:
+                        print(f"✓ 价格{len(price_data)}条 (无净值数据)")
+                else:
+                    print(f"✗ 无数据")
+        
         # 关闭 socks2http 代理
         if socks2http_proc:
             socks2http_proc.terminate()
             socks2http_proc.wait()
             print(f"\n  socks2http 代理已关闭")
 
-        if not all_data:
-            return None, None, []
-
-        # 检查数据源类型
-        sources = set(d['source'].iloc[0] for d in all_data)
-
-        if len(sources) == 1:
-            # 单一数据源：转换为宽格式（向后兼容）
-            all_df = pd.concat(all_data, ignore_index=False)
-            all_df = all_df.reset_index()
-            all_df['date'] = pd.to_datetime(all_df['date'], utc=True).dt.tz_localize(None)
-            etf_data = all_df.pivot_table(
-                index='date',
-                columns='code',
-                values='close',
-                aggfunc='first'
-            )
-            print(f"\n数据整理完成 (单一数据源 {list(sources)[0]}):")
-            print(f"  时间范围: {etf_data.index[0]} ~ {etf_data.index[-1]}")
-            print(f"  交易日数: {len(etf_data)}")
-            print(f"  有效标的: {len(etf_data.columns)} 只")
-        else:
-            # 多数据源：以主市场（Tushare/A股）为基准，其他市场数据前向填充
-            print(f"\n数据整理完成 (多数据源 - 以A股交易日为基准):")
-
-            # 合并所有数据（索引已经是标准化后的日期）
-            all_df = pd.concat(all_data, ignore_index=False)
-            all_df = all_df.reset_index()
-            # 重命名索引列为 date
-            if 'index' in all_df.columns:
-                all_df = all_df.rename(columns={'index': 'date'})
-            # 确保 date 列是日期格式（不含时间）
-            all_df['date'] = pd.to_datetime(all_df['date']).dt.normalize()
+        if not index_data_list:
+            return None, None, None, None, []
 
+        # 处理指数数据
+        print(f"\n整理指数数据（用于因子计算）...")
+        index_df = pd.concat(index_data_list, ignore_index=False)
+        index_df = index_df.reset_index()
+        if 'index' in index_df.columns:
+            index_df = index_df.rename(columns={'index': 'date'})
+        index_df['date'] = pd.to_datetime(index_df['date']).dt.normalize()
+        
+        # 透视为宽格式
+        index_data = index_df.pivot_table(
+            index='date',
+            columns='code',
+            values='close',
+            aggfunc='first'
+        )
+        
+        # 以A股交易日为基准，对齐所有数据
+        tushare_codes = [c for c in valid_codes if self._is_china_index(c)]
+        if tushare_codes:
+            primary_dates = index_data[tushare_codes[0]].dropna().index
+            print(f"  主市场交易日: {len(primary_dates)} 天")
+            
+            # 重新索引到主市场交易日
+            index_data = index_data.reindex(primary_dates)
+            
+            # 对非A股指数进行前向填充
+            non_a_codes = [c for c in valid_codes if not self._is_china_index(c)]
+            for code in non_a_codes:
+                if code in index_data.columns:
+                    index_data[code] = index_data[code].ffill().bfill()
+            
+            print(f"  非A股标的: {len(non_a_codes)} 只 (已前向填充)")
+        
+        print(f"  时间范围: {index_data.index[0]} ~ {index_data.index[-1]}")
+        print(f"  交易日数: {len(index_data)}")
+        
+        # 处理ETF数据
+        if etf_data_list:
+            print(f"\n整理ETF数据（用于收益计算）...")
+            etf_df = pd.concat(etf_data_list, ignore_index=False)
+            etf_df = etf_df.reset_index()
+            if 'index' in etf_df.columns:
+                etf_df = etf_df.rename(columns={'index': 'date'})
+            etf_df['date'] = pd.to_datetime(etf_df['date']).dt.normalize()
+            
             # 透视为宽格式
-            etf_data = all_df.pivot_table(
+            etf_data = etf_df.pivot_table(
                 index='date',
                 columns='code',
                 values='close',
                 aggfunc='first'
             )
-
-            # 获取主市场（Tushare）的交易日历
-            tushare_codes = [c for c in valid_codes if self._is_china_index(c)]
+            
+            # 对齐到主市场交易日
             if tushare_codes:
-                # 使用第一个A股代码的日期作为主市场交易日
-                primary_dates = etf_data[tushare_codes[0]].dropna().index
-                print(f"  主市场交易日: {len(primary_dates)} 天")
-
-                # 重新索引到主市场交易日，使用前向填充
                 etf_data = etf_data.reindex(primary_dates)
-
-                # 对每个非主市场代码进行前向填充
-                yfinance_codes = [c for c in valid_codes if not self._is_china_index(c)]
-                for code in yfinance_codes:
-                    if code in etf_data.columns:
-                        # 前向填充：用最近的有效价格填充休市日的数据
-                        etf_data[code] = etf_data[code].ffill()
-                        # 对于开头的NaN，用后向填充
-                        etf_data[code] = etf_data[code].bfill()
-
-                print(f"  非主市场标的: {len(yfinance_codes)} 只 (已前向填充)")
-
-            print(f"  时间范围: {etf_data.index[0]} ~ {etf_data.index[-1]}")
-            print(f"  交易日数: {len(etf_data)}")
-            print(f"  有效标的: {len(etf_data.columns)} 只")
-
+            
+            print(f"  ETF价格数据: {len(etf_data.columns)} 只")
+        else:
+            # 如果没有ETF数据，使用指数数据代替
+            etf_data = index_data.copy()
+            print(f"\n无ETF映射，使用指数数据代替")
+        
+        # 处理ETF净值数据
+        etf_nav_data = None
+        if etf_nav_data_list:
+            print(f"\n整理ETF净值数据（用于溢价率计算）...")
+            nav_df = pd.concat(etf_nav_data_list, ignore_index=False)
+            nav_df = nav_df.reset_index()
+            if 'index' in nav_df.columns:
+                nav_df = nav_df.rename(columns={'index': 'date'})
+            nav_df['date'] = pd.to_datetime(nav_df['date']).dt.normalize()
+            
+            # 透视为宽格式
+            etf_nav_data = nav_df.pivot_table(
+                index='date',
+                columns='code',
+                values='nav',
+                aggfunc='first'
+            )
+            
+            # 对齐到主市场交易日，并前向填充缺失值（净值数据通常T+1更新）
+            if tushare_codes:
+                etf_nav_data = etf_nav_data.reindex(primary_dates)
+                etf_nav_data = etf_nav_data.ffill()  # 前向填充缺失的净值数据
+            
+            print(f"  ETF净值数据: {len(etf_nav_data.columns)} 只")
+        
         # 获取基准数据
         benchmark_data = self.fetch_single(benchmark_code, start_date, end_date)
         if benchmark_data is not None:
-            # 标准化日期索引（无时区，只保留日期部分）
             benchmark_data.index = pd.to_datetime(benchmark_data.index, utc=True).tz_localize(None).normalize()
-            print(f"  ✓ 基准 {benchmark_code}: {len(benchmark_data)} 条")
+            # 对齐到主市场交易日
+            if tushare_codes:
+                benchmark_data = benchmark_data.reindex(primary_dates)
+            print(f"\n✓ 基准 {benchmark_code}: {len(benchmark_data)} 条")
 
-        return etf_data, benchmark_data, valid_codes
+        return index_data, etf_data, etf_nav_data, benchmark_data, valid_codes
 
     def __enter__(self):
         """上下文管理器入口"""

core/factors/momentum.py

@@ -80,110 +80,74 @@ def calculate_daily_return(price_series: pd.Series) -> pd.Series:
 
 
 def compute_factors(
-    etf_data: pd.DataFrame,
+    index_data: pd.DataFrame,
     code_list: list,
     n: int = 25,
     factor_type: str = "slope_r2",
+    etf_data: pd.DataFrame = None,
+    code_config: dict = None,
 ) -> tuple[pd.DataFrame, list]:
     """
-    计算所有指数的因子和日收益率
-    支持长格式数据（混合数据源：Tushare + YFinance）
-
+    计算所有指数的因子和日收益率（支持指数-ETF双轨数据）
+    
     Args:
-        etf_data: DataFrame, 长格式数据，包含 [code, close, source] 列
+        index_data: 指数价格数据（宽格式，用于因子计算）
         code_list: 指数代码列表
         n: 动量/趋势窗口
         factor_type: 'momentum' 或 'slope_r2'
+        etf_data: ETF价格数据（宽格式，用于收益计算）
+        code_config: 代码配置字典 {code: {name, etf, market}}，用于判断是否为加密货币
 
     Returns:
         tuple: (result_df, valid_codes)
+        - result_df: 包含因子得分和日收益率的DataFrame
+        - valid_codes: 有效代码列表
     """
-    # 检查数据格式
-    if 'code' in etf_data.columns:
-        # 长格式数据 - 按 code 分别计算因子（旧逻辑，保留兼容）
-        all_factors = []
-        valid_codes = []
-
-        for code in code_list:
-            code_data = etf_data[etf_data['code'] == code].copy()
-            if len(code_data) == 0:
-                print(f"  ⚠ 跳过 {code}: 不在数据中")
-                continue
-
-            # 检查缺失值
-            null_pct = code_data['close'].isnull().sum() / len(code_data)
-            if null_pct > 0.2:
-                print(f"  ⚠ 剔除 {code}: 缺失率 {null_pct:.1%} 过高")
-                continue
-
-            # 按日期排序
-            code_data = code_data.sort_index()
-
-            # 计算日收益率和因子
-            code_data[f"日收益率_{code}"] = calculate_daily_return(code_data['close'])
-
-            if factor_type == "momentum":
-                code_data[f"得分_{code}"] = calculate_momentum(code_data['close'], n)
-            elif factor_type == "slope_r2":
-                code_data[f"得分_{code}"] = calculate_slope_r2(code_data['close'], n)
-            else:
-                raise ValueError(f"不支持的因子类型: {factor_type}")
-
-            # 保留需要的列
-            code_data = code_data[[f"日收益率_{code}", f"得分_{code}"]]
-            all_factors.append(code_data)
+    code_config = code_config or {}
+    
+    # 如果没有提供ETF数据，创建一个空的DataFrame
+    if etf_data is None:
+        etf_data = pd.DataFrame()
+    
+    result = index_data.copy()
+    
+    # 过滤掉缺失值过多的指数
+    total_rows = len(result)
+    valid_codes = []
+    for code in code_list:
+        if code not in result.columns:
+            print(f"  ⚠ 跳过 {code}: 不在数据中")
+            continue
+        null_pct = result[code].isnull().sum() / total_rows
+        if null_pct > 0.2:
+            print(f"  ⚠ 剔除 {code}: 缺失率 {null_pct:.1%} 过高")
+            result = result.drop(columns=[code])
+        else:
             valid_codes.append(code)
 
-        if not all_factors:
-            raise ValueError("没有有效的指数数据")
+    # 对有效指数计算因子和收益率
+    for code in valid_codes:
+        # 因子基于指数价格计算
+        if factor_type == "momentum":
+            result[f"得分_{code}"] = calculate_momentum(result[code], n)
+        elif factor_type == "slope_r2":
+            result[f"得分_{code}"] = calculate_slope_r2(result[code], n)
+        else:
+            raise ValueError(f"不支持的因子类型: {factor_type}")
+        
+        # 日收益率基于指数价格计算（回测使用指数价格）
+        result[f"日收益率_{code}"] = calculate_daily_return(result[code])
 
-        # 合并所有因子的数据（按日期内连接 - 只保留所有指数都有数据的日期）
-        result = all_factors[0]
-        for df in all_factors[1:]:
-            result = result.join(df, how='inner')
-
-        # 删除所有得分都是 NaN 的行（即窗口期内的数据）
-        score_cols = [f"得分_{code}" for code in valid_codes]
-        # 只删除完全无法比较的行（所有得分都是NaN）
-        result = result.dropna(subset=score_cols, how='all')
-
-    else:
-        # 宽格式数据（向后兼容）
-        result = etf_data.copy()
-
-        # 过滤掉缺失值过多的指数
-        total_rows = len(result)
-        valid_codes = []
-        for code in code_list:
-            if code not in result.columns:
-                print(f"  ⚠ 跳过 {code}: 不在数据中")
-                continue
-            null_pct = result[code].isnull().sum() / total_rows
-            if null_pct > 0.2:
-                print(f"  ⚠ 剔除 {code}: 缺失率 {null_pct:.1%} 过高")
-                result = result.drop(columns=[code])
-            else:
-                valid_codes.append(code)
-
-        # 对有效指数计算因子
-        for code in valid_codes:
-            result[f"日收益率_{code}"] = calculate_daily_return(result[code])
-
-            if factor_type == "momentum":
-                result[f"得分_{code}"] = calculate_momentum(result[code], n)
-            elif factor_type == "slope_r2":
-                result[f"得分_{code}"] = calculate_slope_r2(result[code], n)
-            else:
-                raise ValueError(f"不支持的因子类型: {factor_type}")
-
-        # 按得分列做 dropna
-        score_cols = [f"得分_{code}" for code in valid_codes]
-        result = result.dropna(subset=score_cols)
+    # 按得分列做 dropna
+    score_cols = [f"得分_{code}" for code in valid_codes]
+    result = result.dropna(subset=score_cols)
 
     print("\n因子计算完成:")
     print(f"  因子类型: {factor_type}")
     print(f"  窗口天数: {n}")
     print(f"  有效指数: {len(valid_codes)}/{len(code_list)}")
     print(f"  有效数据: {len(result)} 行")
+    if etf_data is not index_data:
+        print(f"  使用ETF数据计算收益: ✓")
 
     return result, valid_codes