diff --git a/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts index 831ddb39..9023e569 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts @@ -310,6 +310,7 @@ export function runTests() { ["chengdu", "02L/20R"], ["chongqing", "20R/02L"], ["wuhan", "04/22"], + ["qingdao", "16/34"], ["seoul", "15R/33L"], ] as const; settlementRunwayCases.forEach(([city, settlementRwy]) => { diff --git a/src/analysis/deb_algorithm.py b/src/analysis/deb_algorithm.py index bdd1a345..64672aaa 100644 --- a/src/analysis/deb_algorithm.py +++ b/src/analysis/deb_algorithm.py @@ -1092,6 +1092,27 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec 返回: blended_high (融合预报值), weights_info (权重展示字符串) """ + components = calculate_dynamic_weight_components( + city_name, + current_forecasts, + lookback_days=lookback_days, + decay_factor=decay_factor, + ) + forecasts = components.get("forecasts") or {} + weights = components.get("weights") or {} + if not forecasts or not weights: + return components.get("prediction"), components.get("weights_info") or "暂无模型数据" + blended_high = sum(forecasts[m] * weights[m] for m in weights if m in forecasts) + return round(blended_high, 1), components.get("weights_info") or "权重计算异常" + + +def calculate_dynamic_weight_components( + city_name, + current_forecasts, + lookback_days=7, + decay_factor=0.85, +): + """Return DEB forecast representatives and model weights for reuse by hourly paths.""" project_root = os.path.dirname( os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ) @@ -1105,23 +1126,41 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec if v is not None and not _is_excluded_model_name(k) ] ) - current_forecasts = _collapse_forecasts_for_deb(current_forecasts) - dedup_note = "家族去重" if raw_forecast_count > len(current_forecasts) else "" + forecasts = _collapse_forecasts_for_deb(current_forecasts) + dedup_note = "家族去重" if raw_forecast_count > len(forecasts) else "" + valid_vals = [v for v in forecasts.values() if v is not None] + if not valid_vals: + return { + "prediction": None, + "weights": {}, + "forecasts": {}, + "maes": {}, + "weights_info": "暂无模型数据", + "days_used": 0, + "dedup_note": dedup_note, + } + + def _equal_weight_result(note: str, days_used: int): + weights = {model: 1.0 / len(forecasts) for model in forecasts} + weights_info = f"{note} | {dedup_note}" if dedup_note else note + prediction = sum(forecasts[m] * weights[m] for m in weights) + return { + "prediction": round(prediction, 1), + "weights": weights, + "forecasts": forecasts, + "maes": {}, + "weights_info": weights_info, + "days_used": days_used, + "dedup_note": dedup_note, + } if city_name not in data or not data[city_name]: - valid_vals = [v for v in current_forecasts.values() if v is not None] - if not valid_vals: - return None, "暂无模型数据" - avg = sum(valid_vals) / len(valid_vals) - note = "等权平均(历史数据不足)" - if dedup_note: - note = f"{note} | {dedup_note}" - return round(avg, 1), note + return _equal_weight_result("等权平均(历史数据不足)", 0) city_data = data[city_name] sorted_dates = sorted(city_data.keys(), reverse=True) - errors: dict = {model: [] for model in current_forecasts.keys()} + errors: dict = {model: [] for model in forecasts.keys()} days_used = 0 for date_str in sorted_dates: if date_str == datetime.now().strftime("%Y-%m-%d"): @@ -1137,7 +1176,7 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec decay_weight = decay_factor ** days_used - for model in current_forecasts.keys(): + for model in forecasts.keys(): if model in past_forecasts and past_forecasts[model] is not None: try: pv = float(past_forecasts[model]) @@ -1145,7 +1184,6 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec except (TypeError, ValueError): continue daily_error = abs(pv - av) - # Blend with hourly error when available h_err = ( past_hourly_error.get(model) if isinstance(past_hourly_error, dict) @@ -1159,16 +1197,8 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec break if days_used < 2: - valid_vals = [v for v in current_forecasts.values() if v is not None] - if not valid_vals: - return None, f"暂无有效模型数据(由于仅{days_used}天历史)" - avg = sum(valid_vals) / len(valid_vals) - note = f"等权平均(由于仅{days_used}天历史)" - if dedup_note: - note = f"{note} | {dedup_note}" - return round(avg, 1), note + return _equal_weight_result(f"等权平均(由于仅{days_used}天历史)", days_used) - # 计算加权 MAE（时间衰减） maes = {} for model, err_weighted in errors.items(): if err_weighted: @@ -1180,25 +1210,27 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec else: maes[model] = 2.0 - # 计算权重（用 MAE 的倒数，误差越小权重越大；加 0.1 防止除以0） inverse_errors = { m: 1.0 / (mae + 0.1) for m, mae in maes.items() - if current_forecasts.get(m) is not None + if forecasts.get(m) is not None } total_inv = sum(inverse_errors.values()) if total_inv == 0: - return None, "权重计算异常" + return { + "prediction": None, + "weights": {}, + "forecasts": forecasts, + "maes": maes, + "weights_info": "权重计算异常", + "days_used": days_used, + "dedup_note": dedup_note, + } weights = {m: inv / total_inv for m, inv in inverse_errors.items()} + blended_high = sum(forecasts[m] * weights[m] for m in weights) - # 计算加权最高温 - blended_high = 0.0 - for m in weights.keys(): - blended_high += current_forecasts[m] * weights[m] - - # 格式化权重信息，挑选前权重最高的2-3个模型展示 sorted_models = sorted(weights.items(), key=lambda x: x[1], reverse=True) weight_str_parts = [] for m, w in sorted_models[:3]: @@ -1206,7 +1238,15 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec if dedup_note: weight_str_parts.append(dedup_note) - return round(blended_high, 1), " | ".join(weight_str_parts) + return { + "prediction": round(blended_high, 1), + "weights": weights, + "forecasts": forecasts, + "maes": maes, + "weights_info": " | ".join(weight_str_parts), + "days_used": days_used, + "dedup_note": dedup_note, + } def calculate_deb_prediction( diff --git a/src/analysis/deb_hourly_consensus.py b/src/analysis/deb_hourly_consensus.py new file mode 100644 index 00000000..6c7c8307 --- /dev/null +++ b/src/analysis/deb_hourly_consensus.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from src.analysis.deb_algorithm import calculate_dynamic_weight_components + +DEB_HOURLY_CONSENSUS_VERSION = "deb_hourly_consensus.v1" + + +def _to_float(value: Any) -> Optional[float]: + try: + result = float(value) + except (TypeError, ValueError): + return None + if result != result: + return None + return result + + +def _time_part(value: Any) -> str: + text = str(value or "").strip() + if "T" in text: + text = text.split("T", 1)[1] + if " " in text: + text = text.rsplit(" ", 1)[-1] + return text[:5] + + +def _matches_local_date(value: Any, local_date: Optional[str]) -> bool: + if not local_date: + return True + text = str(value or "").strip() + if "T" not in text and " " not in text: + return True + return text.startswith(local_date) + + +def _weighted_value_at_index( + index: int, + hourly_forecasts: Dict[str, Any], + weights: Dict[str, float], +) -> Optional[float]: + weighted_sum = 0.0 + weight_sum = 0.0 + for model_name, model_weight in weights.items(): + series = hourly_forecasts.get(model_name) + if not isinstance(series, (list, tuple)) or index >= len(series): + continue + value = _to_float(series[index]) + if value is None: + continue + weighted_sum += value * model_weight + weight_sum += model_weight + if weight_sum <= 0: + return None + return weighted_sum / weight_sum + + +def build_deb_hourly_consensus_path( + *, + city: str, + hourly_times: List[Any], + hourly_forecasts: Dict[str, Any], + daily_forecasts: Dict[str, Any], + deb_prediction: Optional[float], + local_date: Optional[str] = None, +) -> Optional[Dict[str, Any]]: + if not hourly_times or not isinstance(hourly_forecasts, dict): + return None + + components = calculate_dynamic_weight_components(city, daily_forecasts or {}) + weights = { + model: float(weight) + for model, weight in (components.get("weights") or {}).items() + if model in hourly_forecasts and _to_float(weight) is not None + } + if not weights: + return None + + times: List[str] = [] + raw_temps: List[Optional[float]] = [] + for idx, raw_time in enumerate(hourly_times): + if not _matches_local_date(raw_time, local_date): + continue + value = _weighted_value_at_index(idx, hourly_forecasts, weights) + if value is None: + continue + times.append(_time_part(raw_time)) + raw_temps.append(round(value, 3)) + + numeric_raw = [value for value in raw_temps if value is not None] + if not times or not numeric_raw: + return None + + deb_value = _to_float(deb_prediction) + anchor_adjustment = 0.0 + if deb_value is not None: + anchor_adjustment = deb_value - max(numeric_raw) + temps = [ + round(value + anchor_adjustment, 1) if value is not None else None + for value in raw_temps + ] + + return { + "version": DEB_HOURLY_CONSENSUS_VERSION, + "source": DEB_HOURLY_CONSENSUS_VERSION, + "base_source": "multi_model_hourly_deb_weights", + "times": times, + "temps": temps, + "raw_temps": [round(value, 1) if value is not None else None for value in raw_temps], + "weights": {model: round(weight, 4) for model, weight in weights.items()}, + "weights_info": components.get("weights_info") or "", + "anchor_adjustment": round(anchor_adjustment, 3), + } diff --git a/src/analysis/deb_hourly_correction.py b/src/analysis/deb_hourly_correction.py index 29a50bb9..1c799d19 100644 --- a/src/analysis/deb_hourly_correction.py +++ b/src/analysis/deb_hourly_correction.py @@ -280,6 +280,7 @@ def build_deb_hourly_path( peak_first_h: Optional[int], peak_last_h: Optional[int], corrector: HourlyPeakCorrector, + base_source: str = "hourly_plus_deb_offset", ) -> Dict[str, Any]: deb_value = _to_float(deb_prediction) numeric_base = [_to_float(value) for value in hourly_temps] @@ -305,7 +306,7 @@ def build_deb_hourly_path( "version": DEB_HOURLY_PEAK_CORRECTED_VERSION, "times": applied["times"], "temps": applied["temps"], - "base_source": "hourly_plus_deb_offset", + "base_source": base_source, "base_offset": round(offset, 3), "correction": { "version": applied["version"], diff --git a/src/analysis/trend_engine.py b/src/analysis/trend_engine.py index d0abf271..1404fa1e 100644 --- a/src/analysis/trend_engine.py +++ b/src/analysis/trend_engine.py @@ -16,6 +16,7 @@ from src.analysis.deb_algorithm import ( update_daily_record, _is_excluded_model_name, ) +from src.analysis.deb_hourly_consensus import build_deb_hourly_consensus_path from src.analysis.settlement_rounding import apply_city_settlement, is_exact_settlement_city from src.data_collection.city_registry import CITY_REGISTRY from src.data_collection.city_risk_profiles import get_city_risk_profile @@ -81,6 +82,29 @@ def _resolve_peak_hours( open_meteo_peak: Optional[Any] = None, ) -> List[str]: """Resolve the local high-temperature window, preferring multi-model hourly consensus.""" + deb = weather_data.get("deb") if isinstance(weather_data, dict) else {} + if isinstance(deb, dict): + consensus = deb.get("hourly_consensus") + if isinstance(consensus, dict): + c_times = consensus.get("times") or [] + c_temps = consensus.get("temps") or [] + hourly_values: List[Tuple[str, float]] = [] + for raw_time, raw_temp in zip(c_times, c_temps): + t_str = str(raw_time or "") + if "T" in t_str and not t_str.startswith(local_date_str): + continue + time_part = t_str.split("T", 1)[1][:5] if "T" in t_str else t_str[:5] + try: + hour = int(time_part[:2]) + except Exception: + continue + value = _sf(raw_temp) + if value is not None and 8 <= hour <= 19: + hourly_values.append((time_part, value)) + peak_hours = _peak_hours_from_hourly_values(hourly_values) + if peak_hours: + return peak_hours + multi_model = weather_data.get("multi_model") if isinstance(weather_data, dict) else {} if isinstance(multi_model, dict): hourly_times = multi_model.get("hourly_times") or [] @@ -420,6 +444,7 @@ def analyze_weather_trend( # === DEB === deb_prediction = None deb_raw_prediction = None + deb_hourly_consensus = None deb_version = None deb_bias_adjustment = 0.0 deb_bias_samples = 0 @@ -506,10 +531,30 @@ def analyze_weather_trend( is_cooling = trend_direction == "falling" om_today = _sf(current_forecasts.get("Open-Meteo")) + if city_name and deb_prediction is not None: + mm = weather_data.get("multi_model") or {} + if isinstance(mm, dict): + deb_hourly_consensus = build_deb_hourly_consensus_path( + city=city_name, + hourly_times=mm.get("hourly_times") or [], + hourly_forecasts=mm.get("hourly_forecasts") or {}, + daily_forecasts=current_forecasts, + deb_prediction=deb_prediction, + local_date=local_date_str, + ) # === Peak hours === + peak_weather_data = weather_data + if deb_hourly_consensus: + peak_weather_data = { + **weather_data, + "deb": { + **(weather_data.get("deb") or {}), + "hourly_consensus": deb_hourly_consensus, + }, + } peak_hours = _resolve_peak_hours( - weather_data, + peak_weather_data, local_date_str, times, temps, @@ -942,6 +987,7 @@ def analyze_weather_trend( "peak_hours": peak_hours, "deb_prediction": deb_prediction, "deb_raw_prediction": deb_raw_prediction, + "deb_hourly_consensus": deb_hourly_consensus, "deb_version": deb_version, "deb_bias_adjustment": deb_bias_adjustment, "deb_bias_samples": deb_bias_samples, diff --git a/tests/test_amsc_awos_sources.py b/tests/test_amsc_awos_sources.py index 1f64a638..d8b92f02 100644 --- a/tests/test_amsc_awos_sources.py +++ b/tests/test_amsc_awos_sources.py @@ -69,6 +69,65 @@ def test_parse_wind_plate_payload_normalizes_runway_point_temperatures(): assert pt0["humidity"] == 67.0 +def test_parse_wind_plate_payload_uses_settlement_runway_endpoint_temperature(): + payload = { + "code": 200, + "data": { + "04/22": { + "RNO": "04/22", + "OTIME": "2026-05-14 17:19:00", + "TDZ_TEMP": "31.2", + "MID_TEMP": "33.4", + "END_TEMP": "32.6", + }, + "05/23": { + "RNO": "05/23", + "OTIME": "2026-05-14 17:19:00", + "TDZ_TEMP": "34.8", + "MID_TEMP": "35.1", + "END_TEMP": "34.2", + }, + }, + } + + parsed = _amsc_parse_wind_plate_payload(payload, city_key="wuhan", icao="ZHHH") + + assert parsed is not None + assert parsed["temp_c"] == 31.2 + assert parsed["temp_source"] == "settlement_runway_endpoint" + assert parsed["settlement_runway"] == "04" + assert parsed["settlement_runway_pair"] == "04/22" + assert parsed["settlement_runway_position"] == "tdz" + settlement_point = parsed["runway_obs"]["point_temperatures"][0] + assert settlement_point["is_settlement"] is True + assert settlement_point["settlement_runway"] == "04" + assert settlement_point["target_runway_max"] == 31.2 + + +def test_parse_wind_plate_payload_uses_end_temperature_when_target_is_second_runway(): + payload = { + "code": 200, + "data": { + "20R/02L": { + "RNO": "20R/02L", + "OTIME": "2026-05-14 17:19:00", + "TDZ_TEMP": "34.4", + "MID_TEMP": "35.2", + "END_TEMP": "33.7", + }, + }, + } + + parsed = _amsc_parse_wind_plate_payload(payload, city_key="chongqing", icao="ZUCK") + + assert parsed is not None + assert parsed["temp_c"] == 33.7 + assert parsed["settlement_runway"] == "02L" + assert parsed["settlement_runway_pair"] == "20R/02L" + assert parsed["settlement_runway_position"] == "end" + assert parsed["runway_obs"]["point_temperatures"][0]["target_runway_max"] == 33.7 + + def test_parse_wind_plate_payload_rejects_unauthorized_or_empty_payloads(): assert _amsc_parse_wind_plate_payload( {"errCode": -12010, "errMsg": "无权访问此接口"}, diff --git a/tests/test_deb_hourly_consensus.py b/tests/test_deb_hourly_consensus.py new file mode 100644 index 00000000..f5f3d68a --- /dev/null +++ b/tests/test_deb_hourly_consensus.py @@ -0,0 +1,72 @@ +from src.analysis.deb_hourly_consensus import ( + DEB_HOURLY_CONSENSUS_VERSION, + build_deb_hourly_consensus_path, +) + + +def test_deb_hourly_consensus_uses_deb_weights_and_anchors_to_daily_prediction(monkeypatch): + monkeypatch.setattr( + "src.analysis.deb_algorithm.load_history", + lambda _: { + "wuhan": { + "2026-05-20": { + "actual_high": 31.0, + "forecasts": {"ECMWF": 31.0, "GFS": 35.0, "ICON": 28.0}, + }, + "2026-05-21": { + "actual_high": 32.0, + "forecasts": {"ECMWF": 32.0, "GFS": 36.0, "ICON": 29.0}, + }, + } + }, + ) + + path = build_deb_hourly_consensus_path( + city="wuhan", + hourly_times=[ + "2026-05-28T09:00", + "2026-05-28T12:00", + "2026-05-28T15:00", + "2026-05-28T18:00", + ], + hourly_forecasts={ + "ECMWF": [25.0, 29.0, 31.0, 27.0], + "GFS": [25.0, 31.0, 35.0, 29.0], + "ICON": [24.0, 27.0, 28.0, 26.0], + }, + daily_forecasts={"ECMWF": 31.0, "GFS": 35.0, "ICON": 28.0}, + deb_prediction=30.0, + local_date="2026-05-28", + ) + + assert path["version"] == DEB_HOURLY_CONSENSUS_VERSION + assert path["times"] == ["09:00", "12:00", "15:00", "18:00"] + assert max(path["temps"]) == 30.0 + assert path["temps"].index(30.0) == 2 + assert path["weights"]["ECMWF"] > path["weights"]["GFS"] + assert path["weights"]["ECMWF"] > path["weights"]["ICON"] + assert path["base_source"] == "multi_model_hourly_deb_weights" + + +def test_deb_hourly_consensus_filters_to_city_local_date(monkeypatch): + monkeypatch.setattr("src.analysis.deb_algorithm.load_history", lambda _: {}) + + path = build_deb_hourly_consensus_path( + city="wuhan", + hourly_times=[ + "2026-05-27T15:00", + "2026-05-28T09:00", + "2026-05-28T15:00", + "2026-05-29T09:00", + ], + hourly_forecasts={ + "ECMWF": [40.0, 24.0, 30.0, 41.0], + "GFS": [40.0, 25.0, 31.0, 41.0], + }, + daily_forecasts={"ECMWF": 30.0, "GFS": 31.0}, + deb_prediction=30.0, + local_date="2026-05-28", + ) + + assert path["times"] == ["09:00", "15:00"] + assert path["temps"] == [24.0, 30.0] diff --git a/tests/test_deb_hourly_peak_correction.py b/tests/test_deb_hourly_peak_correction.py index 21046f67..f30b8ee1 100644 --- a/tests/test_deb_hourly_peak_correction.py +++ b/tests/test_deb_hourly_peak_correction.py @@ -103,7 +103,26 @@ def test_build_deb_hourly_path_anchors_corrected_curve_to_deb_prediction(): ) assert path["source"] == DEB_HOURLY_PEAK_CORRECTED_VERSION + assert path["base_source"] == "hourly_plus_deb_offset" assert max(path["temps"]) == 29.0 assert path["temps"][1] == 29.0 assert path["temps"][0] < 25.5 assert path["correction"]["samples"] == 6 + + +def test_build_deb_hourly_path_preserves_consensus_base_source(): + corrector = build_hourly_peak_corrector([], min_samples=2) + + path = build_deb_hourly_path( + city="wuhan", + hourly_times=["09:00", "15:00"], + hourly_temps=[24.0, 30.0], + deb_prediction=30.0, + peak_first_h=15, + peak_last_h=15, + corrector=corrector, + base_source="deb_hourly_consensus", + ) + + assert path["base_source"] == "deb_hourly_consensus" + assert path["temps"] == [24.0, 30.0] diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index 5068cd2d..a85d896c 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -67,3 +67,13 @@ def test_deploy_script_retries_image_pull_for_registry_propagation(): assert "for pull_attempt in $(seq 1 6)" in script assert "docker compose pull && pull_ok=1 && break" in script + + +def test_city_detail_builds_deb_hourly_consensus_before_peak_window(): + source = (ROOT / "web" / "analysis_service.py").read_text(encoding="utf-8") + + assert "from src.analysis.deb_hourly_consensus import build_deb_hourly_consensus_path" in source + assert "deb_hourly_consensus = build_deb_hourly_consensus_path(" in source + assert '"hourly_consensus": deb_hourly_consensus' in source + assert 'deb_base_source = "deb_hourly_consensus"' in source + assert "base_source=deb_base_source" in source diff --git a/tests/test_trend_engine.py b/tests/test_trend_engine.py index 175f299a..8ba01deb 100644 --- a/tests/test_trend_engine.py +++ b/tests/test_trend_engine.py @@ -206,6 +206,41 @@ class TestMuCalculation: assert sd["peak_status"] == "before" assert sd["mu"] is not None and sd["mu"] >= 29.0 + @patch("src.analysis.trend_engine.calculate_dynamic_weights", return_value=(None, "")) + @patch("src.analysis.trend_engine.get_deb_accuracy", return_value=None) + @patch("src.analysis.trend_engine.update_daily_record") + def test_deb_hourly_consensus_takes_priority_for_peak_window( + self, _udr, _deb_acc, _dw + ): + """The peak window should follow the independent DEB hourly path before raw model medians.""" + data = _make_weather_data( + cur_temp=25.0, + max_so_far=25.0, + om_today_high=31.0, + ens_median=None, + ens_p10=None, + ens_p90=None, + local_time="2026-03-04 10:00", + multi_model_hourly={ + "hourly_times": [f"2026-03-04T{h:02d}:00" for h in range(24)], + "hourly_forecasts": { + "ECMWF": [24.0 if h != 12 else 31.0 for h in range(24)], + "GFS": [24.0 if h != 12 else 31.0 for h in range(24)], + }, + }, + ) + data["deb"] = { + "hourly_consensus": { + "times": ["09:00", "15:00", "16:00"], + "temps": [24.0, 30.0, 30.0], + } + } + + _, _, sd = analyze_weather_trend(data, "°C", "test_city") + + assert sd["peak_hours"] == ["15:00", "16:00"] + assert sd["peak_status"] == "before" + # ─── Tests: Dead Market ─── diff --git a/web/analysis_service.py b/web/analysis_service.py index 7803c322..dc8ebfae 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -25,6 +25,7 @@ from web.core import ( _weather, ) from src.analysis.deb_algorithm import calculate_deb_prediction +from src.analysis.deb_hourly_consensus import build_deb_hourly_consensus_path from src.analysis.deb_hourly_correction import ( build_deb_hourly_path, get_cached_hourly_peak_corrector, @@ -1091,6 +1092,7 @@ def _analyze( deb_raw_val, deb_version = None, None deb_bias_adjustment, deb_bias_samples = 0.0, 0 deb_intraday_adjustment = 0.0 + deb_hourly_consensus = None if current_forecasts: deb_result = calculate_deb_prediction(city, current_forecasts) if deb_result.get("prediction") is not None: @@ -1100,6 +1102,14 @@ def _analyze( deb_bias_adjustment = deb_result.get("bias_adjustment") or 0.0 deb_bias_samples = deb_result.get("bias_samples") or 0 deb_weights = deb_result.get("weights_info") or "" + deb_hourly_consensus = build_deb_hourly_consensus_path( + city=city, + hourly_times=mm.get("hourly_times") or [], + hourly_forecasts=mm.get("hourly_forecasts") or {}, + daily_forecasts=current_forecasts, + deb_prediction=deb_val, + local_date=local_date_str, + ) # ── 7. Ensemble stats ── ens_data = { @@ -1188,7 +1198,16 @@ def _analyze( h_lifted_index = [None for _ in parsed_obs] h_boundary_layer_height = [None for _ in parsed_obs] - peak_hours = _resolve_peak_hours(raw, local_date_str, h_times, h_temps, om_today) + peak_source = raw + if deb_hourly_consensus: + peak_source = { + **raw, + "deb": { + **(raw.get("deb") or {}), + "hourly_consensus": deb_hourly_consensus, + }, + } + peak_hours = _resolve_peak_hours(peak_source, local_date_str, h_times, h_temps, om_today) first_peak_h = int(peak_hours[0].split(":")[0]) if peak_hours else 13 last_peak_h = int(peak_hours[-1].split(":")[0]) if peak_hours else 15 @@ -1242,6 +1261,8 @@ def _analyze( deb_bias_adjustment = sd.get("deb_bias_adjustment") or 0.0 deb_bias_samples = sd.get("deb_bias_samples") or 0 deb_weights = sd.get("deb_weights", "") + if deb_hourly_consensus is None and sd.get("deb_hourly_consensus"): + deb_hourly_consensus = sd.get("deb_hourly_consensus") except Exception as e: logger.warning(f"Structured analysis skipped for {city}: {e}") @@ -1306,16 +1327,27 @@ def _analyze( deb_weights = f"{deb_weights or 'DEB'} + intraday_bias({deb_intraday_adjustment:+.1f})" deb_hourly_path = None - if deb_val is not None and today_hourly.get("times") and today_hourly.get("temps"): + deb_base_source = "hourly_plus_deb_offset" + deb_base_times = [str(item) for item in today_hourly.get("times") or []] + deb_base_temps = today_hourly.get("temps") or [] + if isinstance(deb_hourly_consensus, dict): + consensus_times = deb_hourly_consensus.get("times") or [] + consensus_temps = deb_hourly_consensus.get("temps") or [] + if consensus_times and consensus_temps: + deb_base_source = "deb_hourly_consensus" + deb_base_times = [str(item) for item in consensus_times] + deb_base_temps = consensus_temps + if deb_val is not None and deb_base_times and deb_base_temps: try: deb_hourly_path = build_deb_hourly_path( city=city, - hourly_times=[str(item) for item in today_hourly.get("times") or []], - hourly_temps=today_hourly.get("temps") or [], + hourly_times=deb_base_times, + hourly_temps=deb_base_temps, deb_prediction=deb_val, peak_first_h=first_peak_h, peak_last_h=last_peak_h, corrector=get_cached_hourly_peak_corrector(), + base_source=deb_base_source, ) except Exception as exc: logger.debug(f"DEB hourly path correction skipped for {city}: {exc}") @@ -1741,6 +1773,7 @@ def _analyze( "bias_adjustment": deb_bias_adjustment, "bias_samples": deb_bias_samples, "intraday_adjustment": deb_intraday_adjustment, + "hourly_consensus": deb_hourly_consensus, "hourly_path": deb_hourly_path, "hourly_correction": (deb_hourly_path or {}).get("correction") if isinstance(deb_hourly_path, dict) else None, },