feat(#68): [Ti-1] 氯化车间质量预测特征工程（基于点位字典）

新增 templates/ti-cl4/quality-forecast/features.py：
- PointDict：解析 point_dict CSV（9 列，对齐 core/edge-gateway），检索/存在性校验
- FeatureSpec：声明式特征（source/transform/window/meaning），换行业只改清单
- FeatureExtractor：按清单从时序样本抽取特征矩阵，缺失值 NaN 占位
- 9 算子：raw/mean/std/min/max/range/diff/slope/ratio
- 零依赖 YAML 子集加载（与 recipe-optim/data-bus 同款）
- config/features.template.yaml：7 个默认特征（炉温/配比/CO波动/炉层/TiCl₄纯度）
- 22 用例全通过；纯标准库零运行时依赖。

templates/ti-cl4/quality-forecast/README.md

@@ -0,0 +1,37 @@
+# Ti-1 氯化车间质量预测（quality-forecast）
+
+> 父 Issue：#10「[Template-Ti 一期] ① 质量预测 + ③ 炉层杂质预警」
+> 子任务：#68 特征工程 / #69 模型训练与评估 / #73 模型部署到内核并接入驾驶舱
+
+基于 `templates/ti-cl4/point-dict/point_dict.default.csv` 默认点位集（CLF-01 沸腾氯化炉 /
+RF-01 精制还原 / E-01 能源 / ST-01 蒸汽 / CW-01 循环水），交付「DCS 点表 → 可训练特征
+矩阵 → 质量预测 → 部署接入」全链路，对齐 PRD §5.3 ①「质量预测」。
+
+## 设计要点
+
+- **模板化**：特征清单（`config/features.template.yaml`）外置，换行业/换模板只改特征
+  清单 + 点位字典，特征工程代码零改动（PRD §5.3「换行业只改 Recipe」）。
+- **纯标准库零运行时依赖**：CSV/YAML 子集/统计全部自实现（不依赖 numpy/pandas/pyyaml），
+  便于离线/隔离网部署，与 `recipe-optim` / `impurity-forecast` 内核模块一致。
+- **可解释前置**：每个 `FeatureSpec` 带 `meaning`（工艺含义），供 #69 模型可解释性引用。
+
+## 模块
+
+| 文件 | 说明 | Issue |
+| --- | --- | --- |
+| `features.py` | 点位字典 `PointDict` + 特征规格 `FeatureSpec` + 抽取器 `FeatureExtractor` | #68 |
+| `model.py` | 质量预测模型训练与评估（岭回归 + R²） | #69 |
+| `serve.py` | 模型部署到内核并接入驾驶舱（预测服务） | #73 |
+| `config/features.template.yaml` | 默认特征清单（7 个特征，对齐默认点位集） | #68 |
+
+## 算子
+
+`raw / mean / std / min / max / range / diff / slope / ratio`（`ratio` 需 `denominator`）。
+缺失点位统一用 `NaN` 占位，便于上层判空屏蔽（`FeatureMatrix.drop_nan_rows()`）。
+
+## 运行
+
+```bash
+# 单测（嵌入式 Python 3.12 即可，无第三方依赖）
+python -m unittest discover -s templates/ti-cl4/quality-forecast/tests -p "test_*.py" -v
+```

templates/ti-cl4/quality-forecast/__init__.py

@@ -0,0 +1,14 @@
+# -*- coding: utf-8 -*-
+"""Ti-1 氯化车间质量预测模板包（Issue #68/#69/#73）。
+
+子模块：
+
+- ``features``：基于点位字典的特征工程（#68）；
+- ``model``：质量预测模型训练与评估（#69）；
+- ``serve``：模型部署到内核并接入驾驶舱（#73）。
+
+设计口径：纯标准库、零运行时依赖，便于离线/隔离网部署（与既有
+``recipe-optim`` / ``impurity-forecast`` 内核模块一致）。
+"""
+
+__all__ = ["features", "model", "serve"]

templates/ti-cl4/quality-forecast/_sanity_check.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+"""Ti-1 质量预测 sanity 检查（无构建环境下的离线基本验证）。
+
+检查项：
+1. 默认特征清单 features.template.yaml 可加载且通过校验；
+2. 特征 source/denominator 的点位都在默认点位字典内；
+3. 全部测试用例通过。
+
+用法：python _sanity_check.py
+退出码：0 全通过，非 0 有失败。
+"""
+import os
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TESTS = os.path.join(HERE, "tests")
+
+
+def main() -> int:
+    failures = []
+
+    # 1) 加载特征清单并校验点位
+    try:
+        sys.path.insert(0, TESTS)
+        import _bootstrap  # noqa: F401  挂载 quality_forecast
+        from quality_forecast import features as F
+        ti_cl4 = os.path.dirname(HERE)
+        pd = F.PointDict.from_csv(
+            os.path.join(ti_cl4, "point-dict", "point_dict.default.csv"))
+        with open(os.path.join(HERE, "config", "features.template.yaml"),
+                  "r", encoding="utf-8") as fh:
+            ext = F.load_feature_specs(fh.read(), pd)
+        if not ext.names:
+            failures.append("特征清单为空")
+    except Exception as exc:  # noqa: BLE001
+        failures.append(f"特征清单加载失败: {exc}")
+
+    # 2) 跑测试
+    loader = unittest.TestLoader()
+    suite = loader.discover(TESTS, pattern="test_*.py")
+    runner = unittest.TextTestRunner(verbosity=1)
+    result = runner.run(suite)
+    if not result.wasSuccessful():
+        failures.append(f"{len(result.failures)} 失败, {len(result.errors)} 错误")
+
+    if failures:
+        print("\n[sanity] 失败:")
+        for f in failures:
+            print("  -", f)
+        return 1
+    print("\n[sanity] 全部通过")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

templates/ti-cl4/quality-forecast/config/features.template.yaml

@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+# Ti-1 氯化车间质量预测 · 特征清单模板（Issue #68 / PRD §5.3 ①）。
+#
+# 模板化技术路径（PRD §5.3）：特征清单外置，换行业/换模板只改本文件 + 点位字典，
+# 特征工程代码零改动。本清单基于 templates/ti-cl4/point-dict/point_dict.default.csv
+# 的默认点位集（CLF-01 沸腾氯化炉 / RF-01 精制还原）。
+#
+# 算子说明：
+#   raw     取窗口内最新值
+#   mean    窗口均值
+#   std     窗口标准差（样本无偏）
+#   min/max 窗口极值
+#   range   窗口极差
+#   diff    最近两点差分
+#   slope   最小二乘斜率（值/秒）
+#   ratio   source/denominator 比值（需 denominator）
+#
+# 缺省窗口 60s；meaning 字段供 #69 模型可解释性引用。
+
+features:
+  - name: clf_temp_mean
+    source: CLF-01.TEMP
+    transform: mean
+    window: 60
+    unit: "℃"
+    meaning: 沸腾氯化炉炉温窗口均值（反应强度主控变量）
+  - name: clf_temp_slope
+    source: CLF-01.TEMP
+    transform: slope
+    window: 120
+    unit: "℃/s"
+    meaning: 炉温变化趋势（升温过快影响 TiCl₄纯度）
+  - name: clf_cl2_feed_ratio
+    source: CLF-01.CL2
+    transform: ratio
+    denominator: CLF-01.FEED
+    window: 60
+    unit: "m³/t"
+    meaning: 氯气/进料配比（配比偏离是杂质主因）
+  - name: clf_co_std
+    source: CLF-01.CO
+    transform: std
+    window: 120
+    unit: "%"
+    meaning: CO 含量波动（燃烧不稳信号）
+  - name: clf_bed_range
+    source: CLF-01.BED
+    transform: range
+    window: 300
+    unit: "%"
+    meaning: 炉层状态波动范围（偏钛酸铁预警信号）
+  - name: rf_purity_last
+    source: RF-01.PURITY
+    transform: raw
+    window: 60
+    unit: "%"
+    meaning: TiCl₄ 纯度最新读数（LIMS 低频，质量标签候选）
+  - name: rf_imp_mean
+    source: RF-01.IMP
+    transform: mean
+    window: 60
+    unit: "%"
+    meaning: 杂质含量窗口均值（质量标签候选）

templates/ti-cl4/quality-forecast/features.py

@@ -0,0 +1,629 @@
+# -*- coding: utf-8 -*-
+"""Ti-1 氯化车间质量预测 · 特征工程（基于点位字典）（Issue #68 / PRD §5.3 ①）。
+
+承接 PRD §5.3 ①「① 质量预测」与父 Issue #10「[Template-Ti 一期] ① 质量预测 +
+③ 炉层杂质预警」：把"DCS 点表 → 可训练的特征矩阵"这条链路**模板化、可配置、
+可测试**，且与 #69 模型训练、#73 模型部署解耦。
+
+PRD 设计口径
+------------
+- 架构表（PRD §5.3）：``质量预测 | 预测 | 入：DCS实时数据+LIMS；
+  出：质量指标预测值(纯度/杂质) | ① 质量预测 | 中``。
+- 模板化技术路径：特征清单（``FeatureSpec``）外置为 YAML/JSON 超参包，
+  切换模板/行业只改特征清单，特征工程代码零改动（PRD §5.3「换行业只改 Recipe」）。
+- 数据门槛：一期客户 DCS 点表未到位时启用默认通用点位集完成框架验证
+  （PRD §13 缺省策略），故本模块**不依赖真实历史数据**——用合成/默认点位即可
+  完整跑通特征抽取，单测零外部数据依赖。
+
+本模块交付
+----------
+1. **点位字典加载 ``PointDict``**：解析 ``point_dict.default.csv``
+   （device_id/point_id/name/unit/...，与 ``core/edge-gateway`` 同款 9 列），
+   提供 ``by_point_id`` / ``by_device`` 检索与点位存在性校验。
+2. **特征规格 ``FeatureSpec``**：声明式特征——``name``、``source``（点位 point_id
+   或常量）、``transform``（聚合算子 raw/mean/std/min/max/diff/ratio/…）、
+   ``window``（时间窗，秒）、``meaning``（工艺含义，供 #69/#73 可解释引用）。
+3. **特征抽取器 ``FeatureExtractor``**：按特征清单从时序样本（``Sample`` 列表）
+   抽取特征向量；缺失值用 ``NaN`` 占位（与 impurity-forecast / recipe-optim 一致，
+   便于上层判空屏蔽）；输出有序 ``FeatureMatrix``（行=样本时刻，列=特征）。
+4. **声明式加载**：从 YAML/JSON 特征清单加载（零第三方依赖 YAML 子集解析，
+   与 recipe-optim / data-bus / rag-kb 同款）。
+
+设计要点
+--------
+- **零运行时依赖**（纯标准库）：CSV 用 ``csv``、YAML 子集自实现、统计用 ``math``
+  与手写聚合（不依赖 numpy/pandas），便于隔离网部署。
+- **可解释前置**：特征 ``meaning`` 字段，为 #69 模型可解释性预留引用依据。
+- **可校验**：``FeatureSpec.validate`` 聚合列出全部错误（未知点位/非法算子/负窗
+  口），便于配置台一次性反馈。
+"""
+from __future__ import annotations
+
+import csv
+import math
+import os
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
+
+# 缺失值统一用 float('nan')，与 impurity-forecast / recipe-optim 一致。
+NAN = float("nan")
+
+# 点位字典 CSV 表头（与 core/edge-gateway/config/point_dict.example.csv 对齐，9 列）
+POINT_COLUMNS = [
+    "device_id", "point_id", "name", "unit", "dataType",
+    "sampleRate", "qualityCode", "opcNode", "protocol",
+]
+
+# 允许的特征变换算子（与 impurity-forecast 特征口径对齐）
+ALLOWED_TRANSFORMS = {
+    "raw", "mean", "std", "min", "max", "range", "diff", "ratio", "slope",
+}
+
+
+class FeatureError(ValueError):
+    """特征工程错误（未知点位 / 非法算子 / 窗口非法 / 重复特征名等）。"""
+
+
+# ---------------------------------------------------------------------------
+# 点位字典
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class Point:
+    """点位字典一行。"""
+
+    device_id: str
+    point_id: str
+    name: str
+    unit: str
+    data_type: str = "float"
+    sample_rate: int = 1000
+    quality_code: str = "true"
+    opc_node: str = ""
+    protocol: str = ""
+
+    @classmethod
+    def from_row(cls, row: Dict[str, str]) -> "Point":
+        return cls(
+            device_id=(row.get("device_id") or "").strip(),
+            point_id=(row.get("point_id") or "").strip(),
+            name=(row.get("name") or "").strip(),
+            unit=(row.get("unit") or "").strip(),
+            data_type=(row.get("dataType") or "float").strip(),
+            sample_rate=int(float(row.get("sampleRate") or 1000)),
+            quality_code=(row.get("qualityCode") or "true").strip(),
+            opc_node=(row.get("opcNode") or "").strip(),
+            protocol=(row.get("protocol") or "").strip(),
+        )
+
+
+class PointDict:
+    """点位字典：解析 CSV，提供检索与存在性校验。"""
+
+    def __init__(self, points: Sequence[Point]):
+        self._by_id: Dict[str, Point] = {p.point_id: p for p in points}
+        self._by_device: Dict[str, List[Point]] = {}
+        for p in points:
+            self._by_device.setdefault(p.device_id, []).append(p)
+        self.points: Tuple[Point, ...] = tuple(points)
+
+    @classmethod
+    def from_csv(cls, path: str) -> "PointDict":
+        with open(path, "r", encoding="utf-8") as fh:
+            rows = list(csv.DictReader(fh))
+        if not rows:
+            raise FeatureError(f"点位字典为空: {path}")
+        header = list(rows[0].keys())
+        missing = [c for c in POINT_COLUMNS if c not in header]
+        if missing:
+            raise FeatureError(f"点位字典缺列: {missing}")
+        return cls([Point.from_row(r) for r in rows])
+
+    def has(self, point_id: str) -> bool:
+        return point_id in self._by_id
+
+    def by_point_id(self, point_id: str) -> Point:
+        if point_id not in self._by_id:
+            raise FeatureError(f"未知点位: {point_id}")
+        return self._by_id[point_id]
+
+    def by_device(self, device_id: str) -> List[Point]:
+        return list(self._by_device.get(device_id, []))
+
+    def point_ids(self) -> List[str]:
+        return list(self._by_id.keys())
+
+
+# ---------------------------------------------------------------------------
+# 特征规格
+# ---------------------------------------------------------------------------
+
+
+class Transform(str, Enum):
+    RAW = "raw"
+    MEAN = "mean"
+    STD = "std"
+    MIN = "min"
+    MAX = "max"
+    RANGE = "range"
+    DIFF = "diff"
+    RATIO = "ratio"
+    SLOPE = "slope"
+
+
+@dataclass
+class FeatureSpec:
+    """声明式特征规格。
+
+    - ``source`` 形如 ``CLF-01.TEMP``（点位 point_id）或常量数值；
+    - ``transform`` 聚合算子（raw/mean/std/min/max/range/diff/ratio/slope）；
+    - ``window`` 时间窗（秒，仅滚动窗算子有意义；raw/diff 用最近两点）；
+    - ``denominator`` 仅 ratio 算子使用（另一个 point_id 或常量）；
+    - ``meaning`` 工艺含义（#69/#73 可解释性引用）。
+    """
+
+    name: str
+    source: str
+    transform: str = "raw"
+    window: float = 60.0
+    denominator: Optional[str] = None
+    meaning: str = ""
+    unit: str = ""
+
+    def validate(self, point_dict: Optional[PointDict] = None) -> List[str]:
+        errors: List[str] = []
+        if not self.name:
+            errors.append("特征 name 不能为空")
+        if self.transform not in ALLOWED_TRANSFORMS:
+            errors.append(f"特征 {self.name}: 非法 transform={self.transform}")
+        if self.window < 0:
+            errors.append(f"特征 {self.name}: window 不能为负 (={self.window})")
+        if self.transform == "ratio" and not self.denominator:
+            errors.append(f"特征 {self.name}: ratio 算子需指定 denominator")
+        # 点位存在性（source/denominator 形如 point_id 时校验）
+        if point_dict is not None:
+            for label, val in (("source", self.source),
+                               ("denominator", self.denominator)):
+                if val and not _is_constant(val) and not point_dict.has(val):
+                    errors.append(f"特征 {self.name}: {label}={val} 不在点位字典")
+        return errors
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> "FeatureSpec":
+        return cls(
+            name=str(d.get("name", "")).strip(),
+            source=str(d.get("source", "")).strip(),
+            transform=str(d.get("transform", "raw")).strip(),
+            window=float(d.get("window", 60.0)),
+            denominator=(str(d.get("denominator")).strip()
+                         if d.get("denominator") else None),
+            meaning=str(d.get("meaning", "")).strip(),
+            unit=str(d.get("unit", "")).strip(),
+        )
+
+
+def _is_constant(val: str) -> bool:
+    """source/denominator 是否为常量数值（而非 point_id）。"""
+    try:
+        float(val)
+        return True
+    except (TypeError, ValueError):
+        return False
+
+
+# ---------------------------------------------------------------------------
+# 时序样本
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class Sample:
+    """一个采样时刻的多点位读数。
+
+    - ``ts`` 时间戳（秒，单调不减）；
+    - ``values`` point_id → 数值；缺失点位视为无读数。
+    """
+
+    ts: float
+    values: Dict[str, float] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# 特征抽取
+# ---------------------------------------------------------------------------
+
+
+class FeatureExtractor:
+    """按特征清单从时序样本抽取特征向量。
+
+    用法::
+
+        ext = FeatureExtractor(specs, point_dict)
+        matrix = ext.extract(samples)
+        # matrix.rows[i] 是一个有序特征向量；matrix.names 是列名
+    """
+
+    def __init__(self, specs: Sequence[FeatureSpec],
+                 point_dict: Optional[PointDict] = None,
+                 *, strict: bool = True):
+        self.specs: Tuple[FeatureSpec, ...] = tuple(specs)
+        self.point_dict = point_dict
+        if strict:
+            errors = self.validate()
+            if errors:
+                raise FeatureError("特征清单校验失败:\n  " + "\n  ".join(errors))
+        # 重复特征名检查
+        names = [s.name for s in self.specs]
+        dup = {n for n in names if names.count(n) > 1}
+        if dup and strict:
+            raise FeatureError(f"重复特征名: {sorted(dup)}")
+
+    def validate(self) -> List[str]:
+        errors: List[str] = []
+        for s in self.specs:
+            errors.extend(s.validate(self.point_dict))
+        return errors
+
+    @property
+    def names(self) -> List[str]:
+        return [s.name for s in self.specs]
+
+    def extract(self, samples: Sequence[Sample]) -> "FeatureMatrix":
+        rows: List[List[float]] = []
+        # 滚动窗：按 window 秒选取 <= ts 的历史样本
+        win = [s.window for s in self.specs]
+        max_window = max(win) if win else 0.0
+        ordered = sorted(samples, key=lambda s: s.ts)
+        for cur in ordered:
+            window_samples = [
+                s for s in ordered
+                if cur.ts - max_window <= s.ts <= cur.ts
+            ]
+            row = [self._compute(spec, cur, window_samples)
+                   for spec in self.specs]
+            rows.append(row)
+        return FeatureMatrix(names=self.names, rows=rows)
+
+    # 单特征计算 ------------------------------------------------------------
+    def _compute(self, spec: FeatureSpec, cur: Sample,
+                 window_samples: Sequence[Sample]) -> float:
+        series = _series(spec.source, window_samples, self.point_dict)
+        denom_series = (
+            _series(spec.denominator, window_samples, self.point_dict)
+            if spec.denominator else []
+        )
+        tf = spec.transform
+        if tf == "raw":
+            return _last_or_nan(series)
+        if tf == "mean":
+            return _mean(series)
+        if tf == "std":
+            return _std(series)
+        if tf == "min":
+            return _min(series)
+        if tf == "max":
+            return _max(series)
+        if tf == "range":
+            return _range(series)
+        if tf == "diff":
+            return _diff(series)
+        if tf == "slope":
+            return _slope(series, spec.window)
+        if tf == "ratio":
+            return _ratio(_last_or_nan(series), _last_or_nan(denom_series))
+        # 不应到达（已 validate）
+        return NAN
+
+
+@dataclass
+class FeatureMatrix:
+    """特征抽取结果：有序特征名 + 行向量集合。"""
+
+    names: List[str]
+    rows: List[List[float]]
+
+    def column(self, name: str) -> List[float]:
+        idx = self.names.index(name)
+        return [r[idx] for r in self.rows]
+
+    def to_records(self) -> List[Dict[str, float]]:
+        return [dict(zip(self.names, row)) for row in self.rows]
+
+    def drop_nan_rows(self) -> "FeatureMatrix":
+        """丢弃任一特征为 NaN 的行（数据门槛不足时常用）。"""
+        clean = [r for r in self.rows if not any(math.isnan(v) for v in r)]
+        return FeatureMatrix(names=list(self.names), rows=clean)
+
+
+# ---------------------------------------------------------------------------
+# 聚合算子（纯标准库）
+# ---------------------------------------------------------------------------
+
+
+def _series(source: str, samples: Sequence[Sample],
+            point_dict: Optional[PointDict]) -> List[Tuple[float, float]]:
+    """取一个 source 的 (ts, value) 序列。常量源展开为各样本时刻。"""
+    if _is_constant(source):
+        const = float(source)
+        return [(s.ts, const) for s in samples]
+    return [(s.ts, s.values[source]) for s in samples
+            if source in s.values and not math.isnan(s.values[source])]
+
+
+def _last_or_nan(series: Sequence[Tuple[float, float]]) -> float:
+    return series[-1][1] if series else NAN
+
+
+def _values(series: Sequence[Tuple[float, float]]) -> List[float]:
+    return [v for _, v in series]
+
+
+def _mean(series: Sequence[Tuple[float, float]]) -> float:
+    vs = _values(series)
+    return sum(vs) / len(vs) if vs else NAN
+
+
+def _std(series: Sequence[Tuple[float, float]]) -> float:
+    vs = _values(series)
+    n = len(vs)
+    if n < 2:
+        return NAN if n == 0 else 0.0
+    mu = sum(vs) / n
+    var = sum((v - mu) ** 2 for v in vs) / (n - 1)
+    return math.sqrt(var)
+
+
+def _min(series: Sequence[Tuple[float, float]]) -> float:
+    vs = _values(series)
+    return min(vs) if vs else NAN
+
+
+def _max(series: Sequence[Tuple[float, float]]) -> float:
+    vs = _values(series)
+    return max(vs) if vs else NAN
+
+
+def _range(series: Sequence[Tuple[float, float]]) -> float:
+    vs = _values(series)
+    return (max(vs) - min(vs)) if vs else NAN
+
+
+def _diff(series: Sequence[Tuple[float, float]]) -> float:
+    if len(series) < 2:
+        return NAN
+    return series[-1][1] - series[-2][1]
+
+
+def _slope(series: Sequence[Tuple[float, float]], window: float) -> float:
+    """最小二乘斜率（值/秒）；样本不足返回 NaN。"""
+    if len(series) < 2:
+        return NAN
+    xs = [t for t, _ in series]
+    # 时间窗外的样本不参与（已由 caller 截窗，这里再以 window 收敛）
+    if window and window > 0:
+        tmax = max(xs)
+        kept = [(t, v) for t, v in series if t >= tmax - window]
+        if len(kept) < 2:
+            return NAN
+        xs = [t for t, _ in kept]
+        ys = [v for _, v in kept]
+    else:
+        ys = [v for _, v in series]
+    n = len(xs)
+    xbar = sum(xs) / n
+    ybar = sum(ys) / n
+    num = sum((xs[i] - xbar) * (ys[i] - ybar) for i in range(n))
+    den = sum((xs[i] - xbar) ** 2 for i in range(n))
+    return num / den if den else NAN
+
+
+def _ratio(a: float, b: float) -> float:
+    if math.isnan(a) or math.isnan(b) or b == 0:
+        return NAN
+    return a / b
+
+
+# ---------------------------------------------------------------------------
+# 声明式加载（零第三方依赖 YAML 子集解析，与 recipe-optim 同款）
+# ---------------------------------------------------------------------------
+
+
+def load_feature_specs(text: str,
+                       point_dict: Optional[PointDict] = None,
+                       *, strict: bool = True) -> FeatureExtractor:
+    """从 YAML/JSON 文本加载特征清单并构造 FeatureExtractor。
+
+    支持的 YAML 子集：``features:`` 顶层键，下为 ``- name/source/transform/...``
+    列表项。也兼容 JSON（``{"features": [...]}``）。
+    """
+    text = text.strip()
+    data: Any
+    if text.startswith("{") or text.startswith("["):
+        import json
+        data = json.loads(text)
+    else:
+        data = _parse_yaml_subset(text)
+    if not isinstance(data, dict):
+        raise FeatureError("特征清单顶层应为映射（含 features 键）")
+    raw_features = data.get("features")
+    if not isinstance(raw_features, list):
+        raise FeatureError("特征清单缺少 features 列表")
+    specs = [FeatureSpec.from_dict(f) for f in raw_features if isinstance(f, dict)]
+    if not specs:
+        raise FeatureError("特征清单 features 为空")
+    return FeatureExtractor(specs, point_dict, strict=strict)
+
+
+def _parse_yaml_subset(text: str) -> Any:
+    """极简 YAML 子集解析器（仅供模板资产，非通用 YAML）。
+
+    支持：注释（# ...）、映射（key: value）、列表（- item）、嵌套缩进、
+    基本标量（int/float/str/bool/null）。与 recipe-optim / data-bus 同款。
+    """
+    lines: List[str] = []
+    for raw in text.splitlines():
+        stripped = raw.rstrip()
+        if not stripped.strip():
+            continue
+        if stripped.lstrip().startswith("#"):
+            continue
+        hi = _find_inline_comment(stripped)
+        if hi is not None:
+            stripped = stripped[:hi].rstrip()
+        if stripped:
+            lines.append(stripped)
+    parser = _YamlParser(lines)
+    return parser.parse_block(0)[0] if lines else {}
+
+
+def _find_inline_comment(line: str) -> Optional[int]:
+    depth = 0
+    in_str = False
+    for i, ch in enumerate(line):
+        if ch == '"':
+            in_str = not in_str
+        elif not in_str:
+            if ch in "[{":
+                depth += 1
+            elif ch in "]}":
+                depth = max(0, depth - 1)
+            elif ch == "#" and depth == 0:
+                if i == 0 or line[i - 1] in (" ", "\t"):
+                    return i
+    return None
+
+
+def _parse_scalar(raw: str) -> Any:
+    raw = raw.strip()
+    if not raw:
+        return None
+    if raw.startswith('"') and raw.endswith('"'):
+        return raw[1:-1]
+    if raw.startswith("[") or raw.startswith("{"):
+        import json
+        try:
+            return json.loads(raw)
+        except Exception:
+            return raw
+    low = raw.lower()
+    if low == "true":
+        return True
+    if low == "false":
+        return False
+    if low in ("null", "~", "none"):
+        return None
+    try:
+        return int(raw)
+    except ValueError:
+        pass
+    try:
+        return float(raw)
+    except ValueError:
+        pass
+    return raw
+
+
+class _YamlParser:
+    """递归下降的 YAML 子集解析器（按缩进分层）。"""
+
+    def __init__(self, lines: List[str]) -> None:
+        self.lines = lines
+        self.i = 0
+
+    def _indent(self, line: str) -> int:
+        return len(line) - len(line.lstrip(" "))
+
+    def parse_block(self, indent: int) -> Tuple[Any, bool]:
+        if self.i >= len(self.lines):
+            return {}, False
+        line = self.lines[self.i]
+        cur = self._indent(line)
+        if cur < indent:
+            return {}, False
+        stripped = line.strip()
+        if stripped.startswith("- ") or stripped == "-":
+            return self._parse_list(cur), True
+        return self._parse_mapping(cur), False
+
+    def _parse_mapping(self, indent: int) -> Dict[str, Any]:
+        result: Dict[str, Any] = {}
+        effective = indent
+        if self.i < len(self.lines):
+            first = self._indent(self.lines[self.i])
+            if first > indent:
+                effective = first
+        while self.i < len(self.lines):
+            line = self.lines[self.i]
+            cur = self._indent(line)
+            if cur < effective:
+                break
+            if cur > effective:
+                self.i += 1
+                continue
+            stripped = line.strip()
+            if stripped.startswith("- "):
+                break
+            key, sep, rest = stripped.partition(":")
+            if not sep:
+                self.i += 1
+                continue
+            key = key.strip()
+            rest = rest.strip()
+            self.i += 1
+            if rest:
+                result[key] = _parse_scalar(rest)
+            else:
+                # 子块
+                if self.i < len(self.lines):
+                    nxt = self._indent(self.lines[self.i])
+                    if nxt > effective:
+                        val, _ = self.parse_block(nxt)
+                        result[key] = val
+        return result
+
+    def _parse_list(self, indent: int) -> List[Any]:
+        items: List[Any] = []
+        while self.i < len(self.lines):
+            line = self.lines[self.i]
+            cur = self._indent(line)
+            if cur < indent:
+                break
+            if cur > indent:
+                self.i += 1
+                continue
+            stripped = line.strip()
+            if not stripped.startswith("-"):
+                break
+            item_text = stripped[1:].strip()
+            if not item_text:
+                # 子块（嵌套映射/列表）
+                if self.i + 1 < len(self.lines):
+                    nxt = self._indent(self.lines[self.i + 1])
+                    if nxt > cur:
+                        self.i += 1
+                        val, _ = self.parse_block(nxt)
+                        items.append(val)
+                        continue
+                self.i += 1
+                items.append(None)
+                continue
+            # "- key: value" 形式 → 该 item 是映射
+            if ":" in item_text and not item_text.startswith('"'):
+                k, sep, v = item_text.partition(":")
+                if sep:
+                    item: Dict[str, Any] = {k.strip(): _parse_scalar(v.strip())}
+                    self.i += 1
+                    # 后续同缩进的 key 归入同一 item
+                    if self.i < len(self.lines):
+                        child_indent = self._indent(self.lines[self.i])
+                        if child_indent > cur:
+                            sub, _ = self.parse_block(child_indent)
+                            if isinstance(sub, dict):
+                                item.update(sub)
+                    items.append(item)
+                    continue
+            items.append(_parse_scalar(item_text))
+            self.i += 1
+        return items

templates/ti-cl4/quality-forecast/tests/_bootstrap.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+"""测试引导：把连字符目录挂载为可导入包（与 core 模块同款模式）。
+
+- ``templates/ti-cl4/quality-forecast`` → 包名 ``quality_forecast``。
+本模块零内核依赖（纯标准库），仅挂载自身包即可。
+"""
+import importlib.util
+import os
+import sys
+
+PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _load_package(name: str, path: str) -> None:
+    """按文件路径完整加载一个包（执行其 __init__.py）。"""
+    if name in sys.modules:
+        return
+    init_py = os.path.join(path, "__init__.py")
+    spec = importlib.util.spec_from_file_location(
+        name, init_py, submodule_search_locations=[path])
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+
+
+_load_package("quality_forecast", PKG_DIR)

templates/ti-cl4/quality-forecast/tests/test_features.py

@@ -0,0 +1,228 @@
+# -*- coding: utf-8 -*-
+"""Ti-1 质量预测特征工程测试（Issue #68）。
+
+覆盖：
+1. 点位字典加载（CSV 解析、检索、存在性校验、缺列报错）；
+2. FeatureSpec 校验（非法算子/未知点位/负窗口/ratio 缺 denominator）；
+3. 各 transform 算子（raw/mean/std/min/max/range/diff/slope/ratio）数值正确；
+4. 缺失点位 → NaN 占位；
+5. 滚动窗：window 外的样本不参与；
+6. 声明式加载（YAML 子集 + JSON）；
+7. 重复特征名报错；
+8. 模板资产 features.template.yaml 可加载并通过校验（对齐默认点位集）。
+"""
+import math
+import os
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))            # .../quality-forecast/tests
+PKG_DIR = os.path.dirname(HERE)                              # .../quality-forecast
+TI_CL4_DIR = os.path.dirname(PKG_DIR)                        # .../ti-cl4
+sys.path.insert(0, HERE)
+import _bootstrap  # noqa: F401,E402  挂载 quality_forecast 包
+
+from quality_forecast import features as F  # noqa: E402
+
+PDICT_DEFAULT = os.path.join(
+    TI_CL4_DIR, "point-dict", "point_dict.default.csv")
+FEATURES_TPL = os.path.join(
+    PKG_DIR, "config", "features.template.yaml")
+
+
+def _pdict():
+    return F.PointDict.from_csv(PDICT_DEFAULT)
+
+
+def _samples(values, ts0=0.0, step=10.0):
+    """构造样本序列：values 是 [{point_id: v}, ...]。"""
+    out = []
+    for i, vmap in enumerate(values):
+        out.append(F.Sample(ts=ts0 + i * step, values=dict(vmap)))
+    return out
+
+
+class TestPointDict(unittest.TestCase):
+    def test_load_default_csv(self):
+        pd = _pdict()
+        self.assertTrue(pd.has("CLF-01.TEMP"))
+        self.assertIn("CLF-01", {p.device_id for p in pd.points})
+
+    def test_by_point_id_unknown_raises(self):
+        pd = _pdict()
+        with self.assertRaises(F.FeatureError):
+            pd.by_point_id("NOPE")
+
+    def test_by_device(self):
+        pd = _pdict()
+        clf = pd.by_device("CLF-01")
+        self.assertTrue(all(p.device_id == "CLF-01" for p in clf))
+        self.assertGreater(len(clf), 0)
+
+
+class TestFeatureSpecValidate(unittest.TestCase):
+    def test_bad_transform(self):
+        s = F.FeatureSpec(name="x", source="CLF-01.TEMP", transform="bogus")
+        self.assertIn("非法 transform", "\n".join(s.validate()))
+
+    def test_negative_window(self):
+        s = F.FeatureSpec(name="x", source="CLF-01.TEMP", window=-1)
+        self.assertIn("window 不能为负", "\n".join(s.validate()))
+
+    def test_ratio_needs_denominator(self):
+        s = F.FeatureSpec(name="x", source="CLF-01.CL2", transform="ratio")
+        self.assertIn("denominator", "\n".join(s.validate()))
+
+    def test_unknown_point_with_dict(self):
+        pd = _pdict()
+        s = F.FeatureSpec(name="x", source="UNKNOWN.PT")
+        errs = s.validate(pd)
+        self.assertTrue(any("不在点位字典" in e for e in errs))
+
+    def test_constant_source_ok(self):
+        pd = _pdict()
+        s = F.FeatureSpec(name="x", source="1.5")
+        self.assertEqual(s.validate(pd), [])
+
+
+class TestTransforms(unittest.TestCase):
+    def setUp(self):
+        self.pd = _pdict()
+        # 4 个样本，TEMP 单调上升
+        self.samples = _samples([
+            {"CLF-01.TEMP": 100.0},
+            {"CLF-01.TEMP": 110.0},
+            {"CLF-01.TEMP": 120.0},
+            {"CLF-01.TEMP": 130.0},
+        ], step=10.0)
+
+    def test_raw(self):
+        ext = F.FeatureExtractor(
+            [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], self.pd)
+        m = ext.extract(self.samples)
+        self.assertEqual(m.column("t")[-1], 130.0)
+
+    def test_mean(self):
+        ext = F.FeatureExtractor(
+            [F.FeatureSpec("t", "CLF-01.TEMP", "mean", window=1000)],
+            self.pd)
+        m = ext.extract(self.samples)
+        self.assertAlmostEqual(m.column("t")[-1], 115.0)
+
+    def test_min_max_range(self):
+        ext = F.FeatureExtractor([
+            F.FeatureSpec("mn", "CLF-01.TEMP", "min", window=1000),
+            F.FeatureSpec("mx", "CLF-01.TEMP", "max", window=1000),
+            F.FeatureSpec("rg", "CLF-01.TEMP", "range", window=1000),
+        ], self.pd)
+        m = ext.extract(self.samples)
+        last = m.rows[-1]
+        self.assertEqual(last[0], 100.0)   # min
+        self.assertEqual(last[1], 130.0)   # max
+        self.assertEqual(last[2], 30.0)    # range
+
+    def test_std(self):
+        ext = F.FeatureExtractor(
+            [F.FeatureSpec("s", "CLF-01.TEMP", "std", window=1000)],
+            self.pd)
+        m = ext.extract(self.samples)
+        # 无偏样本标准差：100,110,120,130 → 12.9099...
+        self.assertAlmostEqual(m.column("s")[-1],
+                               math.sqrt(500.0 / 3), places=4)
+
+    def test_diff(self):
+        ext = F.FeatureExtractor(
+            [F.FeatureSpec("d", "CLF-01.TEMP", "diff")], self.pd)
+        m = ext.extract(self.samples)
+        self.assertEqual(m.column("d")[-1], 10.0)
+
+    def test_slope(self):
+        ext = F.FeatureExtractor(
+            [F.FeatureSpec("sl", "CLF-01.TEMP", "slope", window=1000)],
+            self.pd)
+        m = ext.extract(self.samples)
+        # 每 10s +10 → 斜率 1.0
+        self.assertAlmostEqual(m.column("sl")[-1], 1.0, places=6)
+
+    def test_ratio(self):
+        ext = F.FeatureExtractor([
+            F.FeatureSpec("r", "CLF-01.CL2", "ratio",
+                          denominator="CLF-01.FEED"),
+        ], self.pd)
+        samples = _samples([
+            {"CLF-01.CL2": 30.0, "CLF-01.FEED": 10.0},
+            {"CLF-01.CL2": 60.0, "CLF-01.FEED": 20.0},
+        ], step=10.0)
+        m = ext.extract(samples)
+        self.assertAlmostEqual(m.column("r")[-1], 3.0)
+
+
+class TestMissingAndWindow(unittest.TestCase):
+    def test_missing_point_is_nan(self):
+        ext = F.FeatureExtractor(
+            [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], _pdict())
+        # 样本里没有 TEMP → NaN
+        samples = _samples([{"CLF-01.PRES": 1.0}])
+        m = ext.extract(samples)
+        self.assertTrue(math.isnan(m.column("t")[0]))
+
+    def test_window_excludes_old(self):
+        ext = F.FeatureExtractor(
+            [F.FeatureSpec("t", "CLF-01.TEMP", "mean", window=15)],
+            _pdict())
+        # window=15s 只含最近 ≤2 个样本（step=10）
+        samples = _samples([{"CLF-01.TEMP": 0.0},
+                            {"CLF-01.TEMP": 100.0},
+                            {"CLF-01.TEMP": 200.0}], step=10.0)
+        m = ext.extract(samples)
+        # 最后时刻 window=15 → 含 ts=20(100) 与 ts=30(200) → 均值 150
+        self.assertAlmostEqual(m.column("t")[-1], 150.0)
+
+    def test_drop_nan_rows(self):
+        ext = F.FeatureExtractor(
+            [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], _pdict())
+        samples = _samples([
+            {"CLF-01.PRES": 1.0},     # TEMP 缺失 → NaN
+            {"CLF-01.TEMP": 50.0},
+        ])
+        m = ext.extract(samples).drop_nan_rows()
+        self.assertEqual(len(m.rows), 1)
+
+
+class TestLoading(unittest.TestCase):
+    def test_load_template_yaml(self):
+        pd = _pdict()
+        with open(FEATURES_TPL, "r", encoding="utf-8") as fh:
+            text = fh.read()
+        ext = F.load_feature_specs(text, pd)
+        self.assertGreater(len(ext.names), 0)
+        # 抽取一次能跑通（合成样本）
+        samples = _samples([{"CLF-01.TEMP": 850.0, "CLF-01.CL2": 120.0,
+                             "CLF-01.FEED": 4.0, "CLF-01.CO": 2.0,
+                             "CLF-01.BED": 60.0, "RF-01.PURITY": 99.0,
+                             "RF-01.IMP": 0.3}])
+        m = ext.extract(samples)
+        self.assertEqual(len(m.names), len(ext.names))
+        self.assertEqual(len(m.rows), 1)
+
+    def test_load_json(self):
+        import json
+        text = json.dumps({"features": [
+            {"name": "t", "source": "CLF-01.TEMP", "transform": "raw"}]})
+        ext = F.load_feature_specs(text, _pdict())
+        self.assertEqual(ext.names, ["t"])
+
+    def test_duplicate_names_raise(self):
+        with self.assertRaises(F.FeatureError):
+            F.FeatureExtractor([
+                F.FeatureSpec("dup", "CLF-01.TEMP"),
+                F.FeatureSpec("dup", "CLF-01.PRES"),
+            ], _pdict())
+
+    def test_empty_features_raise(self):
+        with self.assertRaises(F.FeatureError):
+            F.load_feature_specs("features: []", _pdict())
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)