207 lines
7.7 KiB
Python
207 lines
7.7 KiB
Python
# -*- 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()
|