Spaces:

sgfw
/

kronos

Running

kronos / webui /output /multifactor_feature_engineering.py

Upload 68 files

459d1c8 verified 5 months ago

22.5 kB

# -*- coding: utf-8 -*-
"""
多因子机器学习股价预测引擎 - 特征工程模块
============================================
四类前瞻因子 + 市场状态特征 + 因子交叉 + 基本面拼接
纯NumPy/Pandas实现，支持CPU环境
"""

import numpy as np
import pandas as pd
from collections import OrderedDict
import warnings
warnings.filterwarnings('ignore')
import json


class MultiFactorFeatureEngineer:
    """多因子特征工程"""
    
    def __init__(self, sequence_length=20):
        self.seq_len = sequence_length
        self.feature_names = []
        self._build_feature_schema()
    
    def _build_feature_schema(self):
        """构建完整特征名列表"""
        self.feature_names = [
            # 市场状态特征 (21维)
            'ma5', 'ma10', 'ma20', 'ma60', 'ma120',
            'macd_dif', 'macd_dea', 'macd_histogram',
            'k_val', 'd_val', 'j_val',
            'atr_14', 'boll_upper', 'boll_middle', 'boll_lower', 'boll_width',
            'vol_ma5', 'vol_ma10', 'vol_ratio', 'obv',
            'wave_position',
            # 订单流因子 (3维)
            'bid_ask_imbalance', 'active_large_buy_ratio', 'vwap_deviation',
            # 资金行为因子 (3维)
            'northbound_net_flow', 'margin_change', 'inst_research_anomaly',
            # 预期修正因子 (3维)
            'eps_revision_ratio', 'news_sentiment_score', 'iv_skew',
            # 政策宏观因子 (2维)
            'industry_policy_heat', 'term_spread_change',
            # 基本面特征 (8维)
            'financial_quality', 'roe', 'debt_ratio', 'revenue_growth',
            'profit_growth', 'pe_ratio', 'pb_ratio', 'dividend_yield',
            # 交叉特征 (6维)
            'cross_rsi_north', 'cross_macd_large', 'cross_vol_eps',
            'cross_atr_research', 'cross_wave_news', 'cross_boll_iv',
        ]
    
    def compute_market_state_features(self, df):
        """计算市场状态特征（传统技术指标降权）"""
        features = pd.DataFrame(index=df.index)
        
        close = df['close'].astype(float)
        high = df['high'].astype(float)
        low = df['low'].astype(float)
        volume = df['volume'].astype(float)
        
        # 均线系统
        for period in [5, 10, 20, 60, 120]:
            features[f'ma{period}'] = close.rolling(period).mean()
        
        # MACD
        ema12 = close.ewm(span=12, adjust=False).mean()
        ema26 = close.ewm(span=26, adjust=False).mean()
        features['macd_dif'] = ema12 - ema26
        features['macd_dea'] = features['macd_dif'].ewm(span=9, adjust=False).mean()
        features['macd_histogram'] = features['macd_dif'] - features['macd_dea']
        
        # KDJ
        low_min = low.rolling(9).min()
        high_max = high.rolling(9).max()
        rsv = (close - low_min) / (high_max - low_min + 1e-10) * 100
        features['k_val'] = rsv.ewm(com=2, adjust=False).mean()
        features['d_val'] = features['k_val'].ewm(com=2, adjust=False).mean()
        features['j_val'] = 3 * features['k_val'] - 2 * features['d_val']
        
        # ATR
        tr = pd.concat([
            high - low,
            (high - close.shift(1)).abs(),
            (low - close.shift(1)).abs()
        ], axis=1).max(axis=1)
        features['atr_14'] = tr.rolling(14).mean()
        
        # 布林带
        features['boll_middle'] = close.rolling(20).mean()
        std20 = close.rolling(20).std()
        features['boll_upper'] = features['boll_middle'] + 2 * std20
        features['boll_lower'] = features['boll_middle'] - 2 * std20
        features['boll_width'] = (features['boll_upper'] - features['boll_lower']) / (features['boll_middle'] + 1e-10)
        
        # 量能特征
        features['vol_ma5'] = volume.rolling(5).mean()
        features['vol_ma10'] = volume.rolling(10).mean()
        features['vol_ratio'] = volume / (features['vol_ma5'] + 1e-10)
        
        # OBV
        obv = pd.Series(0, index=df.index, dtype=float)
        for i in range(1, len(df)):
            if close.iloc[i] > close.iloc[i-1]:
                obv.iloc[i] = obv.iloc[i-1] + volume.iloc[i]
            elif close.iloc[i] < close.iloc[i-1]:
                obv.iloc[i] = obv.iloc[i-1] - volume.iloc[i]
            else:
                obv.iloc[i] = obv.iloc[i-1]
        features['obv'] = obv
        
        # 波浪位置（简化RSI映射）
        delta = close.diff()
        gain = delta.where(delta > 0, 0).rolling(14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
        rs = gain / (loss + 1e-10)
        rsi = 100 - 100 / (1 + rs)
        features['wave_position'] = rsi / 100.0  # 归一化到0-1
        
        return features
    
    def compute_order_flow_features(self, df):
        """
        计算订单流因子
        基于tick数据模拟（用日线量价近似）
        """
        features = pd.DataFrame(index=df.index)
        close = df['close'].astype(float)
        high = df['high'].astype(float)
        low = df['low'].astype(float)
        open_ = df['open'].astype(float)
        volume = df['volume'].astype(float)
        amount = df.get('amount', volume * close).astype(float)
        
        # 1. 买卖盘口不平衡度（日线近似）
        # 用日内振幅与收盘位置模拟
        day_range = high - low + 1e-10
        close_position = (close - low) / day_range  # 0=最低, 1=最高
        features['bid_ask_imbalance'] = 2 * close_position - 1  # -1到1
        features['bid_ask_imbalance'] = features['bid_ask_imbalance'].ewm(span=5).mean()
        
        # 2. 主动大单净买入占比（用大阳线/大阴线比例近似）
        body = close - open_
        body_ratio = body / (day_range + 1e-10)  # 实体占振幅比
        vol_change = volume.pct_change().fillna(0)
        features['active_large_buy_ratio'] = (
            np.sign(body_ratio) * np.abs(body_ratio) * (1 + vol_change)
        ).ewm(span=5).mean()
        
        # 3. VWAP乖离率
        cum_amount = amount.cumsum()
        cum_vol = volume.cumsum()
        vwap = cum_amount / (cum_vol + 1e-10)
        features['vwap_deviation'] = (close - vwap) / (vwap + 1e-10)
        features['vwap_deviation'] = features['vwap_deviation'].rolling(20).mean()
        
        return features
    
    def compute_capital_flow_features(self, df):
        """
        计算资金行为因子
        用价格变动+量能变化模拟
        """
        features = pd.DataFrame(index=df.index)
        close = df['close'].astype(float)
        volume = df['volume'].astype(float)
        
        # 1. 北向资金分时净流入（用大单流入近似）
        # 多日连续上涨+放量 = 模拟北向流入
        price_momentum = close.pct_change(5).fillna(0)
        vol_momentum = volume.rolling(5).mean().pct_change(5).fillna(0)
        features['northbound_net_flow'] = (
            price_momentum * 0.6 + vol_momentum * 0.4
        ).ewm(span=10).mean()
        
        # 2. 融资盘边际变化（用杠杆近似）
        short_term_ma = close.rolling(5).mean()
        long_term_ma = close.rolling(60).mean()
        features['margin_change'] = (
            (short_term_ma - long_term_ma) / (long_term_ma + 1e-10)
        ).ewm(span=10).mean()
        
        # 3. 机构调研异常度（用波动率异常近似）
        daily_ret = close.pct_change().fillna(0)
        vol_20 = daily_ret.rolling(20).std()
        vol_60 = daily_ret.rolling(60).std()
        features['inst_research_anomaly'] = (
            (vol_20 / (vol_60 + 1e-10)) - 1
        ).clip(-3, 3)
        
        return features
    
    def compute_expectation_features(self, df):
        """
        计算预期修正因子
        用价格趋势和动量模拟
        """
        features = pd.DataFrame(index=df.index)
        close = df['close'].astype(float)
        high = df['high'].astype(float)
        low = df['low'].astype(float)
        volume = df['volume'].astype(float)
        
        # 1. 分析师一致预期EPS上调比例（用盈余惊喜近似）
        # 用实际价格走势vs均线的偏离来模拟预期变化
        est_earnings = close.rolling(20).mean()  # 隐含盈利趋势
        actual_vs_est = (close - est_earnings.shift(20)) / (est_earnings.shift(20) + 1e-10)
        features['eps_revision_ratio'] = actual_vs_est.rolling(10).mean().clip(-0.5, 0.5)
        
        # 2. 新闻情感得分（用价格动量和波动率模拟）
        ret_5d = close.pct_change(5).fillna(0)
        ret_20d = close.pct_change(20).fillna(0)
        momentum_shift = ret_5d - ret_20d  # 短期vs长期动量差
        features['news_sentiment_score'] = np.tanh(momentum_shift * 10).clip(-1, 1)
        
        # 3. 期权隐含波动率偏斜（用put/call近似）
        # 用下行风险vs上行空间的比来模拟偏斜
        downside = (low.rolling(10).min() - close) / (close + 1e-10)
        upside = (high.rolling(10).max() - close) / (close + 1e-10)
        features['iv_skew'] = (
            (np.abs(downside) - np.abs(upside)) / (np.abs(downside) + np.abs(upside) + 1e-10)
        ).rolling(5).mean()
        
        return features
    
    def compute_policy_macro_features(self, df):
        """
        计算政策与宏观因子
        用市场广度和板块联动近似
        """
        features = pd.DataFrame(index=df.index)
        close = df['close'].astype(float)
        volume = df['volume'].astype(float)
        
        # 1. 行业政策主题热度（用放量程度+趋势持续性模拟）
        vol_ratio = volume / (volume.rolling(20).mean() + 1e-10)
        trend_persist = (close > close.shift(1)).rolling(10).mean()
        features['industry_policy_heat'] = (
            np.tanh(vol_ratio - 1) * 0.5 + np.tanh(trend_persist - 0.5) * 0.5
        ).clip(-1, 1)
        
        # 2. 期限利差变化（用短期vs长期均线斜率差近似）
        ma5_slope = close.rolling(5).apply(lambda x: np.polyfit(range(len(x)), x, 1)[0] if len(x) >= 5 else 0, raw=True)
        ma60_slope = close.rolling(60).apply(lambda x: np.polyfit(range(len(x)), x, 1)[0] if len(x) >= 60 else 0, raw=True)
        features['term_spread_change'] = (
            ma5_slope - ma60_slope * 0.1
        ).clip(-0.5, 0.5)
        
        return features
    
    def compute_cross_features(self, state_df, forward_df):
        """计算状态类 × 前瞻类交叉特征"""
        cross = pd.DataFrame(index=state_df.index)
        
        # RSI超卖 AND 北向流入加速
        rsi_proxy = state_df['wave_position']  # 已归一化RSI
        north = forward_df.get('northbound_net_flow', pd.Series(0, index=state_df.index))
        cross['cross_rsi_north'] = np.tanh((1 - rsi_proxy) * 3) * np.tanh(north * 10)
        
        # MACD金叉 AND 主动大单净买入
        macd_hist = state_df.get('macd_histogram', pd.Series(0, index=state_df.index))
        large_buy = forward_df.get('active_large_buy_ratio', pd.Series(0, index=state_df.index))
        cross['cross_macd_large'] = np.tanh(macd_hist * 10) * np.tanh(large_buy * 5)
        
        # 量能突破 AND EPS修正
        vol_ratio = state_df.get('vol_ratio', pd.Series(1, index=state_df.index))
        eps_rev = forward_df.get('eps_revision_ratio', pd.Series(0, index=state_df.index))
        cross['cross_vol_eps'] = np.tanh((vol_ratio - 1) * 5) * np.tanh(eps_rev * 10)
        
        # 低波动 AND 机构调研
        atr = state_df.get('atr_14', pd.Series(0, index=state_df.index))
        inst_res = forward_df.get('inst_research_anomaly', pd.Series(0, index=state_df.index))
        cross['cross_atr_research'] = np.tanh(-atr * 50) * np.tanh(inst_res * 3)
        
        # 波浪支撑 AND 新闻情感
        wave = state_df.get('wave_position', pd.Series(0.5, index=state_df.index))
        news = forward_df.get('news_sentiment_score', pd.Series(0, index=state_df.index))
        cross['cross_wave_news'] = np.tanh((0.3 - wave) * 5) * news
        
        # 布林挤压 AND IV偏斜
        boll_w = state_df.get('boll_width', pd.Series(0, index=state_df.index))
        iv_sk = forward_df.get('iv_skew', pd.Series(0, index=state_df.index))
        cross['cross_boll_iv'] = np.tanh(-boll_w * 20) * iv_sk
        
        return cross
    
    def compute_fundamental_features(self, df, financial_data=None):
        """
        计算基本面特征
        使用腾讯财经/东方财富HTTP API获取财务数据
        """
        features = pd.DataFrame(index=df.index)
        close = df['close'].astype(float)
        
        if financial_data is not None and len(financial_data) > 0:
            # 有真实财务数据
            features['financial_quality'] = financial_data.get('roe', 0)
            features['roe'] = financial_data.get('roe', 0)
            features['debt_ratio'] = financial_data.get('debtRatio', 0)
            features['revenue_growth'] = financial_data.get('yoyEquity', 0)
            features['profit_growth'] = financial_data.get('yoyAsset', 0)
            features['pe_ratio'] = financial_data.get('pe', 15)
            features['pb_ratio'] = financial_data.get('pb', 2)
            features['dividend_yield'] = financial_data.get('dv_ttm', 0) / 100
        else:
            # 模拟基本面（用价格长期趋势近似）
            ret_60 = close.pct_change(60).fillna(0)
            ret_250 = close.pct_change(250).fillna(0)
            vol_60 = close.pct_change().rolling(60).std().fillna(0.02)
            
            features['financial_quality'] = np.tanh(ret_60 * 5) * 0.5 + 0.5
            features['roe'] = np.clip(ret_60 * 3 + 0.1, -0.1, 0.3)
            features['debt_ratio'] = np.clip(0.5 + ret_250 * 0.5, 0.1, 0.9)
            features['revenue_growth'] = np.clip(ret_60, -0.5, 1.0)
            features['profit_growth'] = np.clip(ret_60 * 1.5, -0.5, 1.5)
            features['pe_ratio'] = np.clip(15 / (1 + ret_60), 5, 100)
            features['pb_ratio'] = np.clip(2 / (1 + ret_60), 0.5, 10)
            features['dividend_yield'] = np.clip(0.03 / (1 + abs(ret_60)), 0.005, 0.08)
        
        return features
    
    def extract_all_features(self, df, financial_data=None):
        """提取所有特征并标准化"""
        # 各类特征
        market_state = self.compute_market_state_features(df)
        order_flow = self.compute_order_flow_features(df)
        capital_flow = self.compute_capital_flow_features(df)
        expectation = self.compute_expectation_features(df)
        policy_macro = self.compute_policy_macro_features(df)
        cross = self.compute_cross_features(market_state, pd.concat([order_flow, capital_flow, expectation, policy_macro], axis=1))
        fundamental = self.compute_fundamental_features(df, financial_data)
        
        # 合并
        all_features = pd.concat([
            market_state, order_flow, capital_flow,
            expectation, policy_macro, fundamental, cross
        ], axis=1)
        
        # 标准化（滚动Z-score）
        all_features = all_features.replace([np.inf, -np.inf], np.nan)
        for col in all_features.columns:
            rolling_mean = all_features[col].rolling(60, min_periods=20).mean()
            rolling_std = all_features[col].rolling(60, min_periods=20).std()
            all_features[col] = (all_features[col] - rolling_mean) / (rolling_std + 1e-10)
        all_features = all_features.fillna(0).clip(-5, 5)
        
        return all_features
    
    def create_sequences(self, features_df, df, seq_len=None):
        """创建时间序列样本"""
        if seq_len is None:
            seq_len = self.seq_len
        
        close = df['close'].astype(float).values
        
        # 标签：未来5日趋势方向、变化率、波动率
        labels = np.zeros((len(df), 3))
        for i in range(len(df) - 5):
            future_ret = (close[i+5] - close[i]) / (close[i] + 1e-10)
            future_vol = np.std(np.diff(close[max(0,i):i+6]) / (np.diff(close[max(0,i):i+6]) + 1e-10))
            labels[i, 0] = np.sign(future_ret)      # 趋势方向
            labels[i, 1] = np.tanh(future_ret * 5)   # 变化率
            labels[i, 2] = np.clip(future_vol, 0, 1) # 波动率
        
        feat_values = features_df.values
        valid_len = min(len(feat_values), len(close))
        
        X_cont = []
        labels_list = []
        
        for i in range(seq_len, valid_len - 5):
            X_cont.append(feat_values[i-seq_len:i])
            labels_list.append(labels[i])
        
        X_cont = np.array(X_cont, dtype=np.float32)
        labels = np.array(labels_list, dtype=np.float32)
        
        # 无类别特征（纯连续）
        X_cat = np.zeros((len(X_cont), seq_len, 1), dtype=np.int32)
        
        return X_cont, X_cat, labels
    
    def get_feature_info(self):
        """返回特征信息"""
        info = OrderedDict()
        info['total_features'] = len(self.feature_names)
        info['feature_names'] = self.feature_names
        info['categories'] = {
            'market_state': self.feature_names[:21],
            'order_flow': self.feature_names[21:24],
            'capital_flow': self.feature_names[24:27],
            'expectation': self.feature_names[27:30],
            'policy_macro': self.feature_names[30:32],
            'fundamental': self.feature_names[32:40],
            'cross': self.feature_names[40:46],
        }
        return info


class MultiFactorDataFetcher:
    """数据获取器 - 使用腾讯财经HTTP API（稳定快速，纯HTTP API无外部依赖）"""
    
    def fetch_stock_data(self, code, start_date='2020-01-01', end_date='2026-05-31'):
        """通过腾讯财经API获取股票日线数据"""
        import urllib.request
        
        try:
            # 转换代码格式：sh.600519 -> sh600519
            clean_code = code.replace('.', '')
            
            # 判断市场
            if clean_code.startswith('6') or clean_code.startswith('9'):
                market = 'sh'
            else:
                market = 'sz'
            
            param_code = f"{market}{clean_code}"
            
            # 日期格式
            start_fmt = start_date.replace('-', '')
            end_fmt = end_date.replace('-', '')
            
            # 腾讯API不支持日期范围，获取全部后程序端过滤
            url = f'https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={param_code},day,,,1000,qfq'
            req = urllib.request.Request(url, headers={
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            })
            resp = urllib.request.urlopen(req, timeout=10)
            data_json = json.loads(resp.read().decode())
            stock_data = data_json.get('data', {}).get(param_code, {})
            klines = stock_data.get('qfqday') or stock_data.get('day', [])
            
            if not klines or not isinstance(klines, list):
                print(f"[WARN] 腾讯API无数据: {code}")
                return None
            
            # 腾讯qfqday: [date, open, close, high, low, volume(手)]
            records = []
            for item in klines:
                if isinstance(item, list) and len(item) >= 6 and float(item[2]) > 0:
                    records.append({
                        'date': item[0],
                        'open': float(item[1]) if item[1] else 0,
                        'high': float(item[3]) if len(item) > 3 and item[3] else 0,
                        'low': float(item[4]) if len(item) > 4 and item[4] else 0,
                        'close': float(item[2]) if item[2] else 0,
                        'volume': float(item[5]) * 100 if item[5] else 0,  # 手转股
                        'amount': 0,
                        'turn': 0,
                    })
            
            if not records:
                return None
            
            df = pd.DataFrame(records)
            df['date'] = pd.to_datetime(df['date'])
            df = df.sort_values('date').reset_index(drop=True)
            df = df.dropna(subset=['close', 'volume'])
            df = df[df['volume'] > 0].reset_index(drop=True)
            
            return df
            
        except Exception as e:
            print(f"[ERROR] 腾讯API获取失败 {code}: {e}")
            return None
    
    def fetch_financial_data(self, code):
        """获取财务数据（使用东方财富HTTP API）"""
        import urllib.request
        
        try:
            clean_code = code.replace('.', '')
            # 东方财富个股信息接口
            if clean_code.startswith('6'):
                secid = f'1.{clean_code}'
            else:
                secid = f'0.{clean_code}'
            
            url = f'https://push2.eastmoney.com/api/qt/stock/get?secid={secid}&fields=f57,f58,f162,f163,f167,f168,f169,f170,f171,f173'
            req = urllib.request.Request(url, headers={
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            })
            resp = urllib.request.urlopen(req, timeout=10)
            data = json.loads(resp.read().decode())
            d = data.get('data', {})
            
            if not d:
                return None
            
            return {
                'roe': float(d.get('f162', 0)) / 100 if d.get('f162') else 0,
                'pe': float(d.get('f167', 15)) / 100 if d.get('f167') else 15,
                'pb': float(d.get('f168', 2)) / 100 if d.get('f168') else 2,
                'dv_ttm': float(d.get('f173', 0)) if d.get('f173') else 0,
                'total_mv': float(d.get('f116', 0)) if d.get('f116') else 0,
                'circ_mv': float(d.get('f117', 0)) if d.get('f117') else 0,
            }
        except Exception as e:
            print(f"[WARN] 东方财富财务数据获取失败: {e}")
            return None

	# -- coding: utf-8 --
	"""
	多因子机器学习股价预测引擎 - 特征工程模块
	============================================
	四类前瞻因子 + 市场状态特征 + 因子交叉 + 基本面拼接
	纯NumPy/Pandas实现，支持CPU环境
	"""

	import numpy as np
	import pandas as pd
	from collections import OrderedDict
	import warnings
	warnings.filterwarnings('ignore')
	import json


	class MultiFactorFeatureEngineer:
	"""多因子特征工程"""

	def __init__(self, sequence_length=20):
	self.seq_len = sequence_length
	self.feature_names = []
	self._build_feature_schema()

	def _build_feature_schema(self):
	"""构建完整特征名列表"""
	self.feature_names = [
	# 市场状态特征 (21维)
	'ma5', 'ma10', 'ma20', 'ma60', 'ma120',
	'macd_dif', 'macd_dea', 'macd_histogram',
	'k_val', 'd_val', 'j_val',
	'atr_14', 'boll_upper', 'boll_middle', 'boll_lower', 'boll_width',
	'vol_ma5', 'vol_ma10', 'vol_ratio', 'obv',
	'wave_position',
	# 订单流因子 (3维)
	'bid_ask_imbalance', 'active_large_buy_ratio', 'vwap_deviation',
	# 资金行为因子 (3维)
	'northbound_net_flow', 'margin_change', 'inst_research_anomaly',
	# 预期修正因子 (3维)
	'eps_revision_ratio', 'news_sentiment_score', 'iv_skew',
	# 政策宏观因子 (2维)
	'industry_policy_heat', 'term_spread_change',
	# 基本面特征 (8维)
	'financial_quality', 'roe', 'debt_ratio', 'revenue_growth',
	'profit_growth', 'pe_ratio', 'pb_ratio', 'dividend_yield',
	# 交叉特征 (6维)
	'cross_rsi_north', 'cross_macd_large', 'cross_vol_eps',
	'cross_atr_research', 'cross_wave_news', 'cross_boll_iv',
	]

	def compute_market_state_features(self, df):
	"""计算市场状态特征（传统技术指标降权）"""
	features = pd.DataFrame(index=df.index)

	close = df['close'].astype(float)
	high = df['high'].astype(float)
	low = df['low'].astype(float)
	volume = df['volume'].astype(float)

	# 均线系统
	for period in [5, 10, 20, 60, 120]:
	features[f'ma{period}'] = close.rolling(period).mean()

	# MACD
	ema12 = close.ewm(span=12, adjust=False).mean()
	ema26 = close.ewm(span=26, adjust=False).mean()
	features['macd_dif'] = ema12 - ema26
	features['macd_dea'] = features['macd_dif'].ewm(span=9, adjust=False).mean()
	features['macd_histogram'] = features['macd_dif'] - features['macd_dea']

	# KDJ
	low_min = low.rolling(9).min()
	high_max = high.rolling(9).max()
	rsv = (close - low_min) / (high_max - low_min + 1e-10) * 100
	features['k_val'] = rsv.ewm(com=2, adjust=False).mean()
	features['d_val'] = features['k_val'].ewm(com=2, adjust=False).mean()
	features['j_val'] = 3 * features['k_val'] - 2 * features['d_val']

	# ATR
	tr = pd.concat([
	high - low,
	(high - close.shift(1)).abs(),
	(low - close.shift(1)).abs()
	], axis=1).max(axis=1)
	features['atr_14'] = tr.rolling(14).mean()

	# 布林带
	features['boll_middle'] = close.rolling(20).mean()
	std20 = close.rolling(20).std()
	features['boll_upper'] = features['boll_middle'] + 2 * std20
	features['boll_lower'] = features['boll_middle'] - 2 * std20
	features['boll_width'] = (features['boll_upper'] - features['boll_lower']) / (features['boll_middle'] + 1e-10)

	# 量能特征
	features['vol_ma5'] = volume.rolling(5).mean()
	features['vol_ma10'] = volume.rolling(10).mean()
	features['vol_ratio'] = volume / (features['vol_ma5'] + 1e-10)

	# OBV
	obv = pd.Series(0, index=df.index, dtype=float)
	for i in range(1, len(df)):
	if close.iloc[i] > close.iloc[i-1]:
	obv.iloc[i] = obv.iloc[i-1] + volume.iloc[i]
	elif close.iloc[i] < close.iloc[i-1]:
	obv.iloc[i] = obv.iloc[i-1] - volume.iloc[i]
	else:
	obv.iloc[i] = obv.iloc[i-1]
	features['obv'] = obv

	# 波浪位置（简化RSI映射）
	delta = close.diff()
	gain = delta.where(delta > 0, 0).rolling(14).mean()
	loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
	rs = gain / (loss + 1e-10)
	rsi = 100 - 100 / (1 + rs)
	features['wave_position'] = rsi / 100.0 # 归一化到0-1

	return features

	def compute_order_flow_features(self, df):
	"""
	计算订单流因子
	基于tick数据模拟（用日线量价近似）
	"""
	features = pd.DataFrame(index=df.index)
	close = df['close'].astype(float)
	high = df['high'].astype(float)
	low = df['low'].astype(float)
	open_ = df['open'].astype(float)
	volume = df['volume'].astype(float)
	amount = df.get('amount', volume * close).astype(float)

	# 1. 买卖盘口不平衡度（日线近似）
	# 用日内振幅与收盘位置模拟
	day_range = high - low + 1e-10
	close_position = (close - low) / day_range # 0=最低, 1=最高
	features['bid_ask_imbalance'] = 2 * close_position - 1 # -1到1
	features['bid_ask_imbalance'] = features['bid_ask_imbalance'].ewm(span=5).mean()

	# 2. 主动大单净买入占比（用大阳线/大阴线比例近似）
	body = close - open_
	body_ratio = body / (day_range + 1e-10) # 实体占振幅比
	vol_change = volume.pct_change().fillna(0)
	features['active_large_buy_ratio'] = (
	np.sign(body_ratio) * np.abs(body_ratio) * (1 + vol_change)
	).ewm(span=5).mean()

	# 3. VWAP乖离率
	cum_amount = amount.cumsum()
	cum_vol = volume.cumsum()
	vwap = cum_amount / (cum_vol + 1e-10)
	features['vwap_deviation'] = (close - vwap) / (vwap + 1e-10)
	features['vwap_deviation'] = features['vwap_deviation'].rolling(20).mean()

	return features

	def compute_capital_flow_features(self, df):
	"""
	计算资金行为因子
	用价格变动+量能变化模拟
	"""
	features = pd.DataFrame(index=df.index)
	close = df['close'].astype(float)
	volume = df['volume'].astype(float)

	# 1. 北向资金分时净流入（用大单流入近似）
	# 多日连续上涨+放量 = 模拟北向流入
	price_momentum = close.pct_change(5).fillna(0)
	vol_momentum = volume.rolling(5).mean().pct_change(5).fillna(0)
	features['northbound_net_flow'] = (
	price_momentum * 0.6 + vol_momentum * 0.4
	).ewm(span=10).mean()

	# 2. 融资盘边际变化（用杠杆近似）
	short_term_ma = close.rolling(5).mean()
	long_term_ma = close.rolling(60).mean()
	features['margin_change'] = (
	(short_term_ma - long_term_ma) / (long_term_ma + 1e-10)
	).ewm(span=10).mean()

	# 3. 机构调研异常度（用波动率异常近似）
	daily_ret = close.pct_change().fillna(0)
	vol_20 = daily_ret.rolling(20).std()
	vol_60 = daily_ret.rolling(60).std()
	features['inst_research_anomaly'] = (
	(vol_20 / (vol_60 + 1e-10)) - 1
	).clip(-3, 3)

	return features

	def compute_expectation_features(self, df):
	"""
	计算预期修正因子
	用价格趋势和动量模拟
	"""
	features = pd.DataFrame(index=df.index)
	close = df['close'].astype(float)
	high = df['high'].astype(float)
	low = df['low'].astype(float)
	volume = df['volume'].astype(float)

	# 1. 分析师一致预期EPS上调比例（用盈余惊喜近似）
	# 用实际价格走势vs均线的偏离来模拟预期变化
	est_earnings = close.rolling(20).mean() # 隐含盈利趋势
	actual_vs_est = (close - est_earnings.shift(20)) / (est_earnings.shift(20) + 1e-10)
	features['eps_revision_ratio'] = actual_vs_est.rolling(10).mean().clip(-0.5, 0.5)

	# 2. 新闻情感得分（用价格动量和波动率模拟）
	ret_5d = close.pct_change(5).fillna(0)
	ret_20d = close.pct_change(20).fillna(0)
	momentum_shift = ret_5d - ret_20d # 短期vs长期动量差
	features['news_sentiment_score'] = np.tanh(momentum_shift * 10).clip(-1, 1)

	# 3. 期权隐含波动率偏斜（用put/call近似）
	# 用下行风险vs上行空间的比来模拟偏斜
	downside = (low.rolling(10).min() - close) / (close + 1e-10)
	upside = (high.rolling(10).max() - close) / (close + 1e-10)
	features['iv_skew'] = (
	(np.abs(downside) - np.abs(upside)) / (np.abs(downside) + np.abs(upside) + 1e-10)
	).rolling(5).mean()

	return features

	def compute_policy_macro_features(self, df):
	"""
	计算政策与宏观因子
	用市场广度和板块联动近似
	"""
	features = pd.DataFrame(index=df.index)
	close = df['close'].astype(float)
	volume = df['volume'].astype(float)

	# 1. 行业政策主题热度（用放量程度+趋势持续性模拟）
	vol_ratio = volume / (volume.rolling(20).mean() + 1e-10)
	trend_persist = (close > close.shift(1)).rolling(10).mean()
	features['industry_policy_heat'] = (
	np.tanh(vol_ratio - 1) * 0.5 + np.tanh(trend_persist - 0.5) * 0.5
	).clip(-1, 1)

	# 2. 期限利差变化（用短期vs长期均线斜率差近似）
	ma5_slope = close.rolling(5).apply(lambda x: np.polyfit(range(len(x)), x, 1)[0] if len(x) >= 5 else 0, raw=True)
	ma60_slope = close.rolling(60).apply(lambda x: np.polyfit(range(len(x)), x, 1)[0] if len(x) >= 60 else 0, raw=True)
	features['term_spread_change'] = (
	ma5_slope - ma60_slope * 0.1
	).clip(-0.5, 0.5)

	return features

	def compute_cross_features(self, state_df, forward_df):
	"""计算状态类 × 前瞻类交叉特征"""
	cross = pd.DataFrame(index=state_df.index)

	# RSI超卖 AND 北向流入加速
	rsi_proxy = state_df['wave_position'] # 已归一化RSI
	north = forward_df.get('northbound_net_flow', pd.Series(0, index=state_df.index))
	cross['cross_rsi_north'] = np.tanh((1 - rsi_proxy) * 3) * np.tanh(north * 10)

	# MACD金叉 AND 主动大单净买入
	macd_hist = state_df.get('macd_histogram', pd.Series(0, index=state_df.index))
	large_buy = forward_df.get('active_large_buy_ratio', pd.Series(0, index=state_df.index))
	cross['cross_macd_large'] = np.tanh(macd_hist * 10) * np.tanh(large_buy * 5)

	# 量能突破 AND EPS修正
	vol_ratio = state_df.get('vol_ratio', pd.Series(1, index=state_df.index))
	eps_rev = forward_df.get('eps_revision_ratio', pd.Series(0, index=state_df.index))
	cross['cross_vol_eps'] = np.tanh((vol_ratio - 1) * 5) * np.tanh(eps_rev * 10)

	# 低波动 AND 机构调研
	atr = state_df.get('atr_14', pd.Series(0, index=state_df.index))
	inst_res = forward_df.get('inst_research_anomaly', pd.Series(0, index=state_df.index))
	cross['cross_atr_research'] = np.tanh(-atr * 50) * np.tanh(inst_res * 3)

	# 波浪支撑 AND 新闻情感
	wave = state_df.get('wave_position', pd.Series(0.5, index=state_df.index))
	news = forward_df.get('news_sentiment_score', pd.Series(0, index=state_df.index))
	cross['cross_wave_news'] = np.tanh((0.3 - wave) * 5) * news

	# 布林挤压 AND IV偏斜
	boll_w = state_df.get('boll_width', pd.Series(0, index=state_df.index))
	iv_sk = forward_df.get('iv_skew', pd.Series(0, index=state_df.index))
	cross['cross_boll_iv'] = np.tanh(-boll_w * 20) * iv_sk

	return cross

	def compute_fundamental_features(self, df, financial_data=None):
	"""
	计算基本面特征
	使用腾讯财经/东方财富HTTP API获取财务数据
	"""
	features = pd.DataFrame(index=df.index)
	close = df['close'].astype(float)

	if financial_data is not None and len(financial_data) > 0:
	# 有真实财务数据
	features['financial_quality'] = financial_data.get('roe', 0)
	features['roe'] = financial_data.get('roe', 0)
	features['debt_ratio'] = financial_data.get('debtRatio', 0)
	features['revenue_growth'] = financial_data.get('yoyEquity', 0)
	features['profit_growth'] = financial_data.get('yoyAsset', 0)
	features['pe_ratio'] = financial_data.get('pe', 15)
	features['pb_ratio'] = financial_data.get('pb', 2)
	features['dividend_yield'] = financial_data.get('dv_ttm', 0) / 100
	else:
	# 模拟基本面（用价格长期趋势近似）
	ret_60 = close.pct_change(60).fillna(0)
	ret_250 = close.pct_change(250).fillna(0)
	vol_60 = close.pct_change().rolling(60).std().fillna(0.02)

	features['financial_quality'] = np.tanh(ret_60 * 5) * 0.5 + 0.5
	features['roe'] = np.clip(ret_60 * 3 + 0.1, -0.1, 0.3)
	features['debt_ratio'] = np.clip(0.5 + ret_250 * 0.5, 0.1, 0.9)
	features['revenue_growth'] = np.clip(ret_60, -0.5, 1.0)
	features['profit_growth'] = np.clip(ret_60 * 1.5, -0.5, 1.5)
	features['pe_ratio'] = np.clip(15 / (1 + ret_60), 5, 100)
	features['pb_ratio'] = np.clip(2 / (1 + ret_60), 0.5, 10)
	features['dividend_yield'] = np.clip(0.03 / (1 + abs(ret_60)), 0.005, 0.08)

	return features

	def extract_all_features(self, df, financial_data=None):
	"""提取所有特征并标准化"""
	# 各类特征
	market_state = self.compute_market_state_features(df)
	order_flow = self.compute_order_flow_features(df)
	capital_flow = self.compute_capital_flow_features(df)
	expectation = self.compute_expectation_features(df)
	policy_macro = self.compute_policy_macro_features(df)
	cross = self.compute_cross_features(market_state, pd.concat([order_flow, capital_flow, expectation, policy_macro], axis=1))
	fundamental = self.compute_fundamental_features(df, financial_data)

	# 合并
	all_features = pd.concat([
	market_state, order_flow, capital_flow,
	expectation, policy_macro, fundamental, cross
	], axis=1)

	# 标准化（滚动Z-score）
	all_features = all_features.replace([np.inf, -np.inf], np.nan)
	for col in all_features.columns:
	rolling_mean = all_features[col].rolling(60, min_periods=20).mean()
	rolling_std = all_features[col].rolling(60, min_periods=20).std()
	all_features[col] = (all_features[col] - rolling_mean) / (rolling_std + 1e-10)
	all_features = all_features.fillna(0).clip(-5, 5)

	return all_features

	def create_sequences(self, features_df, df, seq_len=None):
	"""创建时间序列样本"""
	if seq_len is None:
	seq_len = self.seq_len

	close = df['close'].astype(float).values

	# 标签：未来5日趋势方向、变化率、波动率
	labels = np.zeros((len(df), 3))
	for i in range(len(df) - 5):
	future_ret = (close[i+5] - close[i]) / (close[i] + 1e-10)
	future_vol = np.std(np.diff(close[max(0,i):i+6]) / (np.diff(close[max(0,i):i+6]) + 1e-10))
	labels[i, 0] = np.sign(future_ret) # 趋势方向
	labels[i, 1] = np.tanh(future_ret * 5) # 变化率
	labels[i, 2] = np.clip(future_vol, 0, 1) # 波动率

	feat_values = features_df.values
	valid_len = min(len(feat_values), len(close))

	X_cont = []
	labels_list = []

	for i in range(seq_len, valid_len - 5):
	X_cont.append(feat_values[i-seq_len:i])
	labels_list.append(labels[i])

	X_cont = np.array(X_cont, dtype=np.float32)
	labels = np.array(labels_list, dtype=np.float32)

	# 无类别特征（纯连续）
	X_cat = np.zeros((len(X_cont), seq_len, 1), dtype=np.int32)

	return X_cont, X_cat, labels

	def get_feature_info(self):
	"""返回特征信息"""
	info = OrderedDict()
	info['total_features'] = len(self.feature_names)
	info['feature_names'] = self.feature_names
	info['categories'] = {
	'market_state': self.feature_names[:21],
	'order_flow': self.feature_names[21:24],
	'capital_flow': self.feature_names[24:27],
	'expectation': self.feature_names[27:30],
	'policy_macro': self.feature_names[30:32],
	'fundamental': self.feature_names[32:40],
	'cross': self.feature_names[40:46],
	}
	return info


	class MultiFactorDataFetcher:
	"""数据获取器 - 使用腾讯财经HTTP API（稳定快速，纯HTTP API无外部依赖）"""

	def fetch_stock_data(self, code, start_date='2020-01-01', end_date='2026-05-31'):
	"""通过腾讯财经API获取股票日线数据"""
	import urllib.request

	try:
	# 转换代码格式：sh.600519 -> sh600519
	clean_code = code.replace('.', '')

	# 判断市场
	if clean_code.startswith('6') or clean_code.startswith('9'):
	market = 'sh'
	else:
	market = 'sz'

	param_code = f"{market}{clean_code}"

	# 日期格式
	start_fmt = start_date.replace('-', '')
	end_fmt = end_date.replace('-', '')

	# 腾讯API不支持日期范围，获取全部后程序端过滤
	url = f'https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={param_code},day,,,1000,qfq'
	req = urllib.request.Request(url, headers={
	'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
	})
	resp = urllib.request.urlopen(req, timeout=10)
	data_json = json.loads(resp.read().decode())
	stock_data = data_json.get('data', {}).get(param_code, {})
	klines = stock_data.get('qfqday') or stock_data.get('day', [])

	if not klines or not isinstance(klines, list):
	print(f"[WARN] 腾讯API无数据: {code}")
	return None

	# 腾讯qfqday: [date, open, close, high, low, volume(手)]
	records = []
	for item in klines:
	if isinstance(item, list) and len(item) >= 6 and float(item[2]) > 0:
	records.append({
	'date': item[0],
	'open': float(item[1]) if item[1] else 0,
	'high': float(item[3]) if len(item) > 3 and item[3] else 0,
	'low': float(item[4]) if len(item) > 4 and item[4] else 0,
	'close': float(item[2]) if item[2] else 0,
	'volume': float(item[5]) * 100 if item[5] else 0, # 手转股
	'amount': 0,
	'turn': 0,
	})

	if not records:
	return None

	df = pd.DataFrame(records)
	df['date'] = pd.to_datetime(df['date'])
	df = df.sort_values('date').reset_index(drop=True)
	df = df.dropna(subset=['close', 'volume'])
	df = df[df['volume'] > 0].reset_index(drop=True)

	return df

	except Exception as e:
	print(f"[ERROR] 腾讯API获取失败 {code}: {e}")
	return None

	def fetch_financial_data(self, code):
	"""获取财务数据（使用东方财富HTTP API）"""
	import urllib.request

	try:
	clean_code = code.replace('.', '')
	# 东方财富个股信息接口
	if clean_code.startswith('6'):
	secid = f'1.{clean_code}'
	else:
	secid = f'0.{clean_code}'

	url = f'https://push2.eastmoney.com/api/qt/stock/get?secid={secid}&fields=f57,f58,f162,f163,f167,f168,f169,f170,f171,f173'
	req = urllib.request.Request(url, headers={
	'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
	})
	resp = urllib.request.urlopen(req, timeout=10)
	data = json.loads(resp.read().decode())
	d = data.get('data', {})

	if not d:
	return None

	return {
	'roe': float(d.get('f162', 0)) / 100 if d.get('f162') else 0,
	'pe': float(d.get('f167', 15)) / 100 if d.get('f167') else 15,
	'pb': float(d.get('f168', 2)) / 100 if d.get('f168') else 2,
	'dv_ttm': float(d.get('f173', 0)) if d.get('f173') else 0,
	'total_mv': float(d.get('f116', 0)) if d.get('f116') else 0,
	'circ_mv': float(d.get('f117', 0)) if d.get('f117') else 0,
	}
	except Exception as e:
	print(f"[WARN] 东方财富财务数据获取失败: {e}")
	return None