gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add EMOS auto retraining and gating

Browse Source
This commit is contained in:
2569718930@qq.com
2026-04-19 04:08:48 +08:00
parent d4cdd76235
commit 7c3b8ec459
5 changed files with 419 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+4
View File
@@ -56,6 +56,10 @@ METAR_CACHE_TTL_SEC=600
JMA_AMEDAS_CACHE_TTL_SEC=120
METEOBLUE_CACHE_TTL_SEC=7200
POLYWEATHER_PROBABILITY_ENGINE=emos_primary
POLYWEATHER_EMOS_AUTO_MIN_SAMPLES=50
POLYWEATHER_EMOS_AUTO_MAX_DELTA_CRPS=0
POLYWEATHER_EMOS_AUTO_MAX_DELTA_MAE=0.05
POLYWEATHER_EMOS_AUTO_MIN_DELTA_BUCKET_HIT_RATE=-0.05
POLYWEATHER_LGBM_ENABLED=false
POLYWEATHER_LGBM_MODEL_PATH=/app/artifacts/models/lgbm_daily_high.txt
POLYWEATHER_LGBM_SCHEMA_PATH=/app/artifacts/models/lgbm_daily_high_schema.json
.gitignore
+2
View File
@@ -11,6 +11,8 @@ data/historical/
data/cache/
data/models/
logs/
artifacts/probability_calibration/auto_retrain_report.json
artifacts/probability_calibration/candidates/
# Python
__pycache__/
docs/EMOS_TRAINING_REPORT_ZH.md
+60 -1
View File
@@ -99,7 +99,66 @@ POLYWEATHER_PROBABILITY_ENGINE=legacy
如果连续回归显示 EMOS 退化，应先切回 `emos_shadow`，保留 shadow 观测，再决定是否回退到 `legacy`。
## 8. 已验证
## 8. 自动重训
已新增自动重训编排脚本：
```text
python scripts\auto_retrain_probability_calibration.py
```
默认行为：
- 生成一个新的 EMOS candidate。
- 对 candidate 跑离线评估。
- 写入候选目录和门禁报告。
- 不覆盖线上 [default.json](/E:/web/PolyWeather/artifacts/probability_calibration/default.json)。
候选产物默认写入：
```text
artifacts/probability_calibration/candidates/<version>/
```
允许门禁通过后自动发布：
```text
python scripts\auto_retrain_probability_calibration.py --promote-if-passed --run-tests
```
门禁默认阈值：
- `POLYWEATHER_EMOS_AUTO_MIN_SAMPLES=50`
- `POLYWEATHER_EMOS_AUTO_MAX_DELTA_CRPS=0`
- `POLYWEATHER_EMOS_AUTO_MAX_DELTA_MAE=0.05`
- `POLYWEATHER_EMOS_AUTO_MIN_DELTA_BUCKET_HIT_RATE=-0.05`
说明：
- `CRPS` 不允许比 legacy 更差。
- `MAE` 最多允许轻微退化 `0.05`。
- `bucket_hit_rate` 只做软门槛，因为它对结算边界过于敏感。
- 如果发布，会先备份旧版 `default.json`。
Docker 手动触发：
```text
docker compose exec -T polyweather_web python scripts/auto_retrain_probability_calibration.py
```
Docker 允许门禁发布：
```text
docker compose exec -T polyweather_web python scripts/auto_retrain_probability_calibration.py --promote-if-passed --run-tests
```
建议后续挂到宿主机 `cron` 或 systemd timer：
```text
0 3 * * * cd /root/PolyWeather && docker compose exec -T polyweather_web python scripts/auto_retrain_probability_calibration.py --promote-if-passed --run-tests
```
## 9. 已验证
本次上线前已执行：
scripts/auto_retrain_probability_calibration.py
+311
View File
@@ -0,0 +1,311 @@
import argparse
import json
import os
import shutil
import subprocess
import sys
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if PROJECT_ROOT not in sys.path:
sys.path.insert(0, PROJECT_ROOT)
from src.analysis.probability_calibration import DEFAULT_CALIBRATION_FILE # noqa: E402
ARTIFACT_DIR = os.path.join(PROJECT_ROOT, "artifacts", "probability_calibration")
DEFAULT_CANDIDATE_ROOT = os.path.join(ARTIFACT_DIR, "candidates")
DEFAULT_DECISION_REPORT = os.path.join(ARTIFACT_DIR, "auto_retrain_report.json")
def _sf(value: Any) -> Optional[float]:
if value is None:
return None
try:
return float(value)
except Exception:
return None
def _env_float(name: str, default: float) -> float:
value = _sf(os.getenv(name))
return value if value is not None else default
def _env_int(name: str, default: int) -> int:
value = _sf(os.getenv(name))
return int(value) if value is not None else default
def _load_json(path: str) -> Dict[str, Any]:
try:
with open(path, "r", encoding="utf-8") as fh:
data = json.load(fh)
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _write_json(path: str, payload: Dict[str, Any]) -> None:
output_dir = os.path.dirname(os.path.abspath(path))
if output_dir:
os.makedirs(output_dir, exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
json.dump(payload, fh, ensure_ascii=False, indent=2)
def _run_python(args: List[str]) -> Dict[str, Any]:
command = [sys.executable, *args]
completed = subprocess.run(
command,
cwd=PROJECT_ROOT,
text=True,
capture_output=True,
check=False,
)
return {
"command": command,
"returncode": completed.returncode,
"stdout": completed.stdout,
"stderr": completed.stderr,
}
def _append_blocker(blockers: List[str], condition: bool, message: str) -> None:
if condition:
blockers.append(message)
def judge_candidate(
evaluation_report: Dict[str, Any],
*,
min_samples: int,
max_delta_crps: float,
max_delta_mae: float,
min_delta_bucket_hit_rate: float,
) -> Dict[str, Any]:
summary = evaluation_report.get("summary") or {}
delta = summary.get("delta") or {}
sample_count = int(summary.get("sample_count") or 0)
delta_crps = _sf(delta.get("crps"))
delta_mae = _sf(delta.get("mae"))
delta_hit = _sf(delta.get("bucket_hit_rate"))
blockers: List[str] = []
_append_blocker(
blockers,
sample_count < min_samples,
f"sample_count {sample_count} < {min_samples}",
)
_append_blocker(
blockers,
delta_crps is None or delta_crps > max_delta_crps,
f"delta_crps {delta_crps} > {max_delta_crps}",
)
_append_blocker(
blockers,
delta_mae is None or delta_mae > max_delta_mae,
f"delta_mae {delta_mae} > {max_delta_mae}",
)
_append_blocker(
blockers,
delta_hit is None or delta_hit < min_delta_bucket_hit_rate,
f"delta_bucket_hit_rate {delta_hit} < {min_delta_bucket_hit_rate}",
)
return {
"decision": "promote" if not blockers else "hold",
"ready_for_promotion": not blockers,
"blocking_reasons": blockers,
"thresholds": {
"min_samples": min_samples,
"max_delta_crps": max_delta_crps,
"max_delta_mae": max_delta_mae,
"min_delta_bucket_hit_rate": min_delta_bucket_hit_rate,
},
"metrics": {
"sample_count": sample_count,
"delta_crps": delta_crps,
"delta_mae": delta_mae,
"delta_bucket_hit_rate": delta_hit,
},
}
def _promote(candidate_path: str, target_path: str) -> str:
target_dir = os.path.dirname(os.path.abspath(target_path))
if target_dir:
os.makedirs(target_dir, exist_ok=True)
backup_path = os.path.join(
target_dir,
"default.backup-{ts}.json".format(
ts=datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
),
)
if os.path.exists(target_path):
shutil.copy2(target_path, backup_path)
shutil.copy2(candidate_path, target_path)
return backup_path
def main() -> int:
parser = argparse.ArgumentParser(
description="Train an EMOS candidate, evaluate it, and optionally promote it behind gates."
)
parser.add_argument("--candidate-root", default=DEFAULT_CANDIDATE_ROOT)
parser.add_argument("--target", default=DEFAULT_CALIBRATION_FILE)
parser.add_argument("--decision-output", default=DEFAULT_DECISION_REPORT)
parser.add_argument(
"--promote-if-passed",
action="store_true",
help="Copy the candidate over the active calibration file only if gates pass.",
)
parser.add_argument(
"--run-tests",
action="store_true",
help="Run focused probability tests before promotion.",
)
parser.add_argument(
"--min-samples",
type=int,
default=_env_int("POLYWEATHER_EMOS_AUTO_MIN_SAMPLES", 50),
)
parser.add_argument(
"--max-delta-crps",
type=float,
default=_env_float("POLYWEATHER_EMOS_AUTO_MAX_DELTA_CRPS", 0.0),
help="Candidate EMOS CRPS may not be worse than legacy by more than this.",
)
parser.add_argument(
"--max-delta-mae",
type=float,
default=_env_float("POLYWEATHER_EMOS_AUTO_MAX_DELTA_MAE", 0.05),
)
parser.add_argument(
"--min-delta-bucket-hit-rate",
type=float,
default=_env_float("POLYWEATHER_EMOS_AUTO_MIN_DELTA_BUCKET_HIT_RATE", -0.05),
help="Soft guard only; bucket hit rate is boundary-sensitive.",
)
args = parser.parse_args()
version = "emos-auto-{ts}".format(
ts=datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
)
candidate_dir = os.path.join(args.candidate_root, version)
os.makedirs(candidate_dir, exist_ok=True)
candidate_path = os.path.join(candidate_dir, "default.json")
evaluation_path = os.path.join(candidate_dir, "evaluation_report.json")
decision_path = os.path.join(candidate_dir, "decision_report.json")
fit_result = _run_python(
[
"scripts/fit_probability_calibration.py",
"--output",
candidate_path,
"--version",
version,
]
)
if fit_result["returncode"] != 0:
payload = {
"ok": False,
"version": version,
"stage": "fit",
"fit": fit_result,
}
_write_json(decision_path, payload)
_write_json(args.decision_output, payload)
print(json.dumps(payload, ensure_ascii=False, indent=2))
return fit_result["returncode"] or 1
eval_result = _run_python(
[
"scripts/evaluate_probability_calibration.py",
"--calibration-file",
candidate_path,
"--output",
evaluation_path,
]
)
if eval_result["returncode"] != 0:
payload = {
"ok": False,
"version": version,
"stage": "evaluate",
"candidate_path": candidate_path,
"fit": fit_result,
"evaluate": eval_result,
}
_write_json(decision_path, payload)
_write_json(args.decision_output, payload)
print(json.dumps(payload, ensure_ascii=False, indent=2))
return eval_result["returncode"] or 1
evaluation_report = _load_json(evaluation_path)
decision = judge_candidate(
evaluation_report,
min_samples=args.min_samples,
max_delta_crps=args.max_delta_crps,
max_delta_mae=args.max_delta_mae,
min_delta_bucket_hit_rate=args.min_delta_bucket_hit_rate,
)
test_result = None
if args.run_tests and decision["ready_for_promotion"]:
test_result = _run_python(
[
"-m",
"pytest",
"tests/test_probability_calibration.py",
"tests/test_probability_rollout.py",
"tests/test_trend_engine.py",
]
)
if test_result["returncode"] != 0:
decision["decision"] = "hold"
decision["ready_for_promotion"] = False
decision.setdefault("blocking_reasons", []).append(
"focused tests failed"
)
promoted = False
backup_path = None
if args.promote_if_passed and decision["ready_for_promotion"]:
backup_path = _promote(candidate_path, args.target)
promoted = True
payload = {
"ok": True,
"version": version,
"generated_at": datetime.now(timezone.utc).isoformat(),
"candidate_dir": candidate_dir,
"candidate_path": candidate_path,
"evaluation_path": evaluation_path,
"target_path": args.target,
"promote_requested": bool(args.promote_if_passed),
"promoted": promoted,
"backup_path": backup_path,
"decision": decision,
"fit": fit_result,
"evaluate": eval_result,
"tests": test_result,
}
_write_json(decision_path, payload)
_write_json(args.decision_output, payload)
print(json.dumps(payload["decision"], ensure_ascii=False, indent=2))
print(f"candidate: {candidate_path}")
print(f"evaluation: {evaluation_path}")
print(f"decision: {decision_path}")
if promoted:
print(f"promoted to {args.target}; backup: {backup_path}")
elif args.promote_if_passed:
print("not promoted; gates did not pass")
else:
print("not promoted; run with --promote-if-passed to allow gated promotion")
return 0
if __name__ == "__main__":
raise SystemExit(main())
tests/test_auto_retrain_probability_calibration.py
+42
View File
@@ -0,0 +1,42 @@
from scripts.auto_retrain_probability_calibration import judge_candidate
def _report(sample_count=80, crps=-0.1, mae=0.0, hit=0.0):
return {
"summary": {
"sample_count": sample_count,
"delta": {
"crps": crps,
"mae": mae,
"bucket_hit_rate": hit,
},
}
}
def test_candidate_gate_promotes_when_metrics_pass():
decision = judge_candidate(
_report(),
min_samples=50,
max_delta_crps=0.0,
max_delta_mae=0.05,
min_delta_bucket_hit_rate=-0.05,
)
assert decision["decision"] == "promote"
assert decision["ready_for_promotion"] is True
assert decision["blocking_reasons"] == []
def test_candidate_gate_holds_when_metrics_regress():
decision = judge_candidate(
_report(sample_count=40, crps=0.1, mae=0.2, hit=-0.2),
min_samples=50,
max_delta_crps=0.0,
max_delta_mae=0.05,
min_delta_bucket_hit_rate=-0.05,
)
assert decision["decision"] == "hold"
assert decision["ready_for_promotion"] is False
assert len(decision["blocking_reasons"]) == 4