diff --git a/.env.example b/.env.example index 7faf6e13..4c5b07e1 100644 --- a/.env.example +++ b/.env.example @@ -60,6 +60,9 @@ 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 +# Optional: cap recent probability snapshots used by EMOS retraining if the +# runtime SQLite snapshot table grows too large. +# POLYWEATHER_EMOS_TRAINING_SNAPSHOT_LIMIT=20000 # Optional: set this to a writable runtime path if you want auto retrain to # promote new EMOS parameters without rebuilding the image. # POLYWEATHER_PROBABILITY_CALIBRATION_FILE=/var/lib/polyweather/probability_calibration/default.json diff --git a/docs/EMOS_TRAINING_REPORT_ZH.md b/docs/EMOS_TRAINING_REPORT_ZH.md index c247b91b..6647dcf2 100644 --- a/docs/EMOS_TRAINING_REPORT_ZH.md +++ b/docs/EMOS_TRAINING_REPORT_ZH.md @@ -149,7 +149,7 @@ python scripts\auto_retrain_probability_calibration.py --promote-if-passed --run Docker 手动触发： ```text -docker compose exec -T polyweather_web python scripts/auto_retrain_probability_calibration.py +docker compose exec -T polyweather_web python scripts/auto_retrain_probability_calibration.py --verbose ``` 查看最新报告： @@ -161,7 +161,19 @@ docker compose exec -T polyweather_web cat /var/lib/polyweather/probability_cali Docker 允许门禁发布： ```text -docker compose exec -T polyweather_web python scripts/auto_retrain_probability_calibration.py --promote-if-passed --run-tests +docker compose exec -T polyweather_web python scripts/auto_retrain_probability_calibration.py --verbose --promote-if-passed --run-tests +``` + +如果线上 SQLite 的 `probability_training_snapshots_store` 已经很大，可以先限制最近 N 条快照： + +```text +docker compose exec -T polyweather_web python scripts/auto_retrain_probability_calibration.py --verbose --snapshot-limit 20000 +``` + +也可以放到 `.env`： + +```text +POLYWEATHER_EMOS_TRAINING_SNAPSHOT_LIMIT=20000 ``` 如果希望自动发布不需要重建镜像，需要在 `.env` 中指定可写参数文件： diff --git a/scripts/auto_retrain_probability_calibration.py b/scripts/auto_retrain_probability_calibration.py index be8b9ebe..b2bf2140 100644 --- a/scripts/auto_retrain_probability_calibration.py +++ b/scripts/auto_retrain_probability_calibration.py @@ -67,8 +67,21 @@ def _write_json(path: str, payload: Dict[str, Any]) -> None: json.dump(payload, fh, ensure_ascii=False, indent=2) -def _run_python(args: List[str]) -> Dict[str, Any]: +def _run_python(args: List[str], *, stream: bool = False) -> Dict[str, Any]: command = [sys.executable, *args] + if stream: + completed = subprocess.run( + command, + cwd=PROJECT_ROOT, + text=True, + check=False, + ) + return { + "command": command, + "returncode": completed.returncode, + "stdout": "", + "stderr": "", + } completed = subprocess.run( command, cwd=PROJECT_ROOT, @@ -178,6 +191,17 @@ def main() -> int: action="store_true", help="Run focused probability tests before promotion.", ) + parser.add_argument( + "--verbose", + action="store_true", + help="Print child script progress while training and evaluating.", + ) + parser.add_argument( + "--snapshot-limit", + type=int, + default=_env_int("POLYWEATHER_EMOS_TRAINING_SNAPSHOT_LIMIT", 0), + help="Optional max number of recent probability snapshots to load from SQLite.", + ) parser.add_argument( "--min-samples", type=int, @@ -211,14 +235,20 @@ def main() -> int: evaluation_path = os.path.join(candidate_dir, "evaluation_report.json") decision_path = os.path.join(candidate_dir, "decision_report.json") - fit_result = _run_python( - [ + fit_args = [ "scripts/fit_probability_calibration.py", "--output", candidate_path, "--version", version, - ] + ] + if args.verbose: + fit_args.append("--verbose") + if args.snapshot_limit and args.snapshot_limit > 0: + fit_args.extend(["--snapshot-limit", str(args.snapshot_limit)]) + fit_result = _run_python( + fit_args, + stream=args.verbose, ) if fit_result["returncode"] != 0: payload = { @@ -239,7 +269,8 @@ def main() -> int: candidate_path, "--output", evaluation_path, - ] + ], + stream=args.verbose, ) if eval_result["returncode"] != 0: payload = { diff --git a/scripts/fit_probability_calibration.py b/scripts/fit_probability_calibration.py index 530fa7df..0c2f73fc 100644 --- a/scripts/fit_probability_calibration.py +++ b/scripts/fit_probability_calibration.py @@ -34,6 +34,21 @@ def _sf(value): return None +def _env_int(name, default=None): + try: + value = os.getenv(name) + if value is None or str(value).strip() == "": + return default + return int(value) + except Exception: + return default + + +def _log(enabled, message): + if enabled: + print(f"[fit_probability_calibration] {message}", flush=True) + + def _load_json_if_exists(path): if not path or not os.path.exists(path): return {} @@ -96,9 +111,28 @@ def _load_training_feature_history(): return {} -def _load_snapshot_rows(path): +def _load_snapshot_rows(path, limit=None): if get_state_storage_mode() == STATE_STORAGE_SQLITE: - return ProbabilitySnapshotRepository().load_all_rows() + repo = ProbabilitySnapshotRepository() + if limit is not None and int(limit) > 0: + with repo.db.connect() as conn: + rows = conn.execute( + """ + SELECT payload_json + FROM probability_training_snapshots_store + ORDER BY id DESC + LIMIT ? + """, + (int(limit),), + ).fetchall() + out = [] + for row in reversed(rows): + try: + out.append(json.loads(row["payload_json"])) + except Exception: + continue + return out + return repo.load_all_rows() rows = [] if not path or not os.path.exists(path): return rows @@ -410,14 +444,42 @@ def main(): default=None, help="Optional explicit calibration version.", ) + parser.add_argument( + "--snapshot-limit", + type=int, + default=_env_int("POLYWEATHER_EMOS_TRAINING_SNAPSHOT_LIMIT"), + help="Optional max number of recent probability snapshots to load from SQLite.", + ) + parser.add_argument( + "--verbose", + action="store_true", + help="Print data loading and fitting progress.", + ) args = parser.parse_args() + _log(args.verbose, "loading daily records") history = _load_history_with_fallback(args.history_file) + _log(args.verbose, f"loaded daily record cities={len(history or {})}") + _log(args.verbose, "loading training feature history") training_feature_history = _load_training_feature_history() + _log(args.verbose, f"loaded training feature cities={len(training_feature_history or {})}") + _log(args.verbose, "loading truth history") truth_history = _load_truth_history() + _log(args.verbose, f"loaded truth cities={len(truth_history or {})}") + _log(args.verbose, "loading settlement history") settlement_history = _load_json_if_exists(args.settlement_history) - snapshot_rows = _load_snapshot_rows(args.snapshot_file) + _log(args.verbose, f"loaded settlement history cities={len(settlement_history or {})}") + _log( + args.verbose, + "loading probability snapshots" + + (f" limit={args.snapshot_limit}" if args.snapshot_limit else ""), + ) + snapshot_rows = _load_snapshot_rows(args.snapshot_file, limit=args.snapshot_limit) + _log(args.verbose, f"loaded probability snapshots={len(snapshot_rows or [])}") + _log(args.verbose, "loading legacy training archive") legacy_training_samples = _load_legacy_training_samples() + _log(args.verbose, f"loaded legacy training samples={len(legacy_training_samples or [])}") + _log(args.verbose, "extracting EMOS samples") samples, filled_actual_from_history = _extract_samples( history, training_feature_history=training_feature_history, @@ -426,6 +488,7 @@ def main(): snapshot_rows=snapshot_rows, ) samples = merge_samples_with_legacy_archive(samples, legacy_training_samples) + _log(args.verbose, f"fitting calibration samples={len(samples or [])}") calibration = fit_calibration(samples, version=args.version) if not samples: calibration = default_calibration_payload( @@ -447,6 +510,7 @@ def main(): with open(args.output, "w", encoding="utf-8") as fh: json.dump(calibration, fh, ensure_ascii=False, indent=2) + _log(args.verbose, "done") print( "saved calibration to {path} with {count} samples".format( path=args.output,