diff --git a/src/analysis/deb_algorithm.py b/src/analysis/deb_algorithm.py index 291f20a1..1e62c7b4 100644 --- a/src/analysis/deb_algorithm.py +++ b/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 diff --git a/src/analysis/trend_engine.py b/src/analysis/trend_engine.py index 113acde1..0371c1f1 100644 --- a/src/analysis/trend_engine.py +++ b/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 diff --git a/tests/test_deb_training_truth.py b/tests/test_deb_training_truth.py new file mode 100644 index 00000000..38be41d5 --- /dev/null +++ b/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