feat(#39): 超参包 JSON Schema 校验器(PRD 5.3 模型框架配置化)

新增 core/model-framework/hyperparam.py:超参包(Hyperparam Pack)结构化
校验器,零外部依赖(不依赖 jsonschema),对齐 PRD 5.3「超参包驱动」与 EPIC #5。

校验维度(10 类):
- 顶层结构:必填字段 model_id/template/algorithm/features/target;
- algorithm:已知算法集合校验(xgboost/lightgbm/dnn/lstm/gnn/
  isolation_forest/zscore/linear/ridge),未知项报错并列出已知项
  (对齐 PRD「新增结构走 Recipe 插件注册」);
- features:非空列表、name 唯一且合法、spec 非空(FeatureSpec 文本
  存在性校验,语法由 #35 引擎解释);
- objective/hyperparams/train_window/alarm_threshold/drift_check 可选
  字段的类型、取值与格式校验(train_window 形如 180d/4w;drift limit
  在 (0,1];alarm type 含对应阈值键)。

设计:
- validate_pack 返回 ValidationReport(逐条问题,不抛异常),便于配置台
  聚合展示;load_pack 校验失败抛 ValueError 供训练/推理流水线 fail-fast。
- validate_pack_file 复用结构校验,文件/JSON 解析错误也记入报告。

测试:tests/test_hyperparam.py 25 用例全绿(合法包/必填缺失/未知算法/
特征重复与缺 spec/目标函数/train_window 格式/alarm/drift/文件加载与
JSON 解析错误);python -m unittest discover -s tests -v → 25 passed。

close #39
This commit is contained in:
2026-08-04 21:35:44 +08:00
parent 691a812fcf
commit 5fa1103e3f
5 changed files with 647 additions and 0 deletions
+16
View File
@@ -0,0 +1,16 @@
# -*- coding: utf-8 -*-
"""测试引导:把 `core/model-framework` 以包名 `model_framework` 挂载到 sys.modules。
目录名 `model-framework` 含连字符,无法直接以包名 import;挂载后模块内相对导入
(`from .hyperparam import ...`)在 unittest 发现机制下可正常解析。
"""
import os
import sys
import types
MF_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, MF_DIR)
if "model_framework" not in sys.modules:
pkg = types.ModuleType("model_framework")
pkg.__path__ = [MF_DIR]
sys.modules["model_framework"] = pkg
@@ -0,0 +1,239 @@
# -*- coding: utf-8 -*-
"""超参包 JSON Schema 校验器测试(issue #39)。
覆盖:合法包通过 / 必填缺失 / 未知算法 / 特征重复与缺 spec / 目标函数 /
train_window 格式 / alarm_threshold / drift_check / 文件加载与 JSON 解析错误。
"""
import os
import tempfile
import unittest
import _bootstrap # noqa: F401 挂载包名
from model_framework.hyperparam import (
HyperparamPack,
load_pack,
validate_pack,
validate_pack_file,
)
def _good_pack():
"""PRD 5.3 示例的合法超参包。"""
return {
"model_id": "quality_predict_ti",
"template": "iAOP-Template-Ti",
"algorithm": "xgboost",
"features": [
{"name": "EMA_CLF_TEMP_5m", "spec": "EMA(CLF-01.TEMP, 5m)"},
{"name": "RollingStd_CL2_10", "spec": "RollingStd(CLF-01.CL2, 10)"},
],
"target": "Ti_purity",
"objective": "reg:squarederror",
"hyperparams": {"max_depth": 6, "eta": 0.1, "n_estimators": 300},
"train_window": "180d",
"alarm_threshold": {"type": "zscore", "k": 3.0},
"drift_check": {"method": "psi", "limit": 0.2},
}
class ValidatePackTest(unittest.TestCase):
def test_good_pack_passes(self):
report = validate_pack(_good_pack())
self.assertTrue(report.ok, report.summary())
self.assertEqual(report.issues, [])
def test_minimal_pack_passes(self):
report = validate_pack(
{
"model_id": "anomaly_resin",
"template": "iAOP-Template-Resin",
"algorithm": "isolation_forest",
"features": [{"name": "F1", "spec": "RateOfChange(炉压)"}],
"target": "anomaly_score",
}
)
self.assertTrue(report.ok, report.summary())
def test_missing_required_field(self):
pack = _good_pack()
del pack["algorithm"]
report = validate_pack(pack)
self.assertFalse(report.ok)
codes = [i.code for i in report.issues]
self.assertIn("missing_field", codes)
self.assertTrue(any(i.path == "algorithm" for i in report.issues))
def test_missing_all_required(self):
report = validate_pack({})
self.assertFalse(report.ok)
codes = [i.code for i in report.issues]
# 5 个必填字段全部缺失
self.assertEqual(codes.count("missing_field"), 5)
def test_unknown_algorithm_lists_known(self):
pack = _good_pack()
pack["algorithm"] = "magic_boost"
report = validate_pack(pack)
self.assertFalse(report.ok)
issue = next(i for i in report.issues if i.code == "unknown_algorithm")
self.assertIn("xgboost", issue.message)
self.assertIn("magic_boost", issue.message)
def test_empty_features_rejected(self):
pack = _good_pack()
pack["features"] = []
report = validate_pack(pack)
codes = [i.code for i in report.issues]
self.assertIn("empty_features", codes)
def test_duplicate_feature_name(self):
pack = _good_pack()
pack["features"] = [
{"name": "F1", "spec": "EMA(A, 5m)"},
{"name": "F1", "spec": "EMA(B, 5m)"},
]
report = validate_pack(pack)
codes = [i.code for i in report.issues]
self.assertIn("dup_feature", codes)
def test_feature_missing_spec(self):
pack = _good_pack()
pack["features"] = [{"name": "F1", "spec": ""}]
report = validate_pack(pack)
codes = [i.code for i in report.issues]
self.assertIn("missing_field", codes)
def test_feature_bad_name_format(self):
pack = _good_pack()
pack["features"] = [{"name": "1bad", "spec": "EMA(A,5m)"}]
report = validate_pack(pack)
codes = [i.code for i in report.issues]
self.assertIn("bad_format", codes)
def test_unknown_objective(self):
pack = _good_pack()
pack["objective"] = "reg:magic"
report = validate_pack(pack)
self.assertTrue(any(i.code == "unknown_objective" for i in report.issues))
def test_bad_train_window_format(self):
for bad in ["180", "180days", "abc", "", "1y"]:
pack = _good_pack()
pack["train_window"] = bad
report = validate_pack(pack)
self.assertTrue(
any(i.code == "bad_format" and i.path == "train_window" for i in report.issues),
f"应拒绝非法 train_window: {bad!r}",
)
def test_good_train_window_units(self):
for good in ["180d", "4w", "72h", "30m"]:
pack = _good_pack()
pack["train_window"] = good
report = validate_pack(pack)
self.assertTrue(report.ok, f"应接受合法 train_window: {good!r} -> {report.summary()}")
def test_alarm_threshold_zscore_missing_k(self):
pack = _good_pack()
pack["alarm_threshold"] = {"type": "zscore"}
report = validate_pack(pack)
self.assertTrue(any(i.code == "missing_field" and i.path == "alarm_threshold.k" for i in report.issues))
def test_alarm_threshold_unknown_type(self):
pack = _good_pack()
pack["alarm_threshold"] = {"type": "voodoo"}
report = validate_pack(pack)
self.assertTrue(any(i.code == "unknown_alarm_type" for i in report.issues))
def test_drift_check_limit_range(self):
for bad_limit in [0, 1.5, -0.1]:
pack = _good_pack()
pack["drift_check"] = {"method": "psi", "limit": bad_limit}
report = validate_pack(pack)
self.assertTrue(
any(i.code == "bad_range" for i in report.issues),
f"应拒绝非法 drift_check.limit: {bad_limit}",
)
def test_drift_check_unknown_method(self):
pack = _good_pack()
pack["drift_check"] = {"method": "magic", "limit": 0.2}
report = validate_pack(pack)
self.assertTrue(any(i.code == "unknown_drift_method" for i in report.issues))
def test_root_not_object(self):
report = validate_pack([1, 2, 3]) # type: ignore[arg-type]
self.assertFalse(report.ok)
self.assertEqual(report.issues[0].code, "bad_type")
def test_report_summary_ok(self):
self.assertEqual(validate_pack(_good_pack()).summary(), "超参包校验通过")
def test_report_summary_aggregates(self):
pack = _good_pack()
del pack["algorithm"]
del pack["target"]
summary = validate_pack(pack).summary()
self.assertIn("2 个问题", summary)
self.assertIn("missing_field", summary)
class LoadPackTest(unittest.TestCase):
def test_load_good_returns_pack(self):
pack = load_pack(_good_pack())
self.assertIsInstance(pack, HyperparamPack)
self.assertEqual(pack.model_id, "quality_predict_ti")
self.assertEqual(pack.algorithm, "xgboost")
self.assertEqual(pack.feature_names, ["EMA_CLF_TEMP_5m", "RollingStd_CL2_10"])
self.assertEqual(pack.hyperparams["max_depth"], 6)
def test_load_bad_raises(self):
pack = _good_pack()
del pack["features"]
with self.assertRaises(ValueError) as cm:
load_pack(pack)
self.assertIn("超参包校验未通过", str(cm.exception))
class ValidatePackFileTest(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.mkdtemp()
self.path = os.path.join(self._tmp, "pack.json")
def _write(self, text):
with open(self.path, "w", encoding="utf-8") as fh:
fh.write(text)
return self.path
def test_valid_file_passes(self):
import json
self._write(json.dumps(_good_pack()))
report = validate_pack_file(self.path)
self.assertTrue(report.ok, report.summary())
def test_missing_file(self):
report = validate_pack_file(os.path.join(self._tmp, "nope.json"))
self.assertFalse(report.ok)
self.assertEqual(report.issues[0].code, "file_not_found")
def test_bad_json(self):
self._write("{ not json ")
report = validate_pack_file(self.path)
self.assertFalse(report.ok)
self.assertEqual(report.issues[0].code, "json_parse_error")
def test_file_with_schema_errors(self):
import json
bad = _good_pack()
bad["algorithm"] = "unknown_thing"
self._write(json.dumps(bad))
report = validate_pack_file(self.path)
self.assertFalse(report.ok)
self.assertEqual(report.issues[0].code, "unknown_algorithm")
if __name__ == "__main__":
unittest.main()