# -*- coding: utf-8 -*- """ 希希量化分析系统 - 因子表达式引擎 基于QuantGPT expression_parser.py适配，去除Rust依赖和外部模块引用 支持60+算子：时序、截面、技术指标、基本面、算术/比较/逻辑运算、三元运算符 """ import numpy as np import pandas as pd import re import warnings warnings.filterwarnings('ignore') # ================================================================ # 允许的基本面列名（替代QuantGPT的fundamental_data模块） # ================================================================ _FUNDAMENTAL_COLUMNS = { # 估值 'pe', 'pe_ttm', 'pb', 'ps', 'ps_ttm', 'pcf', 'pcf_ttm', 'ev_ebitda', # 盈利能力 'roe', 'roa', 'gross_margin', 'net_margin', 'ebitda_margin', 'operating_margin', 'ebit_margin', # 成长性 'revenue_yoy', 'net_profit_yoy', 'eps_yoy', 'operating_revenue_yoy', 'revenue_qoq', 'net_profit_qoq', # 每股指标 'eps', 'eps_ttm', 'bvps', 'dps', 'operating_cashflow_ps', # 资产负债 'debt_ratio', 'current_ratio', 'quick_ratio', 'asset_turnover', 'equity_multiplier', 'cash_ratio', # 现金流 'operating_cashflow', 'free_cashflow', 'cashflow_to_debt', 'fcf_yoy', 'ocf_yoy', # 其他 'market_cap', 'float_cap', 'total_shares', 'float_shares', 'dividend_yield', 'retention_ratio', 'payout_ratio', 'institutional_ratio', 'north_flow', 'main_net_inflow', 'volume_ratio', 'amplitude', } # 价格/量相关列 _PRICE_COLUMNS = { 'open', 'high', 'low', 'close', 'volume', 'amount', 'pre_close', 'turnover', 'turnover_rate', 'change_pct', 'vwap', 'adj_factor', } # 所有允许的列名 ALLOWED_COLUMNS = _FUNDAMENTAL_COLUMNS | _PRICE_COLUMNS # Alpha101 别名映射 _ALPHA101_ALIASES = { 'delta': 'ts_delta', 'delay': 'ts_shift', 'correlation': 'ts_corr', 'covariance': 'ts_cov', 'rank': 'ts_rank', 'scale': 'ts_scale', 'signed_power': 'ts_signed_power', 'product': 'ts_product', 'stddev': 'ts_std', 'log': 'ts_log', 'sum': 'ts_sum', 'adv': 'ts_mean', 'min': 'ts_min', 'max': 'ts_max', 'argmax': 'ts_argmax', 'argmin': 'ts_argmin', 'av_diff': 'ts_av_diff', 'mean': 'ts_mean', 'indneutralize': 'group_neutralize', 'ts_decay_linear': 'decay_linear', 'ts_std_dev': 'ts_std', 'ts_regression': 'ts_corr', } class ExpressionParser: """因子表达式解析引擎 - 支持嵌套运算、时序/截面算子""" def __init__(self): self._operators = { # 时序算子 'ts_mean': self._calc_ts_mean, 'ts_std': self._calc_ts_std, 'ts_delta': self._calc_ts_delta, 'ts_rank': self._calc_ts_rank, 'ts_argmax': self._calc_ts_argmax, 'ts_argmin': self._calc_ts_argmin, 'ts_corr': self._calc_ts_corr, 'ts_cov': self._calc_ts_cov, 'decay_linear': self._calc_decay_linear, 'product': self._calc_product, 'ts_av_diff': self._calc_ts_av_diff, 'ts_zscore': self._calc_ts_zscore, 'ts_sum': self._calc_ts_sum, 'ts_min': self._calc_ts_min, 'ts_max': self._calc_ts_max, 'ts_shift': self._calc_ts_shift, 'ts_skewness': self._calc_ts_skewness, 'ts_kurtosis': self._calc_ts_kurtosis, 'ts_median': self._calc_ts_median, 'ts_quantile': self._calc_ts_quantile, 'ts_log': self._calc_ts_log, 'ts_abs': self._calc_ts_abs, 'ts_sign': self._calc_ts_sign, 'ts_pow': self._calc_ts_pow, 'ts_prod': self._calc_product, 'ts_co_kurtosis': lambda f, w: f, # 截面算子 'rank': self._calc_cross_rank, 'zscore': self._calc_cross_zscore, 'scale': self._calc_cross_scale, 'group_rank': self._calc_group_rank, 'group_zscore': self._calc_group_zscore, 'group_neutralize': self._calc_group_neutralize, # 技术指标 'ema': self._calc_ema, 'sma': self._calc_sma, 'wma': self._calc_wma, 'rsi': self._calc_rsi, 'macd': self._calc_macd, 'obv': self._calc_obv, 'atr': self._calc_atr, 'boll_upper': self._calc_boll_upper, 'boll_lower': self._calc_boll_lower, 'boll_mid': self._calc_boll_mid, 'kdj_k': self._calc_kdj_k, 'kdj_d': self._calc_kdj_d, 'wr': self._calc_wr, 'cci': self._calc_cci, # Alpha101别名已在上层解析时替换 } def parse(self, expression, data, group_col=None): """ 解析并计算因子表达式 Parameters ---------- expression : str 因子表达式，如 "ts_rank(ts_delta(close, 5), 20)" data : pd.DataFrame 包含价格/基本面列的DataFrame，columns中须有表达式引用的列 group_col : str, optional 分组列名（用于group_rank等算子） Returns ------- pd.Series : 计算结果 """ # 1. 替换Alpha101别名 expr = self._replace_aliases(expression) # 2. 替换三元运算符 a ? b : c → (a) if (a, b, c) expr = self._convert_ternary_operators(expr) # 3. 解析并计算 result = self._evaluate(expr, data, group_col) return result def parse_expression(expression, data, group_col=None): """便捷函数""" parser = ExpressionParser() return parser.parse(expression, data, group_col) # ──── 别名替换 ──── def _replace_aliases(self, expr): for alias, target in _ALPHA101_ALIASES.items(): pattern = rf'\b{alias}\s*\(' expr = re.sub(pattern, f'{target}(', expr) return expr # ──── 三元运算符转换 ──── def _convert_ternary_operators(self, expr): """将 a ? b : c 转换为嵌套函数调用""" while '?' in expr: # 找到最内层的 ?...:... idx = expr.rfind('?') colon_idx = expr.find(':', idx) if colon_idx == -1: break condition = expr[:idx].rstrip() true_val = expr[idx+1:colon_idx].strip() false_val = expr[colon_idx+1:].strip() expr = f'ternary({condition},{true_val},{false_val})' return expr # ──── 核心解析器 ──── def _evaluate(self, expr, data, group_col=None): """递归解析表达式树""" expr = expr.strip() # 括号包裹的表达式 if expr.startswith('(') and self._find_matching_paren(expr, 0) == len(expr) - 1: return self._evaluate(expr[1:-1], data, group_col) # 函数调用 match = re.match(r'^(\w+)\s*\(', expr) if match: func_name = match.group(1) left_paren_pos = match.end() - 1 # 左括号位置 # 用_find_matching_paren从左括号位置找匹配的右括号 paren_end = self._find_matching_paren(expr, left_paren_pos) if paren_end is not None and paren_end >= 0: args_str = expr[left_paren_pos+1:paren_end] else: # 退化处理 args_str = expr[match.end():] if args_str.endswith(')'): args_str = args_str[:-1] args = self._split_args(args_str) return self._call_function(func_name, args, data, group_col) # 算术/比较/逻辑运算符（从低优先级开始解析） for op in ['&&', '||']: idx = self._find_operator(expr, op, data) if idx is not None: left = self._evaluate(expr[:idx], data, group_col) right = self._evaluate(expr[idx+len(op):], data, group_col) if op == '&&': return left.astype(float) * right.astype(float) else: return (left.astype(float) + right.astype(float)).clip(upper=1) for op in ['>=', '<=', '!=', '==', '>', '<']: idx = self._find_operator(expr, op, data) if idx is not None: left = self._evaluate(expr[:idx], data, group_col) right = self._evaluate(expr[idx+len(op):], data, group_col) if op == '>': return (left > right).astype(float) elif op == '<': return (left < right).astype(float) elif op == '>=': return (left >= right).astype(float) elif op == '<=': return (left <= right).astype(float) elif op == '==': return (left == right).astype(float) elif op == '!=': return (left != right).astype(float) for op in ['+', '-']: idx = self._find_operator(expr, op, data) if idx is not None and idx > 0: left = self._evaluate(expr[:idx], data, group_col) right = self._evaluate(expr[idx+1:], data, group_col) if op == '+': return left.astype(float) + right.astype(float) else: return left.astype(float) - right.astype(float) for op in ['*', '/']: idx = self._find_operator(expr, op, data) if idx is not None and idx > 0: left = self._evaluate(expr[:idx], data, group_col) right = self._evaluate(expr[idx+1:], data, group_col) if op == '*': return left.astype(float) * right.astype(float) else: return left.astype(float) / right.astype(float).replace(0, np.nan) # 一元运算符 if expr.startswith('-'): return -self._evaluate(expr[1:], data, group_col) if expr.startswith('!'): return (1 - self._evaluate(expr[1:], data, group_col)).astype(float) if expr.startswith('abs(') or expr.startswith('Abs('): inner = expr[4:-1] if expr.endswith(')') else expr[4:] return self._evaluate(inner, data, group_col).abs() if expr.startswith('log('): inner = expr[4:-1] if expr.endswith(')') else expr[4:] return np.log(self._evaluate(inner, data, group_col).clip(lower=1e-10)) # 特殊变量 if expr == 'ternary': return pd.Series(0, index=data.index) # 列名 if expr in data.columns: return data[expr].astype(float) # 数值 try: return pd.Series(float(expr), index=data.index) except ValueError: pass raise ValueError(f"无法解析表达式: {expr}") def _find_matching_paren(self, expr, start): """找到与start位置左括号匹配的右括号位置""" depth = 0 for i in range(start, len(expr)): if expr[i] == '(': depth += 1 elif expr[i] == ')': depth -= 1 if depth == 0: return i return -1 def _split_args(self, args_str): """按逗号分割参数（尊重嵌套括号）""" args = [] depth = 0 current = '' for ch in args_str: if ch == '(': depth += 1 current += ch elif ch == ')': depth -= 1 current += ch elif ch == ',' and depth == 0: args.append(current.strip()) current = '' else: current += ch if current.strip(): args.append(current.strip()) return args def _find_operator(self, expr, op, data): """找到非括号内的运算符位置""" depth = 0 in_func = False for i in range(len(expr)): ch = expr[i] if ch == '(': depth += 1 elif ch == ')': depth -= 1 elif depth == 0: if expr[i:i+len(op)] == op: # 确保不是函数名的一部分 if i > 0 and expr[i-1].isalpha(): continue if i + len(op) < len(expr) and expr[i+len(op)].isalpha(): continue return i return None def _call_function(self, func_name, args, data, group_col): """调用算子函数""" if func_name == 'ternary': if len(args) == 3: cond = self._evaluate(args[0], data, group_col) true_v = self._evaluate(args[1], data, group_col) false_v = self._evaluate(args[2], data, group_col) return np.where(cond > 0, true_v, false_v) raise ValueError("ternary需要3个参数: condition, true_val, false_val") if func_name == 'group_rank' and group_col and group_col not in data.columns: args_eval = [self._evaluate(a, data, group_col) for a in args] return args_eval[0] if len(args_eval) == 1 else pd.concat(args_eval, axis=1).iloc[:, -1] if func_name in self._operators: args_eval = [] for a in args: a = a.strip() # 如果参数是纯数字，直接转float try: args_eval.append(float(a)) continue except (ValueError, TypeError): pass # 否则作为表达式解析 try: v = self._evaluate(a, data, group_col) # 如果结果是长度为1的Series，取标量 if isinstance(v, pd.Series) and len(v) == 1: v = float(v.iloc[0]) args_eval.append(v) except (ValueError, KeyError) as e: args_eval.append(a) # 处理group_col参数 gc = group_col if func_name.startswith('group_') else None try: return self._operators[func_name](*args_eval, group_col=gc) except TypeError as e: # 可能不需要group_col参数 try: return self._operators[func_name](*args_eval) except TypeError: raise ValueError(f"算子 {func_name} 调用失败: {e}, 参数: {args_eval}") raise ValueError(f"未知算子: {func_name}") # ================================================================ # 时序算子 (Time-Series Operators) # ================================================================ def _calc_ts_mean(self, x, window, **kw): if isinstance(window, (int, float)): return x.rolling(int(window), min_periods=1).mean() return x.mean() def _calc_ts_std(self, x, window, **kw): if isinstance(window, (int, float)): return x.rolling(int(window), min_periods=1).std() return x.std() def _calc_ts_delta(self, x, period, **kw): if isinstance(period, (int, float)): return x - x.shift(int(period)) return x.diff() def _calc_ts_rank(self, x, window, **kw): if isinstance(window, (int, float)): return x.rolling(int(window), min_periods=1).apply( lambda s: pd.Series(s).rank(pct=True).iloc[-1] if len(s) > 1 else 0.5, raw=False) return x.rank(pct=True) def _calc_ts_argmax(self, x, window, **kw): if isinstance(window, (int, float)): return x.rolling(int(window), min_periods=1).apply( lambda s: np.argmax(s.values) if len(s) > 0 else 0, raw=True) return pd.Series(np.argmax(x.values), index=x.index) def _calc_ts_argmin(self, x, window, **kw): if isinstance(window, (int, float)): return x.rolling(int(window), min_periods=1).apply( lambda s: np.argmin(s.values) if len(s) > 0 else 0, raw=True) return pd.Series(np.argmin(x.values), index=x.index) def _calc_ts_corr(self, x, y, window, **kw): if isinstance(window, (int, float)): df = pd.concat([x, y], axis=1) return df.iloc[:,0].rolling(int(window), min_periods=2).corr(df.iloc[:,1]) return x.corr(y) def _calc_ts_cov(self, x, y, window, **kw): if isinstance(window, (int, float)): df = pd.concat([x, y], axis=1) return df.iloc[:,0].rolling(int(window), min_periods=2).cov(df.iloc[:,1]) return x.cov(y) def _calc_decay_linear(self, x, window, **kw): if isinstance(window, (int, float)): w = int(window) def weighted_mean(s): weights = np.arange(1, len(s)+1, dtype=float) weights /= weights.sum() return np.nansum(s.values * weights) return x.rolling(w, min_periods=1).apply(weighted_mean, raw=False) return x def _calc_product(self, x, window, **kw): if isinstance(window, (int, float)): return x.rolling(int(window), min_periods=1).apply( lambda s: np.nanprod(s.clip(lower=-1e10, upper=1e10).values), raw=True) return x.cumprod() def _calc_ts_av_diff(self, x, window, **kw): mean_val = x.rolling(int(window), min_periods=1).mean() return x - mean_val def _calc_ts_zscore(self, x, window, **kw): if isinstance(window, (int, float)): m = x.rolling(int(window), min_periods=1).mean() s = x.rolling(int(window), min_periods=1).std() return (x - m) / s.replace(0, np.nan) return (x - x.mean()) / x.std().replace(0, np.nan) def _calc_ts_sum(self, x, window, **kw): return x.rolling(int(window), min_periods=1).sum() def _calc_ts_min(self, x, window, **kw): return x.rolling(int(window), min_periods=1).min() def _calc_ts_max(self, x, window, **kw): return x.rolling(int(window), min_periods=1).max() def _calc_ts_shift(self, x, period, **kw): return x.shift(int(period)) def _calc_ts_skewness(self, x, window, **kw): return x.rolling(int(window), min_periods=3).skew() def _calc_ts_kurtosis(self, x, window, **kw): return x.rolling(int(window), min_periods=4).kurt() def _calc_ts_median(self, x, window, **kw): return x.rolling(int(window), min_periods=1).median() def _calc_ts_quantile(self, x, window, q=0.5, **kw): return x.rolling(int(window), min_periods=1).quantile(float(q)) def _calc_ts_log(self, x, **kw): return np.log(x.clip(lower=1e-10)) def _calc_ts_abs(self, x, **kw): return x.abs() def _calc_ts_sign(self, x, **kw): return np.sign(x) def _calc_ts_pow(self, x, power, **kw): return x ** float(power) # ================================================================ # 截面算子 (Cross-Sectional Operators) # ================================================================ def _calc_cross_rank(self, x, **kw): return x.rank(pct=True) def _calc_cross_zscore(self, x, **kw): m = x.mean() s = x.std() return (x - m) / s if s > 0 else pd.Series(0, index=x.index) def _calc_cross_scale(self, x, **kw): s = x.abs().sum() return x / s if s > 0 else pd.Series(0, index=x.index) def _calc_group_rank(self, x, group_col=None, **kw): if group_col is not None and hasattr(x, 'index'): # 尝试从data的index获取分组 return x.rank(pct=True) return x.rank(pct=True) def _calc_group_zscore(self, x, group_col=None, **kw): return self._calc_cross_zscore(x) def _calc_group_neutralize(self, x, group_col=None, **kw): return self._calc_cross_zscore(x) # ================================================================ # 技术指标 (Technical Indicators) # ================================================================ def _calc_ema(self, x, period, **kw): return x.ewm(span=int(period), adjust=False).mean() def _calc_sma(self, x, period, **kw): return x.rolling(int(period), min_periods=1).mean() def _calc_wma(self, x, period, **kw): w = int(period) def weighted_avg(s): weights = np.arange(1, len(s)+1, dtype=float) return np.nansum(s.values * weights) / weights.sum() return x.rolling(w, min_periods=1).apply(weighted_avg, raw=False) def _calc_rsi(self, x, period=14, **kw): delta = x.diff() gain = delta.where(delta > 0, 0) loss = (-delta).where(delta < 0, 0) avg_gain = gain.ewm(alpha=1.0/int(period), min_periods=int(period)).mean() avg_loss = loss.ewm(alpha=1.0/int(period), min_periods=int(period)).mean() rs = avg_gain / avg_loss.replace(0, 1e-10) return 100 - (100 / (1 + rs)) def _calc_macd(self, x, fast=12, slow=26, signal=9, **kw): ema_fast = x.ewm(span=int(fast), adjust=False).mean() ema_slow = x.ewm(span=int(slow), adjust=False).mean() dif = ema_fast - ema_slow dea = dif.ewm(span=int(signal), adjust=False).mean() histogram = (dif - dea) * 2 return dif def _calc_obv(self, x, volume=None, **kw): if volume is not None: return x * volume return x def _calc_atr(self, x, period=14, **kw): # 简化：用close的rolling范围近似 return x.rolling(int(period), min_periods=1).apply( lambda s: s.max() - s.min(), raw=True) def _calc_boll_upper(self, x, period=20, num_std=2, **kw): mid = x.rolling(int(period), min_periods=1).mean() std = x.rolling(int(period), min_periods=1).std() return mid + float(num_std) * std def _calc_boll_lower(self, x, period=20, num_std=2, **kw): mid = x.rolling(int(period), min_periods=1).mean() std = x.rolling(int(period), min_periods=1).std() return mid - float(num_std) * std def _calc_boll_mid(self, x, period=20, **kw): return x.rolling(int(period), min_periods=1).mean() def _calc_kdj_k(self, x, period=9, **kw): lowest = x.rolling(int(period), min_periods=1).min() highest = x.rolling(int(period), min_periods=1).max() return (x - lowest) / (highest - lowest).replace(0, np.nan) * 100 def _calc_kdj_d(self, x, period=9, m=3, **kw): k = self._calc_kdj_k(x, period) return k.rolling(int(m), min_periods=1).mean() def _calc_wr(self, x, period=14, **kw): highest = x.rolling(int(period), min_periods=1).max() lowest = x.rolling(int(period), min_periods=1).min() return (highest - x) / (highest - lowest).replace(0, np.nan) * -100 def _calc_cci(self, x, period=20, **kw): tp = x # simplified: use close as typical price ma = tp.rolling(int(period), min_periods=1).mean() md = tp.rolling(int(period), min_periods=1).apply( lambda s: np.abs(s - s.mean()).mean(), raw=True) return (tp - ma) / (0.015 * md.replace(0, np.nan)) # 便捷函数 def parse_expression(expression, data, group_col=None): """ 解析因子表达式（全局便捷入口） Parameters ---------- expression : str - 因子表达式 data : pd.DataFrame - 包含所需列的数据 group_col : str, optional - 分组列名 Returns ------- pd.Series 示例: >>> parse_expression("ts_rank(ts_delta(close, 5), 20)", df) >>> parse_expression("rank(rsi(close, 14))", df) >>> parse_expression("ts_zscore(close / ts_mean(close, 20), 60)", df) >>> parse_expression("close > ts_mean(close, 20) ? 1 : -1", df) """ parser = ExpressionParser() return parser.parse(expression, data, group_col) # 验证表达式语法 def validate_expression(expression): """ 验证表达式是否包含允许的算子和列名 返回 (is_valid, error_message) """ try: parser = ExpressionParser() # 替换别名 expr = parser._replace_aliases(expression) expr = parser._convert_ternary_operators(expr) # 检查是否有未知函数调用 func_calls = re.findall(r'(\w+)\s*\(', expr) known_ops = set(parser._operators.keys()) | {'ternary'} for fc in func_calls: if fc not in known_ops and fc not in ALLOWED_COLUMNS: return False, f"未知算子或列名: {fc}" return True, None except Exception as e: return False, str(e) # 常用因子模板 FACTOR_TEMPLATES = { 'momentum_5d': {'expr': 'ts_delta(close, 5) / ts_delay(close, 5)', 'name': '5日动量', 'category': '动量'}, 'momentum_20d': {'expr': 'ts_delta(close, 20) / ts_delay(close, 20)', 'name': '20日动量', 'category': '动量'}, 'volatility_20d': {'expr': 'ts_std(close/ts_delay(close,1) - 1, 20)', 'name': '20日波动率', 'category': '波动率'}, 'rsi_14': {'expr': 'rsi(close, 14)', 'name': 'RSI(14)', 'category': '技术'}, 'turnover_rate': {'expr': 'ts_mean(volume, 5)', 'name': '5日均量', 'category': '量价'}, 'boll_deviation': {'expr': '(close - boll_mid(close, 20)) / ts_std(close, 20)', 'name': '布林偏离度', 'category': '技术'}, 'price_momentum_rank': {'expr': 'ts_rank(ts_delta(close, 10) / ts_delay(close, 10), 60)', 'name': '价格动量排名', 'category': '动量'}, 'volume_price_corr': {'expr': 'ts_corr(close, volume, 20)', 'name': '量价相关系数', 'category': '量价'}, 'mean_reversion_5d': {'expr': '-ts_delta(close, 5) / ts_delay(close, 5)', 'name': '5日均值回归', 'category': '反转'}, 'high_low_ratio': {'expr': 'ts_mean(close/ts_delay(close,1) - 1, 5) / ts_std(close/ts_delay(close,1) - 1, 20)', 'name': '高低比率', 'category': '风险'}, 'ema_cross': {'expr': 'ema(close, 12) - ema(close, 26)', 'name': 'EMA交叉', 'category': '趋势'}, 'kurtosis_20d': {'expr': 'ts_kurtosis(close/ts_delay(close,1) - 1, 20)', 'name': '20日峰度', 'category': '分布'}, 'skewness_20d': {'expr': 'ts_skewness(close/ts_delay(close,1) - 1, 20)', 'name': '20日偏度', 'category': '分布'}, 'decay_volatility': {'expr': 'decay_linear(ts_std(close/ts_delay(close,1) - 1, 5), 20)', 'name': '衰减波动率', 'category': '波动率'}, 'zscore_60d': {'expr': 'ts_zscore(close, 60)', 'name': '60日Z-Score', 'category': '统计'}, 'cumulative_return_20d': {'expr': 'product(1 + (close - ts_delay(close, 1))/ts_delay(close, 1), 20)', 'name': '20日累计收益', 'category': '动量'}, 'max_drawdown_20d': {'expr': 'ts_min(close/ts_delay(close,1) - 1, 20)', 'name': '20日最大回撤', 'category': '风险'}, 'turnover_acceleration': {'expr': 'ts_delta(volume, 5) - ts_delta(volume, 20)', 'name': '换手加速度', 'category': '量价'}, }