# -*- 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()