feat: Implement reconciliation of recent actual high temperatures using METAR data and integrate it into the DEB command.

bot_listener.py

@@ -132,7 +132,11 @@ def start_bot():
             from datetime import datetime as _dt, timedelta as _td
             import os as _os
 
-            from src.analysis.deb_algorithm import load_history, _is_excluded_model_name
+            from src.analysis.deb_algorithm import (
+                load_history,
+                _is_excluded_model_name,
+                reconcile_recent_actual_highs,
+            )
             from src.data_collection.city_registry import ALIASES
 
             city_input = parts[1].strip().lower()
@@ -153,6 +157,9 @@ def start_bot():
             if not _ensure_query_points(message, DEB_QUERY_COST, "/deb"):
                 return
 
+            reconcile_info = reconcile_recent_actual_highs(city_name, lookback_days=7)
+            # Reload in case reconciliation updated the file
+            data = load_history(history_file)
             city_data = data[city_name]
             today = _dt.now().date()
             today_str = today.strftime("%Y-%m-%d")
@@ -174,6 +181,17 @@ def start_bot():
                 "",
                 "📅 <b>近日记录：</b>",
             ]
+            if (
+                isinstance(reconcile_info, dict)
+                and reconcile_info.get("ok")
+                and int(reconcile_info.get("updated") or 0) > 0
+            ):
+                lines.extend(
+                    [
+                        f"🔁 已用 METAR 历史回填修正 {int(reconcile_info.get('updated'))} 天实测最高温",
+                        "",
+                    ]
+                )
             total_days = 0
             hits = 0
             deb_errors = []

src/analysis/deb_algorithm.py

@@ -1,6 +1,7 @@
 import os
 import json
 from datetime import datetime, timedelta
+import requests
 from src.analysis.settlement_rounding import wu_round
 
 # Cross-platform file locking
@@ -80,6 +81,141 @@ def save_history(filepath, data):
         print(f"Error saving history: {e}")
 
 
+def _parse_metar_row_time(row):
+    """Parse METAR row timestamp from aviationweather API payload."""
+    candidates = [
+        row.get("reportTime"),
+        row.get("receiptTime"),
+        row.get("observation_time"),
+    ]
+    for raw in candidates:
+        if not raw:
+            continue
+        try:
+            return datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
+        except Exception:
+            continue
+    obs_epoch = row.get("obsTime")
+    if obs_epoch is not None:
+        try:
+            return datetime.utcfromtimestamp(int(obs_epoch))
+        except Exception:
+            pass
+    return None
+
+
+def reconcile_recent_actual_highs(city_name: str, lookback_days: int = 7):
+    """
+    Reconcile recent `actual_high` values using historical METAR data from
+    aviationweather.gov to fix stale/wrong daily records.
+    """
+    try:
+        from src.data_collection.city_registry import CITY_REGISTRY, ALIASES
+
+        city_key = str(city_name or "").strip().lower()
+        city_key = ALIASES.get(city_key, city_key)
+        city_meta = CITY_REGISTRY.get(city_key)
+        if not isinstance(city_meta, dict):
+            return {"ok": False, "reason": "unknown_city", "updated": 0}
+
+        icao = str(city_meta.get("icao") or "").strip().upper()
+        if not icao:
+            return {"ok": False, "reason": "missing_icao", "updated": 0}
+
+        tz_offset = int(city_meta.get("tz_offset") or 0)
+        use_fahrenheit = bool(city_meta.get("use_fahrenheit"))
+
+        project_root = os.path.dirname(
+            os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        )
+        history_file = os.path.join(project_root, "data", "daily_records.json")
+        data = load_history(history_file)
+        city_data = data.get(city_key) or {}
+        if not isinstance(city_data, dict) or not city_data:
+            return {"ok": True, "reason": "no_city_history", "updated": 0}
+
+        local_now = datetime.utcnow() + timedelta(seconds=tz_offset)
+        local_today = local_now.strftime("%Y-%m-%d")
+        cutoff = (local_now - timedelta(days=max(lookback_days, 1) + 1)).strftime(
+            "%Y-%m-%d"
+        )
+        target_dates = sorted(
+            d for d in city_data.keys() if isinstance(d, str) and cutoff <= d < local_today
+        )
+        if not target_dates:
+            return {"ok": True, "reason": "no_target_dates", "updated": 0}
+
+        try:
+            min_target = datetime.strptime(target_dates[0], "%Y-%m-%d")
+            span_hours = int((local_now - min_target).total_seconds() / 3600) + 12
+        except Exception:
+            span_hours = (lookback_days + 3) * 24
+        span_hours = max(72, min(240, span_hours))
+
+        url = (
+            f"https://aviationweather.gov/api/data/metar"
+            f"?ids={icao}&format=json&hours={span_hours}"
+        )
+        resp = requests.get(url, timeout=12)
+        resp.raise_for_status()
+        rows = resp.json() or []
+        if not isinstance(rows, list):
+            rows = []
+
+        daily_max_c = {}
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            temp = row.get("temp")
+            if temp is None:
+                continue
+            obs_dt = _parse_metar_row_time(row)
+            if obs_dt is None:
+                continue
+            local_dt = obs_dt + timedelta(seconds=tz_offset)
+            d = local_dt.strftime("%Y-%m-%d")
+            if d < cutoff or d >= local_today:
+                continue
+            try:
+                t = float(temp)
+            except Exception:
+                continue
+            prev = daily_max_c.get(d)
+            if prev is None or t > prev:
+                daily_max_c[d] = t
+
+        updated = 0
+        for d in target_dates:
+            t_c = daily_max_c.get(d)
+            if t_c is None:
+                continue
+            corrected = round(t_c * 9 / 5 + 32, 1) if use_fahrenheit else round(t_c, 1)
+            rec = city_data.get(d) or {}
+            old = rec.get("actual_high")
+            try:
+                old_val = float(old) if old is not None else None
+            except Exception:
+                old_val = None
+            if old_val is None or abs(old_val - corrected) >= 0.1:
+                rec["actual_high"] = corrected
+                city_data[d] = rec
+                updated += 1
+
+        if updated > 0:
+            data[city_key] = city_data
+            save_history(history_file, data)
+
+        return {
+            "ok": True,
+            "updated": updated,
+            "scanned_dates": len(target_dates),
+            "metar_rows": len(rows),
+            "icao": icao,
+        }
+    except Exception as e:
+        return {"ok": False, "reason": str(e), "updated": 0}
+
+
 def update_daily_record(
     city_name, date_str, forecasts, actual_high, deb_prediction=None,
     mu=None, probabilities=None