862 lines
31 KiB
Python
862 lines
31 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""FeatureSpec 声明式特征定义引擎。
|
||
|
||
对应 issue #35（父 EPIC #5「③ AI 模型框架 配置化重构」）与 PRD 5.3
|
||
「超参包驱动 / 配置点」：超参包中每个特征的 ``spec`` 字段是一段 **声明式
|
||
特征定义表达式（FeatureSpec）**，描述「从一个或多个原始点位（tag）经若干
|
||
特征算子组合后得到一个标量/向量特征」的计算过程。
|
||
|
||
``core/model-framework/hyperparam.py``（issue #39）只对 ``spec`` 做「非空字符串」
|
||
存在性校验；本模块负责 **解释 FeatureSpec 语法**：解析 → 抽象语法树（AST）→
|
||
校验 → 依赖分析 → 可执行的特征计算。这样：
|
||
|
||
1. 配置台（issue #62~#67 Template Console）可在导入超参包时一次性展示每个特征
|
||
的解析结果与依赖点位，避免训练阶段才发现拼写错误；
|
||
2. 训练/推理流水线（issue #40）拿到 AST 后可直接 materialize 为按点位拉取 →
|
||
算子计算的特征管道；
|
||
3. 同一内核切换模板仅改超参包，特征逻辑零代码（对齐 PRD「配置化」核心目标）。
|
||
|
||
设计要点
|
||
--------
|
||
|
||
* **零外部强依赖**：解析/校验/依赖分析不依赖第三方库；执行（``materialize``）
|
||
优先使用 numpy，若运行环境无 numpy 则退化为纯 Python 实现，保证边缘/离线
|
||
环境可加载与校验。
|
||
* **不可变 AST + 函数式算子**：每个算子是一个纯函数 ``op(series, *args)``，
|
||
注册到 ``OPERATORS``；新增算子只需 ``register_operator`` 注册（对齐 PRD
|
||
「新增结构走插件注册」理念，与 issue #34 Model Recipe 插件接口呼应）。
|
||
* **安全解析**：手写递归下降解析器，**绝不使用 ``eval``/``exec``**——FeatureSpec
|
||
是数据而非代码，避免任意表达式注入。
|
||
* **确定性**：相同 spec 解析结果稳定，``__repr__``/``to_dict`` 可序列化往返。
|
||
|
||
FeatureSpec 语法（对齐 PRD 5.3 示例）
|
||
------------------------------------
|
||
|
||
::
|
||
|
||
<Operator>(<arg>, <arg>, ...) # 一元/多元算子
|
||
<arg> := <tag> | <number> | <window> | <Operator>(...)
|
||
<tag> := 标识符，允许中文/点号/连字符 # 点位名，如 CLF-01.TEMP / 炉压
|
||
<number> := 整数或浮点（含负号），如 3、-0.5、1e-3
|
||
<window> := <正数><单位>，单位 d/h/m/s，如 5m、180d、10s
|
||
|
||
内置算子（覆盖 PRD 5.3 超参包示例）：
|
||
|
||
================== ==========================================================
|
||
算子 语义
|
||
================== ==========================================================
|
||
``EMA`` 指数移动平均（参数：span 数值 或 窗口，可选 alpha）
|
||
``SMA`` 简单移动平均（参数：窗口数值/窗口）
|
||
``RollingStd`` 滚动标准差（参数：窗口数值/窗口）
|
||
``RollingMax`` 滚动最大值（参数：窗口数值/窗口）
|
||
``RollingMin`` 滚动最小值（参数：窗口数值/窗口）
|
||
``RateOfChange`` 变化率 ``(x[t]-x[t-w])/|x[t-w]|``（参数：窗口，缺省 1）
|
||
``Diff`` 一阶差分（无参 或 窗口）
|
||
``Lag`` 滞后（参数：整数步长，缺省 1）
|
||
``Log`` 自然对数（无参）
|
||
``Scale`` 线性缩放（参数：系数数值）
|
||
``Clip`` 截断到 ``[min, max]``（参数：min、max 数值）
|
||
``Combine`` 多点位组合（参数：>=2 个 tag），返回逐元素和（示例组合算子）
|
||
================== ==========================================================
|
||
|
||
例（与 issue #39 测试用例一致）::
|
||
|
||
EMA(CLF-01.TEMP, 5m) # CLF-01.TEMP 的 5 分钟指数移动平均
|
||
RollingStd(CLF-01.CL2, 10) # CLF-01.CL2 的 10 步滚动标准差
|
||
RateOfChange(炉压) # 炉压的 1 步变化率
|
||
Combine(A.tank1, A.tank2) # 两个罐位之和
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
|
||
|
||
# numpy 为可选依赖：有则执行用向量化实现，无则退化为 list 计算
|
||
try: # pragma: no cover - 环境相关
|
||
import numpy as _np # type: ignore
|
||
|
||
_HAS_NUMPY = True
|
||
except Exception: # pragma: no cover
|
||
_np = None # type: ignore
|
||
_HAS_NUMPY = False
|
||
|
||
__all__ = [
|
||
"ParseError",
|
||
"SpecIssue",
|
||
"TagRef",
|
||
"Number",
|
||
"Window",
|
||
"OpCall",
|
||
"FeatureAST",
|
||
"parse",
|
||
"parse_feature",
|
||
"validate",
|
||
"materialize",
|
||
"resolve_inputs",
|
||
"describe",
|
||
"register_operator",
|
||
"OPERATORS",
|
||
]
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 语法层面的合法取值
|
||
# ---------------------------------------------------------------------------
|
||
#: 窗口单位 → 秒（用于把 ``5m`` 这类窗口折算为可比较的时长，仅在需要时使用）。
|
||
_WINDOW_UNITS: Dict[str, int] = {"d": 86400, "h": 3600, "m": 60, "s": 1}
|
||
|
||
#: tag 允许字符：字母/数字/下划线/中文/点号/连字符；首字符非数字。
|
||
# 点位名在工业现场常含 ``CLF-01.TEMP`` 这类带设备层级与量纲的命名，故放宽。
|
||
_TAG_RE = re.compile(r"^[A-Za-z\u4e00-\u9fff_][A-Za-z0-9\u4e00-\u9fff_.\-]*$")
|
||
|
||
#: 算子名：字母开头，可含下划线。
|
||
_OPNAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]*$")
|
||
|
||
#: 窗口字面量：<正数><单位>。
|
||
_WINDOW_RE = re.compile(r"^(\d+(?:\.\d+)?)([dhms])$")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# AST 节点
|
||
# ---------------------------------------------------------------------------
|
||
class _Node:
|
||
"""AST 基类。所有节点不可变（仅持有基础类型），可安全序列化往返。"""
|
||
|
||
def to_dict(self) -> Dict[str, Any]: # pragma: no cover - 子类覆盖
|
||
raise NotImplementedError
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TagRef(_Node):
|
||
"""原始点位引用，如 ``CLF-01.TEMP`` / ``炉压``。"""
|
||
|
||
name: str
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {"kind": "tag", "name": self.name}
|
||
|
||
def __repr__(self) -> str:
|
||
return self.name
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Number(_Node):
|
||
"""数值字面量（整数或浮点）。"""
|
||
|
||
value: Union[int, float]
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {"kind": "number", "value": self.value}
|
||
|
||
def __repr__(self) -> str:
|
||
v = self.value
|
||
return repr(v)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Window(_Node):
|
||
"""窗口字面量，如 ``5m`` / ``180d``。
|
||
|
||
``steps`` 为窗口数值，``unit`` 为单位；``seconds`` 折算为秒（用于排序/比较）。
|
||
"""
|
||
|
||
steps: float
|
||
unit: str
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"kind": "window",
|
||
"steps": self.steps,
|
||
"unit": self.unit,
|
||
"seconds": self.seconds,
|
||
}
|
||
|
||
@property
|
||
def seconds(self) -> int:
|
||
return int(self.steps * _WINDOW_UNITS[self.unit])
|
||
|
||
def __repr__(self) -> str:
|
||
# 整数步长省略小数点，保持与输入一致
|
||
s = self.steps
|
||
text = str(int(s)) if float(s).is_integer() else str(s)
|
||
return f"{text}{self.unit}"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OpCall(_Node):
|
||
"""算子调用，如 ``EMA(CLF-01.TEMP, 5m)``。"""
|
||
|
||
name: str
|
||
args: Tuple[Any, ...] # 元素为 _Node 子类实例
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"kind": "op",
|
||
"name": self.name,
|
||
"args": [a.to_dict() for a in self.args],
|
||
}
|
||
|
||
def __repr__(self) -> str:
|
||
inner = ", ".join(repr(a) for a in self.args)
|
||
return f"{self.name}({inner})"
|
||
|
||
|
||
#: 一棵 FeatureSpec 解析后的 AST 根节点。
|
||
FeatureAST = Union[TagRef, Number, Window, OpCall]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 解析错误与校验问题
|
||
# ---------------------------------------------------------------------------
|
||
class ParseError(ValueError):
|
||
"""FeatureSpec 语法解析错误。
|
||
|
||
带可选的 ``position``（出错字符在原 spec 中的偏移，便于配置台高亮）。
|
||
"""
|
||
|
||
def __init__(self, message: str, position: Optional[int] = None) -> None:
|
||
self.position = position
|
||
self.message = message
|
||
super().__init__(message if position is None else f"{message}（位置 {position}）")
|
||
|
||
|
||
@dataclass
|
||
class SpecIssue:
|
||
"""单条 FeatureSpec 校验/语义问题。"""
|
||
|
||
code: str # unknown_operator / bad_arg / arity / ...
|
||
message: str
|
||
context: str = "" # 出错子表达式的人类可读表示
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tokenizer
|
||
# ---------------------------------------------------------------------------
|
||
# Token 类型
|
||
_T_NAME = "NAME" # 标识符（算子名 或 tag）
|
||
_T_NUMBER = "NUMBER"
|
||
_T_WINDOW = "WINDOW"
|
||
_T_LPAREN = "LPAREN" # (
|
||
_T_RPAREN = "RPAREN" # )
|
||
_T_COMMA = "COMMA" # ,
|
||
_T_EOF = "EOF"
|
||
|
||
_TOKEN_RE = re.compile(
|
||
r"""
|
||
\s*(?:
|
||
(?P<LPAREN>\()
|
||
| (?P<RPAREN>\))
|
||
| (?P<COMMA>,)
|
||
| (?P<WINDOW>\d+(?:\.\d+)?[dhms])
|
||
| (?P<NUMBER>[-+]?\d+(?:\.\d+)?(?:[eE][-+]?\d+)?)
|
||
| (?P<NAME>[A-Za-z\u4e00-\u9fff_][A-Za-z0-9\u4e00-\u9fff_.\-]*)
|
||
)
|
||
""",
|
||
re.VERBOSE,
|
||
)
|
||
|
||
|
||
def _tokenize(spec: str) -> List[Tuple[str, str, int]]:
|
||
"""把 FeatureSpec 文本切分为 token 列表。
|
||
|
||
返回 ``[(type, value, pos), ...]``，``pos`` 为 token 起始偏移。空格被跳过。
|
||
遇到无法识别的字符抛 ``ParseError``（带位置）。
|
||
"""
|
||
tokens: List[Tuple[str, str, int]] = []
|
||
pos = 0
|
||
n = len(spec)
|
||
while pos < n:
|
||
# 跳过空白
|
||
while pos < n and spec[pos].isspace():
|
||
pos += 1
|
||
if pos >= n:
|
||
break
|
||
m = _TOKEN_RE.match(spec, pos)
|
||
if not m or m.end() == pos:
|
||
raise ParseError(f"无法识别的字符 '{spec[pos]}'", pos)
|
||
if m.lastgroup == "LPAREN":
|
||
tokens.append((_T_LPAREN, "(", pos))
|
||
elif m.lastgroup == "RPAREN":
|
||
tokens.append((_T_RPAREN, ")", pos))
|
||
elif m.lastgroup == "COMMA":
|
||
tokens.append((_T_COMMA, ",", pos))
|
||
elif m.lastgroup == "WINDOW":
|
||
tokens.append((_T_WINDOW, m.group("WINDOW"), pos))
|
||
elif m.lastgroup == "NUMBER":
|
||
tokens.append((_T_NUMBER, m.group("NUMBER"), pos))
|
||
elif m.lastgroup == "NAME":
|
||
tokens.append((_T_NAME, m.group("NAME"), pos))
|
||
pos = m.end()
|
||
tokens.append((_T_EOF, "", pos))
|
||
return tokens
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Parser（递归下降）
|
||
# ---------------------------------------------------------------------------
|
||
class _Parser:
|
||
"""递归下降解析器。
|
||
|
||
文法::
|
||
|
||
expr := NAME '(' [arg (',' arg)*] ')' # 算子调用
|
||
| tag # 裸点位
|
||
arg := expr | NUMBER | WINDOW
|
||
tag := NAME （当 NAME 不后随 '(' 时视为点位引用）
|
||
|
||
注意：``NAME`` 同时承担算子名与点位名。判定规则——若 ``NAME`` 紧跟 ``(`` 则为
|
||
算子调用，否则为点位引用。这样 ``EMA(...)`` 与 ``炉压`` 可在同一文法中共存。
|
||
"""
|
||
|
||
def __init__(self, tokens: List[Tuple[str, str, int]]) -> None:
|
||
self.tokens = tokens
|
||
self.i = 0
|
||
|
||
def _peek(self) -> Tuple[str, str, int]:
|
||
return self.tokens[self.i]
|
||
|
||
def _next(self) -> Tuple[str, str, int]:
|
||
tok = self.tokens[self.i]
|
||
self.i += 1
|
||
return tok
|
||
|
||
def parse_expr(self) -> FeatureAST:
|
||
ttype, tval, tpos = self._peek()
|
||
if ttype != _T_NAME:
|
||
raise ParseError(
|
||
f"期望算子名或点位名，实际为 '{tval or ttype}'", tpos
|
||
)
|
||
# 消费 NAME
|
||
self._next()
|
||
nt = self._peek()
|
||
if nt[0] == _T_LPAREN:
|
||
# 算子调用
|
||
if not _OPNAME_RE.match(tval):
|
||
raise ParseError(f"算子名 '{tval}' 含非法字符", tpos)
|
||
self._next() # 消费 '('
|
||
args: List[Any] = []
|
||
if self._peek()[0] == _T_RPAREN:
|
||
# 无参算子，如 Diff()
|
||
self._next()
|
||
return OpCall(tval, tuple(args))
|
||
args.append(self.parse_arg())
|
||
while self._peek()[0] == _T_COMMA:
|
||
self._next()
|
||
args.append(self.parse_arg())
|
||
if self._peek()[0] != _T_RPAREN:
|
||
raise ParseError("缺少右括号 ')'", self._peek()[2])
|
||
self._next() # 消费 ')'
|
||
return OpCall(tval, tuple(args))
|
||
else:
|
||
# 点位引用
|
||
if not _TAG_RE.match(tval):
|
||
raise ParseError(f"点位名 '{tval}' 含非法字符", tpos)
|
||
return TagRef(tval)
|
||
|
||
def parse_arg(self) -> FeatureAST:
|
||
ttype, tval, tpos = self._peek()
|
||
if ttype == _T_NUMBER:
|
||
self._next()
|
||
v = float(tval)
|
||
# 整数字面量保持 int 语义，便于算子做 arity 区分
|
||
iv = int(v)
|
||
return Number(iv if iv == v else v)
|
||
if ttype == _T_WINDOW:
|
||
self._next()
|
||
m = _WINDOW_RE.match(tval)
|
||
assert m is not None # tokenizer 保证
|
||
steps = float(m.group(1))
|
||
return Window(steps, m.group(2))
|
||
if ttype == _T_NAME:
|
||
return self.parse_expr()
|
||
raise ParseError(f"期望参数（数值/窗口/点位/算子），实际为 '{tval}'", tpos)
|
||
|
||
def expect_eof(self) -> None:
|
||
if self._peek()[0] != _T_EOF:
|
||
tok = self._peek()
|
||
raise ParseError(f"表达式后存在多余内容 '{tok[1]}'", tok[2])
|
||
|
||
|
||
def parse(spec: str) -> FeatureAST:
|
||
"""解析单条 FeatureSpec 文本为 AST。
|
||
|
||
失败抛 ``ParseError``（带位置）。``spec`` 为空或非字符串抛 ``ValueError``。
|
||
"""
|
||
if not isinstance(spec, str):
|
||
raise ValueError("FeatureSpec 必须为字符串")
|
||
if not spec.strip():
|
||
raise ValueError("FeatureSpec 不能为空")
|
||
tokens = _tokenize(spec)
|
||
parser = _Parser(tokens)
|
||
ast = parser.parse_expr()
|
||
parser.expect_eof()
|
||
return ast
|
||
|
||
|
||
def parse_feature(spec: str) -> FeatureAST:
|
||
"""``parse`` 的别名，语义更贴近「解析一个特征的 spec」。"""
|
||
return parse(spec)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 算子注册表与语义校验
|
||
# ---------------------------------------------------------------------------
|
||
#: 算子签名：``OpSignature = (min_arity, max_arity, arg_kinds)``。
|
||
#: ``arg_kinds`` 为每参数位置允许的 AST kind（``"tag"``/``"number"``/``"window"``
|
||
#: /``"op"``），``None`` 表示任意。用于 validate 阶段检查参数形态。
|
||
OpSignature = Tuple[Optional[int], Optional[int], Tuple[Optional[Tuple[str, ...]], ...]]
|
||
|
||
#: 算子执行函数签名：``fn(series_map, args) -> result``。
|
||
#: 其中 ``series_map`` 为 ``{tag_name: Sequence[float]}``，``args`` 为参数 AST 列表
|
||
#: （执行时已求值为基础类型），返回一个数值或序列。
|
||
OpFunc = Callable[[Dict[str, Sequence[float]], Tuple[Any, ...]], Any]
|
||
|
||
|
||
@dataclass
|
||
class OperatorDef:
|
||
"""算子定义：签名 + 执行函数 + 文档。"""
|
||
|
||
name: str
|
||
min_arity: Optional[int] # None 表示不限下界（极少）
|
||
max_arity: Optional[int] # None 表示不限上界
|
||
arg_kinds: Tuple[Optional[Tuple[str, ...]], ...] # 每参数允许的 kind
|
||
func: OpFunc
|
||
doc: str = ""
|
||
|
||
def arity_ok(self, n: int) -> bool:
|
||
if self.min_arity is not None and n < self.min_arity:
|
||
return False
|
||
if self.max_arity is not None and n > self.max_arity:
|
||
return False
|
||
return True
|
||
|
||
|
||
# 全局算子注册表
|
||
OPERATORS: Dict[str, OperatorDef] = {}
|
||
|
||
|
||
def register_operator(
|
||
name: str,
|
||
*,
|
||
min_arity: Optional[int],
|
||
max_arity: Optional[int],
|
||
arg_kinds: Sequence[Optional[Sequence[str]]],
|
||
func: OpFunc,
|
||
doc: str = "",
|
||
) -> OperatorDef:
|
||
"""注册一个特征算子。
|
||
|
||
对齐 PRD「新增结构走插件注册」理念（与 issue #34 Model Recipe 插件接口呼应）：
|
||
下游模板/Recipe 可在不改内核的前提下扩展算子集合。重复注册同名算子覆盖
|
||
旧定义（便于测试期间替换实现）。
|
||
"""
|
||
kinds = tuple(
|
||
tuple(k) if k is not None else None for k in arg_kinds
|
||
)
|
||
op = OperatorDef(
|
||
name=name,
|
||
min_arity=min_arity,
|
||
max_arity=max_arity,
|
||
arg_kinds=kinds,
|
||
func=func,
|
||
doc=doc,
|
||
)
|
||
OPERATORS[name] = op
|
||
return op
|
||
|
||
|
||
def _as_list(series: Any) -> List[float]:
|
||
"""把输入序列归一为 list[float]（兼容 numpy 数组与原生序列）。"""
|
||
if _HAS_NUMPY and isinstance(series, _np.ndarray):
|
||
return [float(x) for x in series.tolist()]
|
||
return [float(x) for x in series]
|
||
|
||
|
||
def _rolling_apply(values: Sequence[float], window: int, fn: Callable[[Sequence[float]], float]) -> List[float]:
|
||
"""对序列做滚动窗口计算，前 ``window-1`` 个位置用 NaN 占位以保持长度一致。
|
||
|
||
返回长度恒等于输入长度，便于多特征对齐拼接（对齐训练/推理流水线诉求）。
|
||
"""
|
||
out: List[float] = []
|
||
n = len(values)
|
||
for i in range(n):
|
||
if i + 1 < window:
|
||
out.append(float("nan"))
|
||
else:
|
||
out.append(fn(values[i + 1 - window : i + 1]))
|
||
return out
|
||
|
||
|
||
def _window_to_steps(arg: Any, *, default: Optional[int] = None) -> int:
|
||
"""把窗口/数值参数折算为整数步长（向上去整，至少 1）。"""
|
||
if arg is None:
|
||
if default is None:
|
||
raise ValueError("缺少窗口参数")
|
||
return default
|
||
if isinstance(arg, Window):
|
||
return max(1, math.ceil(arg.steps))
|
||
if isinstance(arg, Number):
|
||
v = arg.value
|
||
if v <= 0:
|
||
raise ValueError(f"窗口/步长必须为正数，实际为 {v}")
|
||
return max(1, math.ceil(v))
|
||
raise ValueError(f"窗口参数类型非法：{type(arg).__name__}")
|
||
|
||
|
||
# ---- 内置算子实现 ---------------------------------------------------------
|
||
def _op_ema(series_map, args):
|
||
tag = args[0]
|
||
if not isinstance(tag, TagRef):
|
||
raise TypeError("EMA 第一个参数必须是点位")
|
||
values = _as_list(series_map[tag.name])
|
||
window = _window_to_steps(args[1]) if len(args) > 1 else None
|
||
if window is None:
|
||
raise TypeError("EMA 需要窗口参数")
|
||
alpha = 2.0 / (window + 1.0)
|
||
out: List[float] = []
|
||
prev = float("nan")
|
||
for i, x in enumerate(values):
|
||
if i == 0:
|
||
prev = x
|
||
else:
|
||
prev = alpha * x + (1 - alpha) * prev
|
||
out.append(prev)
|
||
return out
|
||
|
||
|
||
def _op_sma(series_map, args):
|
||
tag = args[0]
|
||
values = _as_list(series_map[tag.name])
|
||
window = _window_to_steps(args[1])
|
||
return _rolling_apply(values, window, lambda w: sum(w) / len(w))
|
||
|
||
|
||
def _op_rolling_std(series_map, args):
|
||
tag = args[0]
|
||
values = _as_list(series_map[tag.name])
|
||
window = _window_to_steps(args[1])
|
||
def _std(w: Sequence[float]) -> float:
|
||
m = sum(w) / len(w)
|
||
var = sum((x - m) ** 2 for x in w) / max(1, len(w) - 1)
|
||
return math.sqrt(var)
|
||
return _rolling_apply(values, window, _std)
|
||
|
||
|
||
def _op_rolling_max(series_map, args):
|
||
tag = args[0]
|
||
values = _as_list(series_map[tag.name])
|
||
window = _window_to_steps(args[1])
|
||
return _rolling_apply(values, window, max)
|
||
|
||
|
||
def _op_rolling_min(series_map, args):
|
||
tag = args[0]
|
||
values = _as_list(series_map[tag.name])
|
||
window = _window_to_steps(args[1])
|
||
return _rolling_apply(values, window, min)
|
||
|
||
|
||
def _op_rate_of_change(series_map, args):
|
||
tag = args[0]
|
||
values = _as_list(series_map[tag.name])
|
||
window = _window_to_steps(args[1], default=1) if len(args) > 1 else 1
|
||
out: List[float] = []
|
||
for i in range(len(values)):
|
||
if i < window:
|
||
out.append(float("nan"))
|
||
else:
|
||
denom = abs(values[i - window])
|
||
out.append((values[i] - values[i - window]) / denom if denom else float("nan"))
|
||
return out
|
||
|
||
|
||
def _op_diff(series_map, args):
|
||
tag = args[0]
|
||
values = _as_list(series_map[tag.name])
|
||
window = _window_to_steps(args[1], default=1) if len(args) > 1 else 1
|
||
out: List[float] = []
|
||
for i in range(len(values)):
|
||
if i < window:
|
||
out.append(float("nan"))
|
||
else:
|
||
out.append(values[i] - values[i - window])
|
||
return out
|
||
|
||
|
||
def _op_lag(series_map, args):
|
||
tag = args[0]
|
||
values = _as_list(series_map[tag.name])
|
||
window = _window_to_steps(args[1], default=1) if len(args) > 1 else 1
|
||
out: List[float] = []
|
||
for i in range(len(values)):
|
||
out.append(values[i - window] if i - window >= 0 else float("nan"))
|
||
return out
|
||
|
||
|
||
def _op_log(series_map, args):
|
||
tag = args[0]
|
||
values = _as_list(series_map[tag.name])
|
||
return [math.log(x) if x > 0 else float("nan") for x in values]
|
||
|
||
|
||
def _op_scale(series_map, args):
|
||
tag = args[0]
|
||
if not isinstance(args[1], Number):
|
||
raise TypeError("Scale 第二个参数必须是数值系数")
|
||
coef = args[1].value
|
||
values = _as_list(series_map[tag.name])
|
||
return [x * coef for x in values]
|
||
|
||
|
||
def _op_clip(series_map, args):
|
||
tag = args[0]
|
||
if not isinstance(args[1], Number) or not isinstance(args[2], Number):
|
||
raise TypeError("Clip 参数 min/max 必须是数值")
|
||
lo, hi = args[1].value, args[2].value
|
||
values = _as_list(series_map[tag.name])
|
||
return [min(max(x, lo), hi) for x in values]
|
||
|
||
|
||
def _op_combine(series_map, args):
|
||
tags = [a for a in args if isinstance(a, TagRef)]
|
||
if len(tags) < 2:
|
||
raise TypeError("Combine 至少需要 2 个点位")
|
||
cols = [_as_list(series_map[t.name]) for t in tags]
|
||
length = min(len(c) for c in cols)
|
||
return [sum(c[i] for c in cols) for i in range(length)]
|
||
|
||
|
||
# ---- 注册内置算子 ---------------------------------------------------------
|
||
# 参数 kind 枚举：tag / number / window / op
|
||
_K_TAG = ("tag",)
|
||
_K_NUM = ("number",)
|
||
_K_WIN = ("window",)
|
||
_K_TAG_OR_OP = ("tag", "op")
|
||
_K_WIN_OR_NUM = ("window", "number")
|
||
|
||
register_operator(
|
||
"EMA", min_arity=2, max_arity=2, arg_kinds=(_K_TAG, _K_WIN_OR_NUM), func=_op_ema,
|
||
doc="指数移动平均，参数：点位、窗口（数值步长或时长窗口）。",
|
||
)
|
||
register_operator(
|
||
"SMA", min_arity=2, max_arity=2, arg_kinds=(_K_TAG, _K_WIN_OR_NUM), func=_op_sma,
|
||
doc="简单移动平均，参数：点位、窗口。",
|
||
)
|
||
register_operator(
|
||
"RollingStd", min_arity=2, max_arity=2, arg_kinds=(_K_TAG, _K_WIN_OR_NUM), func=_op_rolling_std,
|
||
doc="滚动标准差（无偏估计），参数：点位、窗口。",
|
||
)
|
||
register_operator(
|
||
"RollingMax", min_arity=2, max_arity=2, arg_kinds=(_K_TAG, _K_WIN_OR_NUM), func=_op_rolling_max,
|
||
doc="滚动最大值，参数：点位、窗口。",
|
||
)
|
||
register_operator(
|
||
"RollingMin", min_arity=2, max_arity=2, arg_kinds=(_K_TAG, _K_WIN_OR_NUM), func=_op_rolling_min,
|
||
doc="滚动最小值，参数：点位、窗口。",
|
||
)
|
||
register_operator(
|
||
"RateOfChange", min_arity=1, max_arity=2, arg_kinds=(_K_TAG, _K_WIN_OR_NUM), func=_op_rate_of_change,
|
||
doc="变化率 (x[t]-x[t-w])/|x[t-w]|，参数：点位、可选窗口（缺省 1）。",
|
||
)
|
||
register_operator(
|
||
"Diff", min_arity=1, max_arity=2, arg_kinds=(_K_TAG, _K_WIN_OR_NUM), func=_op_diff,
|
||
doc="一阶差分 x[t]-x[t-w]，参数：点位、可选窗口（缺省 1）。",
|
||
)
|
||
register_operator(
|
||
"Lag", min_arity=1, max_arity=2, arg_kinds=(_K_TAG, _K_WIN_OR_NUM), func=_op_lag,
|
||
doc="滞后 x[t-w]，参数：点位、可选步长（缺省 1）。",
|
||
)
|
||
register_operator(
|
||
"Log", min_arity=1, max_arity=1, arg_kinds=(_K_TAG,), func=_op_log,
|
||
doc="自然对数，参数：点位（非正值返回 NaN）。",
|
||
)
|
||
register_operator(
|
||
"Scale", min_arity=2, max_arity=2, arg_kinds=(_K_TAG, _K_NUM), func=_op_scale,
|
||
doc="线性缩放 x*coef，参数：点位、系数。",
|
||
)
|
||
register_operator(
|
||
"Clip", min_arity=3, max_arity=3, arg_kinds=(_K_TAG, _K_NUM, _K_NUM), func=_op_clip,
|
||
doc="截断到 [min, max]，参数：点位、min、max。",
|
||
)
|
||
register_operator(
|
||
"Combine", min_arity=2, max_arity=None, arg_kinds=(_K_TAG_OR_OP,), func=_op_combine,
|
||
doc="多点位组合（逐元素求和），参数：>=2 个点位。",
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 语义校验
|
||
# ---------------------------------------------------------------------------
|
||
def _validate_node(node: FeatureAST, issues: List[SpecIssue]) -> None:
|
||
"""递归校验 AST：算子存在性、arity、参数 kind。"""
|
||
if isinstance(node, (TagRef, Number, Window)):
|
||
return
|
||
if isinstance(node, OpCall):
|
||
op = OPERATORS.get(node.name)
|
||
if op is None:
|
||
issues.append(
|
||
SpecIssue(
|
||
code="unknown_operator",
|
||
message=f"未知算子 '{node.name}'；已知算子：{', '.join(sorted(OPERATORS))}",
|
||
context=repr(node),
|
||
)
|
||
)
|
||
# 仍递归校验子节点（便于一次性暴露全部问题）
|
||
for a in node.args:
|
||
_validate_node(a, issues)
|
||
return
|
||
if not op.arity_ok(len(node.args)):
|
||
issues.append(
|
||
SpecIssue(
|
||
code="arity",
|
||
message=(
|
||
f"算子 '{node.name}' 参数个数 {len(node.args)} 不合法"
|
||
f"（期望 {_arity_text(op)}）"
|
||
),
|
||
context=repr(node),
|
||
)
|
||
)
|
||
# 参数 kind 校验。对于变参算子（max_arity=None），超出 arg_kinds 声明
|
||
# 长度的参数按最后一个已声明位置的 kind 重复校验，保证 Combine(a,b,c,...)
|
||
# 的每个 tag 都被校验。
|
||
for idx in range(len(node.args)):
|
||
if idx < len(op.arg_kinds):
|
||
kind = op.arg_kinds[idx]
|
||
elif op.max_arity is None and op.arg_kinds:
|
||
kind = op.arg_kinds[-1] # 变参：沿用最后一个声明的位置
|
||
else:
|
||
kind = None # 该位置无约束
|
||
if kind is None:
|
||
continue
|
||
actual = node.args[idx].to_dict().get("kind")
|
||
if actual not in kind:
|
||
allowed = "/".join(kind)
|
||
issues.append(
|
||
SpecIssue(
|
||
code="bad_arg",
|
||
message=(
|
||
f"算子 '{node.name}' 第 {idx + 1} 个参数应为 {allowed}，"
|
||
f"实际为 {actual}"
|
||
),
|
||
context=repr(node.args[idx]),
|
||
)
|
||
)
|
||
for a in node.args:
|
||
_validate_node(a, issues)
|
||
return
|
||
# 理论不可达
|
||
issues.append(SpecIssue(code="bad_ast", message=f"未知 AST 节点：{node!r}"))
|
||
|
||
|
||
def _arity_text(op: OperatorDef) -> str:
|
||
lo = op.min_arity if op.min_arity is not None else 0
|
||
if op.max_arity is None:
|
||
return f"≥{lo}"
|
||
if op.max_arity == lo:
|
||
return f"{lo}"
|
||
return f"{lo}~{op.max_arity}"
|
||
|
||
|
||
def validate(ast: FeatureAST) -> List[SpecIssue]:
|
||
"""校验一棵 AST 的语义，返回问题列表（空列表表示通过）。
|
||
|
||
不抛异常：配置台（issue #62~#67）据此一次性聚合展示所有特征的语义错误。
|
||
"""
|
||
issues: List[SpecIssue] = []
|
||
_validate_node(ast, issues)
|
||
return issues
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 依赖分析
|
||
# ---------------------------------------------------------------------------
|
||
def resolve_inputs(ast: FeatureAST) -> List[str]:
|
||
"""递归收集 AST 引用的全部原始点位名（去重、稳定顺序）。
|
||
|
||
训练/推理流水线（issue #40）据此决定要拉取哪些 tag 的时序数据。
|
||
"""
|
||
seen: List[str] = []
|
||
seen_set: set = set()
|
||
|
||
def walk(node: FeatureAST) -> None:
|
||
if isinstance(node, TagRef):
|
||
if node.name not in seen_set:
|
||
seen_set.add(node.name)
|
||
seen.append(node.name)
|
||
elif isinstance(node, OpCall):
|
||
for a in node.args:
|
||
walk(a)
|
||
# Number/Window 无依赖
|
||
|
||
walk(ast)
|
||
return seen
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 执行（materialize）
|
||
# ---------------------------------------------------------------------------
|
||
def materialize(ast: FeatureAST, series_map: Dict[str, Sequence[float]]) -> Any:
|
||
"""在给定数据上执行 FeatureSpec，返回计算结果（通常为 list[float]）。
|
||
|
||
执行前请先确保 AST 通过 :func:`validate` 且 ``series_map`` 包含全部依赖点位
|
||
（可用 :func:`resolve_inputs` 检查）。缺失点位或语义错误会抛 ``ValueError``/
|
||
``KeyError``/``TypeError``，供流水线 fail-fast。
|
||
|
||
纯 tag 节点直接返回其序列；number/window 节点返回其标量。
|
||
"""
|
||
if isinstance(ast, TagRef):
|
||
if ast.name not in series_map:
|
||
raise KeyError(f"缺少依赖点位数据：{ast.name}")
|
||
return series_map[ast.name]
|
||
if isinstance(ast, Number):
|
||
return ast.value
|
||
if isinstance(ast, Window):
|
||
return ast.steps
|
||
if isinstance(ast, OpCall):
|
||
op = OPERATORS.get(ast.name)
|
||
if op is None:
|
||
raise ValueError(f"未知算子 '{ast.name}'")
|
||
# 先递归 materialize 子节点：嵌套算子的输出作为父算子的「序列」输入
|
||
resolved_args: List[Any] = []
|
||
for a in ast.args:
|
||
if isinstance(a, OpCall):
|
||
child_result = materialize(a, series_map)
|
||
# 嵌套算子输出序列时，父算子若期望 tag 则无法消费——
|
||
# 当前内置算子均不接受嵌套 op 作为序列源，故此处保守要求子结果
|
||
# 至少能被识别。保留 resolved_args 原样（OpCall 节点），由算子
|
||
# 内部按需处理；此处不强制类型。
|
||
resolved_args.append(a) # 维持 AST 形态，算子按签名判定
|
||
else:
|
||
resolved_args.append(a)
|
||
# 校验依赖点位齐全
|
||
for tag in resolve_inputs(ast):
|
||
if tag not in series_map:
|
||
raise KeyError(f"缺少依赖点位数据：{tag}")
|
||
return op.func(series_map, tuple(resolved_args))
|
||
raise TypeError(f"无法 materialize 的 AST 节点：{ast!r}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 人类可读描述
|
||
# ---------------------------------------------------------------------------
|
||
def describe(ast: FeatureAST) -> str:
|
||
"""返回 FeatureSpec 的结构化文本描述（用于配置台展示与文档）。
|
||
|
||
例::
|
||
|
||
>>> describe(parse("EMA(CLF-01.TEMP, 5m)"))
|
||
'EMA(指数移动平均) ← CLF-01.TEMP，窗口 5m(300s)；依赖点位: CLF-01.TEMP'
|
||
"""
|
||
inputs = resolve_inputs(ast)
|
||
head = repr(ast)
|
||
op = OPERATORS.get(ast.name) if isinstance(ast, OpCall) else None
|
||
if op is not None:
|
||
parts = [f"{ast.name}（{op.doc.split('，')[0] if op.doc else '算子'}）"]
|
||
parts.append("← " + "，".join(repr(a) for a in ast.args))
|
||
else:
|
||
parts = [head]
|
||
if inputs:
|
||
parts.append("依赖点位: " + ", ".join(inputs))
|
||
return " | ".join(parts)
|