Fix DEB training truth finality handling

src/analysis/deb_algorithm.py

@@ -880,6 +880,7 @@ def update_daily_record(
     shadow_probabilities=None,
     calibration_summary=None,
     hourly_error=None,
+    actual_is_final=True,
 ):
     """
     保存/更新某城市某天的各个模型预报与最终实测值
@@ -986,7 +987,7 @@ def update_daily_record(
 
     next_mu = round(mu, 2) if mu is not None else None
     if (
-        old_actual == actual_high
+        old_actual == (actual_high if actual_is_final else old_actual)
         and old_forecasts == merged_forecasts
         and (deb_prediction is None or old_deb == deb_prediction)
         and (mu is None or old_mu == next_mu)
@@ -1007,15 +1008,21 @@ def update_daily_record(
     ):
         return
 
-    # actual_high 应该是日内最高温，理论上不应下降；防止异常写入覆盖已确认高值
-    if old_actual is not None and actual_high is not None:
-        try:
-            actual_high = max(float(old_actual), float(actual_high))
-        except Exception:
-            pass
+    # Only final settlement truth may update actual_high. Intraday max_so_far is
+    # useful context, but treating it as settled truth poisons DEB training.
+    next_actual_high = old_actual
+    if actual_is_final:
+        next_actual_high = actual_high
+        # actual_high 应该是日内最高温，理论上不应下降；防止异常写入覆盖已确认高值
+        if old_actual is not None and actual_high is not None:
+            try:
+                next_actual_high = max(float(old_actual), float(actual_high))
+            except Exception:
+                pass
 
     existing["forecasts"] = merged_forecasts
-    existing["actual_high"] = actual_high
+    if actual_is_final or next_actual_high is not None:
+        existing["actual_high"] = next_actual_high
     if deb_prediction is not None:
         existing["deb_prediction"] = deb_prediction
     if mu is not None:
@@ -1031,16 +1038,16 @@ def update_daily_record(
     if next_hourly_error is not None:
         existing["hourly_error"] = next_hourly_error
 
-    if actual_high is not None:
+    if actual_is_final and next_actual_high is not None:
         try:
             _persist_truth_record(
                 city_name,
                 date_str,
-                float(actual_high),
+                float(next_actual_high),
                 updated_by="runtime:update_daily_record",
                 reason="update_daily_record",
                 source_payload={
-                    "actual_high": actual_high,
+                    "actual_high": next_actual_high,
                     "deb_prediction": deb_prediction,
                     "mu": next_mu,
                 },
@@ -1161,8 +1168,12 @@ def calculate_dynamic_weight_components(
     sorted_dates = sorted(city_data.keys(), reverse=True)
     today_str = datetime.now().strftime("%Y-%m-%d")
     available_days = sum(
-        1 for d in sorted_dates
-        if d != today_str and city_data[d].get("actual_high") is not None
+        1
+        for d in sorted_dates
+        if d != today_str
+        and city_data[d].get("actual_high") is not None
+        and isinstance(city_data[d].get("forecasts"), dict)
+        and any(v is not None for v in city_data[d].get("forecasts", {}).values())
     )
 
     # ── 改进3: 自适应 lookback — 数据多的城市用更多历史 ──
@@ -1187,6 +1198,7 @@ def calculate_dynamic_weight_components(
         if actual is None:
             continue
 
+        usable_day = False
         for model in forecasts.keys():
             if model in past_forecasts and past_forecasts[model] is not None:
                 try:
@@ -1194,6 +1206,7 @@ def calculate_dynamic_weight_components(
                     av = float(actual)
                 except (TypeError, ValueError):
                     continue
+                usable_day = True
                 # Track signed error for bias
                 model_biases[model] += (pv - av)  # positive = model overpredicts
                 bias_samples[model] += 1
@@ -1208,6 +1221,8 @@ def calculate_dynamic_weight_components(
                 decay_weight = decay_factor ** days_used
                 errors[model].append((blended_error, decay_weight))
 
+        if not usable_day:
+            continue
         days_used += 1
         if days_used >= effective_lookback:
             break

src/analysis/trend_engine.py

@@ -974,6 +974,7 @@ def analyze_weather_trend(
             deb_prediction=_deb_to_save,
             mu=mu,
             probabilities=_prob_list,
+            actual_is_final=False,
         )
     except Exception:
         pass

tests/test_deb_training_truth.py

@@ -0,0 +1,88 @@
+from src.database.runtime_state import (
+    DailyRecordRepository,
+    RuntimeStateDB,
+    TrainingFeatureRecordRepository,
+    TruthRecordRepository,
+)
+
+
+def _wire_runtime_repos(monkeypatch, tmp_path):
+    from src.analysis import deb_algorithm
+
+    db = RuntimeStateDB(str(tmp_path / "polyweather.db"))
+    daily_repo = DailyRecordRepository(db)
+    truth_repo = TruthRecordRepository(db)
+    feature_repo = TrainingFeatureRecordRepository(db)
+    monkeypatch.setattr(deb_algorithm, "_daily_record_repo", daily_repo)
+    monkeypatch.setattr(deb_algorithm, "_truth_record_repo", truth_repo)
+    monkeypatch.setattr(deb_algorithm, "_training_feature_repo", feature_repo)
+    monkeypatch.setattr(deb_algorithm, "_history_cache", {})
+    monkeypatch.setattr(deb_algorithm, "_history_mtime", 0)
+    monkeypatch.setenv("POLYWEATHER_STATE_STORAGE_MODE", "sqlite")
+    return db
+
+
+def test_intraday_daily_record_does_not_persist_unfinal_actual_as_truth(monkeypatch, tmp_path):
+    from src.analysis.deb_algorithm import update_daily_record
+
+    db = _wire_runtime_repos(monkeypatch, tmp_path)
+
+    update_daily_record(
+        "ankara",
+        "2026-06-24",
+        {"ECMWF": 28.0, "GFS": 27.0},
+        24.0,
+        deb_prediction=27.5,
+        mu=27.2,
+        actual_is_final=False,
+    )
+
+    with db.connect() as conn:
+        daily = conn.execute(
+            """
+            SELECT actual_high, deb_prediction, mu, payload_json
+            FROM daily_records_store
+            WHERE city = ? AND target_date = ?
+            """,
+            ("ankara", "2026-06-24"),
+        ).fetchone()
+        truth_count = conn.execute(
+            """
+            SELECT count(*) AS n
+            FROM truth_records_store
+            WHERE city = ? AND target_date = ?
+            """,
+            ("ankara", "2026-06-24"),
+        ).fetchone()["n"]
+
+    assert daily is not None
+    assert daily["actual_high"] is None
+    assert daily["deb_prediction"] == 27.5
+    assert daily["mu"] == 27.2
+    assert truth_count == 0
+
+
+def test_dynamic_weights_skip_actual_only_rows_when_selecting_recent_training_days(monkeypatch):
+    from src.analysis.deb_algorithm import calculate_dynamic_weight_components
+
+    history = {"ankara": {}}
+    for day in range(23, 15, -1):
+        history["ankara"][f"2026-06-{day:02d}"] = {"actual_high": 20.0}
+    history["ankara"]["2026-06-15"] = {
+        "actual_high": 20.0,
+        "forecasts": {"ECMWF": 20.0, "GFS": 30.0},
+    }
+    history["ankara"]["2026-06-14"] = {
+        "actual_high": 20.0,
+        "forecasts": {"ECMWF": 20.0, "GFS": 30.0},
+    }
+    monkeypatch.setattr("src.analysis.deb_algorithm.load_history", lambda _path: history)
+
+    components = calculate_dynamic_weight_components(
+        "ankara",
+        {"ECMWF": 20.0, "GFS": 30.0},
+        lookback_days=7,
+    )
+
+    assert components["weights"]["ECMWF"] > components["weights"]["GFS"]
+    assert components["prediction"] < 25.0