bot_dev1
5fa1103e3f
新增 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
340 lines
14 KiB
Python
340 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""超参包（Hyperparam Pack）JSON Schema 校验器。
|
||
|
||
对应 issue #39（父 EPIC #5「③ AI 模型框架 配置化重构」）与 PRD 5.3
|
||
「超参包驱动」：所有可变量（输入特征清单、算法选型、超参、训练窗口、告警阈值、
|
||
目标函数）外置为 JSON 超参包，同一框架切换模板仅改此包。
|
||
|
||
本模块对超参包做 **结构化校验**（不依赖 jsonschema 第三方库，零外部依赖，
|
||
便于边缘 / 离线环境运行），返回逐条问题的校验报告，便于配置台一次性展示全部问题、
|
||
也便于训练/推理流水线在加载包时 fail-fast。
|
||
|
||
校验维度（对齐 PRD 5.3 超参包示例与 5.3「配置点」）：
|
||
1. 顶层结构：必填字段 model_id / template / algorithm / features / target；
|
||
2. model_id / template：非空字符串；
|
||
3. algorithm：必须在合法算法集合内（xgboost / lightgbm / dnn / lstm / gnn /
|
||
isolation_forest / zscore / linear / ridge）；未知算法报错并提示已知项
|
||
（对齐 PRD「新增结构走插件注册」——校验期即暴露非法选型）；
|
||
4. features：非空列表；每项含 name（非空、包内唯一）与 spec（非空字符串，
|
||
FeatureSpec 文本，由 issue #35 引擎解释，此处只做存在性校验）；
|
||
5. target：非空字符串；
|
||
6. objective（可选）：若填写必须为已知目标函数（reg:squarederror /
|
||
binary:logistic / multi:softmax / regression / classification 等）；
|
||
7. hyperparams（可选）：若提供必须为对象（dict），且不含空键；
|
||
8. train_window（可选）：若填写必须形如 ``<正整数><单位>``（单位 d/w/h/m），
|
||
如 ``180d`` / ``4w``；
|
||
9. alarm_threshold（可选）：若提供必须为对象，且含 ``type``
|
||
（zscore / quantile / absolute）与对应阈值键；
|
||
10. drift_check（可选）：若提供必须为对象，含 ``method``（psi / ks / chi2）
|
||
与 ``limit``（0~1 之间）。
|
||
|
||
注：返回报告而非抛异常，便于配置台聚合展示；``load_pack`` 在校验失败时抛
|
||
``ValueError`` 供流水线 fail-fast。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 合法取值集合（对齐 PRD 5.3「四类模型模板」+ 默认主干网络 + Recipe 插件扩展点）
|
||
# ---------------------------------------------------------------------------
|
||
#: 已注册算法（默认主干 + Recipe 可选结构）。新增结构应走插件注册（issue #34/#41），
|
||
#: 此处枚举的是内核自带项；校验期遇到未知 algorithm 会报错并列出已知项，避免
|
||
#: 静默落到错误分支。
|
||
KNOWN_ALGORITHMS: Tuple[str, ...] = (
|
||
# 监督回归 / 分类（质量预测）
|
||
"xgboost",
|
||
"lightgbm",
|
||
"linear",
|
||
"ridge",
|
||
# 神经网络主干（默认 DNN；Recipe 可选 LSTM/GNN）
|
||
"dnn",
|
||
"lstm",
|
||
"gnn",
|
||
# 无监督异常 / 杂质预警
|
||
"isolation_forest",
|
||
"zscore",
|
||
)
|
||
|
||
#: 已知目标函数（xgboost 风格 + 通用风格）。
|
||
KNOWN_OBJECTIVES: Tuple[str, ...] = (
|
||
"reg:squarederror",
|
||
"reg:squaredlogerror",
|
||
"binary:logistic",
|
||
"multi:softmax",
|
||
"multi:softprob",
|
||
"regression",
|
||
"classification",
|
||
)
|
||
|
||
#: 训练窗口合法时间单位。
|
||
_TRAIN_WINDOW_RE = re.compile(r"^\d+(\.\d+)?[dwhm]$")
|
||
#: FeatureSpec 仅校验非空文本（具体语法由 issue #35 引擎解释）。
|
||
_FEATURE_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
||
|
||
|
||
@dataclass
|
||
class ValidationIssue:
|
||
"""单条校验问题。"""
|
||
|
||
code: str # 错误码：missing_field / bad_type / unknown_algorithm / ...
|
||
path: str # JSON 路径，如 ``features[1].name`` / ``algorithm``
|
||
message: str # 人类可读描述
|
||
|
||
|
||
@dataclass
|
||
class ValidationReport:
|
||
"""校验报告。"""
|
||
|
||
issues: List[ValidationIssue] = field(default_factory=list)
|
||
|
||
@property
|
||
def ok(self) -> bool:
|
||
return not self.issues
|
||
|
||
def summary(self) -> str:
|
||
if self.ok:
|
||
return "超参包校验通过"
|
||
by_code: Dict[str, int] = {}
|
||
for it in self.issues:
|
||
by_code[it.code] = by_code.get(it.code, 0) + 1
|
||
parts = [f"{code}×{n}" for code, n in sorted(by_code.items())]
|
||
return f"超参包校验未通过（{len(self.issues)} 个问题：" + "，".join(parts) + "）"
|
||
|
||
|
||
@dataclass
|
||
class HyperparamPack:
|
||
"""解析后的超参包（校验通过的视图）。
|
||
|
||
字段对齐 PRD 5.3 超参包示例；可选字段缺失时为 ``None``。
|
||
"""
|
||
|
||
model_id: str
|
||
template: str
|
||
algorithm: str
|
||
features: List[Dict[str, str]]
|
||
target: str
|
||
objective: Optional[str] = None
|
||
hyperparams: Optional[Dict[str, Any]] = None
|
||
train_window: Optional[str] = None
|
||
alarm_threshold: Optional[Dict[str, Any]] = None
|
||
drift_check: Optional[Dict[str, Any]] = None
|
||
raw: Dict[str, Any] = field(default_factory=dict)
|
||
|
||
@property
|
||
def feature_names(self) -> List[str]:
|
||
return [f["name"] for f in self.features]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 校验主逻辑
|
||
# ---------------------------------------------------------------------------
|
||
def validate_pack(pack: Dict[str, Any]) -> ValidationReport:
|
||
"""校验一个已解析为 dict 的超参包，返回报告（不抛异常）。"""
|
||
report = ValidationReport()
|
||
|
||
def add(code: str, path: str, message: str) -> None:
|
||
report.issues.append(ValidationIssue(code, path, message))
|
||
|
||
# 0. 顶层必须是对象
|
||
if not isinstance(pack, dict):
|
||
add("bad_type", "$", f"超参包根节点必须为 JSON 对象，实际为 {type(pack).__name__}")
|
||
return report
|
||
|
||
# 1. 必填字段
|
||
required: Tuple[str, ...] = ("model_id", "template", "algorithm", "features", "target")
|
||
for key in required:
|
||
if key not in pack:
|
||
add("missing_field", key, f"缺少必填字段：{key}")
|
||
|
||
# 2. 标量字段类型与取值
|
||
model_id = pack.get("model_id")
|
||
if "model_id" in pack:
|
||
if not isinstance(model_id, str) or not model_id.strip():
|
||
add("bad_type", "model_id", "model_id 必须为非空字符串")
|
||
elif not _FEATURE_NAME_RE.match(model_id):
|
||
add("bad_format", "model_id", "model_id 含非法字符（仅字母/数字/下划线，首字符非数字）")
|
||
|
||
template = pack.get("template")
|
||
if "template" in pack and (not isinstance(template, str) or not template.strip()):
|
||
add("bad_type", "template", "template 必须为非空字符串")
|
||
|
||
algorithm = pack.get("algorithm")
|
||
if "algorithm" in pack:
|
||
if not isinstance(algorithm, str) or not algorithm.strip():
|
||
add("bad_type", "algorithm", "algorithm 必须为非空字符串")
|
||
elif algorithm not in KNOWN_ALGORITHMS:
|
||
known = "、".join(KNOWN_ALGORITHMS)
|
||
add(
|
||
"unknown_algorithm",
|
||
"algorithm",
|
||
f"未知算法 '{algorithm}'；已知项：{known}（新增结构请走 Model Recipe 插件注册）",
|
||
)
|
||
|
||
target = pack.get("target")
|
||
if "target" in pack and (not isinstance(target, str) or not target.strip()):
|
||
add("bad_type", "target", "target 必须为非空字符串")
|
||
|
||
# 3. objective（可选）
|
||
objective = pack.get("objective")
|
||
if objective is not None:
|
||
if not isinstance(objective, str) or not objective.strip():
|
||
add("bad_type", "objective", "objective 若填写必须为非空字符串")
|
||
elif objective not in KNOWN_OBJECTIVES:
|
||
known = "、".join(KNOWN_OBJECTIVES)
|
||
add("unknown_objective", "objective", f"未知目标函数 '{objective}'；已知项：{known}")
|
||
|
||
# 4. features（必填，非空列表）
|
||
features = pack.get("features")
|
||
if features is None:
|
||
# missing_field 已在步骤 1 记录
|
||
pass
|
||
elif not isinstance(features, list):
|
||
add("bad_type", "features", f"features 必须为数组，实际为 {type(features).__name__}")
|
||
elif len(features) == 0:
|
||
add("empty_features", "features", "features 不能为空（模型至少需要一个输入特征）")
|
||
else:
|
||
seen_names: Dict[str, int] = {}
|
||
for i, feat in enumerate(features):
|
||
fpath = f"features[{i}]"
|
||
if not isinstance(feat, dict):
|
||
add("bad_type", fpath, f"特征项必须为对象，实际为 {type(feat).__name__}")
|
||
continue
|
||
name = feat.get("name")
|
||
spec = feat.get("spec")
|
||
if not isinstance(name, str) or not name.strip():
|
||
add("missing_field", f"{fpath}.name", "特征缺少 name 或为空")
|
||
else:
|
||
if not _FEATURE_NAME_RE.match(name):
|
||
add("bad_format", f"{fpath}.name", f"特征名 '{name}' 含非法字符")
|
||
if name in seen_names:
|
||
add(
|
||
"dup_feature",
|
||
f"{fpath}.name",
|
||
f"特征名 '{name}' 重复（首次出现在 features[{seen_names[name]}]）",
|
||
)
|
||
else:
|
||
seen_names[name] = i
|
||
if not isinstance(spec, str) or not spec.strip():
|
||
add("missing_field", f"{fpath}.spec", f"特征 '{name}' 缺少 spec（FeatureSpec 声明）或为空")
|
||
|
||
# 5. hyperparams（可选，对象）
|
||
hyperparams = pack.get("hyperparams")
|
||
if hyperparams is not None:
|
||
if not isinstance(hyperparams, dict):
|
||
add("bad_type", "hyperparams", f"hyperparams 必须为对象，实际为 {type(hyperparams).__name__}")
|
||
else:
|
||
for k, v in hyperparams.items():
|
||
if not isinstance(k, str) or not k.strip():
|
||
add("bad_format", "hyperparams", "hyperparams 含空键")
|
||
|
||
# 6. train_window（可选，<数><单位>）
|
||
train_window = pack.get("train_window")
|
||
if train_window is not None:
|
||
if not isinstance(train_window, str) or not _TRAIN_WINDOW_RE.match(train_window):
|
||
add(
|
||
"bad_format",
|
||
"train_window",
|
||
"train_window 必须形如 '<正数><单位>'（单位 d/w/h/m），如 '180d'、'4w'",
|
||
)
|
||
|
||
# 7. alarm_threshold（可选，对象，含 type）
|
||
alarm = pack.get("alarm_threshold")
|
||
if alarm is not None:
|
||
if not isinstance(alarm, dict):
|
||
add("bad_type", "alarm_threshold", f"alarm_threshold 必须为对象，实际为 {type(alarm).__name__}")
|
||
else:
|
||
atype = alarm.get("type")
|
||
known_alarm_types = ("zscore", "quantile", "absolute")
|
||
if not isinstance(atype, str) or atype not in known_alarm_types:
|
||
add(
|
||
"unknown_alarm_type",
|
||
"alarm_threshold.type",
|
||
f"alarm_threshold.type 必须为 {known_alarm_types} 之一",
|
||
)
|
||
if atype == "zscore" and "k" not in alarm:
|
||
add("missing_field", "alarm_threshold.k", "zscore 阈值缺少 k")
|
||
if atype == "quantile" and "q" not in alarm:
|
||
add("missing_field", "alarm_threshold.q", "quantile 阈值缺少 q")
|
||
if atype == "absolute" and "value" not in alarm:
|
||
add("missing_field", "alarm_threshold.value", "absolute 阈值缺少 value")
|
||
|
||
# 8. drift_check（可选，对象，method + limit）
|
||
drift = pack.get("drift_check")
|
||
if drift is not None:
|
||
if not isinstance(drift, dict):
|
||
add("bad_type", "drift_check", f"drift_check 必须为对象，实际为 {type(drift).__name__}")
|
||
else:
|
||
method = drift.get("method")
|
||
known_methods = ("psi", "ks", "chi2")
|
||
if not isinstance(method, str) or method not in known_methods:
|
||
add(
|
||
"unknown_drift_method",
|
||
"drift_check.method",
|
||
f"drift_check.method 必须为 {known_methods} 之一",
|
||
)
|
||
limit = drift.get("limit")
|
||
if limit is None:
|
||
add("missing_field", "drift_check.limit", "drift_check 缺少 limit")
|
||
elif not isinstance(limit, (int, float)) or isinstance(limit, bool):
|
||
add("bad_type", "drift_check.limit", "drift_check.limit 必须为数值")
|
||
elif not (0 < limit <= 1):
|
||
add("bad_range", "drift_check.limit", "drift_check.limit 必须在 (0, 1] 范围内")
|
||
|
||
return report
|
||
|
||
|
||
def load_pack(pack: Dict[str, Any]) -> HyperparamPack:
|
||
"""校验并把 dict 装配为 :class:`HyperparamPack`；校验失败抛 ``ValueError``。
|
||
|
||
供训练 / 推理流水线在加载超参包时 fail-fast 使用。
|
||
"""
|
||
report = validate_pack(pack)
|
||
if not report.ok:
|
||
raise ValueError(report.summary())
|
||
|
||
return HyperparamPack(
|
||
model_id=pack["model_id"],
|
||
template=pack["template"],
|
||
algorithm=pack["algorithm"],
|
||
features=list(pack["features"]),
|
||
target=pack["target"],
|
||
objective=pack.get("objective"),
|
||
hyperparams=pack.get("hyperparams"),
|
||
train_window=pack.get("train_window"),
|
||
alarm_threshold=pack.get("alarm_threshold"),
|
||
drift_check=pack.get("drift_check"),
|
||
raw=dict(pack),
|
||
)
|
||
|
||
|
||
def validate_pack_file(path: str) -> ValidationReport:
|
||
"""读取 JSON 文件并校验；文件 / JSON 解析错误也记入报告（不抛异常）。"""
|
||
report = ValidationReport()
|
||
|
||
def add(code: str, msg: str) -> None:
|
||
report.issues.append(ValidationIssue(code, "$", msg))
|
||
|
||
if not os.path.exists(path):
|
||
add("file_not_found", f"超参包文件不存在：{path}")
|
||
return report
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
text = fh.read()
|
||
except OSError as exc:
|
||
add("file_read_error", f"读取超参包失败：{exc}")
|
||
return report
|
||
try:
|
||
pack = json.loads(text)
|
||
except json.JSONDecodeError as exc:
|
||
add("json_parse_error", f"超参包不是合法 JSON：{exc}")
|
||
return report
|
||
|
||
inner = validate_pack(pack)
|
||
report.issues.extend(inner.issues)
|
||
return report
|