Add EMOS retraining snapshot limit and verbose logging

.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

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` 中指定可写参数文件：

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 = {

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,