# -*- 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