Blame for webui/factor_engine.py

# -*- 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': '量价'},
}