From 1a7d2a487fed2b57df30a720577a290aaa8a5dfe Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Wed, 27 May 2026 22:41:16 +0800 Subject: [PATCH] feat: add hourly peak correction for DEB charts --- .../__tests__/temperatureChartData.test.ts | 60 +++ .../scan-terminal/temperature-chart-logic.ts | 30 +- frontend/lib/chart-utils.ts | 21 +- frontend/lib/dashboard-types.ts | 12 + src/analysis/deb_hourly_correction.py | 350 ++++++++++++++++++ src/database/runtime_state.py | 20 + tests/test_deb_hourly_peak_correction.py | 109 ++++++ web/analysis_service.py | 21 ++ 8 files changed, 614 insertions(+), 9 deletions(-) create mode 100644 src/analysis/deb_hourly_correction.py create mode 100644 tests/test_deb_hourly_peak_correction.py diff --git a/frontend/components/dashboard/scan-terminal/__tests__/temperatureChartData.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/temperatureChartData.test.ts index 9be53fa2..223b4d2f 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/temperatureChartData.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/temperatureChartData.test.ts @@ -1,6 +1,7 @@ import { getTemperatureChartData } from "@/lib/chart-utils"; import type { CityDetail } from "@/lib/dashboard-types"; import { buildDebBaselinePath } from "@/lib/temperature-chart-paths"; +import { buildFullDayChartData } from "@/components/dashboard/scan-terminal/temperature-chart-logic"; function assert(condition: unknown, message: string) { if (!condition) throw new Error(message); @@ -46,6 +47,65 @@ export function runTests() { "temperature chart should keep normalized hourly temperatures shifted by offset", ); + const correctedHourlyPathChart = buildFullDayChartData( + { + city: "shanghai", + local_date: "2026-05-16", + local_time: "12:00", + temp_symbol: "°C", + deb_prediction: 30, + tz_offset_seconds: 8 * 3600, + } as any, + { + forecastTodayHigh: 29, + debPrediction: 30, + localDate: "2026-05-16", + localTime: "12:00", + times: ["10:00", "12:00", "14:00"], + temps: [24, 29, 25], + debHourlyPath: { + source: "deb_hourly_peak_corrected.v1", + times: ["10:00", "12:00", "14:00"], + temps: [25.1, 28.0, 26.0], + }, + } as any, + true, + ); + const correctedDebSeries = correctedHourlyPathChart.series.find((item) => item.key === "hourly_forecast"); + const correctedDebValues = correctedHourlyPathChart.data + .map((item) => item.hourly_forecast) + .filter((value) => value !== null && value !== undefined); + assert(correctedDebSeries, "corrected hourly DEB path should still render as DEB Forecast"); + assert(correctedDebValues.includes(28.0), `chart should use backend deb.hourly_path before rebuilding a shifted curve; got ${correctedDebValues.join(",")}`); + + const correctedDetailChart = getTemperatureChartData( + { + name: "shanghai", + display_name: "Shanghai", + local_date: "2026-05-16", + local_time: "12:00", + temp_symbol: "°C", + hourly: { + times: ["10:00", "12:00", "14:00"], + temps: [24, 29, 25], + }, + forecast: { today_high: 29 }, + deb: { + prediction: 30, + hourly_path: { + source: "deb_hourly_peak_corrected.v1", + times: ["10:00", "12:00", "14:00"], + temps: [25.1, 28.0, 26.0], + }, + }, + } as CityDetail, + "zh-CN", + ); + assert( + correctedDetailChart?.datasets.debSeries.some((point) => point.labelTime === "12:00" && point.y === 28.0), + "detail temperature chart should use backend deb.hourly_path before fallback baseline", + ); + const ankaraChartData = getTemperatureChartData( { name: "ankara", diff --git a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts index 0ae33667..f9190e73 100644 --- a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts +++ b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts @@ -5,6 +5,7 @@ import type { ScanOpportunityRow, ForecastDay, DailyModelForecast, + DebHourlyPath, ProbabilityBucket, } from "@/lib/dashboard-types"; import { buildDebBaselinePath } from "@/lib/temperature-chart-paths"; @@ -735,6 +736,7 @@ function runwayPatchPointsFromRunwayObs(runwayObs: any) { type HourlyForecast = { forecastTodayHigh?: number | null; debPrediction?: number | null; + debHourlyPath?: DebHourlyPath | null; localDate?: string | null; localTime?: string | null; times: string[]; @@ -759,6 +761,7 @@ function seedHourlyForecastFromRow(row: ScanOpportunityRow | null): HourlyForeca return { forecastTodayHigh: null, debPrediction: validNumber(row.deb_prediction), + debHourlyPath: null, localDate: row.local_date || null, localTime: row.local_time || null, times: [], @@ -793,6 +796,7 @@ function parseHourlyForecastFromCityDetail(json: CityDetail | null): HourlyForec return { forecastTodayHigh: json.forecast?.today_high ?? null, debPrediction: json.deb?.prediction ?? (json as any)?.overview?.deb_prediction ?? null, + debHourlyPath: json.deb?.hourly_path || null, localDate: json.local_date || (json as any)?.overview?.local_date || null, localTime: json.local_time || null, times: hourlySource.times || [], @@ -1591,16 +1595,27 @@ function buildFullDayChartData( if (shouldRenderMetar) metarObs.forEach((point) => timelineSet.add(point.ts)); addLocalDayAxisSlots(timelineSet, localDayBounds); - let debPath: ReturnType | null = null; - if (hourly?.times?.length && hourly?.temps?.length) { - debPath = buildDebBaselinePath( + const correctedDebPath = hourly?.debHourlyPath; + const correctedDebTimes = Array.isArray(correctedDebPath?.times) ? correctedDebPath?.times || [] : []; + const correctedDebTemps = Array.isArray(correctedDebPath?.temps) ? correctedDebPath?.temps || [] : []; + let debTimes: string[] = []; + let debTemps: Array = []; + if (correctedDebTimes.length && correctedDebTemps.length) { + debTimes = correctedDebTimes; + debTemps = correctedDebTemps; + } else if (hourly?.times?.length && hourly?.temps?.length) { + const debPath = buildDebBaselinePath( hourly.times, hourly.temps, validNumber(hourly?.debPrediction) ?? row?.deb_prediction, hourly.localTime || row?.local_time, hourly.forecastTodayHigh, ); - addHourlyTimesToTimeline(timelineSet, hourly.times, debPath.debTemps, tzOffset, localDateStr, localDayBounds); + debTimes = hourly.times; + debTemps = debPath.debTemps; + } + if (debTimes.length && debTemps.length) { + addHourlyTimesToTimeline(timelineSet, debTimes, debTemps, tzOffset, localDateStr, localDayBounds); } if (hourly?.times?.length && hourly?.modelCurves) { Object.values(hourly.modelCurves).forEach((modelTemps) => { @@ -1693,8 +1708,8 @@ function buildFullDayChartData( } // ── DEB forecast curve ── - if (hourly?.times?.length && debPath?.debTemps.length) { - const debVals = valuesForHourlyTimes(n, indexByTs, hourly.times, debPath.debTemps, tzOffset, localDateStr, localDayBounds); + if (debTimes.length && debTemps.length) { + const debVals = valuesForHourlyTimes(n, indexByTs, debTimes, debTemps, tzOffset, localDateStr, localDayBounds); if (debVals.some((v) => v !== null)) { series.push({ key: "hourly_forecast", @@ -1708,7 +1723,7 @@ function buildFullDayChartData( } // Per-model curves - if (hourly.modelCurves) { + if (hourly?.times?.length && hourly.modelCurves) { const modelColors = ["#2563eb", "#7c3aed", "#059669", "#d97706", "#dc2626", "#0891b2"]; Object.keys(hourly.modelCurves).forEach((model, idx) => { const modelTemps = hourly.modelCurves![model]; @@ -1795,7 +1810,6 @@ function probabilityOverlayValues(probabilityOverlay?: ProbabilityOverlay | null ...probabilityOverlay.bands.flatMap((band) => [band.lower, band.upper]), ]; } - function buildIntDegreeTicks( series: EvidenceSeries[], data?: Array>, diff --git a/frontend/lib/chart-utils.ts b/frontend/lib/chart-utils.ts index 5ca9a518..14d4cf49 100644 --- a/frontend/lib/chart-utils.ts +++ b/frontend/lib/chart-utils.ts @@ -175,13 +175,32 @@ export function getTemperatureChartData( detail.forecast?.today_high, mgmHourlyMax, ); - const { + let { debTemps, debPast, debFuture, currentIndex, offset, } = baseline; + const correctedDebPath = detail.deb?.hourly_path; + const correctedDebTimes = Array.isArray(correctedDebPath?.times) ? correctedDebPath?.times || [] : []; + const correctedDebRawTemps = Array.isArray(correctedDebPath?.temps) ? correctedDebPath?.temps || [] : []; + if (correctedDebTimes.length && correctedDebRawTemps.length) { + const mappedDebTemps = times.map((time) => { + const minute = hmToMinutes(time); + if (minute == null) return null; + return interpolateSeriesAtMinutes(correctedDebTimes, correctedDebRawTemps, minute); + }); + if (mappedDebTemps.some((value) => value != null)) { + debTemps = mappedDebTemps; + debPast = debTemps.map((value, index) => (index <= currentIndex ? value : null)); + debFuture = debTemps.map((value, index) => (index >= currentIndex ? value : null)); + const correctedOffset = Number(correctedDebPath?.base_offset); + if (Number.isFinite(correctedOffset)) { + offset = correctedOffset; + } + } + } const suppressAnkaraMgmObservation = isTurkishMgmCity(detail); const observationTag = getRealtimeObservationTag(detail); diff --git a/frontend/lib/dashboard-types.ts b/frontend/lib/dashboard-types.ts index 28ecccfd..540d66bb 100644 --- a/frontend/lib/dashboard-types.ts +++ b/frontend/lib/dashboard-types.ts @@ -204,6 +204,16 @@ export interface ForecastData { sunshine_hours?: number | null; } +export interface DebHourlyPath { + source?: string | null; + version?: string | null; + times?: string[]; + temps?: Array; + base_source?: string | null; + base_offset?: number | null; + correction?: Record | null; +} + export interface DebForecast { prediction: number | null; raw_prediction?: number | null; @@ -212,6 +222,8 @@ export interface DebForecast { bias_adjustment?: number | null; bias_samples?: number | null; intraday_adjustment?: number | null; + hourly_path?: DebHourlyPath | null; + hourly_correction?: Record | null; } export interface CitySummary { diff --git a/src/analysis/deb_hourly_correction.py b/src/analysis/deb_hourly_correction.py new file mode 100644 index 00000000..29a50bb9 --- /dev/null +++ b/src/analysis/deb_hourly_correction.py @@ -0,0 +1,350 @@ +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Any, Dict, Iterable, List, Optional, Tuple + +from loguru import logger + +DEB_HOURLY_PEAK_CORRECTED_VERSION = "deb_hourly_peak_corrected.v1" + +_DEFAULT_MAX_NEAREST_MINUTES = 75 +_DEFAULT_MAX_ADJUSTMENT = 3.0 + + +def _to_float(value: Any) -> Optional[float]: + try: + result = float(value) + except (TypeError, ValueError): + return None + if result != result: + return None + return result + + +def _parse_minutes(value: Any) -> Optional[int]: + text = str(value or "").strip() + if not text: + return None + if "T" in text: + text = text.split("T", 1)[1] + if " " in text: + text = text.rsplit(" ", 1)[-1] + text = text.replace("Z", "") + if "+" in text: + text = text.split("+", 1)[0] + if "-" in text and text.count(":") >= 1 and text[0:1].isdigit(): + text = text.split("-", 1)[0] + parts = text.split(":") + if len(parts) < 2: + return None + try: + hour = int(parts[0]) + minute = int(parts[1]) + except ValueError: + return None + if not (0 <= hour <= 23 and 0 <= minute <= 59): + return None + return hour * 60 + minute + + +def _phase_for_minute(minute: int, first_h: Optional[int], last_h: Optional[int]) -> str: + hour = minute // 60 + first = int(first_h if first_h is not None else 13) + last = int(last_h if last_h is not None else 15) + if hour < first: + return "before_peak" + if hour <= last: + return "peak_window" + return "after_peak" + + +def _nearest_value( + target_minute: int, + base_points: List[Tuple[int, float]], + max_minutes: int, +) -> Optional[float]: + best: Optional[Tuple[int, float]] = None + for minute, value in base_points: + distance = abs(minute - target_minute) + if distance > max_minutes: + continue + if best is None or distance < best[0]: + best = (distance, value) + return best[1] if best is not None else None + + +def _mean(values: Iterable[float]) -> float: + items = list(values) + if not items: + return 0.0 + return sum(items) / len(items) + + +def _clamp(value: float, limit: float) -> float: + return max(-limit, min(limit, value)) + + +def _stat(values: List[float], max_adjustment: float) -> Dict[str, Any]: + average = _clamp(_mean(values), max_adjustment) + return { + "adjustment": round(average, 3), + "samples": len(values), + } + + +def _normalize_city(value: Any) -> str: + return str(value or "").strip().lower() + + +def _iter_observations(snapshot: Dict[str, Any]) -> Tuple[str, List[Any]]: + settlement_rows = snapshot.get("settlement_today_obs") + if isinstance(settlement_rows, list) and settlement_rows: + return "settlement", settlement_rows + metar_rows = snapshot.get("metar_today_obs") + if isinstance(metar_rows, list) and metar_rows: + return "metar", metar_rows + return "", [] + + +def _obs_time_temp(item: Any) -> Tuple[Optional[int], Optional[float]]: + if isinstance(item, dict): + minute = _parse_minutes( + item.get("time") + or item.get("obs_time") + or item.get("observation_time") + or item.get("timestamp") + ) + value = _to_float(item.get("temp") if "temp" in item else item.get("value")) + return minute, value + if isinstance(item, (list, tuple)) and len(item) >= 2: + return _parse_minutes(item[0]), _to_float(item[1]) + return None, None + + +@dataclass(frozen=True) +class HourlyPeakCorrector: + city_hour_adjustments: Dict[str, Dict[int, Dict[str, Any]]] + city_phase_adjustments: Dict[str, Dict[str, Dict[str, Any]]] + min_samples: int + max_adjustment: float + sample_count: int + + def _adjustment_for(self, city: str, minute: int, first_h: Optional[int], last_h: Optional[int]) -> Tuple[float, str]: + city_key = _normalize_city(city) + hour = minute // 60 + hour_stats = self.city_hour_adjustments.get(city_key, {}).get(hour) + if hour_stats and int(hour_stats.get("samples") or 0) >= self.min_samples: + return float(hour_stats.get("adjustment") or 0.0), "hour" + phase = _phase_for_minute(minute, first_h, last_h) + phase_stats = self.city_phase_adjustments.get(city_key, {}).get(phase) + if phase_stats and int(phase_stats.get("samples") or 0) >= self.min_samples: + return float(phase_stats.get("adjustment") or 0.0), phase + return 0.0, "none" + + def apply( + self, + city: str, + times: List[str], + temps: List[Optional[float]], + *, + peak_first_h: Optional[int], + peak_last_h: Optional[int], + deb_prediction: Optional[float] = None, + ) -> Dict[str, Any]: + corrected: List[Optional[float]] = [] + applied_sources: Dict[str, int] = {} + for index, raw in enumerate(temps): + base_value = _to_float(raw) + minute = _parse_minutes(times[index] if index < len(times) else None) + if base_value is None or minute is None: + corrected.append(None) + continue + adjustment, source = self._adjustment_for(city, minute, peak_first_h, peak_last_h) + applied_sources[source] = applied_sources.get(source, 0) + 1 + corrected.append(round(base_value + adjustment, 1)) + + anchor_adjustment = 0.0 + deb_value = _to_float(deb_prediction) + numeric_values = [value for value in corrected if value is not None] + if deb_value is not None and numeric_values: + anchor_adjustment = deb_value - max(numeric_values) + corrected = [ + round(value + anchor_adjustment, 1) if value is not None else None + for value in corrected + ] + + city_key = _normalize_city(city) + return { + "version": DEB_HOURLY_PEAK_CORRECTED_VERSION, + "source": DEB_HOURLY_PEAK_CORRECTED_VERSION, + "times": list(times), + "temps": corrected, + "samples": self.sample_count, + "phase_adjustments": self.city_phase_adjustments.get(city_key, {}), + "hour_adjustments": self.city_hour_adjustments.get(city_key, {}), + "applied_sources": applied_sources, + "anchor_adjustment": round(anchor_adjustment, 3), + } + + +def build_hourly_peak_corrector( + snapshots: Iterable[Dict[str, Any]], + *, + min_samples: int = 6, + max_adjustment: float = _DEFAULT_MAX_ADJUSTMENT, + nearest_minutes: int = _DEFAULT_MAX_NEAREST_MINUTES, +) -> HourlyPeakCorrector: + hour_errors: Dict[str, Dict[int, List[float]]] = {} + phase_errors: Dict[str, Dict[str, List[float]]] = {} + seen_observations: set[Tuple[str, str, str, int]] = set() + sample_count = 0 + + for snapshot in snapshots: + if not isinstance(snapshot, dict): + continue + city_key = _normalize_city(snapshot.get("city")) + if not city_key: + continue + base_path = snapshot.get("deb_base_path") or {} + base_times = base_path.get("times") if isinstance(base_path, dict) else [] + base_temps = base_path.get("temps") if isinstance(base_path, dict) else [] + if not isinstance(base_times, list) or not isinstance(base_temps, list): + continue + base_points: List[Tuple[int, float]] = [] + for index, time_value in enumerate(base_times): + minute = _parse_minutes(time_value) + value = _to_float(base_temps[index] if index < len(base_temps) else None) + if minute is not None and value is not None: + base_points.append((minute, value)) + if not base_points: + continue + + peak = snapshot.get("peak") or {} + first_h = peak.get("first_h") if isinstance(peak, dict) else None + last_h = peak.get("last_h") if isinstance(peak, dict) else None + source_key, obs_rows = _iter_observations(snapshot) + if not obs_rows: + continue + + target_date = str(snapshot.get("target_date") or snapshot.get("local_date") or "").strip() + for item in obs_rows: + minute, observed = _obs_time_temp(item) + if minute is None or observed is None: + continue + dedupe_key = (city_key, target_date, source_key, minute) + if dedupe_key in seen_observations: + continue + seen_observations.add(dedupe_key) + base_value = _nearest_value(minute, base_points, nearest_minutes) + if base_value is None: + continue + error = _clamp(observed - base_value, max_adjustment) + hour = minute // 60 + phase = _phase_for_minute(minute, first_h, last_h) + hour_errors.setdefault(city_key, {}).setdefault(hour, []).append(error) + phase_errors.setdefault(city_key, {}).setdefault(phase, []).append(error) + sample_count += 1 + + city_hour_adjustments = { + city: { + hour: _stat(values, max_adjustment) + for hour, values in hours.items() + if len(values) >= min_samples + } + for city, hours in hour_errors.items() + } + city_phase_adjustments = { + city: { + phase: _stat(values, max_adjustment) + for phase, values in phases.items() + if len(values) >= min_samples + } + for city, phases in phase_errors.items() + } + return HourlyPeakCorrector( + city_hour_adjustments=city_hour_adjustments, + city_phase_adjustments=city_phase_adjustments, + min_samples=min_samples, + max_adjustment=max_adjustment, + sample_count=sample_count, + ) + + +def build_deb_hourly_path( + *, + city: str, + hourly_times: List[str], + hourly_temps: List[Optional[float]], + deb_prediction: Optional[float], + peak_first_h: Optional[int], + peak_last_h: Optional[int], + corrector: HourlyPeakCorrector, +) -> Dict[str, Any]: + deb_value = _to_float(deb_prediction) + numeric_base = [_to_float(value) for value in hourly_temps] + numeric_only = [value for value in numeric_base if value is not None] + if deb_value is not None and numeric_only: + offset = deb_value - max(numeric_only) + else: + offset = 0.0 + base_temps = [ + round(value + offset, 1) if value is not None else None + for value in numeric_base + ] + applied = corrector.apply( + city, + list(hourly_times), + base_temps, + peak_first_h=peak_first_h, + peak_last_h=peak_last_h, + deb_prediction=deb_value, + ) + return { + "source": DEB_HOURLY_PEAK_CORRECTED_VERSION, + "version": DEB_HOURLY_PEAK_CORRECTED_VERSION, + "times": applied["times"], + "temps": applied["temps"], + "base_source": "hourly_plus_deb_offset", + "base_offset": round(offset, 3), + "correction": { + "version": applied["version"], + "samples": applied["samples"], + "phase_adjustments": applied["phase_adjustments"], + "hour_adjustments": applied["hour_adjustments"], + "applied_sources": applied["applied_sources"], + "anchor_adjustment": applied["anchor_adjustment"], + }, + } + + +_CORRECTOR_CACHE: Dict[str, Any] = {"loaded_at": 0.0, "corrector": None} + + +def get_cached_hourly_peak_corrector( + *, + ttl_seconds: int = 600, + max_rows: int = 20000, + min_samples: int = 6, +) -> HourlyPeakCorrector: + now = time.time() + cached = _CORRECTOR_CACHE.get("corrector") + if cached is not None and now - float(_CORRECTOR_CACHE.get("loaded_at") or 0) < ttl_seconds: + return cached + + rows: List[Dict[str, Any]] = [] + try: + from src.database.runtime_state import IntradayPathSnapshotRepository + + repo = IntradayPathSnapshotRepository() + if hasattr(repo, "load_recent_rows"): + rows = repo.load_recent_rows(limit=max_rows) + else: + rows = repo.load_all_rows()[-max_rows:] + except Exception as exc: + logger.debug(f"DEB hourly peak corrector load skipped: {exc}") + + corrector = build_hourly_peak_corrector(rows, min_samples=min_samples) + _CORRECTOR_CACHE["loaded_at"] = now + _CORRECTOR_CACHE["corrector"] = corrector + return corrector diff --git a/src/database/runtime_state.py b/src/database/runtime_state.py index 7b9fa64a..970ade83 100644 --- a/src/database/runtime_state.py +++ b/src/database/runtime_state.py @@ -1186,6 +1186,26 @@ class IntradayPathSnapshotRepository: continue return out + def load_recent_rows(self, limit: int = 20000) -> List[Dict[str, Any]]: + safe_limit = max(1, int(limit or 1)) + with self.db.connect() as conn: + rows = conn.execute( + """ + SELECT payload_json + FROM intraday_path_snapshots_store + ORDER BY id DESC + LIMIT ? + """, + (safe_limit,), + ).fetchall() + out: List[Dict[str, Any]] = [] + for row in reversed(rows): + try: + out.append(json.loads(row["payload_json"])) + except Exception: + continue + return out + def _top_bucket(snapshot: Optional[List[Dict[str, Any]]]) -> Optional[int]: best_value = None best_prob = -1.0 diff --git a/tests/test_deb_hourly_peak_correction.py b/tests/test_deb_hourly_peak_correction.py new file mode 100644 index 00000000..21046f67 --- /dev/null +++ b/tests/test_deb_hourly_peak_correction.py @@ -0,0 +1,109 @@ +from src.analysis.deb_hourly_correction import ( + DEB_HOURLY_PEAK_CORRECTED_VERSION, + build_deb_hourly_path, + build_hourly_peak_corrector, +) + + +def test_hourly_peak_corrector_learns_peak_window_shape_from_snapshots(): + snapshots = [ + { + "city": "ankara", + "target_date": "2026-05-20", + "deb_prediction": 30.0, + "deb_base_path": { + "times": ["10:00", "12:00", "14:00", "16:00"], + "temps": [25.0, 27.0, 30.0, 28.0], + }, + "settlement_today_obs": [ + {"time": "12:00", "temp": 27.5}, + {"time": "14:00", "temp": 32.0}, + {"time": "16:00", "temp": 28.5}, + ], + "peak": {"first_h": 13, "last_h": 15}, + }, + { + "city": "ankara", + "target_date": "2026-05-21", + "deb_prediction": 29.0, + "deb_base_path": { + "times": ["10:00", "12:00", "14:00", "16:00"], + "temps": [24.0, 26.0, 29.0, 27.0], + }, + "settlement_today_obs": [ + {"time": "12:00", "temp": 26.5}, + {"time": "14:00", "temp": 31.0}, + {"time": "16:00", "temp": 27.5}, + ], + "peak": {"first_h": 13, "last_h": 15}, + }, + ] + + corrector = build_hourly_peak_corrector(snapshots, min_samples=2) + result = corrector.apply( + "ankara", + ["10:00", "12:00", "14:00", "16:00"], + [24.0, 26.0, 29.0, 27.0], + peak_first_h=13, + peak_last_h=15, + deb_prediction=30.0, + ) + + assert result["version"] == DEB_HOURLY_PEAK_CORRECTED_VERSION + assert result["samples"] == 6 + assert result["temps"][2] == 30.0 + assert result["temps"][1] < result["temps"][2] + assert result["temps"][3] < result["temps"][2] + assert result["phase_adjustments"]["peak_window"]["samples"] == 2 + + +def test_build_deb_hourly_path_anchors_corrected_curve_to_deb_prediction(): + corrector = build_hourly_peak_corrector( + [ + { + "city": "shanghai", + "target_date": "2026-05-20", + "deb_base_path": { + "times": ["10:00", "14:00", "18:00"], + "temps": [25.0, 28.0, 24.0], + }, + "settlement_today_obs": [ + {"time": "10:00", "temp": 24.0}, + {"time": "14:00", "temp": 30.0}, + {"time": "18:00", "temp": 23.0}, + ], + "peak": {"first_h": 13, "last_h": 15}, + }, + { + "city": "shanghai", + "target_date": "2026-05-21", + "deb_base_path": { + "times": ["10:00", "14:00", "18:00"], + "temps": [26.0, 29.0, 25.0], + }, + "settlement_today_obs": [ + {"time": "10:00", "temp": 25.0}, + {"time": "14:00", "temp": 31.0}, + {"time": "18:00", "temp": 24.0}, + ], + "peak": {"first_h": 13, "last_h": 15}, + }, + ], + min_samples=2, + ) + + path = build_deb_hourly_path( + city="shanghai", + hourly_times=["10:00", "14:00", "18:00"], + hourly_temps=[24.0, 28.0, 23.0], + deb_prediction=29.0, + peak_first_h=13, + peak_last_h=15, + corrector=corrector, + ) + + assert path["source"] == DEB_HOURLY_PEAK_CORRECTED_VERSION + assert max(path["temps"]) == 29.0 + assert path["temps"][1] == 29.0 + assert path["temps"][0] < 25.5 + assert path["correction"]["samples"] == 6 diff --git a/web/analysis_service.py b/web/analysis_service.py index 1de79c5e..c8997516 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -25,6 +25,10 @@ from web.core import ( _weather, ) from src.analysis.deb_algorithm import calculate_deb_prediction +from src.analysis.deb_hourly_correction import ( + build_deb_hourly_path, + get_cached_hourly_peak_corrector, +) from src.analysis.settlement_rounding import apply_city_settlement from src.data_collection.country_networks import build_country_network_snapshot from src.data_collection.city_registry import ALIASES, CITY_REGISTRY @@ -1305,6 +1309,21 @@ def _analyze( mu = round(mu + blended_correction, 1) 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"): + 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 [], + deb_prediction=deb_val, + peak_first_h=first_peak_h, + peak_last_h=last_peak_h, + corrector=get_cached_hourly_peak_corrector(), + ) + except Exception as exc: + logger.debug(f"DEB hourly path correction skipped for {city}: {exc}") + # ── 12b. Next 48h hourly block for future-date analysis modal ── next_48h_hourly = { "times": [], @@ -1726,6 +1745,8 @@ def _analyze( "bias_adjustment": deb_bias_adjustment, "bias_samples": deb_bias_samples, "intraday_adjustment": deb_intraday_adjustment, + "hourly_path": deb_hourly_path, + "hourly_correction": (deb_hourly_path or {}).get("correction") if isinstance(deb_hourly_path, dict) else None, }, "deviation_monitor": deviation_monitor, "ensemble": ens_data,