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
|