92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
"""Validation for persisted ML scoring artifacts."""
|
|
|
|
import math
|
|
|
|
from scoring.policy import SCORE_VERSION
|
|
|
|
ML_ARTIFACT_SCHEMA_VERSION = 2
|
|
REQUIRED_WEIGHT_KEYS = frozenset({
|
|
"puell_multiple",
|
|
"mvrv_zscore",
|
|
"reserve_risk",
|
|
"rhodl_ratio",
|
|
"nupl",
|
|
"fear_greed",
|
|
"drawdown",
|
|
"pct_above_200w_sma",
|
|
"pct_above_lth_rp",
|
|
})
|
|
|
|
|
|
def _weights_valid(weights):
|
|
if not isinstance(weights, dict) or not REQUIRED_WEIGHT_KEYS.issubset(weights):
|
|
return False
|
|
values = [weights[key] for key in REQUIRED_WEIGHT_KEYS]
|
|
return all(
|
|
isinstance(value, (int, float))
|
|
and not isinstance(value, bool)
|
|
and math.isfinite(value)
|
|
and value >= 0
|
|
for value in values
|
|
) and sum(values) > 0
|
|
|
|
|
|
def _has_oos_fold_weights(artifact):
|
|
folds = artifact.get("cv_results", {}).get("folds", [])
|
|
if not isinstance(folds, list) or not folds:
|
|
return False
|
|
for fold in folds:
|
|
validation_range = fold.get("date_ranges", {}).get("validation")
|
|
if not validation_range or not _weights_valid(fold.get("weights")):
|
|
return False
|
|
return True
|
|
|
|
|
|
def validate_ml_artifact(artifact):
|
|
"""Return machine-readable validity and provenance for an ML artifact."""
|
|
errors = []
|
|
if not isinstance(artifact, dict):
|
|
artifact = {}
|
|
errors.append("artifact_object")
|
|
|
|
schema_version = artifact.get("artifact_schema_version")
|
|
if schema_version != ML_ARTIFACT_SCHEMA_VERSION:
|
|
errors.append("artifact_schema_version")
|
|
|
|
score_version = artifact.get("score_version")
|
|
if score_version != SCORE_VERSION:
|
|
errors.append("score_version")
|
|
|
|
if not _weights_valid(artifact.get("weights")):
|
|
errors.append("weights")
|
|
|
|
provenance = artifact.get("provenance")
|
|
if not isinstance(provenance, dict):
|
|
provenance = {}
|
|
errors.append("provenance")
|
|
else:
|
|
required_provenance = {
|
|
"validation_method",
|
|
"label_horizon_days",
|
|
"weight_scope",
|
|
"training_date_range",
|
|
"trained_at",
|
|
}
|
|
if not required_provenance.issubset(provenance):
|
|
errors.append("provenance")
|
|
if provenance.get("validation_method") != "purged_expanding_window":
|
|
errors.append("purged_validation")
|
|
if provenance.get("label_horizon_days") != 365:
|
|
errors.append("label_horizon_days")
|
|
if provenance.get("weight_scope") != "full_history_fit":
|
|
errors.append("weight_scope")
|
|
|
|
return {
|
|
"valid": not errors,
|
|
"schema_version": schema_version,
|
|
"score_version": score_version,
|
|
"weight_scope": provenance.get("weight_scope"),
|
|
"has_oos_fold_weights": _has_oos_fold_weights(artifact),
|
|
"errors": list(dict.fromkeys(errors)),
|
|
}
|