bot_ym/iAOP
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(#35): FeatureSpec 声明式特征定义引擎（PRD 5.3 特征 spec 语义解释）

Browse Source 
新增 core/model-framework/feature_spec.py：把超参包中特征 spec 字段从
「非空字符串存在性校验」（issue #39）升级为可解释的声明式特征定义引擎。

- 手写递归下降解析器（tokenizer+parser），安全无 eval/exec；支持算子调用、
  裸点位、数值/窗口字面量、嵌套、中文点位、带符号/科学计数数值。
- 不可变 AST（TagRef/Number/Window/OpCall），to_dict/repr 可序列化往返。
- 语义校验：未知算子、arity、参数 kind；返回 SpecIssue 列表供配置台聚合展示。
- 依赖分析 resolve_inputs：去重、稳定顺序，供训练/推理流水线按 tag 拉取数据。
- materialize 执行：EMA/SMA/RollingStd/Max/Min/RateOfChange/Diff/Lag/Log/
  Scale/Clip/Combine 共 12 个内置算子，缺数据 fail-fast；numpy 可选依赖。
- 插件注册 register_operator：对齐 PRD「新增结构走插件注册」（呼应 #34 Recipe）。
- 49 个单元测试全通过；零内核改动，纯新增。
This commit is contained in:
bot_dev1
2026-08-04 22:31:54 +08:00
parent 33c2586dd7
commit bec0bad97a
4 changed files with 1291 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/model-framework/README.md
+80
View File
@@ -0,0 +1,80 @@
# iAOP-Core · FeatureSpec 声明式特征定义引擎
对应 issue #35（父 EPIC #5「③ AI 模型框架 配置化重构」）与 PRD 5.3「超参包驱动」：
超参包中每个特征的 `spec` 字段是一段**声明式特征定义表达式（FeatureSpec）**，本
模块负责把这段文本**解释**为可执行的特征计算：
> `EMA(CLF-01.TEMP, 5m)` → 解析为 AST → 校验 → 依赖分析 → materialize 为时序
`core/model-framework/hyperparam.py`（issue #39）只对 `spec` 做「非空字符串」存在性
校验；本模块补齐其语义层，使：
1. **配置台**（issue #62~#67 Template Console）导入超参包时即可一次性展示每个特征
的解析结果与依赖点位，训练前暴露拼写/语义错误；
2. **训练/推理流水线**（issue #40）拿到 AST 后直接 materialize 为「按点位拉取 → 算子
计算」的特征管道；
3. **切换模板仅改超参包**，特征逻辑零代码（对齐 PRD「配置化」核心目标）。
## 模块结构
```
core/model-framework/
├── feature_spec.py FeatureSpec 引擎（解析/校验/依赖/执行/算子注册）
└── tests/
├── _bootstrap.py 测试引导（目录含连字符，挂载包名 model_framework）
└── test_feature_spec.py 解析/校验/依赖/执行/插件注册单元测试
```
## 设计要点
* **零外部强依赖**：解析/校验/依赖分析不依赖第三方库；执行（`materialize`）优先用
numpy 向量化，无 numpy 时退化为纯 Python（边缘/离线环境可加载与校验）。
* **不可变 AST + 函数式算子**：每个算子是纯函数 `op(series_map, args)`，注册到
`OPERATORS`；新增算子只需 `register_operator`（对齐 PRD「新增结构走插件注册」，
与 issue #34 Model Recipe 插件接口呼应）。
* **安全解析**：手写递归下降解析器，**绝不使用 `eval`/`exec`**——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
```
内置算子：`EMA / SMA / RollingStd / RollingMax / RollingMin / RateOfChange / Diff /
Lag / Log / Scale / Clip / Combine`，覆盖 PRD 5.3 超参包示例的全部特征形态。
## 使用
```python
from model_framework.feature_spec import (
parse, validate, resolve_inputs, materialize, describe, register_operator,
)
ast = parse("EMA(CLF-01.TEMP, 5m)") # 1. 解析
issues = validate(ast) # 2. 语义校验（空列表=通过）
tags = resolve_inputs(ast) # 3. 依赖点位：['CLF-01.TEMP']
feat = materialize(ast, {"CLF-01.TEMP": xs}) # 4. 执行 → list[float]
print(describe(ast)) # 人类可读描述
```
扩展算子（插件注册）：
```python
register_operator(
"Double", min_arity=1, max_arity=1, arg_kinds=(("tag",),),
func=lambda s, a: [x * 2 for x in s[a[0].name]], doc="翻倍",
)
materialize(parse("Double(A)"), {"A": [1.0, 2.0]}) # → [2.0, 4.0]
```
## 测试
```bash
cd core/model-framework/tests
python -m unittest test_feature_spec
```
core/model-framework/feature_spec.py
+861
View File
@@ -0,0 +1,861 @@
# -*- 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)
core/model-framework/tests/_bootstrap.py
+16
View File
@@ -0,0 +1,16 @@
# -*- coding: utf-8 -*-
"""测试引导：把 `core/model-framework` 以包名 `model_framework` 挂载到 sys.modules。
目录名 `model-framework` 含连字符，无法直接以包名 import；挂载后模块内相对导入
（`from .feature_spec import ...`）在 unittest 发现机制下可正常解析。
"""
import os
import sys
import types
MF_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, MF_DIR)
if "model_framework" not in sys.modules:
pkg = types.ModuleType("model_framework")
pkg.__path__ = [MF_DIR]
sys.modules["model_framework"] = pkg
core/model-framework/tests/test_feature_spec.py
+334
View File
@@ -0,0 +1,334 @@
# -*- coding: utf-8 -*-
"""FeatureSpec 声明式特征定义引擎测试（issue #35）。
覆盖：
1. 解析：算子调用、裸点位、数值/窗口字面量、嵌套、中文点位、带符号数值；
2. 解析错误：空 spec、非法字符、括号不匹配、多余内容、参数缺失；
3. 语义校验：未知算子、arity 不匹配、参数 kind 错误；
4. 依赖分析：resolve_inputs 去重与顺序、嵌套算子依赖汇总；
5. 执行：EMA/SMA/RollingStd/RateOfChange/Diff/Lag/Log/Scale/Clip/Combine 的
数值正确性，缺失点位 fail-fast；
6. 插件注册：register_operator 扩展新算子；
7. 往返：to_dict/repr 稳定。
"""
import math
import unittest
import _bootstrap # noqa: F401 挂载包名
from model_framework.feature_spec import (
FeatureAST,
Number,
OpCall,
OPERATORS,
ParseError,
SpecIssue,
TagRef,
Window,
describe,
materialize,
parse,
register_operator,
resolve_inputs,
validate,
)
# ---------------------------------------------------------------------------
# 解析
# ---------------------------------------------------------------------------
class ParseTest(unittest.TestCase):
def test_simple_op_with_window(self):
ast = parse("EMA(CLF-01.TEMP, 5m)")
self.assertEqual(
ast, OpCall("EMA", (TagRef("CLF-01.TEMP"), Window(5.0, "m")))
)
def test_simple_op_with_number_window(self):
ast = parse("RollingStd(CLF-01.CL2, 10)")
self.assertEqual(ast, OpCall("RollingStd", (TagRef("CLF-01.CL2"), Number(10))))
def test_bare_tag(self):
self.assertEqual(parse("炉压"), TagRef("炉压"))
def test_tag_with_dots_and_dash(self):
self.assertEqual(parse("A.B-C_01"), TagRef("A.B-C_01"))
def test_signed_and_scientific_number(self):
ast = parse("Scale(A, -0.5)")
self.assertEqual(ast, OpCall("Scale", (TagRef("A"), Number(-0.5))))
ast2 = parse("Scale(A, 1e-3)")
self.assertAlmostEqual(ast2.args[1].value, 0.001)
def test_nested_op(self):
# 嵌套：外层 Scale，内层 EMA 作为第一个参数点位位置（语法合法，语义由算子判定）
ast = parse("Combine(EMA(A, 5m), B)")
self.assertEqual(ast.name, "Combine")
self.assertEqual(len(ast.args), 2)
self.assertEqual(ast.args[0].name, "EMA")
def test_no_arg_op(self):
ast = parse("Diff()")
self.assertEqual(ast, OpCall("Diff", ()))
def test_integer_window_vs_number(self):
self.assertEqual(parse("Lag(A, 3)").args[1], Number(3))
self.assertEqual(parse("Lag(A, 3m)").args[1], Window(3.0, "m"))
def test_repr_roundtrip(self):
for spec in ["EMA(CLF-01.TEMP, 5m)", "RateOfChange(炉压)", "Clip(P, -1, 1)"]:
self.assertEqual(repr(parse(spec)).replace(" ", ""), spec.replace(" ", ""))
# ---- 解析错误 ----
def test_empty_raises(self):
with self.assertRaises((ValueError, ParseError)):
parse("")
with self.assertRaises((ValueError, ParseError)):
parse(" ")
def test_non_string_raises(self):
with self.assertRaises(ValueError):
parse(123) # type: ignore[arg-type]
def test_unrecognized_char(self):
with self.assertRaises(ParseError) as cm:
parse("EMA(A, 5m) @")
self.assertIsNotNone(cm.exception.position)
def test_missing_rparen(self):
with self.assertRaises(ParseError):
parse("EMA(A, 5m")
def test_missing_rparen_inner(self):
with self.assertRaises(ParseError):
parse("EMA(A, (5m)")
def test_trailing_garbage(self):
with self.assertRaises(ParseError):
parse("EMA(A, 5m) B")
def test_missing_arg_after_comma(self):
with self.assertRaises(ParseError):
parse("EMA(A, )")
def test_starts_with_paren(self):
with self.assertRaises(ParseError):
parse("(A)")
# ---------------------------------------------------------------------------
# 语义校验
# ---------------------------------------------------------------------------
class ValidateTest(unittest.TestCase):
def test_known_op_valid(self):
self.assertEqual(validate(parse("EMA(A, 5m)")), [])
def test_unknown_operator(self):
issues = validate(parse("FooBar(A, 5m)"))
self.assertEqual(len(issues), 1)
self.assertEqual(issues[0].code, "unknown_operator")
def test_arity_too_few(self):
issues = validate(parse("EMA(A)"))
self.assertTrue(any(i.code == "arity" for i in issues))
def test_arity_too_many(self):
issues = validate(parse("EMA(A, 5m, 7)"))
self.assertTrue(any(i.code == "arity" for i in issues))
def test_bad_arg_kind_number_where_window(self):
# EMA 第二参数允许 window/number，故合法
self.assertEqual(validate(parse("EMA(A, 7)")), [])
# 但 tag 位置传 number 非法
issues = validate(parse("EMA(5, 7)"))
self.assertTrue(any(i.code == "bad_arg" for i in issues))
def test_combine_varargs(self):
self.assertEqual(validate(parse("Combine(A, B, C)")), [])
issues = validate(parse("Combine(A)"))
self.assertTrue(any(i.code == "arity" for i in issues))
def test_nested_unknown(self):
issues = validate(parse("Combine(Foo(A), B)"))
self.assertTrue(any(i.code == "unknown_operator" for i in issues))
# ---------------------------------------------------------------------------
# 依赖分析
# ---------------------------------------------------------------------------
class ResolveInputsTest(unittest.TestCase):
def test_single_tag(self):
self.assertEqual(resolve_inputs(parse("炉压")), ["炉压"])
def test_dedup_order(self):
# 同一点位重复出现，去重且保持首次出现顺序
self.assertEqual(resolve_inputs(parse("Combine(A, A)")), ["A"])
def test_multiple_tags(self):
self.assertEqual(resolve_inputs(parse("Combine(A.tank1, A.tank2)")), ["A.tank1", "A.tank2"])
def test_op_collects_input(self):
self.assertEqual(resolve_inputs(parse("EMA(CLF-01.TEMP, 5m)")), ["CLF-01.TEMP"])
def test_number_window_no_inputs(self):
# 裸数值/窗口虽不是合法特征根，但 resolve_inputs 不报错
self.assertEqual(resolve_inputs(Number(3)), [])
self.assertEqual(resolve_inputs(Window(5.0, "m")), [])
# ---------------------------------------------------------------------------
# 执行
# ---------------------------------------------------------------------------
class MaterializeTest(unittest.TestCase):
def setUp(self):
# 一个稳定的伪时序：1..10
self.series = {"A": [float(i) for i in range(1, 11)]} # 1..10
def test_bare_tag(self):
self.assertEqual(materialize(parse("A"), self.series), self.series["A"])
def test_number(self):
self.assertEqual(materialize(Number(3), {}), 3)
def test_sma_window3(self):
out = materialize(parse("SMA(A, 3)"), self.series)
# 前 2 个 NaN，第 3 个 = (1+2+3)/3 = 2.0
self.assertTrue(math.isnan(out[0]) and math.isnan(out[1]))
self.assertAlmostEqual(out[2], 2.0)
self.assertAlmostEqual(out[9], (8 + 9 + 10) / 3)
def test_ema_decreasing_weight(self):
out = materialize(parse("EMA(A, 5)"), self.series)
# EMA 单调（输入单调增），首值 = 首个观测
self.assertAlmostEqual(out[0], 1.0)
self.assertTrue(all(out[i] <= out[i + 1] for i in range(len(out) - 1)))
def test_rolling_std(self):
out = materialize(parse("RollingStd(A, 2)"), self.series)
self.assertTrue(math.isnan(out[0]))
# std(1,2) 无偏 = 0.7071...
self.assertAlmostEqual(out[1], math.sqrt(0.5))
def test_rolling_max_min(self):
mx = materialize(parse("RollingMax(A, 3)"), self.series)
mn = materialize(parse("RollingMin(A, 3)"), self.series)
self.assertEqual(mx[2], 3.0)
self.assertEqual(mn[2], 1.0)
def test_diff(self):
out = materialize(parse("Diff(A)"), self.series)
self.assertTrue(math.isnan(out[0]))
self.assertTrue(all(out[i] == 1.0 for i in range(1, len(out))))
def test_lag(self):
out = materialize(parse("Lag(A, 2)"), self.series)
self.assertTrue(math.isnan(out[0]) and math.isnan(out[1]))
self.assertEqual(out[2], 1.0)
def test_rate_of_change(self):
# 常数序列 → 变化率为 0（非 NaN；NaN 仅出现在前 window 步预热）
const = {"C": [5.0] * 6}
out = materialize(parse("RateOfChange(C)"), const)
self.assertTrue(math.isnan(out[0])) # 预热步 NaN
self.assertEqual(out[1], 0.0)
# 含 0 的序列 → 分母为 0 → NaN
zero_denom = {"Z": [0.0, 1.0, 2.0]}
outz = materialize(parse("RateOfChange(Z)"), zero_denom)
self.assertTrue(math.isnan(outz[1]))
# 线性序列 ROC 步长1 = 1/prev
out2 = materialize(parse("RateOfChange(A)"), self.series)
self.assertAlmostEqual(out2[1], 1.0 / 1.0)
self.assertAlmostEqual(out2[5], 1.0 / 5.0)
def test_log_negative_nan(self):
data = {"P": [1.0, -2.0, math.e]}
out = materialize(parse("Log(P)"), data)
self.assertAlmostEqual(out[0], 0.0)
self.assertTrue(math.isnan(out[1]))
self.assertAlmostEqual(out[2], 1.0)
def test_scale(self):
out = materialize(parse("Scale(A, 10)"), self.series)
self.assertEqual(out[0], 10.0)
self.assertEqual(out[9], 100.0)
def test_clip(self):
out = materialize(parse("Clip(A, 3, 7)"), self.series)
self.assertEqual(out, [3.0, 3.0, 3.0, 4.0, 5.0, 6.0, 7.0, 7.0, 7.0, 7.0])
def test_combine(self):
data = {"A": [1.0, 2.0, 3.0], "B": [10.0, 20.0, 30.0]}
self.assertEqual(materialize(parse("Combine(A, B)"), data), [11.0, 22.0, 33.0])
def test_missing_input_fails_fast(self):
with self.assertRaises(KeyError):
materialize(parse("EMA(Missing, 3)"), {"A": [1.0, 2.0, 3.0]})
def test_unknown_op_fails_fast(self):
with self.assertRaises(ValueError):
materialize(OpCall("NoSuchOp", (TagRef("A"),)), self.series)
# ---------------------------------------------------------------------------
# 插件注册
# ---------------------------------------------------------------------------
class RegisterOperatorTest(unittest.TestCase):
def test_register_then_parse_and_run(self):
def _double(series_map, args):
tag = args[0]
return [x * 2 for x in series_map[tag.name]]
register_operator(
"Double",
min_arity=1,
max_arity=1,
arg_kinds=(("tag",),),
func=_double,
doc="示例自定义算子：翻倍",
)
try:
self.assertIn("Double", OPERATORS)
self.assertEqual(validate(parse("Double(A)")), [])
self.assertEqual(
materialize(parse("Double(A)"), {"A": [1.0, 2.0]}), [2.0, 4.0]
)
finally:
OPERATORS.pop("Double", None)
def test_register_overrides(self):
register_operator(
"Stub", min_arity=0, max_arity=0, arg_kinds=(), func=lambda s, a: 1, doc="v1"
)
register_operator(
"Stub", min_arity=0, max_arity=0, arg_kinds=(), func=lambda s, a: 2, doc="v2"
)
try:
self.assertEqual(OPERATORS["Stub"].doc, "v2")
finally:
OPERATORS.pop("Stub", None)
# ---------------------------------------------------------------------------
# 描述 / 往返
# ---------------------------------------------------------------------------
class DescribeAndSerializeTest(unittest.TestCase):
def test_describe_contains_inputs(self):
d = describe(parse("EMA(CLF-01.TEMP, 5m)"))
self.assertIn("CLF-01.TEMP", d)
self.assertIn("EMA", d)
def test_to_dict_roundtrip_shape(self):
ast = parse("RateOfChange(炉压)")
d = ast.to_dict()
self.assertEqual(d["kind"], "op")
self.assertEqual(d["name"], "RateOfChange")
self.assertEqual(d["args"][0], {"kind": "tag", "name": "炉压"})
def test_window_seconds(self):
w = Window(5.0, "m")
self.assertEqual(w.seconds, 300)
self.assertEqual(Window(2.0, "h").seconds, 7200)
if __name__ == "__main__":
unittest.main()