diff --git a/templates/ti-cl4/recipe-optim/README.md b/templates/ti-cl4/recipe-optim/README.md index d065ae5..8cd5175 100644 --- a/templates/ti-cl4/recipe-optim/README.md +++ b/templates/ti-cl4/recipe-optim/README.md @@ -10,9 +10,15 @@ - `problem.py` — 优化问题建模（**#78**）：决策变量 / 目标 / 约束的声明式规格 + 校验 + 可行性判定 + 零依赖 YAML 子集加载。 +- `solver.py` — 求解器集成（**#79**）：`SolverConfig` + `Solution` + 网格枚举/ + 坐标下降轻量求解器 + `solve()` 统一入口，求解器无关契约。 +- `cross_process.py` — 跨工序关联寻优（**#80**）：纯标准库岭回归 + + `CrossProcessModel`（上游指标→下游质量，fit/predict/evaluate R²/可解释权重/序列化）。 +- `advisor.py` — 优化建议生成与可解释性（**#81**）：整合 #78/#79/#80 输出 + 可溯源建议报告（变量级/跨工序佐证/风险提示/溯源链路）。 - `config/recipe_optim.template.yaml` — Template-Ti 配方优化模板资产。 -- `tests/` — 单元测试（`python -m unittest discover -s tests`）。 -- `_sanity_check.py` — 部署期一键自检（5 能力点）。 +- `tests/` — 单元测试（`python -m unittest discover -s tests`，76 用例）。 +- `_sanity_check.py` — 部署期一键自检（8 能力点）。 ## 设计 diff --git a/templates/ti-cl4/recipe-optim/_sanity_check.py b/templates/ti-cl4/recipe-optim/_sanity_check.py index eac776c..3451699 100644 --- a/templates/ti-cl4/recipe-optim/_sanity_check.py +++ b/templates/ti-cl4/recipe-optim/_sanity_check.py @@ -8,6 +8,9 @@ import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from problem import ConstraintKind, OptimizationProblem, load_problem # noqa: E402 +from solver import SolverConfig, solve # noqa: E402 +from cross_process import CrossProcessModel, CrossProcessModelConfig, CrossProcessSample # noqa: E402 +from advisor import generate_advice # noqa: E402 CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config", "recipe_optim.template.yaml") @@ -48,12 +51,45 @@ def main() -> int: if [v.name for v in rt.variables] != [v.name for v in p.variables]: failures.append("序列化往返丢失变量") + # 6) 求解器（#79）端到端：加载模板后能求出可行解 + sol = solve(p, SolverConfig(grid_steps=7, max_combinations=200000)) + if not sol.feasible: + failures.append(f"求解器未求出可行解: {sol.message}") + if sol.strategy != "grid": + failures.append(f"求解策略非 grid: {sol.strategy}") + + # 7) 跨工序关联模型（#80）端到端：合成线性数据训练 + R² 评估 + cfg = CrossProcessModelConfig( + upstream_features=["up"], downstream_targets=["down"], + alpha=0.0, min_samples=8) + samples = [CrossProcessSample(upstream={"up": float(i)}, + downstream={"down": 2.0 * float(i) + 1.0}) + for i in range(12)] + cm = CrossProcessModel(cfg).fit(samples) + report = cm.evaluate(samples) + if not cm.fitted: + failures.append("跨工序模型未训练成功") + if not (report.get("r2_down", 0.0) > 0.99): + failures.append(f"跨工序模型 R² 过低: {report}") + + # 8) 优化建议生成（#81）端到端：可解释、可溯源建议 + advice = generate_advice(p, sol, cross_process_weights={ + "Ti_purity": {"clf_temp": 0.8, "cl2_ratio": 1.2}}) + if not advice.feasible: + failures.append("建议生成器标记不可行") + if len(advice.items) != len(p.variables): + failures.append("建议条目数与变量数不一致") + if not all(it.evidence for it in advice.items): + failures.append("存在无依据的建议条目（违反可溯源要求）") + if not advice.trace: + failures.append("溯源链路为空") + if failures: print("❌ recipe-optim 自检失败:") for f in failures: print(" -", f) return 1 - print("✅ recipe-optim 自检通过（5 能力点）") + print("✅ recipe-optim 自检通过（8 能力点）") return 0 diff --git a/templates/ti-cl4/recipe-optim/advisor.py b/templates/ti-cl4/recipe-optim/advisor.py new file mode 100644 index 0000000..942214c --- /dev/null +++ b/templates/ti-cl4/recipe-optim/advisor.py @@ -0,0 +1,253 @@ +# -*- coding: utf-8 -*- +"""Ti-2 配方优化 · 优化建议生成与可解释性（Issue #81 / PRD 5.3 ② + 5.4）。 + +整合 #78（问题建模）/ #79（求解器）/ #80（跨工序关联），把"求解结果"翻译成 +**工艺工程师可读、可溯源**的优化建议（PRD：李工"要求结果可解释、可溯源，要引用 +依据"）。 + +PRD 设计口径 +------------ +- 场景B（优化）：「下一批次质量目标下达 → 工艺优化模型给出参数建议 → 李工 review + → 下发 DCS → 实际质量反馈回流训练」（PRD §2.2）。 +- 架构表：``出：参数/配方建议``；用户画像："要求结果可解释、可溯源（要引用依据）"。 +- 风险表：二期。故本期交付**确定性、可测试**的建议生成器，把上游链路结构化输出 + 汇编成建议条目；数据/LLM 就绪后可再叠加自然语言润色（注入 llm-gateway）。 + +本模块交付 +---------- +1. **``AdviceItem``**：单条建议（变量、当前值、建议值、变化方向/幅度、依据来源 + `source`、工艺含义 `meaning`、可溯源引用 `evidence`）。 +2. **``AdviceReport``**：建议报告（条目列表 + 摘要 + 是否达标 + 风险提示 + 溯源 + 链路），可序列化。 +3. **``AdviceConfig``**：建议生成配置（变化阈值、是否提示风险、溯源前缀）。 +4. **``generate_advice``**：核心生成函数——输入 #79 的 ``Solution`` + #78 的 + ``OptimizationProblem`` +（可选）#80 的 ``CrossProcessModel`` 特征权重 + 当前 + 配方，产出 ``AdviceReport``，每条建议带： + - **变量级**：建议调整 X 从 a→b（变化幅度/方向），引用变量 meaning； + - **依据级**：若被求解过程约束收紧/禁止组合影响，引用约束 reason； + - **跨工序级**（可选）：引用 #80 上游→下游影响权重作为佐证。 + +设计要点 +-------- +- **零第三方依赖**（纯标准库）；可注入 LLM 做润色但非必需（保证可用性）。 +- **可溯源**：每条建议标注 ``source``（problem/solver/cross_process）与 ``evidence`` + （具体约束/权重值），对齐 PRD"引用依据"。 +- **风险前置**：违反约束或未达标时在报告 ``warnings`` 列出，需人工 review（PRD + 场景B 的"李工 review"环节）。 +""" +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +# 复用上游契约 +try: # pragma: no cover + from recipe_optim.problem import ( # type: ignore[import-not-found] + ConstraintSpec, + DecisionVariable, + OptimizationProblem, + Sense, + _is_num, + ) + from recipe_optim.solver import Solution # type: ignore[import-not-found] +except ImportError: # pragma: no cover + from problem import ( # type: ignore[import-not-found] + ConstraintSpec, + DecisionVariable, + OptimizationProblem, + Sense, + _is_num, + ) + from solver import Solution # type: ignore[import-not-found] + + +class AdvisorError(ValueError): + """建议生成错误。""" + + +@dataclass +class AdviceItem: + """单条优化建议（可解释、可溯源）。""" + + variable: str + current_value: Any + suggested_value: Any + direction: str # "↑" / "↓" / "→"（不变） + delta: float = 0.0 # 建议值 - 当前值（数值变量） + meaning: str = "" # 工艺含义（来自 DecisionVariable.meaning） + unit: str = "" + source: str = "solver" # solver / cross_process / problem + evidence: str = "" # 可溯源依据（约束 reason / 权重值） + reason_text: str = "" # 人话依据 + + def to_dict(self) -> Dict[str, Any]: + return { + "variable": self.variable, + "current_value": self.current_value, + "suggested_value": self.suggested_value, + "direction": self.direction, + "delta": self.delta, + "meaning": self.meaning, + "unit": self.unit, + "source": self.source, + "evidence": self.evidence, + "reason_text": self.reason_text, + } + + +@dataclass +class AdviceReport: + """优化建议报告（多条建议 + 摘要 + 风险提示）。""" + + items: List[AdviceItem] = field(default_factory=list) + summary: str = "" + target_met: bool = False + objective_value: float = 0.0 + feasible: bool = False + warnings: List[str] = field(default_factory=list) + trace: List[str] = field(default_factory=list) # 溯源链路（PRD"引用依据"） + + def to_dict(self) -> Dict[str, Any]: + return { + "items": [i.to_dict() for i in self.items], + "summary": self.summary, + "target_met": self.target_met, + "objective_value": self.objective_value, + "feasible": self.feasible, + "warnings": list(self.warnings), + "trace": list(self.trace), + } + + +@dataclass +class AdviceConfig: + """建议生成配置。""" + + change_threshold: float = 1e-6 # 变化幅度低于此值视为"不变" + show_warnings: bool = True + cross_process_prefix: str = "跨工序关联" + + +def _direction_and_delta(cur: Any, sug: Any) -> tuple: + """计算变化方向与幅度（数值变量）。""" + if _is_num(cur) and _is_num(sug): + delta = float(sug) - float(cur) + if delta > 1e-12: + return "↑", delta + if delta < -1e-12: + return "↓", delta + return "→", 0.0 + return "→" if cur == sug else "≠", 0.0 + + +def generate_advice( + problem: OptimizationProblem, + solution: Solution, + current: Optional[Dict[str, Any]] = None, + cross_process_weights: Optional[Dict[str, Dict[str, float]]] = None, + config: Optional[AdviceConfig] = None, +) -> AdviceReport: + """根据求解结果生成可解释、可溯源的优化建议。 + + 参数 + ---- + problem : #78 的优化问题（取变量 meaning/unit + 约束 reason 作依据）。 + solution : #79 的求解结果（取建议取值 + 可行性 + 违反约束）。 + current : 当前配方/工况取值（缺省取各变量 ``initial``）；用于计算"从 a→b"。 + cross_process_weights : #80 的 ``feature_weights``（目标→{特征:权重}）， + 作为跨工序佐证（可选）。 + config : 建议生成配置。 + """ + cfg = config or AdviceConfig() + cur = dict(current or {}) + report = AdviceReport( + objective_value=solution.objective_value, + feasible=solution.feasible, + target_met=solution.target_met, + ) + report.trace.append("建议生成依据链：#78 问题建模 → #79 求解 → #80 跨工序关联（可选）") + + if not solution.feasible: + report.warnings.append( + "求解器未找到可行解，下列建议仅供参考，需人工复核（PRD 场景B「李工 review」）") + report.summary = solution.message or "无可行解" + # 仍输出违反约束作为风险依据 + for c in solution.violated: + if c.reason: + report.warnings.append(f"违反约束：{c.reason}") + return report + + vmap = problem.variable_map + # 1) 变量级建议 + for var in problem.variables: + sug = solution.assignment.get(var.name) + base = cur.get(var.name, var.initial) + if sug is None: + continue + direction, delta = _direction_and_delta(base, sug) + if abs(delta) < cfg.change_threshold and direction == "→": + # 无变化也输出一条"保持"，便于完整呈现配方 + item = AdviceItem( + variable=var.name, current_value=base, suggested_value=sug, + direction="→", delta=0.0, meaning=var.meaning, unit=var.unit, + source="solver", evidence="求解器最优解保持当前值", + reason_text=f"保持 {var.name}（{var.meaning}）不变：最优解与当前一致") + else: + item = AdviceItem( + variable=var.name, current_value=base, suggested_value=sug, + direction=direction, delta=delta, meaning=var.meaning, unit=var.unit, + source="solver", evidence=f"目标 {problem.objective.sense.value} 下最优", + reason_text=_var_reason(var, direction, delta, problem.objective.sense)) + report.items.append(item) + + # 2) 跨工序佐证（可选）：把 #80 权重作为依据附加到相关变量 + if cross_process_weights: + for target, weights in cross_process_weights.items(): + for var in problem.variables: + w = weights.get(var.name) + if _is_num(w) and abs(w) > 1e-9: + # 找到该变量的已有建议，追加跨工序证据 + for item in report.items: + if item.variable == var.name: + sign = "正向" if w > 0 else "负向" + extra = (f"{cfg.cross_process_prefix}：{var.name} 对下游 " + f"{target} 影响 {sign}（权重 {w:.4g}）") + item.evidence = (item.evidence + "；" + extra) if item.evidence else extra + item.reason_text = item.reason_text + "。" + extra + report.trace.append(extra) + break + + # 3) 风险与达标提示 + if cfg.show_warnings: + for var in problem.variables: + sug = solution.assignment.get(var.name) + if sug is not None and not var.contains(sug): + report.warnings.append( + f"{var.name}（{var.meaning}）建议值 {sug} 越出合法域，需人工复核") + for c in problem.constraints: + if c.reason and not c.satisfied_by(solution.assignment): + report.warnings.append(f"约束风险：{c.reason}") + + # 4) 摘要 + n_change = sum(1 for it in report.items if it.direction in ("↑", "↓", "≠")) + if problem.objective.target_value is not None: + report.summary = ( + f"目标 {problem.objective.target} {'已达成' if solution.target_met else '未达成'}" + f"（目标值 {problem.objective.target_value}，预测 {solution.objective_value:.4g}）；" + f"共 {len(report.items)} 项参数，其中 {n_change} 项建议调整") + else: + report.summary = ( + f"预测目标值 {solution.objective_value:.4g}（{problem.objective.sense.value}）；" + f"共 {len(report.items)} 项参数，其中 {n_change} 项建议调整") + return report + + +def _var_reason(var: DecisionVariable, direction: str, delta: float, + sense: Sense) -> str: + """构造变量级人话依据。""" + arrow = {"↑": "提高", "↓": "降低", "≠": "调整为"}[direction] if direction in ("↑", "↓", "≠") else "调整" + verb = "有利于" if (sense == Sense.MAXIMIZE) == (delta > 0) else "换取" + target_word = "最大化" if sense == Sense.MAXIMIZE else "最小化" + return (f"{arrow} {var.name}（{var.meaning}）{abs(delta):.4g}{var.unit}：" + f"{verb}{target_word}目标") diff --git a/templates/ti-cl4/recipe-optim/cross_process.py b/templates/ti-cl4/recipe-optim/cross_process.py new file mode 100644 index 0000000..8066943 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/cross_process.py @@ -0,0 +1,397 @@ +# -*- coding: utf-8 -*- +"""Ti-2 跨工序关联寻优 · 模型训练（Issue #80 / PRD §5.3 ④）。 + +承接 #78/#79 的配方优化建模与求解器，本模块把"上游工序指标 → 下游工序质量" +的**跨工序关联**建成**可训练、可评估、可序列化**的轻量模型，为「数据就绪后」 +的跨工序寻优（PRD 架构表：``入：上游(TiCl₄)指标；出：下游(海绵钛)寻优建议``） +提供量化基础。 + +PRD 设计口径 +------------ +- 架构表（PRD §5.3 ④）：``跨工序关联寻优 | 关联建模 | 入：上游(TiCl₄)指标； + 出：下游(海绵钛)寻优建议 | 高（需闭环反馈）``。 +- 模板化技术路径：默认「固定主干 + 可配置超参」；**新增结构走插件注册而非改 + 内核**。故本模块主干为**线性 / 岭回归（纯标准库）**，跨工序的强非线性关联 + （LSTM/GNN）走 recipe 插件，不在本期内核。 +- 里程碑表：二期交付（标注数据 ≥ 6 个月，LIMS 对接后补标）。故本期交付**可跑通、 + 可测试**的关联模型与训练/评估闭环，数据就绪后即可上线。 + +本模块交付 +---------- +1. **``CrossProcessSample``**：跨工序样本（上游特征 Dict + 下游目标 + 批次/时间）， + 鸭子类型，便于独立测试。 +2. **``RidgeRegression``**：纯标准库岭回归（含截距、L2 正则、闭式解）， + ``fit`` / ``predict`` / 评估（MSE、R²）。 +3. **``CrossProcessModel``**：跨工序关联模型——把上游指标映射到下游质量，聚合 + 多个下游目标的回归器；``fit`` / ``predict`` / ``evaluate``（R² 报告）/ 序列化 + （零依赖 JSON，便于版本化保存与 #41 模型模板注册机制对接）。 +4. **``CrossProcessModelConfig``**：声明式配置（特征清单、目标清单、正则强度、 + 训练最小样本数），对齐 PRD「超参包驱动」。 + +设计要点 +-------- +- **零第三方依赖**（纯标准库）：与内核既有模块一致，便于离线/隔离网部署。 +- **训练/评估分离**：``evaluate`` 在测试段产出每个目标的 R²（拟合优度），对齐 + PRD「关键质量指标预测准确率 ≥ 90%」的评估口径。 +- **数据门槛前置校验**：``min_samples`` 不足时拒绝训练（对齐 PRD「监督模型需 + ≥ 6 个月标注」的数据门槛约束，避免低质上线）。 +- **与 #78/#79 解耦**：模型只依赖样本的 Dict 特征，输出下游质量预测；上层 + （#81）可把预测喂回 #78 的目标函数做跨工序寻优。 +""" +from __future__ import annotations + +import json +import math +import os +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Sequence, Tuple + +NAN = float("nan") + + +class CrossProcessError(ValueError): + """跨工序关联模型错误（特征缺失/样本不足/未训练等）。""" + + +def _is_num(x: object) -> bool: + return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x)) + + +# --------------------------------------------------------------------------- +# 样本 +# --------------------------------------------------------------------------- + + +@dataclass +class CrossProcessSample: + """跨工序样本：上游特征 + 下游目标（鸭子类型，便于独立测试）。 + + ``upstream`` 的 key 为上游测点/特征名（如 ``TiCl4_purity``、``TiCl4_impurity``）， + ``downstream`` 的 key 为下游质量指标（如 ``sponge_titanium_grade``）。 + """ + + upstream: Dict[str, float] = field(default_factory=dict) + downstream: Dict[str, float] = field(default_factory=dict) + batch: str = "" # 批次号（可溯源，供 #81 引用） + timestamp: float = 0.0 + + +# --------------------------------------------------------------------------- +# 岭回归（纯标准库闭式解） +# --------------------------------------------------------------------------- + + +class RidgeRegression: + """单目标岭回归（含截距 + L2 正则），闭式解（零第三方依赖）。 + + 解析解：``w = (XᵀX + λI)⁻¹ Xᵀ y``（``X`` 已含截距列）；用高斯消元解线性 + 方程组避免 numpy 依赖。``λ=0`` 即普通最小二乘。 + """ + + def __init__(self, alpha: float = 1.0) -> None: + if alpha < 0: + raise CrossProcessError("岭回归正则强度 alpha 必须 ≥ 0") + self.alpha = alpha + self._feature_names: List[str] = [] + self._weights: List[float] = [] # 末位为截距 + self._fitted = False + + @property + def fitted(self) -> bool: + return self._fitted + + @property + def feature_names(self) -> List[str]: + return list(self._feature_names) + + def fit(self, X: Sequence[Sequence[float]], y: Sequence[float], + feature_names: Sequence[str]) -> "RidgeRegression": + """在训练段拟合（X 不含截距列，内部补）。""" + n = len(X) + if n == 0: + raise CrossProcessError("RidgeRegression.fit 至少需要 1 条样本") + if n != len(y): + raise CrossProcessError("X 与 y 样本数不一致") + p = len(feature_names) + if any(len(row) != p for row in X): + raise CrossProcessError("X 列数与 feature_names 不一致") + # 设计矩阵加截距列（末列恒为 1） + Xa = [list(row) + [1.0] for row in X] + # XtX + λI（不对截距正则：最后一行/列不加 λ） + dim = p + 1 + A = [[0.0] * dim for _ in range(dim)] + for row in Xa: + for i in range(dim): + for j in range(dim): + A[i][j] += row[i] * row[j] + for i in range(p): # 不对截距正则 + A[i][i] += self.alpha + # Xty + b = [0.0] * dim + for k, row in enumerate(Xa): + for i in range(dim): + b[i] += row[i] * y[k] + # 解 A w = b（高斯消元 + 回代，带部分主元） + self._weights = _solve_linear(A, b) + self._feature_names = list(feature_names) + self._fitted = True + return self + + def predict_one(self, x: Sequence[float]) -> float: + if not self._fitted: + raise CrossProcessError("RidgeRegression 未 fit") + if len(x) != len(self._feature_names): + raise CrossProcessError("预测特征数与训练不一致") + return sum(w * v for w, v in zip(self._weights[:-1], x)) + self._weights[-1] + + def weights_dict(self) -> Dict[str, float]: + """特征权重 + 截距（可解释，供 #81 引用）。""" + if not self._fitted: + return {} + d = {name: self._weights[i] for i, name in enumerate(self._feature_names)} + d["__intercept__"] = self._weights[-1] + return d + + def to_dict(self) -> Dict[str, Any]: + return { + "alpha": self.alpha, + "feature_names": list(self._feature_names), + "weights": list(self._weights), + } + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "RidgeRegression": + m = cls(alpha=float(d.get("alpha", 1.0))) + m._feature_names = list(d.get("feature_names", [])) + m._weights = [float(w) for w in d.get("weights", [])] + m._fitted = bool(m._weights) + return m + + +# --------------------------------------------------------------------------- +# 线性方程组求解（高斯消元，部分主元） +# --------------------------------------------------------------------------- + + +def _solve_linear(A: List[List[float]], b: List[float]) -> List[float]: + """解 A w = b（方阵），高斯消元 + 回代，带部分主元选元。""" + n = len(A) + # 增广矩阵 + M = [list(A[i]) + [b[i]] for i in range(n)] + for col in range(n): + # 部分主元 + pivot = max(range(col, n), key=lambda r: abs(M[r][col])) + if abs(M[pivot][col]) < 1e-12: + raise CrossProcessError("正规方程奇异（特征共线性或样本不足）") + M[col], M[pivot] = M[pivot], M[col] + # 消元 + piv = M[col][col] + for r in range(col + 1, n): + factor = M[r][col] / piv + if factor != 0.0: + for c in range(col, n + 1): + M[r][c] -= factor * M[col][c] + # 回代 + w = [0.0] * n + for i in range(n - 1, -1, -1): + s = M[i][n] - sum(M[i][j] * w[j] for j in range(i + 1, n)) + w[i] = s / M[i][i] + return w + + +# --------------------------------------------------------------------------- +# 跨工序关联模型 +# --------------------------------------------------------------------------- + + +@dataclass +class CrossProcessModelConfig: + """跨工序关联模型声明式配置（对齐 PRD 超参包驱动）。""" + + upstream_features: List[str] = field(default_factory=list) + downstream_targets: List[str] = field(default_factory=list) + alpha: float = 1.0 # 岭回归正则强度 + min_samples: int = 10 # 训练最小样本数（数据门槛前置校验） + + def __post_init__(self) -> None: + if not self.upstream_features: + raise CrossProcessError("upstream_features 不能为空") + if not self.downstream_targets: + raise CrossProcessError("downstream_targets 不能为空") + if self.alpha < 0: + raise CrossProcessError("alpha 必须 ≥ 0") + if self.min_samples < 2: + raise CrossProcessError("min_samples 必须 ≥ 2") + + +class CrossProcessModel: + """跨工序关联模型：上游指标 → 下游质量（多目标，每目标一个岭回归）。 + + 训练阶段对每个下游目标拟合一个岭回归；推理阶段给定上游指标预测全部下游目标； + 评估阶段在测试段产出每个目标的 R²（拟合优度）。 + """ + + def __init__(self, config: CrossProcessModelConfig) -> None: + self.config = config + self._regressors: Dict[str, RidgeRegression] = {} + self._fitted = False + # 训练统计（可解释，供 #81 引用） + self.train_n: int = 0 + self.train_means: Dict[str, float] = {} + + @property + def fitted(self) -> bool: + return self._fitted + + # ---- 训练 -------------------------------------------------------- + + def fit(self, samples: Sequence[CrossProcessSample]) -> "CrossProcessModel": + """在训练段对每个下游目标拟合岭回归。""" + cfg = self.config + if len(samples) < cfg.min_samples: + raise CrossProcessError( + f"训练样本不足：{len(samples)} < min_samples={cfg.min_samples}" + "（对齐 PRD 监督模型数据门槛）") + # 构造 X / 每目标 y + X: List[List[float]] = [] + per_target_y: Dict[str, List[float]] = {t: [] for t in cfg.downstream_targets} + for s in samples: + row = [] + ok = True + for f in cfg.upstream_features: + v = s.upstream.get(f) + if not _is_num(v): + ok = False + break + row.append(float(v)) + if not ok: + continue + # 每个目标都要有值，否则跳过该样本（保持对齐） + target_vals = {} + for t in cfg.downstream_targets: + tv = s.downstream.get(t) + if not _is_num(tv): + ok = False + break + target_vals[t] = float(tv) + if not ok: + continue + X.append(row) + for t in cfg.downstream_targets: + per_target_y[t].append(target_vals[t]) + if len(X) < cfg.min_samples: + raise CrossProcessError( + f"有效训练样本不足：{len(X)} < {cfg.min_samples}（含缺失值过滤后）") + self._regressors = {} + for t in cfg.downstream_targets: + reg = RidgeRegression(alpha=cfg.alpha) + reg.fit(X, per_target_y[t], cfg.upstream_features) + self._regressors[t] = reg + self.train_n = len(X) + # 训练段上游特征均值（漂移检测/可解释输入） + self.train_means = { + f: sum(row[i] for row in X) / len(X) + for i, f in enumerate(cfg.upstream_features) + } + self._fitted = True + return self + + # ---- 推理 -------------------------------------------------------- + + def predict(self, upstream: Dict[str, float]) -> Dict[str, float]: + """给定上游指标，预测全部下游目标。""" + if not self._fitted: + raise CrossProcessError("CrossProcessModel 未 fit") + row = [] + for f in self.config.upstream_features: + v = upstream.get(f) + if not _is_num(v): + raise CrossProcessError(f"预测缺少上游特征 {f!r}") + row.append(float(v)) + return {t: reg.predict_one(row) for t, reg in self._regressors.items()} + + # ---- 评估 -------------------------------------------------------- + + def evaluate(self, samples: Sequence[CrossProcessSample]) -> Dict[str, float]: + """在测试段产出每个目标的 R²（拟合优度）+ 总体 MSE。""" + if not self._fitted: + raise CrossProcessError("CrossProcessModel 未 fit，无法 evaluate") + report: Dict[str, float] = {} + # 收集每个目标的 真实/预测 + per_target: Dict[str, Tuple[List[float], List[float]]] = { + t: ([], []) for t in self.config.downstream_targets + } + for s in samples: + try: + pred = self.predict(s.upstream) + except CrossProcessError: + continue + for t in self.config.downstream_targets: + truth = s.downstream.get(t) + if not _is_num(truth): + continue + per_target[t][0].append(float(truth)) + per_target[t][1].append(pred[t]) + all_sq_err = [] + for t, (truths, preds) in per_target.items(): + if not truths: + report[f"r2_{t}"] = NAN + continue + mean_t = sum(truths) / len(truths) + ss_res = sum((tr - pr) ** 2 for tr, pr in zip(truths, preds)) + ss_tot = sum((tr - mean_t) ** 2 for tr in truths) + r2 = 1.0 - ss_res / ss_tot if ss_tot > 1e-12 else (1.0 if ss_res < 1e-12 else NAN) + report[f"r2_{t}"] = r2 + all_sq_err.extend((tr - pr) ** 2 for tr, pr in zip(truths, preds)) + report["mse_overall"] = (sum(all_sq_err) / len(all_sq_err)) if all_sq_err else NAN + report["n_eval"] = float(sum(len(v[0]) for v in per_target.values())) + return report + + # ---- 可解释（供 #81） ------------------------------------------- + + def feature_weights(self, target: str) -> Dict[str, float]: + """某下游目标的上游特征权重 + 截距（可解释：上游对下游的影响）。""" + reg = self._regressors.get(target) + if reg is None: + raise CrossProcessError(f"未知下游目标 {target!r}") + return reg.weights_dict() + + # ---- 序列化 ------------------------------------------------------ + + def to_dict(self) -> Dict[str, Any]: + return { + "config": { + "upstream_features": list(self.config.upstream_features), + "downstream_targets": list(self.config.downstream_targets), + "alpha": self.config.alpha, + "min_samples": self.config.min_samples, + }, + "train_n": self.train_n, + "train_means": dict(self.train_means), + "regressors": {t: r.to_dict() for t, r in self._regressors.items()}, + } + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "CrossProcessModel": + cfg = CrossProcessModelConfig( + upstream_features=list(d["config"]["upstream_features"]), + downstream_targets=list(d["config"]["downstream_targets"]), + alpha=float(d["config"].get("alpha", 1.0)), + min_samples=int(d["config"].get("min_samples", 10)), + ) + m = cls(cfg) + m.train_n = int(d.get("train_n", 0)) + m.train_means = dict(d.get("train_means", {})) + m._regressors = {t: RidgeRegression.from_dict(rd) + for t, rd in d.get("regressors", {}).items()} + m._fitted = bool(m._regressors) + return m + + def save(self, path: str) -> None: + with open(path, "w", encoding="utf-8") as fh: + json.dump(self.to_dict(), fh, ensure_ascii=False, indent=2) + + @classmethod + def load(cls, path: str) -> "CrossProcessModel": + with open(path, "r", encoding="utf-8") as fh: + return cls.from_dict(json.load(fh)) diff --git a/templates/ti-cl4/recipe-optim/solver.py b/templates/ti-cl4/recipe-optim/solver.py new file mode 100644 index 0000000..09bbc82 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/solver.py @@ -0,0 +1,304 @@ +# -*- coding: utf-8 -*- +"""Ti-2 配方动态优化 · 求解器集成（Issue #79 / PRD 5.3 ②）。 + +承接 #78 的 ``OptimizationProblem``：把"问题模型"喂给**求解器**，产出满足全部 +约束、逼近目标最优的**配方/参数取值**，并给出可解释的求解报告。 + +PRD 设计口径 +------------ +- 架构表（PRD §5.3）：``出：参数/配方建议``；``高（需闭环反馈）``。 +- 模板化技术路径：默认「固定主干 + 可配置超参」；新增结构走插件注册而非改内核。 +- 风险表：二期交付（数据门槛高）。故本期求解器采用**纯标准库、零第三方依赖**的 + 轻量策略（坐标下降 + 网格采样），数据就绪/精度不足时可注入更强的外部求解器 + （PuLP/scipy/optuna，走 #78 预留的 ``solve`` 扩展点），**内核不绑优化库**。 + +本模块交付 +---------- +1. **``SolverConfig``**：求解策略声明式配置（网格粒度、迭代轮数、随机种子、 + 是否枚举离散选择），对齐 PRD「超参包驱动」。 +2. **``Solution``**：求解结果（取值 ``assignment``、目标值、是否可行、是否达成 + ``target_value``、迭代轨迹、违反约束枚举），为 #81 可解释建议提供结构化输入。 +3. **``GridSolver``**：确定性网格 + 坐标下降求解器（纯标准库）： + - 连续域变量按 ``grid_steps`` 等分离散化； + - 离散域变量枚举 ``choices``； + - 笛卡尔积里筛可行解、按目标 ``sense`` 选最优（全局最优保证）； + - 规模过大时退化为坐标下降（贪心）保可用性（``max_combinations`` 阈值）。 +4. **``solve(problem, config=None)``**：统一入口，便于 #80/#81 调用。 + +设计要点 +-------- +- **确定性可复现**：``random_seed`` 固定，同输入同输出（对齐 PRD"结论可复现"）。 +- **可行优先**：无任何可行解时返回 ``feasible=False`` 的 Solution，不抛异常， + 便于上层降级（对齐 PRD"可用性 ≥ 99.8%"）。 +- **求解器无关契约**：``solve`` 是薄入口，可被外部更强求解器替换；本模块的 + ``Solution`` 结构即外部求解器需返回的契约。 +""" +from __future__ import annotations + +import itertools +import math +import random +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + +# 复用 #78 的问题模型。作为包成员导入用 ``recipe_optim.problem``；当本文件被直接 +# 执行（与 problem.py 同目录）时回落到裸名 ``problem``。 +try: # pragma: no cover - 分支取决于导入方式 + from recipe_optim.problem import ( # type: ignore[import-not-found] + ConstraintSpec, + DecisionVariable, + DomainKind, + ObjectiveSpec, + OptimizationProblem, + Sense, + _is_num, + ) +except ImportError: # pragma: no cover + from problem import ( # type: ignore[import-not-found,no-redef] + ConstraintSpec, + DecisionVariable, + DomainKind, + ObjectiveSpec, + OptimizationProblem, + Sense, + _is_num, + ) + + +class SolverError(ValueError): + """求解器配置或执行错误（网格粒度非法、变量规模溢出等）。""" + + +@dataclass +class SolverConfig: + """求解策略声明式配置（对齐 PRD 超参包驱动）。""" + + grid_steps: int = 11 # 连续域每个变量等分点数（含端点） + max_combinations: int = 200000 # 笛卡尔积规模上限，超过则退化为坐标下降 + random_seed: int = 20260805 # 固定随机种子，保证确定性可复现 + enumerate_choices: bool = True # 是否完整枚举离散 choices（False 时取首个） + + def __post_init__(self) -> None: + if self.grid_steps < 2: + raise SolverError("grid_steps 必须 ≥ 2（至少含两端点）") + if self.max_combinations < 1: + raise SolverError("max_combinations 必须 ≥ 1") + + +@dataclass +class Solution: + """求解结果（#81 可解释建议的结构化输入）。""" + + assignment: Dict[str, Any] = field(default_factory=dict) + objective_value: float = 0.0 + feasible: bool = False + target_met: bool = False + violated: List[ConstraintSpec] = field(default_factory=list) + iterations: int = 0 + evaluated: int = 0 + strategy: str = "" # "grid" / "coordinate_descent" + message: str = "" + + def to_dict(self) -> Dict[str, Any]: + return { + "assignment": dict(self.assignment), + "objective_value": self.objective_value, + "feasible": self.feasible, + "target_met": self.target_met, + "violated": [c.to_dict() for c in self.violated], + "iterations": self.iterations, + "evaluated": self.evaluated, + "strategy": self.strategy, + "message": self.message, + } + + +# --------------------------------------------------------------------------- +# 变量取值候选生成 +# --------------------------------------------------------------------------- + + +def candidate_values(var: DecisionVariable, config: SolverConfig) -> List[Any]: + """为单个变量生成求解候选取值集合。""" + if var.kind == DomainKind.CHOICES: + return list(var.choices) if config.enumerate_choices else [var.choices[0]] + # bounds 连续域：等分离散化 + if var.bounds is None: + return [] + low, high = var.bounds + step = (high - low) / (config.grid_steps - 1) + vals = [low + i * step for i in range(config.grid_steps)] + if var.integer: + vals = [float(round(v)) for v in vals] + # 去重保序 + seen: set = set() + uniq: List[Any] = [] + for v in vals: + iv = int(v) + if iv not in seen: + seen.add(iv) + uniq.append(iv) + return uniq + return vals + + +def _grid_size(problem: OptimizationProblem, config: SolverConfig) -> int: + total = 1 + for v in problem.variables: + total *= len(candidate_values(v, config)) + return total + + +# --------------------------------------------------------------------------- +# 求解器 +# --------------------------------------------------------------------------- + + +def _better(new: float, best: float, sense: Sense) -> bool: + """判断 new 是否比 best 更优。""" + if sense == Sense.MAXIMIZE: + return new > best + return new < best + + +def _initial_objective(sense: Sense) -> float: + return -math.inf if sense == Sense.MAXIMIZE else math.inf + + +def _solve_grid(problem: OptimizationProblem, config: SolverConfig) -> Solution: + """完整网格枚举：笛卡尔积里筛可行、选最优（全局最优保证）。""" + rng = random.Random(config.random_seed) + per_var = [candidate_values(v, config) for v in problem.variables] + names = [v.name for v in problem.variables] + sense = problem.objective.sense + best_obj = _initial_objective(sense) + best_assign: Optional[Dict[str, Any]] = None + evaluated = 0 + iterations = 0 + # 为控制内存，逐组合判定，不一次性 materialize + for combo in itertools.product(*per_var): + evaluated += 1 + iterations += 1 + assignment = dict(zip(names, combo)) + if not problem.is_feasible(assignment): + continue + obj = problem.objective.evaluate(assignment) + if best_assign is None or _better(obj, best_obj, sense): + best_obj = obj + best_assign = assignment + feasible = best_assign is not None + return _build_solution(problem, config, best_assign or {}, best_obj, + feasible, iterations, evaluated, "grid", + "网格枚举完成" if feasible else "无可行解（约束过紧或域为空）") + + +def _solve_coordinate_descent( + problem: OptimizationProblem, config: SolverConfig +) -> Solution: + """坐标下降：固定其余变量、逐维选当前最优取值（贪心，规模过大时降级用）。 + + 从初值（``initial`` 缺省取域中点）出发，反复扫描各变量、在候选值里取使目标 + 最优且保持可行者；迭代至收敛或达 ``max_rounds``。非全局最优，但保可用性。 + """ + sense = problem.objective.sense + names = [v.name for v in problem.variables] + per_var = {v.name: candidate_values(v, config) for v in problem.variables} + # 初值 + assignment: Dict[str, Any] = {} + for v in problem.variables: + if v.initial is not None and v.contains(v.initial): + assignment[v.name] = v.initial + elif v.kind == DomainKind.CHOICES and v.choices: + assignment[v.name] = v.choices[0] + elif v.bounds is not None: + assignment[v.name] = (v.bounds[0] + v.bounds[1]) / 2.0 + else: # pragma: no cover - 防御 + assignment[v.name] = None + max_rounds = max(3, len(names)) + evaluated = 0 + iterations = 0 + for _round in range(max_rounds): + improved = False + for name in names: + cur_best = assignment[name] + cur_assign = dict(assignment) + cur_obj = problem.objective.evaluate(cur_assign) if problem.is_feasible(cur_assign) else None + best_val = cur_best + best_obj = cur_obj if cur_obj is not None else _initial_objective(sense) + for cand in per_var[name]: + evaluated += 1 + trial = dict(assignment) + trial[name] = cand + if not problem.is_feasible(trial): + continue + obj = problem.objective.evaluate(trial) + if cur_obj is None or _better(obj, best_obj, sense): + best_obj = obj + best_val = cand + if best_val != cur_best: + assignment[name] = best_val + improved = True + iterations += 1 + if not improved: + break + feasible = problem.is_feasible(assignment) + final_obj = problem.objective.evaluate(assignment) if feasible else 0.0 + return _build_solution(problem, config, assignment, final_obj, feasible, + iterations, evaluated, "coordinate_descent", + "坐标下降完成" if feasible else "坐标下降未找到可行解") + + +def _build_solution( + problem: OptimizationProblem, + config: SolverConfig, + assignment: Dict[str, Any], + obj: float, + feasible: bool, + iterations: int, + evaluated: int, + strategy: str, + message: str, +) -> Solution: + violated = problem.violated_constraints(assignment) if assignment else [] + target_met = False + if feasible and problem.objective.target_value is not None: + if problem.objective.sense == Sense.MAXIMIZE: + target_met = obj >= problem.objective.target_value + else: + target_met = obj <= problem.objective.target_value + elif feasible and problem.objective.target_value is None: + target_met = True # 未设达标量则视为达成 + return Solution( + assignment=assignment, + objective_value=obj, + feasible=feasible, + target_met=target_met, + violated=violated, + iterations=iterations, + evaluated=evaluated, + strategy=strategy, + message=message, + ) + + +def solve(problem: OptimizationProblem, + config: Optional[SolverConfig] = None) -> Solution: + """统一求解入口。 + + 自动按规模选择策略：网格规模 ≤ ``max_combinations`` 用全局网格枚举， + 否则退化为坐标下降（保可用性）。先做静态校验，校验失败直接返回不可行解。 + """ + cfg = config or SolverConfig() + # 静态校验 + errs = problem.validate() + if errs: + return Solution(feasible=False, strategy="validate", + message="问题校验失败: " + "; ".join(errs)) + # 空问题：无可调变量 + if not problem.variables: + return Solution(feasible=True, target_met=True, strategy="empty", + message="无决策变量，视为平凡可行") + size = _grid_size(problem, cfg) + if size <= cfg.max_combinations: + return _solve_grid(problem, cfg) + return _solve_coordinate_descent(problem, cfg) diff --git a/templates/ti-cl4/recipe-optim/tests/test_advisor.py b/templates/ti-cl4/recipe-optim/tests/test_advisor.py new file mode 100644 index 0000000..f837d3d --- /dev/null +++ b/templates/ti-cl4/recipe-optim/tests/test_advisor.py @@ -0,0 +1,158 @@ +# -*- coding: utf-8 -*- +"""Ti-2 优化建议生成与可解释性 单元测试（Issue #81）。 + +覆盖： +- 单条建议方向/幅度计算； +- generate_advice：变量级建议、跨工序佐证、风险与达标提示、不可行降级、摘要； +- 序列化； +- 端到端（#78→#79→#81 链路 + 跨工序权重注入）。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: E402 + +from recipe_optim.problem import ( # noqa: E402 + ConstraintKind, + ConstraintSpec, + DecisionVariable, + DomainKind, + ObjectiveSpec, + ObjectiveTerm, + OptimizationProblem, + Sense, + load_problem, +) +from recipe_optim.solver import Solution, SolverConfig, solve # noqa: E402 +from recipe_optim.advisor import ( # noqa: E402 + AdviceConfig, + AdviceItem, + AdviceReport, + AdvisorError, + generate_advice, +) + + +def _problem() -> OptimizationProblem: + return OptimizationProblem( + variables=[ + DecisionVariable("clf_temp", DomainKind.BOUNDS, "反应温度", "℃", + bounds=(800.0, 920.0), initial=860.0), + DecisionVariable("cl2_ratio", DomainKind.BOUNDS, "氯气配比", "ratio", + bounds=(0.8, 1.4), initial=1.0), + ], + objective=ObjectiveSpec(Sense.MAXIMIZE, target="Ti_purity", target_value=10.0, + terms=[ObjectiveTerm("clf_temp", 0.01), + ObjectiveTerm("cl2_ratio", 2.0)]), + constraints=[ConstraintSpec(ConstraintKind.BOX, variable="clf_temp", + bounds=(820.0, 900.0), reason="温度安全区间")], + ) + + +class TestDirectionDelta(unittest.TestCase): + def test_up(self): + from recipe_optim.advisor import _direction_and_delta + self.assertEqual(_direction_and_delta(1.0, 1.5), ("↑", 0.5)) + + def test_down(self): + from recipe_optim.advisor import _direction_and_delta + self.assertEqual(_direction_and_delta(2.0, 1.0), ("↓", -1.0)) + + def test_equal(self): + from recipe_optim.advisor import _direction_and_delta + self.assertEqual(_direction_and_delta(1.0, 1.0), ("→", 0.0)) + + def test_non_numeric(self): + from recipe_optim.advisor import _direction_and_delta + d, delta = _direction_and_delta("A", "B") + self.assertEqual(d, "≠") + self.assertEqual(delta, 0.0) + + +class TestGenerateAdvice(unittest.TestCase): + def test_variable_level_advice(self): + p = _problem() + sol = solve(p, SolverConfig(grid_steps=11)) + report = generate_advice(p, sol, current={"clf_temp": 860.0, "cl2_ratio": 1.0}) + self.assertTrue(report.feasible) + self.assertEqual(len(report.items), 2) + # 应当有变化项（求解器会爬到温度/配比上界附近） + changes = [it for it in report.items if it.direction in ("↑", "↓")] + self.assertGreater(len(changes), 0) + # 含工艺含义 + meanings = {it.meaning for it in report.items} + self.assertIn("反应温度", meanings) + + def test_target_met_summary(self): + p = _problem() + sol = solve(p, SolverConfig(grid_steps=11)) + report = generate_advice(p, sol) + self.assertIn("Ti_purity", report.summary) + + def test_cross_process_evidence_appended(self): + p = _problem() + sol = solve(p, SolverConfig(grid_steps=11)) + weights = {"sponge_titanium_grade": {"clf_temp": 0.5, "cl2_ratio": -0.3}} + report = generate_advice(p, sol, cross_process_weights=weights) + joined = " ".join(it.evidence for it in report.items) + self.assertIn("跨工序关联", joined) + self.assertTrue(any("sponge_titanium_grade" in t for t in report.trace)) + + def test_warnings_on_infeasible(self): + p = _problem() + # 构造一个不可行 Solution + sol = Solution(feasible=False, target_met=False, + violated=[ConstraintSpec(ConstraintKind.BOX, variable="clf_temp", + bounds=(820.0, 900.0), reason="温度安全区间")], + message="无可行解（约束过紧）") + report = generate_advice(p, sol) + self.assertFalse(report.feasible) + self.assertTrue(any("可行" in w for w in report.warnings)) + self.assertIn("温度安全区间", " ".join(report.warnings)) + + def test_keep_unchanged_item(self): + p = OptimizationProblem( + variables=[DecisionVariable("x", DomainKind.BOUNDS, "X", "", + bounds=(0.0, 10.0), initial=5.0)], + objective=ObjectiveSpec(Sense.MAXIMIZE, terms=[ObjectiveTerm("x", 0.0)]), + constraints=[ConstraintSpec(ConstraintKind.BOX, variable="x", bounds=(5.0, 5.0))], + ) + sol = solve(p, SolverConfig(grid_steps=3)) + report = generate_advice(p, sol, current={"x": 5.0}) + self.assertEqual(len(report.items), 1) + self.assertEqual(report.items[0].direction, "→") + + def test_serialization(self): + p = _problem() + sol = solve(p, SolverConfig(grid_steps=5)) + report = generate_advice(p, sol) + d = report.to_dict() + self.assertIn("items", d) + self.assertIn("summary", d) + self.assertTrue(d["feasible"]) + # item dict 完整 + if d["items"]: + self.assertIn("reason_text", d["items"][0]) + + +class TestEndToEndFromTemplate(unittest.TestCase): + def test_template_chain(self): + cfg_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "recipe_optim.template.yaml") + p = load_problem(cfg_path) + sol = solve(p, SolverConfig(grid_steps=7, max_combinations=200000)) + report = generate_advice(p, sol, cross_process_weights={ + "Ti_purity": {"clf_temp": 0.8, "cl2_ratio": 1.2, "feed_rate": 0.1}}) + self.assertTrue(report.feasible) + self.assertEqual(len(report.items), len(p.variables)) + # 每条建议都有依据 + for it in report.items: + self.assertTrue(it.evidence) + # 溯源链路非空 + self.assertGreater(len(report.trace), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/templates/ti-cl4/recipe-optim/tests/test_cross_process.py b/templates/ti-cl4/recipe-optim/tests/test_cross_process.py new file mode 100644 index 0000000..17c1401 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/tests/test_cross_process.py @@ -0,0 +1,206 @@ +# -*- coding: utf-8 -*- +"""Ti-2 跨工序关联寻优模型训练 单元测试（Issue #80）。 + +覆盖： +- RidgeRegression（拟合/预测/正则/权重/序列化、奇异矩阵处理）； +- 线性求解器； +- CrossProcessModelConfig（合法性校验）； +- CrossProcessModel（fit/predict/evaluate R²/特征权重可解释/序列化往返）； +- 数据门槛（min_samples 拒绝、缺失值过滤）。 +""" +import math +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: E402 + +from recipe_optim.cross_process import ( # noqa: E402 + CrossProcessError, + CrossProcessModel, + CrossProcessModelConfig, + CrossProcessSample, + RidgeRegression, + _solve_linear, +) + + +class TestSolveLinear(unittest.TestCase): + def test_basic(self): + # x + y = 3, 2x - y = 0 → x=1, y=2 + w = _solve_linear([[1, 1], [2, -1]], [3, 0]) + self.assertAlmostEqual(w[0], 1.0) + self.assertAlmostEqual(w[1], 2.0) + + def test_singular_raises(self): + with self.assertRaises(CrossProcessError): + _solve_linear([[1, 1], [1, 1]], [1, 1]) + + +class TestRidgeRegression(unittest.TestCase): + def test_fit_predict_linear(self): + # y = 2x + 1（精确线性） + reg = RidgeRegression(alpha=0.0) + reg.fit([[0], [1], [2], [3]], [1, 3, 5, 7], ["x"]) + self.assertAlmostEqual(reg.predict_one([4]), 9.0) + d = reg.weights_dict() + self.assertAlmostEqual(d["x"], 2.0, places=6) + self.assertAlmostEqual(d["__intercept__"], 1.0, places=6) + + def test_multivariate(self): + # y = x0 + 2*x1 + reg = RidgeRegression(alpha=0.0) + reg.fit([[0, 0], [1, 0], [0, 1], [1, 1], [2, 3]], + [0, 1, 2, 3, 8], ["x0", "x1"]) + self.assertAlmostEqual(reg.predict_one([1, 1]), 3.0, places=5) + + def test_regularization_smooths(self): + # 强正则下权重被压缩向 0 + X = [[0], [1], [2], [3]] + y = [1, 3, 5, 7] + r0 = RidgeRegression(alpha=0.0); r0.fit(X, y, ["x"]) + rbig = RidgeRegression(alpha=1000.0); rbig.fit(X, y, ["x"]) + self.assertLess(abs(rbig.weights_dict()["x"]), abs(r0.weights_dict()["x"])) + + def test_negative_alpha_rejected(self): + with self.assertRaises(CrossProcessError): + RidgeRegression(alpha=-1) + + def test_mismatched_columns_rejected(self): + reg = RidgeRegression() + with self.assertRaises(CrossProcessError): + reg.fit([[1, 2]], [3], ["x"]) + + def test_predict_before_fit(self): + with self.assertRaises(CrossProcessError): + RidgeRegression().predict_one([1]) + + def test_roundtrip(self): + reg = RidgeRegression(alpha=0.5) + reg.fit([[0], [1], [2]], [1, 3, 5], ["x"]) + reg2 = RidgeRegression.from_dict(reg.to_dict()) + self.assertAlmostEqual(reg2.predict_one([3]), reg.predict_one([3])) + + +class TestConfig(unittest.TestCase): + def test_empty_features_rejected(self): + with self.assertRaises(CrossProcessError): + CrossProcessModelConfig(upstream_features=[], downstream_targets=["t"]) + + def test_empty_targets_rejected(self): + with self.assertRaises(CrossProcessError): + CrossProcessModelConfig(upstream_features=["f"], downstream_targets=[]) + + def test_bad_min_samples(self): + with self.assertRaises(CrossProcessError): + CrossProcessModelConfig(upstream_features=["f"], downstream_targets=["t"], + min_samples=1) + + +def _gen_samples(n=20, seed=42): + """生成 y = 2*x + 3 的合成样本（上游 x，下游 y），用于训练/评估。""" + import random + rng = random.Random(seed) + samples = [] + for i in range(n): + x = rng.uniform(0, 10) + samples.append(CrossProcessSample( + upstream={"TiCl4_purity": x}, + downstream={"sponge_titanium_grade": 2.0 * x + 3.0}, + batch=f"B{i}", timestamp=float(i), + )) + return samples + + +class TestCrossProcessModel(unittest.TestCase): + def test_fit_predict_evaluate(self): + cfg = CrossProcessModelConfig( + upstream_features=["TiCl4_purity"], + downstream_targets=["sponge_titanium_grade"], + alpha=0.0, min_samples=10, + ) + m = CrossProcessModel(cfg) + train = _gen_samples(15, seed=1) + m.fit(train) + # 预测接近真实 + pred = m.predict({"TiCl4_purity": 5.0}) + self.assertAlmostEqual(pred["sponge_titanium_grade"], 2 * 5 + 3, places=3) + # R² 接近 1（线性可精确拟合） + report = m.evaluate(_gen_samples(20, seed=2)) + self.assertGreater(report["r2_sponge_titanium_grade"], 0.99) + self.assertIn("mse_overall", report) + + def test_min_samples_enforced(self): + cfg = CrossProcessModelConfig( + upstream_features=["f"], downstream_targets=["t"], min_samples=10) + m = CrossProcessModel(cfg) + with self.assertRaises(CrossProcessError): + m.fit([CrossProcessSample(upstream={"f": 1}, downstream={"t": 2})] * 3) + + def test_missing_values_filtered(self): + cfg = CrossProcessModelConfig( + upstream_features=["f1", "f2"], downstream_targets=["t"], min_samples=5) + samples = [] + for i in range(10): + s = CrossProcessSample( + upstream={"f1": float(i), "f2": float(i)}, + downstream={"t": float(i) + float(i)}, + batch=f"B{i}") + samples.append(s) + # 给部分样本注入缺失值（应被过滤，但剩余 ≥ min_samples 仍可训练） + samples[0].upstream["f1"] = float("nan") + m = CrossProcessModel(cfg) + m.fit(samples) + self.assertTrue(m.fitted) + + def test_predict_missing_feature(self): + cfg = CrossProcessModelConfig( + upstream_features=["f"], downstream_targets=["t"], min_samples=5) + m = CrossProcessModel(cfg) + m.fit([CrossProcessSample(upstream={"f": float(i)}, + downstream={"t": float(i)}) for i in range(6)]) + with self.assertRaises(CrossProcessError): + m.predict({}) # 缺 f + + def test_feature_weights_explainable(self): + cfg = CrossProcessModelConfig( + upstream_features=["f"], downstream_targets=["t"], + alpha=0.0, min_samples=5) + m = CrossProcessModel(cfg) + m.fit([CrossProcessSample(upstream={"f": float(i)}, + downstream={"t": 2 * float(i) + 1}) + for i in range(6)]) + w = m.feature_weights("t") + self.assertAlmostEqual(w["f"], 2.0, places=4) + self.assertIn("__intercept__", w) + with self.assertRaises(CrossProcessError): + m.feature_weights("ghost") + + def test_evaluate_before_fit(self): + cfg = CrossProcessModelConfig( + upstream_features=["f"], downstream_targets=["t"], min_samples=5) + with self.assertRaises(CrossProcessError): + CrossProcessModel(cfg).evaluate([]) + + def test_save_load_roundtrip(self): + cfg = CrossProcessModelConfig( + upstream_features=["TiCl4_purity"], + downstream_targets=["sponge_titanium_grade"], + alpha=0.1, min_samples=5) + m = CrossProcessModel(cfg) + m.fit(_gen_samples(10, seed=3)) + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "model.json") + m.save(path) + m2 = CrossProcessModel.load(path) + self.assertTrue(m2.fitted) + p1 = m.predict({"TiCl4_purity": 4.0}) + p2 = m2.predict({"TiCl4_purity": 4.0}) + self.assertAlmostEqual(p1["sponge_titanium_grade"], + p2["sponge_titanium_grade"], places=6) + + +if __name__ == "__main__": + unittest.main() diff --git a/templates/ti-cl4/recipe-optim/tests/test_solver.py b/templates/ti-cl4/recipe-optim/tests/test_solver.py new file mode 100644 index 0000000..ab73846 --- /dev/null +++ b/templates/ti-cl4/recipe-optim/tests/test_solver.py @@ -0,0 +1,207 @@ +# -*- coding: utf-8 -*- +"""Ti-2 配方优化求解器集成 单元测试（Issue #79）。 + +覆盖： +- 求解器配置（grid_steps/max_combinations 合法性）； +- 候选取值生成（bounds 等分/integer 去重/choices 枚举）； +- 网格求解（全局最优、可行性、target 达成、无可行解降级）； +- 坐标下降（规模超限降级、收敛）； +- solve 统一入口（自动选策略、静态校验失败、空问题）； +- Solution 序列化； +- 加载 #78 模板后端到端求解。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: E402 挂载 recipe_optim 包 + +from recipe_optim.problem import ( # noqa: E402 + ConstraintKind, + ConstraintSpec, + DecisionVariable, + DomainKind, + ObjectiveSpec, + ObjectiveTerm, + OptimizationProblem, + Sense, + load_problem, +) +from recipe_optim.solver import ( # noqa: E402 + SolverConfig, + SolverError, + Solution, + candidate_values, + solve, +) + +CONFIG_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "recipe_optim.template.yaml", +) + + +def _toy_problem() -> OptimizationProblem: + """minimize -(x+y)，x∈[0,4] 5 点，y∈[0,4] 5 点 → 网格 25 组合。""" + return OptimizationProblem( + variables=[ + DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 4.0)), + DecisionVariable("y", DomainKind.BOUNDS, bounds=(0.0, 4.0)), + ], + objective=ObjectiveSpec(Sense.MAXIMIZE, + terms=[ObjectiveTerm("x", 1.0), ObjectiveTerm("y", 1.0)]), + ) + + +class TestSolverConfig(unittest.TestCase): + def test_defaults(self): + c = SolverConfig() + self.assertGreaterEqual(c.grid_steps, 2) + self.assertGreaterEqual(c.max_combinations, 1) + + def test_invalid_grid_steps(self): + with self.assertRaises(SolverError): + SolverConfig(grid_steps=1) + + def test_invalid_max_combinations(self): + with self.assertRaises(SolverError): + SolverConfig(max_combinations=0) + + +class TestCandidateValues(unittest.TestCase): + def test_bounds_grid(self): + v = DecisionVariable("t", DomainKind.BOUNDS, bounds=(0.0, 4.0)) + vals = candidate_values(v, SolverConfig(grid_steps=5)) + self.assertEqual(vals[0], 0.0) + self.assertEqual(vals[-1], 4.0) + self.assertEqual(len(vals), 5) + + def test_integer_dedupe(self): + v = DecisionVariable("n", DomainKind.BOUNDS, bounds=(0.0, 4.0), integer=True) + vals = candidate_values(v, SolverConfig(grid_steps=5)) + self.assertEqual(vals, [0, 1, 2, 3, 4]) + + def test_choices_enum(self): + v = DecisionVariable("c", DomainKind.CHOICES, choices=["A", "B", "C"]) + self.assertEqual(candidate_values(v, SolverConfig()), ["A", "B", "C"]) + self.assertEqual(candidate_values(v, SolverConfig(enumerate_choices=False)), ["A"]) + + +class TestSolveGrid(unittest.TestCase): + def test_global_optimum_maximize(self): + p = _toy_problem() + sol = solve(p, SolverConfig(grid_steps=5)) + self.assertTrue(sol.feasible) + # 最优 x=y=4 → obj=8 + self.assertAlmostEqual(sol.objective_value, 8.0) + self.assertEqual(sol.assignment["x"], 4.0) + self.assertEqual(sol.assignment["y"], 4.0) + self.assertEqual(sol.strategy, "grid") + + def test_minimize(self): + p = OptimizationProblem( + variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 4.0))], + objective=ObjectiveSpec(Sense.MINIMIZE, terms=[ObjectiveTerm("x", 1.0)]), + ) + sol = solve(p, SolverConfig(grid_steps=5)) + self.assertTrue(sol.feasible) + self.assertEqual(sol.assignment["x"], 0.0) + self.assertAlmostEqual(sol.objective_value, 0.0) + + def test_target_met(self): + p = OptimizationProblem( + variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 10.0))], + objective=ObjectiveSpec(Sense.MAXIMIZE, target_value=8.0, + terms=[ObjectiveTerm("x", 1.0)]), + ) + sol = solve(p, SolverConfig(grid_steps=11)) + self.assertTrue(sol.feasible) + self.assertTrue(sol.target_met) # x=10 >= 8 + + def test_target_not_met(self): + p = OptimizationProblem( + variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 5.0))], + objective=ObjectiveSpec(Sense.MAXIMIZE, target_value=99.0, + terms=[ObjectiveTerm("x", 1.0)]), + ) + sol = solve(p, SolverConfig(grid_steps=6)) + self.assertTrue(sol.feasible) + self.assertFalse(sol.target_met) + + def test_no_feasible_solution(self): + # box 收紧到与域不交 → 无可行 + p = OptimizationProblem( + variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 1.0))], + constraints=[ConstraintSpec(ConstraintKind.BOX, variable="x", bounds=(5.0, 6.0))], + objective=ObjectiveSpec(Sense.MAXIMIZE, terms=[ObjectiveTerm("x", 1.0)]), + ) + sol = solve(p, SolverConfig(grid_steps=3)) + self.assertFalse(sol.feasible) + self.assertIn("无可行解", sol.message) + + +class TestCoordinateDescent(unittest.TestCase): + def test_falls_back_when_grid_too_large(self): + # 三个变量 × grid_steps=5 = 125；设 max_combinations=10 → 降级 + p = OptimizationProblem( + variables=[ + DecisionVariable(f"v{i}", DomainKind.BOUNDS, bounds=(0.0, 4.0)) + for i in range(3) + ], + objective=ObjectiveSpec(Sense.MAXIMIZE, + terms=[ObjectiveTerm(f"v{i}", 1.0) for i in range(3)]), + ) + sol = solve(p, SolverConfig(grid_steps=5, max_combinations=10)) + self.assertEqual(sol.strategy, "coordinate_descent") + # 坐标下降应能爬到各维上界附近（贪心可收敛到此线性目标的全局最优） + self.assertTrue(sol.feasible) + self.assertAlmostEqual(sol.objective_value, 12.0, places=6) + + +class TestSolveEntry(unittest.TestCase): + def test_validate_failure_returns_infeasible(self): + p = OptimizationProblem( + variables=[DecisionVariable("x", DomainKind.BOUNDS, bounds=(0.0, 1.0))], + objective=ObjectiveSpec(Sense.MAXIMIZE, + terms=[ObjectiveTerm("ghost", 1.0)]), + ) + sol = solve(p, SolverConfig()) + self.assertFalse(sol.feasible) + self.assertEqual(sol.strategy, "validate") + self.assertIn("校验失败", sol.message) + + def test_empty_problem_trivially_feasible(self): + p = OptimizationProblem() + sol = solve(p, SolverConfig()) + self.assertTrue(sol.feasible) + self.assertEqual(sol.strategy, "empty") + + def test_solution_to_dict(self): + p = _toy_problem() + sol = solve(p, SolverConfig(grid_steps=3)) + d = sol.to_dict() + self.assertIn("assignment", d) + self.assertIn("objective_value", d) + self.assertIn("evaluated", d) + self.assertTrue(d["feasible"]) + + +class TestEndToEndFromTemplate(unittest.TestCase): + def test_solve_loaded_problem(self): + p = load_problem(CONFIG_PATH) + sol = solve(p, SolverConfig(grid_steps=7, max_combinations=200000)) + # 模板含 4 变量；7^3 * 3 = 1029 组合 < 上限 → 网格 + self.assertEqual(sol.strategy, "grid") + self.assertTrue(sol.feasible) + # 取值应满足 box 收紧（clf_temp∈[820,900]）与 forbidden（非 C@910） + self.assertGreaterEqual(sol.assignment["clf_temp"], 820 - 1e-6) + self.assertLessEqual(sol.assignment["clf_temp"], 900 + 1e-6) + self.assertFalse(sol.assignment["catalyst"] == "C" + and sol.assignment.get("clf_temp") == 910) + # evaluated>0 + self.assertGreater(sol.evaluated, 0) + + +if __name__ == "__main__": + unittest.main()