From 7329dd728ea3950f38894c58a51b0f14af086525 Mon Sep 17 00:00:00 2001
From: "2569718930@qq.com" <2569718930@qq.com>
Date: Wed, 4 Mar 2026 02:52:31 +0800
Subject: [PATCH] refactor: extract trend_engine, unify probability engine
across bot and web, add unit tests
---
bot_listener.py | 490 +------------------------------
src/analysis/trend_engine.py | 538 +++++++++++++++++++++++++++++++++++
tests/test_trend_engine.py | 262 +++++++++++++++++
web/app.py | 164 ++---------
4 files changed, 833 insertions(+), 621 deletions(-)
create mode 100644 src/analysis/trend_engine.py
create mode 100644 tests/test_trend_engine.py
diff --git a/bot_listener.py b/bot_listener.py
index 974d2fa6..721229c9 100644
--- a/bot_listener.py
+++ b/bot_listener.py
@@ -16,492 +16,10 @@ from src.analysis.deb_algorithm import calculate_dynamic_weights, update_daily_r
def analyze_weather_trend(weather_data, temp_symbol, city_name=None):
- """根据实测与预测分析气温态势，增加峰值时刻预测"""
- insights: List[str] = []
- ai_features: List[str] = []
- mu = None
- sorted_probs = []
- _deb_to_save = None
-
- metar = weather_data.get("metar", {})
- open_meteo = weather_data.get("open-meteo", {})
- mb = weather_data.get("meteoblue", {})
- nws = weather_data.get("nws", {})
- weather_data.get("mgm", {})
-
- if not metar or not open_meteo:
- return "", ""
-
- # 数值归一化：防止 JSON 反序列化后的 str 类型炸数学运算
- def _sf(v):
- """safe float"""
- if v is None:
- return None
- try:
- return float(v)
- except Exception:
- return None
-
- _sf(metar.get("current", {}).get("temp"))
- max_so_far = _sf(metar.get("current", {}).get("max_temp_so_far"))
- daily = open_meteo.get("daily", {})
- hourly = open_meteo.get("hourly", {})
- times = hourly.get("time", [])
- temps = hourly.get("temperature_2m", [])
-
- # === 新核心：动态集合权重预报 (DEB) ===
- # 抽取各个确定性预报值构成的字典
- current_forecasts = {}
- if daily.get("temperature_2m_max"):
- current_forecasts["Open-Meteo"] = _sf(daily.get("temperature_2m_max")[0])
- if mb.get("today_high") is not None:
- current_forecasts["Meteoblue"] = _sf(mb.get("today_high"))
- if nws.get("today_high") is not None:
- current_forecasts["NWS"] = _sf(nws.get("today_high"))
-
- mm_forecasts = weather_data.get("multi_model", {}).get("forecasts", {})
- for m_name, m_val in mm_forecasts.items():
- if m_val is not None:
- current_forecasts[m_name] = _sf(m_val)
-
- # 从 URL/入参里我们暂时拿不到城名，为了 DEB 追溯我们在后方的总控那里提取。这里的 analyze_weather_trend 主要计算最高预留。
- forecast_highs = [h for h in current_forecasts.values() if h is not None]
- forecast_high = max(forecast_highs) if forecast_highs else None
- min(forecast_highs) if forecast_highs else forecast_high
- forecast_median = (
- sorted(forecast_highs)[len(forecast_highs) // 2] if forecast_highs else None
- )
-
- wind_speed = metar.get("current", {}).get("wind_speed_kt", 0)
-
- # 获取当地时间小时和分钟
- local_time_full = open_meteo.get("current", {}).get("local_time", "")
- try:
- local_date_str = local_time_full.split(" ")[0]
- time_parts = local_time_full.split(" ")[1].split(":")
- local_hour = int(time_parts[0])
- local_minute = int(time_parts[1]) if len(time_parts) > 1 else 0
- except Exception:
- from datetime import datetime
-
- local_date_str = datetime.now().strftime("%Y-%m-%d")
- local_hour = datetime.now().hour
- local_minute = datetime.now().minute
- local_hour_frac = local_hour + local_minute / 60 # 含分钟的精确小时
-
- # === DEB 融合渲染 ===
- if city_name and current_forecasts:
- blended_high, weight_info = calculate_dynamic_weights(
- city_name, current_forecasts
- )
- if blended_high is not None:
- insights.insert(
- 0,
- f"🧬 DEB 融合预测：{blended_high}{temp_symbol} ({weight_info})",
- )
- ai_features.append(
- f"🧬 DEB系统已通过历史偏差矫正算出期待点是: {blended_high}{temp_symbol}。"
- )
-
- # daily_record 将在函数末尾统一保存（含 μ 和概率快照）
- _deb_to_save = blended_high
-
- # === METAR 趋势分析 (移到前部判断降温) ===
- recent_temps = metar.get("recent_temps", [])
- trend_desc = ""
- if len(recent_temps) >= 2:
- temps_only = [t for _, t in recent_temps]
- latest_val = temps_only[0]
- prev_val = temps_only[1]
- diff = latest_val - prev_val
- if len(temps_only) >= 3:
- all_same = all(t == latest_val for t in temps_only[:3])
- all_rising = all(
- temps_only[i] >= temps_only[i + 1]
- for i in range(min(3, len(temps_only)) - 1)
- )
- all_falling = all(
- temps_only[i] <= temps_only[i + 1]
- for i in range(min(3, len(temps_only)) - 1)
- )
- trend_display = " → ".join(
- [f"{t}{temp_symbol}@{tm}" for tm, t in recent_temps[:3]]
- )
- if all_same:
- trend_desc = f"📉 温度已停滞（{trend_display}），大概率到顶。"
- elif all_rising and diff > 0:
- trend_desc = f"📈 仍在升温（{trend_display}）。"
- elif all_falling and diff < 0:
- trend_desc = f"📉 已开始降温（{trend_display}）。"
- else:
- trend_desc = f"📊 温度波动中（{trend_display}）。"
- elif diff == 0:
- trend_desc = f"📉 温度持平（最近两条都是 {latest_val}{temp_symbol}）。"
- elif diff > 0:
- trend_desc = f"📈 仍在升温（{prev_val} → {latest_val}{temp_symbol}）。"
- else:
- trend_desc = f"📉 已开始降温（{prev_val} → {latest_val}{temp_symbol}）。"
-
- is_cooling = "降温" in trend_desc
-
- om_today = daily.get("temperature_2m_max", [None])[0]
-
- # === 峰值时刻预测（提前计算，供概率引擎使用）===
- # 只在白天时段 (8:00-19:00) 内搜索，避免夜间温差小时把凌晨误判为峰值
- peak_hours = []
- if times and temps and om_today is not None:
- for t_str, temp in zip(times, temps):
- if t_str.startswith(local_date_str) and abs(temp - om_today) <= 0.2:
- hour = int(t_str.split("T")[1][:2])
- if 8 <= hour <= 19: # 只考虑白天
- peak_hours.append(t_str.split("T")[1][:5])
- if peak_hours:
- first_peak_h = int(peak_hours[0].split(":")[0])
- last_peak_h = int(peak_hours[-1].split(":")[0])
- else:
- first_peak_h, last_peak_h = 13, 15
-
- # === 集合预报区间 ===
- ensemble = weather_data.get("ensemble", {})
- ens_p10 = _sf(ensemble.get("p10"))
- ens_p90 = _sf(ensemble.get("p90"))
- ens_median = _sf(ensemble.get("median"))
- if ens_p10 is not None and ens_p90 is not None and ens_median is not None:
- msg1 = f"📊 集合预报：中位数 {ens_median}{temp_symbol}，90% 区间 [{ens_p10}{temp_symbol} - {ens_p90}{temp_symbol}]。"
- if not is_cooling:
- insights.append(msg1)
- ai_features.append(msg1)
-
- if om_today is not None:
- if om_today > ens_p90 and (
- max_so_far is None or max_so_far < om_today - 0.5
- ):
- msg2 = f"⚡ 预报偏高：确定性预报 {om_today}{temp_symbol} 超集合90%上限，更可能接近 {ens_median}{temp_symbol}。"
- ai_features.append(msg2)
- elif om_today < ens_p10 and (max_so_far is None or max_so_far < ens_median):
- msg2 = f"⚡ 预报偏低：确定性预报 {om_today}{temp_symbol} 低于集合90%下限，更可能接近 {ens_median}{temp_symbol}。"
- ai_features.append(msg2)
-
- # === 数学概率计算（基于集合预报正态分布拟合）===
- import math as _math
-
- # 用 P10/P90 反推标准差: P10 = median - 1.28*sigma, P90 = median + 1.28*sigma
- sigma = (ens_p90 - ens_p10) / 2.56
- if sigma < 0.1:
- sigma = 0.1 # 防止除以零
-
- # 用 DEB 历史 MAE 作为 σ 的下限
- # 如果模型过去的平均误差远大于集合预报的 σ，说明集合低估了真实不确定性
- if city_name:
- from src.analysis.deb_algorithm import get_deb_accuracy
-
- acc = get_deb_accuracy(city_name)
- if acc:
- _, hist_mae, _, _ = acc
- if hist_mae > sigma:
- sigma = hist_mae
-
- # === Shock Score: 气象突变软评分 (0~1) ===
- # 用近 4 条 METAR 的风向/云量/气压变化评估环境稳定性
- # 越高 = 越不稳定 = σ 放宽
- shock_score = 0.0
- recent_obs = metar.get("recent_obs", [])
- if len(recent_obs) >= 2:
- oldest = recent_obs[-1] # 最早
- newest = recent_obs[0] # 最新
-
- # ① 风向变化项 (0~0.4)
- # 角度差 × 风速放大系数（风速 > 10kt 时权重高，弱风忽略噪声）
- wdir_old = _sf(oldest.get("wdir"))
- wdir_new = _sf(newest.get("wdir"))
- wspd_new = _sf(newest.get("wspd")) or 0
- if wdir_old is not None and wdir_new is not None:
- angle_diff = abs(wdir_new - wdir_old)
- if angle_diff > 180:
- angle_diff = 360 - angle_diff
- wind_weight = min(wspd_new / 15.0, 1.0)
- wind_shock = min(angle_diff / 90.0, 1.0) * wind_weight * 0.4
- shock_score += wind_shock
-
- # ② 云量阶跃项 (0~0.35)
- # CLR=0, FEW=1, SCT=2, BKN=3, OVC=4
- cloud_old = oldest.get("cloud_rank", 0)
- cloud_new = newest.get("cloud_rank", 0)
- cloud_jump = abs(cloud_new - cloud_old)
- cloud_shock = min(cloud_jump / 3.0, 1.0) * 0.35
- shock_score += cloud_shock
-
- # ③ 气压变化项 (0~0.25)
- # 2h 内气压变化超过 2hPa 视为异常
- altim_old = _sf(oldest.get("altim"))
- altim_new = _sf(newest.get("altim"))
- if altim_old is not None and altim_new is not None:
- press_diff = abs(altim_new - altim_old)
- press_shock = min(press_diff / 4.0, 1.0) * 0.25
- shock_score += press_shock
-
- # 应用 shock_score: 放宽 σ
- if shock_score > 0.05:
- sigma *= 1 + 0.5 * shock_score
-
- # 时间修正：根据当前时间距峰值的位置调整 σ
- # 峰值前：σ 不变（不确定性最大）
- # 峰值窗口内：σ 缩小 30%（正在定型）
- # 峰值后：σ 缩小 70%（基本确定）
- if local_hour_frac > last_peak_h:
- sigma *= 0.3 # 峰值已过，结果基本锁定
- elif first_peak_h <= local_hour_frac <= last_peak_h:
- sigma *= 0.7 # 正在峰值窗口
- # === 判定是否为“死盘” (Dead Market) ===
- # 逻辑：深夜且气温大幅回落，或者已过峰值时段且明显降温
- is_dead_market = False
- current_temp = _sf(metar.get("current", {}).get("temp"))
- if max_so_far is not None and current_temp is not None:
- # 深夜死盘：21:00 后，回落超过 3°C
- if local_hour >= 21 and max_so_far - current_temp >= 3.0:
- is_dead_market = True
- # 峰值后死盘：已过最热窗口，回落超过 1.5°C
- elif local_hour > last_peak_h and max_so_far - current_temp >= 1.5:
- is_dead_market = True
-
- if ens_p10 is not None and ens_p90 is not None and not is_dead_market:
- # --- Reality-anchored μ ---
- # Determine peak status
- if local_hour_frac > last_peak_h:
- _peak_status = "past"
- elif first_peak_h <= local_hour_frac <= last_peak_h:
- _peak_status = "in_window"
- else:
- _peak_status = "before"
-
- # Compute forecast miss magnitude
- forecast_miss_deg = 0
- if max_so_far is not None and forecast_median is not None:
- forecast_miss_deg = round(forecast_median - max_so_far, 1)
-
- # If peak is past/in_window AND actual max is significantly below
- # forecasts, anchor μ on actual max, not on failed predictions
- if (
- max_so_far is not None
- and forecast_median is not None
- and _peak_status in ("past", "in_window")
- and max_so_far < forecast_median - 2.0
- ):
- if is_cooling or _peak_status == "past":
- mu = max_so_far
- else:
- mu = max_so_far + 0.5
- else:
- mu = (
- forecast_median * 0.7 + ens_median * 0.3
- if forecast_median is not None
- else ens_median
- )
- if max_so_far is not None and max_so_far > mu:
- mu = max_so_far + (0.3 if not is_cooling else 0.0)
-
- # Inject forecast miss severity for AI
- if forecast_miss_deg > 2.0 and _peak_status in ("past", "in_window"):
- if forecast_miss_deg > 5.0:
- severity = "重"
- elif forecast_miss_deg > 3.0:
- severity = "中"
- else:
- severity = "轻"
- min_fc = min(
- (v for v in forecast_highs if v is not None), default=None
- )
- _trend_dir = "降温" if is_cooling else ("停滞" if "停滞" in trend_desc else "升温")
- ai_features.append(
- f"🚨 预报崩盘 [{severity}级失准]: 最低预报 {min_fc}{temp_symbol} vs 实测最高 {max_so_far}{temp_symbol}，"
- f"偏差 {forecast_miss_deg}°。当前趋势: {_trend_dir}。"
- )
-
- def _norm_cdf(x, m, s):
- return 0.5 * (1 + _math.erf((x - m) / (s * _math.sqrt(2))))
-
- min_possible_wu = round(max_so_far) if max_so_far is not None else -999
- probs = {}
- for n in range(round(mu) - 2, round(mu) + 3):
- if n < min_possible_wu:
- continue
- p = _norm_cdf(n + 0.5, mu, sigma) - _norm_cdf(n - 0.5, mu, sigma)
- if p > 0.01:
- probs[n] = p
-
- total_p = sum(probs.values())
- if total_p > 0:
- probs = {k: v / total_p for k, v in probs.items()}
- sorted_probs = sorted(probs.items(), key=lambda x: x[1], reverse=True)
- prob_parts = [
- f"{int(t)}{temp_symbol} [{t - 0.5}~{t + 0.5}) {p * 100:.0f}%"
- for t, p in sorted_probs[:4]
- ]
- if prob_parts:
- prob_str = " | ".join(prob_parts)
- insights.append(f"🎲 结算概率 (μ={mu:.1f})：{prob_str}")
- ai_features.append(f"🎲 数学概率分布：{prob_str}")
- elif is_dead_market:
- settled_wu = round(max_so_far) if max_so_far is not None else "N/A"
- dead_msg = f"🎲 结算预测：已锁定 {settled_wu}{temp_symbol} (死盘确认)"
- insights.append(dead_msg)
- ai_features.append("🎲 状态: 确认死盘，结算已无悬念。")
-
- # === 实测已超预报 & 趋势输出 ===
- if max_so_far is not None and forecast_high is not None:
- if max_so_far > forecast_high + 0.5:
- exceed_by = max_so_far - forecast_high
- bt_msg = f"🚨 实测已超预报：{max_so_far}{temp_symbol} 超过上限 {forecast_high}{temp_symbol}（+{exceed_by:.1f}°）。"
- insights.append(bt_msg)
- ai_features.append(
- f"🚨 异常: 实测已冲破所有预报上限 ({max_so_far}{temp_symbol} vs {forecast_high}{temp_symbol})。"
- )
- if trend_desc:
- ai_features.append(trend_desc)
- else:
- if trend_desc:
- ai_features.append(trend_desc)
- elif trend_desc:
- ai_features.append(trend_desc)
-
- # === 结算取整分析 ===
- if max_so_far is not None:
- settled = round(max_so_far)
- fractional = max_so_far - int(max_so_far)
- dist_to_boundary = abs(fractional - 0.5)
- if dist_to_boundary <= 0.3:
- if fractional < 0.5:
- msg = f"⚖️ 结算边界：当前最高 {max_so_far}{temp_symbol} → WU 结算 {settled}{temp_symbol}，但只差 {0.5 - fractional:.1f}° 就会进位到 {settled + 1}{temp_symbol}！"
- else:
- msg = f"⚖️ 结算边界：当前最高 {max_so_far}{temp_symbol} → WU 结算 {settled}{temp_symbol}，刚刚越过进位线，再降 {fractional - 0.5:.1f}° 就会回落到 {settled - 1}{temp_symbol}。"
- insights.append(msg)
- ai_features.append(msg)
-
- # === 峰值时刻 AI 提示 ===
- if peak_hours:
- window = (
- f"{peak_hours[0]} - {peak_hours[-1]}"
- if len(peak_hours) > 1
- else peak_hours[0]
- )
-
- if local_hour <= last_peak_h:
- if last_peak_h < 6:
- ai_features.append(
- "⚠️ 提示：预测最热在凌晨，后续气温可能一路走低。"
- )
- elif local_hour < first_peak_h and (
- max_so_far is None or max_so_far < forecast_high
- ):
- target_temp = om_today if om_today is not None else forecast_high
- ai_features.append(
- f"🎯 关注重点：看看那个时段能否涨到 {target_temp}{temp_symbol}。"
- )
-
- # 写给AI（使用精确到分钟的时间）
- remain_hrs = first_peak_h - local_hour_frac
- if local_hour_frac > last_peak_h:
- ai_features.append(f"⏱️ 状态: 预报峰值时段已过 ({window})。")
- elif first_peak_h <= local_hour_frac <= last_peak_h:
- remain_in_window = last_peak_h - local_hour_frac
- if remain_in_window < 1:
- ai_features.append(
- f"⏱️ 状态: 正处于预报最热窗口 ({window})内，距窗口结束约 {int(remain_in_window * 60)} 分钟。"
- )
- else:
- ai_features.append(
- f"⏱️ 状态: 正处于预报最热窗口 ({window})内，距窗口结束约 {remain_in_window:.1f}h。"
- )
- elif remain_hrs < 1:
- ai_features.append(
- f"⏱️ 状态: 距最热时段开始还有约 {int(remain_hrs * 60)} 分钟 ({window})，尚未进入峰值窗口。"
- )
- else:
- ai_features.append(
- f"⏱️ 状态: 距最热时段开始还有约 {remain_hrs:.1f}h ({window})。"
- )
-
- # === 其他 AI 专供的事实特征 ===
- # 明确告知 AI 当前实测温度和今日最高温，避免 AI 从趋势数据中误读
- current_temp = metar.get("current", {}).get("temp")
- if current_temp is not None:
- ai_features.append(f"🌡️ 当前实测温度: {current_temp}{temp_symbol}。")
- if max_so_far is not None:
- ai_features.append(
- f"🏔️ 今日实测最高温: {max_so_far}{temp_symbol} (WU结算={round(max_so_far)}{temp_symbol})。"
- )
-
- # 传递城市的 METAR 取整特性给 AI
- from src.data_collection.city_risk_profiles import get_city_risk_profile
-
- if city_name:
- _profile = get_city_risk_profile(city_name)
- if _profile and _profile.get("metar_rounding"):
- ai_features.append(f"⚠️ METAR特性: {_profile['metar_rounding']}")
- if wind_speed:
- wind_dir = metar.get("current", {}).get("wind_dir", "未知")
- ai_features.append(f"🌬️ 当下风况: 约 {wind_speed}kt (方向 {wind_dir}°)。")
- humidity = metar.get("current", {}).get("humidity")
- if humidity and humidity > 80:
- ai_features.append(f"💦 湿度极高 ({humidity}%)。")
-
- clouds = metar.get("current", {}).get("clouds", [])
- if clouds:
- cover = clouds[-1].get("cover", "")
- c_desc = {"OVC": "全阴", "BKN": "多云", "SCT": "散云", "FEW": "少云"}.get(
- cover, cover
- )
- ai_features.append(f"☁️ 天空状况: {c_desc}。")
-
- wx_desc = metar.get("current", {}).get("wx_desc")
- if wx_desc:
- ai_features.append(f"🌧️ 天气现象: {wx_desc}。")
-
- max_temp_time_str = metar.get("current", {}).get("max_temp_time", "")
- if max_so_far is not None and max_temp_time_str:
- try:
- max_h = int(max_temp_time_str.split(":")[0])
- max_temp_rad = 0.0
- hourly_rad = hourly.get("shortwave_radiation", [])
- for t_str, rad in zip(times, hourly_rad):
- if (
- t_str.startswith(local_date_str)
- and int(t_str.split("T")[1][:2]) == max_h
- ):
- max_temp_rad = rad if rad is not None else 0.0
- break
- if max_temp_rad < 50:
- ai_features.append(
- f"🌙 动力事实: 最高温出现在低辐射时段 ({max_temp_time_str}, 辐射{max_temp_rad:.0f}W/m²)。"
- )
- except Exception:
- pass
-
- # === 统一保存今日记录（含 μ 和概率快照）===
- try:
- _prob_list = None
- if sorted_probs:
- _prob_list = [
- {"value": int(t), "probability": round(p, 3)}
- for t, p in sorted_probs[:4]
- ]
- elif is_dead_market and max_so_far is not None:
- _prob_list = [{"value": round(max_so_far), "probability": 1.0}]
-
- update_daily_record(
- city_name,
- local_date_str,
- current_forecasts,
- max_so_far,
- deb_prediction=_deb_to_save,
- mu=mu,
- probabilities=_prob_list,
- )
- except Exception:
- pass
-
- display_str = "\n".join(insights) if insights else ""
- return display_str, "\n".join(ai_features)
+ """Thin wrapper — delegates to shared trend_engine module."""
+ from src.analysis.trend_engine import analyze_weather_trend as _analyze
+ display_str, ai_context, _structured = _analyze(weather_data, temp_symbol, city_name)
+ return display_str, ai_context
def start_bot():
diff --git a/src/analysis/trend_engine.py b/src/analysis/trend_engine.py
new file mode 100644
index 00000000..51c98326
--- /dev/null
+++ b/src/analysis/trend_engine.py
@@ -0,0 +1,538 @@
+"""
+Trend Engine — Shared weather analysis module
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Extracted from bot_listener.py to provide a single source of truth
+for both Telegram bot and web dashboard.
+"""
+
+import math
+from typing import List, Optional, Tuple, Dict, Any
+
+from src.analysis.deb_algorithm import (
+ calculate_dynamic_weights,
+ get_deb_accuracy,
+ update_daily_record,
+)
+from src.data_collection.city_risk_profiles import get_city_risk_profile
+
+
+def _sf(v):
+ """Safe float conversion — prevents JSON str types from breaking math."""
+ if v is None:
+ return None
+ try:
+ return float(v)
+ except Exception:
+ return None
+
+
+def analyze_weather_trend(
+ weather_data: dict,
+ temp_symbol: str,
+ city_name: Optional[str] = None,
+) -> Tuple[str, str, Dict[str, Any]]:
+ """
+ Analyze weather trend from multi-source data.
+
+ Returns:
+ (display_str, ai_context, structured_data)
+
+ display_str: HTML-formatted insights for Telegram display
+ ai_context: plain-text context for AI analysis
+ structured_data: dict with computed values for direct use:
+ - mu: probability center
+ - probabilities: [{value, range, probability}, ...]
+ - trend_info: {direction, recent, is_cooling, is_dead_market}
+ - peak_status: "before" / "in_window" / "past"
+ - peak_hours: list of peak hour strings
+ - deb_prediction: DEB blended value
+ - current_forecasts: {model: temp, ...}
+ - forecast_miss_deg: float
+ - max_so_far: float
+ - cur_temp: float
+ - wu_settle: int
+ """
+ insights: List[str] = []
+ ai_features: List[str] = []
+ mu = None
+ sorted_probs = []
+ _deb_to_save = None
+
+ metar = weather_data.get("metar", {})
+ open_meteo = weather_data.get("open-meteo", {})
+ mb = weather_data.get("meteoblue", {})
+ nws = weather_data.get("nws", {})
+
+ empty_result = ("", "", {})
+ if not metar or not open_meteo:
+ return empty_result
+
+ max_so_far = _sf(metar.get("current", {}).get("max_temp_so_far"))
+ cur_temp = _sf(metar.get("current", {}).get("temp"))
+ daily = open_meteo.get("daily", {})
+ hourly = open_meteo.get("hourly", {})
+ times = hourly.get("time", [])
+ temps = hourly.get("temperature_2m", [])
+
+ # === Forecasts ===
+ current_forecasts: Dict[str, Optional[float]] = {}
+ if daily.get("temperature_2m_max"):
+ current_forecasts["Open-Meteo"] = _sf(daily.get("temperature_2m_max")[0])
+ if mb.get("today_high") is not None:
+ current_forecasts["Meteoblue"] = _sf(mb.get("today_high"))
+ if nws.get("today_high") is not None:
+ current_forecasts["NWS"] = _sf(nws.get("today_high"))
+
+ mm_forecasts = weather_data.get("multi_model", {}).get("forecasts", {})
+ for m_name, m_val in mm_forecasts.items():
+ if m_val is not None:
+ current_forecasts[m_name] = _sf(m_val)
+
+ forecast_highs = [h for h in current_forecasts.values() if h is not None]
+ forecast_high = max(forecast_highs) if forecast_highs else None
+ forecast_median = (
+ sorted(forecast_highs)[len(forecast_highs) // 2] if forecast_highs else None
+ )
+
+ wind_speed = metar.get("current", {}).get("wind_speed_kt", 0)
+
+ # === Local time ===
+ local_time_full = open_meteo.get("current", {}).get("local_time", "")
+ try:
+ local_date_str = local_time_full.split(" ")[0]
+ time_parts = local_time_full.split(" ")[1].split(":")
+ local_hour = int(time_parts[0])
+ local_minute = int(time_parts[1]) if len(time_parts) > 1 else 0
+ except Exception:
+ from datetime import datetime
+ local_date_str = datetime.now().strftime("%Y-%m-%d")
+ local_hour = datetime.now().hour
+ local_minute = datetime.now().minute
+ local_hour_frac = local_hour + local_minute / 60
+
+ # === DEB ===
+ deb_prediction = None
+ deb_weights = ""
+ if city_name and current_forecasts:
+ blended_high, weight_info = calculate_dynamic_weights(
+ city_name, current_forecasts
+ )
+ if blended_high is not None:
+ deb_prediction = blended_high
+ deb_weights = weight_info
+ insights.insert(
+ 0,
+ f"🧬 DEB 融合预测：{blended_high}{temp_symbol} ({weight_info})",
+ )
+ ai_features.append(
+ f"🧬 DEB系统已通过历史偏差矫正算出期待点是: {blended_high}{temp_symbol}。"
+ )
+ _deb_to_save = blended_high
+
+ # === METAR trend ===
+ recent_temps = metar.get("recent_temps", [])
+ trend_desc = ""
+ trend_direction = "unknown"
+ if len(recent_temps) >= 2:
+ temps_only = [t for _, t in recent_temps]
+ latest_val = temps_only[0]
+ prev_val = temps_only[1]
+ diff = latest_val - prev_val
+ if len(temps_only) >= 3:
+ all_same = all(t == latest_val for t in temps_only[:3])
+ all_rising = all(
+ temps_only[i] >= temps_only[i + 1]
+ for i in range(min(3, len(temps_only)) - 1)
+ )
+ all_falling = all(
+ temps_only[i] <= temps_only[i + 1]
+ for i in range(min(3, len(temps_only)) - 1)
+ )
+ trend_display = " → ".join(
+ [f"{t}{temp_symbol}@{tm}" for tm, t in recent_temps[:3]]
+ )
+ if all_same:
+ trend_desc = f"📉 温度已停滞（{trend_display}），大概率到顶。"
+ trend_direction = "stagnant"
+ elif all_rising and diff > 0:
+ trend_desc = f"📈 仍在升温（{trend_display}）。"
+ trend_direction = "rising"
+ elif all_falling and diff < 0:
+ trend_desc = f"📉 已开始降温（{trend_display}）。"
+ trend_direction = "falling"
+ else:
+ trend_desc = f"📊 温度波动中（{trend_display}）。"
+ trend_direction = "mixed"
+ elif diff == 0:
+ trend_desc = f"📉 温度持平（最近两条都是 {latest_val}{temp_symbol}）。"
+ trend_direction = "stagnant"
+ elif diff > 0:
+ trend_desc = f"📈 仍在升温（{prev_val} → {latest_val}{temp_symbol}）。"
+ trend_direction = "rising"
+ else:
+ trend_desc = f"📉 已开始降温（{prev_val} → {latest_val}{temp_symbol}）。"
+ trend_direction = "falling"
+
+ is_cooling = "降温" in trend_desc
+
+ om_today = daily.get("temperature_2m_max", [None])[0]
+
+ # === Peak hours ===
+ peak_hours = []
+ if times and temps and om_today is not None:
+ for t_str, temp in zip(times, temps):
+ if t_str.startswith(local_date_str) and abs(temp - om_today) <= 0.2:
+ hour = int(t_str.split("T")[1][:2])
+ if 8 <= hour <= 19:
+ peak_hours.append(t_str.split("T")[1][:5])
+ if peak_hours:
+ first_peak_h = int(peak_hours[0].split(":")[0])
+ last_peak_h = int(peak_hours[-1].split(":")[0])
+ else:
+ first_peak_h, last_peak_h = 13, 15
+
+ # Peak status
+ if local_hour_frac > last_peak_h:
+ peak_status = "past"
+ elif first_peak_h <= local_hour_frac <= last_peak_h:
+ peak_status = "in_window"
+ else:
+ peak_status = "before"
+
+ # === Ensemble ===
+ ensemble = weather_data.get("ensemble", {})
+ ens_p10 = _sf(ensemble.get("p10"))
+ ens_p90 = _sf(ensemble.get("p90"))
+ ens_median = _sf(ensemble.get("median"))
+ ens_data = {"p10": ens_p10, "p90": ens_p90, "median": ens_median}
+
+ sigma = None
+ if ens_p10 is not None and ens_p90 is not None and ens_median is not None:
+ msg1 = (
+ f"📊 集合预报：中位数 {ens_median}{temp_symbol}，"
+ f"90% 区间 [{ens_p10}{temp_symbol} - {ens_p90}{temp_symbol}]。"
+ )
+ if not is_cooling:
+ insights.append(msg1)
+ ai_features.append(msg1)
+
+ if om_today is not None:
+ if om_today > ens_p90 and (
+ max_so_far is None or max_so_far < om_today - 0.5
+ ):
+ ai_features.append(
+ f"⚡ 预报偏高：确定性预报 {om_today}{temp_symbol} 超集合90%上限，"
+ f"更可能接近 {ens_median}{temp_symbol}。"
+ )
+ elif om_today < ens_p10 and (
+ max_so_far is None or max_so_far < ens_median
+ ):
+ ai_features.append(
+ f"⚡ 预报偏低：确定性预报 {om_today}{temp_symbol} 低于集合90%下限，"
+ f"更可能接近 {ens_median}{temp_symbol}。"
+ )
+
+ # === Sigma calculation ===
+ sigma = (ens_p90 - ens_p10) / 2.56
+ if sigma < 0.1:
+ sigma = 0.1
+
+ # MAE floor
+ if city_name:
+ acc = get_deb_accuracy(city_name)
+ if acc:
+ _, hist_mae, _, _ = acc
+ if hist_mae > sigma:
+ sigma = hist_mae
+
+ # Shock Score
+ shock_score = 0.0
+ recent_obs = metar.get("recent_obs", [])
+ if len(recent_obs) >= 2:
+ oldest = recent_obs[-1]
+ newest = recent_obs[0]
+ wdir_old = _sf(oldest.get("wdir"))
+ wdir_new = _sf(newest.get("wdir"))
+ wspd_new = _sf(newest.get("wspd")) or 0
+ if wdir_old is not None and wdir_new is not None:
+ angle_diff = abs(wdir_new - wdir_old)
+ if angle_diff > 180:
+ angle_diff = 360 - angle_diff
+ wind_weight = min(wspd_new / 15.0, 1.0)
+ shock_score += min(angle_diff / 90.0, 1.0) * wind_weight * 0.4
+ cloud_old = oldest.get("cloud_rank", 0)
+ cloud_new = newest.get("cloud_rank", 0)
+ shock_score += min(abs(cloud_new - cloud_old) / 3.0, 1.0) * 0.35
+ altim_old = _sf(oldest.get("altim"))
+ altim_new = _sf(newest.get("altim"))
+ if altim_old is not None and altim_new is not None:
+ shock_score += min(abs(altim_new - altim_old) / 4.0, 1.0) * 0.25
+
+ if shock_score > 0.05:
+ sigma *= 1 + 0.5 * shock_score
+
+ # Time decay
+ if local_hour_frac > last_peak_h:
+ sigma *= 0.3
+ elif first_peak_h <= local_hour_frac <= last_peak_h:
+ sigma *= 0.7
+
+ # === Dead Market ===
+ is_dead_market = False
+ if max_so_far is not None and cur_temp is not None:
+ if local_hour >= 21 and max_so_far - cur_temp >= 3.0:
+ is_dead_market = True
+ elif local_hour > last_peak_h and max_so_far - cur_temp >= 1.5:
+ is_dead_market = True
+
+ # === Probability Engine ===
+ probabilities: List[Dict[str, Any]] = []
+ forecast_miss_deg = 0.0
+
+ if ens_p10 is not None and ens_p90 is not None and not is_dead_market:
+ # Forecast miss magnitude
+ if max_so_far is not None and forecast_median is not None:
+ forecast_miss_deg = round(forecast_median - max_so_far, 1)
+
+ # Reality-anchored μ
+ if (
+ max_so_far is not None
+ and forecast_median is not None
+ and peak_status in ("past", "in_window")
+ and max_so_far < forecast_median - 2.0
+ ):
+ if is_cooling or peak_status == "past":
+ mu = max_so_far
+ else:
+ mu = max_so_far + 0.5
+ else:
+ mu = (
+ forecast_median * 0.7 + ens_median * 0.3
+ if forecast_median is not None
+ else ens_median
+ )
+ if max_so_far is not None and max_so_far > mu:
+ mu = max_so_far + (0.3 if not is_cooling else 0.0)
+
+ # Forecast miss severity for AI
+ if forecast_miss_deg > 2.0 and peak_status in ("past", "in_window"):
+ severity = "重" if forecast_miss_deg > 5.0 else ("中" if forecast_miss_deg > 3.0 else "轻")
+ min_fc = min((v for v in forecast_highs if v is not None), default=None)
+ _trend_dir = "降温" if is_cooling else ("停滞" if "停滞" in trend_desc else "升温")
+ ai_features.append(
+ f"🚨 预报崩盘 [{severity}级失准]: 最低预报 {min_fc}{temp_symbol} vs "
+ f"实测最高 {max_so_far}{temp_symbol}，偏差 {forecast_miss_deg}°。当前趋势: {_trend_dir}。"
+ )
+
+ # Gaussian CDF
+ def _norm_cdf(x, m, s):
+ return 0.5 * (1 + math.erf((x - m) / (s * math.sqrt(2))))
+
+ min_possible_wu = round(max_so_far) if max_so_far is not None else -999
+ probs = {}
+ for n in range(round(mu) - 2, round(mu) + 3):
+ if n < min_possible_wu:
+ continue
+ p = _norm_cdf(n + 0.5, mu, sigma) - _norm_cdf(n - 0.5, mu, sigma)
+ if p > 0.01:
+ probs[n] = p
+
+ total_p = sum(probs.values())
+ if total_p > 0:
+ probs = {k: v / total_p for k, v in probs.items()}
+ sorted_probs = sorted(probs.items(), key=lambda x: x[1], reverse=True)
+ prob_parts = [
+ f"{int(t)}{temp_symbol} [{t - 0.5}~{t + 0.5}) {p * 100:.0f}%"
+ for t, p in sorted_probs[:4]
+ ]
+ if prob_parts:
+ prob_str = " | ".join(prob_parts)
+ insights.append(f"🎲 结算概率 (μ={mu:.1f})：{prob_str}")
+ ai_features.append(f"🎲 数学概率分布：{prob_str}")
+ for t, p in sorted_probs[:4]:
+ probabilities.append(
+ {"value": int(t), "range": f"[{t-0.5}~{t+0.5})", "probability": round(p, 3)}
+ )
+
+ elif is_dead_market:
+ settled_wu = round(max_so_far) if max_so_far is not None else 0
+ dead_msg = f"🎲 结算预测：已锁定 {settled_wu}{temp_symbol} (死盘确认)"
+ insights.append(dead_msg)
+ ai_features.append("🎲 状态: 确认死盘，结算已无悬念。")
+ if max_so_far is not None:
+ mu = max_so_far
+ probabilities = [
+ {"value": settled_wu, "range": f"[{settled_wu-0.5}~{settled_wu+0.5})", "probability": 1.0}
+ ]
+
+ # === Actual exceeds forecast ===
+ if max_so_far is not None and forecast_high is not None:
+ if max_so_far > forecast_high + 0.5:
+ exceed_by = max_so_far - forecast_high
+ bt_msg = (
+ f"🚨 实测已超预报：{max_so_far}{temp_symbol} 超过上限 "
+ f"{forecast_high}{temp_symbol}（+{exceed_by:.1f}°）。"
+ )
+ insights.append(bt_msg)
+ ai_features.append(
+ f"🚨 异常: 实测已冲破所有预报上限 ({max_so_far}{temp_symbol} vs {forecast_high}{temp_symbol})。"
+ )
+ if trend_desc:
+ ai_features.append(trend_desc)
+
+ # === Settlement boundary ===
+ if max_so_far is not None:
+ settled = round(max_so_far)
+ fractional = max_so_far - int(max_so_far)
+ dist_to_boundary = abs(fractional - 0.5)
+ if dist_to_boundary <= 0.3:
+ if fractional < 0.5:
+ msg = (
+ f"⚖️ 结算边界：当前最高 {max_so_far}{temp_symbol} → WU 结算 "
+ f"{settled}{temp_symbol}，但只差 {0.5 - fractional:.1f}° "
+ f"就会进位到 {settled + 1}{temp_symbol}！"
+ )
+ else:
+ msg = (
+ f"⚖️ 结算边界：当前最高 {max_so_far}{temp_symbol} → WU 结算 "
+ f"{settled}{temp_symbol}，刚刚越过进位线，再降 "
+ f"{fractional - 0.5:.1f}° 就会回落到 {settled - 1}{temp_symbol}。"
+ )
+ insights.append(msg)
+ ai_features.append(msg)
+
+ # === Peak window AI hints ===
+ if peak_hours:
+ window = (
+ f"{peak_hours[0]} - {peak_hours[-1]}"
+ if len(peak_hours) > 1
+ else peak_hours[0]
+ )
+ if local_hour <= last_peak_h:
+ if last_peak_h < 6:
+ ai_features.append("⚠️ 提示：预测最热在凌晨，后续气温可能一路走低。")
+ elif local_hour < first_peak_h and (
+ max_so_far is None or max_so_far < forecast_high
+ ):
+ target_temp = om_today if om_today is not None else forecast_high
+ ai_features.append(
+ f"🎯 关注重点：看看那个时段能否涨到 {target_temp}{temp_symbol}。"
+ )
+
+ remain_hrs = first_peak_h - local_hour_frac
+ if local_hour_frac > last_peak_h:
+ ai_features.append(f"⏱️ 状态: 预报峰值时段已过 ({window})。")
+ elif first_peak_h <= local_hour_frac <= last_peak_h:
+ remain_in_window = last_peak_h - local_hour_frac
+ if remain_in_window < 1:
+ ai_features.append(
+ f"⏱️ 状态: 正处于预报最热窗口 ({window})内，距窗口结束约 {int(remain_in_window * 60)} 分钟。"
+ )
+ else:
+ ai_features.append(
+ f"⏱️ 状态: 正处于预报最热窗口 ({window})内，距窗口结束约 {remain_in_window:.1f}h。"
+ )
+ elif remain_hrs < 1:
+ ai_features.append(
+ f"⏱️ 状态: 距最热时段开始还有约 {int(remain_hrs * 60)} 分钟 ({window})，尚未进入峰值窗口。"
+ )
+ else:
+ ai_features.append(f"⏱️ 状态: 距最热时段开始还有约 {remain_hrs:.1f}h ({window})。")
+
+ # === AI fact features ===
+ if cur_temp is not None:
+ ai_features.append(f"🌡️ 当前实测温度: {cur_temp}{temp_symbol}。")
+ if max_so_far is not None:
+ ai_features.append(
+ f"🏔️ 今日实测最高温: {max_so_far}{temp_symbol} (WU结算={round(max_so_far)}{temp_symbol})。"
+ )
+ if city_name:
+ _profile = get_city_risk_profile(city_name)
+ if _profile and _profile.get("metar_rounding"):
+ ai_features.append(f"⚠️ METAR特性: {_profile['metar_rounding']}")
+ if wind_speed:
+ wind_dir = metar.get("current", {}).get("wind_dir", "未知")
+ ai_features.append(f"🌬️ 当下风况: 约 {wind_speed}kt (方向 {wind_dir}°)。")
+ humidity = metar.get("current", {}).get("humidity")
+ if humidity and humidity > 80:
+ ai_features.append(f"💦 湿度极高 ({humidity}%)。")
+
+ clouds = metar.get("current", {}).get("clouds", [])
+ if clouds:
+ cover = clouds[-1].get("cover", "")
+ c_desc = {"OVC": "全阴", "BKN": "多云", "SCT": "散云", "FEW": "少云"}.get(cover, cover)
+ ai_features.append(f"☁️ 天空状况: {c_desc}。")
+
+ wx_desc = metar.get("current", {}).get("wx_desc")
+ if wx_desc:
+ ai_features.append(f"🌧️ 天气现象: {wx_desc}。")
+
+ max_temp_time_str = metar.get("current", {}).get("max_temp_time", "")
+ if max_so_far is not None and max_temp_time_str:
+ try:
+ max_h = int(max_temp_time_str.split(":")[0])
+ max_temp_rad = 0.0
+ hourly_rad = hourly.get("shortwave_radiation", [])
+ for t_str, rad in zip(times, hourly_rad):
+ if t_str.startswith(local_date_str) and int(t_str.split("T")[1][:2]) == max_h:
+ max_temp_rad = rad if rad is not None else 0.0
+ break
+ if max_temp_rad < 50:
+ ai_features.append(
+ f"🌙 动力事实: 最高温出现在低辐射时段 ({max_temp_time_str}, 辐射{max_temp_rad:.0f}W/m²)。"
+ )
+ except Exception:
+ pass
+
+ # === Save daily record (with μ + prob snapshot) ===
+ try:
+ _prob_list = None
+ if sorted_probs:
+ _prob_list = [
+ {"value": int(t), "probability": round(p, 3)}
+ for t, p in sorted_probs[:4]
+ ]
+ elif is_dead_market and max_so_far is not None:
+ _prob_list = [{"value": round(max_so_far), "probability": 1.0}]
+
+ update_daily_record(
+ city_name,
+ local_date_str,
+ current_forecasts,
+ max_so_far,
+ deb_prediction=_deb_to_save,
+ mu=mu,
+ probabilities=_prob_list,
+ )
+ except Exception:
+ pass
+
+ # === Build recent list for trend_info ===
+ recent_list = []
+ for tm, t in recent_temps[:4]:
+ recent_list.append({"time": tm, "temp": t})
+
+ # === Structured result ===
+ structured = {
+ "mu": mu,
+ "probabilities": probabilities,
+ "trend_info": {
+ "direction": trend_direction if 'trend_direction' in dir() else "unknown",
+ "recent": recent_list,
+ "is_cooling": is_cooling,
+ "is_dead_market": is_dead_market,
+ },
+ "peak_status": peak_status,
+ "peak_hours": peak_hours,
+ "deb_prediction": deb_prediction,
+ "deb_weights": deb_weights,
+ "current_forecasts": current_forecasts,
+ "ens_data": ens_data,
+ "forecast_miss_deg": forecast_miss_deg,
+ "max_so_far": max_so_far,
+ "cur_temp": cur_temp,
+ "wu_settle": round(max_so_far) if max_so_far is not None else None,
+ }
+
+ display_str = "\n".join(insights) if insights else ""
+ return display_str, "\n".join(ai_features), structured
diff --git a/tests/test_trend_engine.py b/tests/test_trend_engine.py
new file mode 100644
index 00000000..c2fecc4e
--- /dev/null
+++ b/tests/test_trend_engine.py
@@ -0,0 +1,262 @@
+"""
+Unit tests for trend_engine core logic.
+Tests: μ/σ calculation, dead market detection, forecast bust grading.
+"""
+
+import sys
+import os
+
+# Ensure project root is in path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import pytest
+from unittest.mock import patch
+from src.analysis.trend_engine import analyze_weather_trend, _sf
+
+
+# ─── Helpers ───
+
+def _make_weather_data(
+ cur_temp=25.0,
+ max_so_far=28.0,
+ om_today_high=30.0,
+ ens_median=29.0,
+ ens_p10=27.0,
+ ens_p90=31.0,
+ local_time="2026-03-04 14:30",
+ recent_temps=None,
+ multi_model=None,
+ recent_obs=None,
+):
+ """Build a minimal weather_data dict for testing."""
+ data = {
+ "metar": {
+ "current": {
+ "temp": cur_temp,
+ "max_temp_so_far": max_so_far,
+ "max_temp_time": "14:00",
+ "wind_speed_kt": 5,
+ "wind_dir": 180,
+ "humidity": 50,
+ "clouds": [{"cover": "SCT", "base": 5000}],
+ },
+ "recent_temps": recent_temps or [("14:00", 27.0), ("13:00", 26.0), ("12:00", 25.0)],
+ "recent_obs": recent_obs or [],
+ },
+ "open-meteo": {
+ "current": {"local_time": local_time},
+ "daily": {
+ "temperature_2m_max": [om_today_high],
+ "sunrise": ["06:30"],
+ "sunset": ["18:30"],
+ },
+ "hourly": {
+ "time": [f"2026-03-04T{h:02d}:00" for h in range(24)],
+ "temperature_2m": [15 + (h - 6) * 1.5 if 6 <= h <= 14 else 20 - (h - 14) * 0.5 for h in range(24)],
+ "shortwave_radiation": [0 if h < 6 or h > 18 else 200 + h * 20 for h in range(24)],
+ },
+ },
+ "ensemble": {
+ "median": ens_median,
+ "p10": ens_p10,
+ "p90": ens_p90,
+ },
+ "multi_model": {"forecasts": multi_model or {}},
+ "meteoblue": {},
+ "nws": {},
+ }
+ return data
+
+
+# ─── Tests: _sf ───
+
+class TestSafeFloat:
+ def test_none(self):
+ assert _sf(None) is None
+
+ def test_int(self):
+ assert _sf(5) == 5.0
+
+ def test_str_number(self):
+ assert _sf("3.14") == 3.14
+
+ def test_invalid_str(self):
+ assert _sf("abc") is None
+
+
+# ─── Tests: μ Calculation ───
+
+class TestMuCalculation:
+ @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_normal_mu_blends_forecast_and_ensemble(self, _udr, _deb_acc, _dw):
+ """Normal case: μ = forecast_median * 0.7 + ens_median * 0.3"""
+ data = _make_weather_data(
+ cur_temp=25.0, max_so_far=26.0,
+ om_today_high=30.0, ens_median=29.0,
+ local_time="2026-03-04 10:00" # Before peak window to prevent early bust override
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+
+ mu = sd["mu"]
+ # forecast_median = 30.0, ens_median = 29.0 → 30*0.7 + 29*0.3 = 29.7
+ assert mu is not None
+ assert 29.0 <= mu <= 31.0 # Reasonable range
+
+ @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_reality_anchored_mu_on_forecast_bust(self, _udr, _deb_acc, _dw):
+ """When past peak and actual << forecasts, μ anchors on actual max."""
+ data = _make_weather_data(
+ cur_temp=22.0, max_so_far=23.0,
+ om_today_high=30.0, ens_median=29.0,
+ local_time="2026-03-04 17:00", # Past peak
+ recent_temps=[("17:00", 22.0), ("16:00", 23.0), ("15:00", 23.0)],
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+
+ mu = sd["mu"]
+ # max_so_far=23 vs forecast_median=30 → bust → μ ≈ 23
+ assert mu is not None
+ assert mu <= 24.0, f"μ should anchor on actual max (23°C), got {mu}"
+
+ @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_mu_rises_when_actual_exceeds_forecast(self, _udr, _deb_acc, _dw):
+ """When actual max exceeds μ, μ adjusts upward."""
+ data = _make_weather_data(
+ cur_temp=32.0, max_so_far=33.0,
+ om_today_high=30.0, ens_median=29.0,
+ local_time="2026-03-04 14:00",
+ recent_temps=[("14:00", 32.0), ("13:00", 31.0), ("12:00", 30.0)],
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+
+ mu = sd["mu"]
+ assert mu is not None
+ assert mu >= 33.0, f"μ should be >= actual max (33°C), got {mu}"
+
+
+# ─── Tests: Dead Market ───
+
+class TestDeadMarket:
+ @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_dead_market_after_peak_with_cooling(self, _udr, _deb_acc, _dw):
+ """Past peak + 1.5°C drop → dead market."""
+ data = _make_weather_data(
+ cur_temp=26.0, max_so_far=28.0,
+ local_time="2026-03-04 17:00",
+ recent_temps=[("17:00", 26.0), ("16:00", 27.0), ("15:00", 28.0)],
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+
+ ti = sd["trend_info"]
+ assert ti["is_dead_market"] is True
+
+ @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_not_dead_market_during_peak_warming(self, _udr, _deb_acc, _dw):
+ """During peak window while still warming → NOT dead market."""
+ data = _make_weather_data(
+ cur_temp=28.0, max_so_far=28.0,
+ local_time="2026-03-04 14:00",
+ recent_temps=[("14:00", 28.0), ("13:00", 27.0), ("12:00", 26.0)],
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+
+ ti = sd["trend_info"]
+ assert ti["is_dead_market"] is False
+
+ @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_dead_market_probability_is_100_percent(self, _udr, _deb_acc, _dw):
+ """When dead market, probabilities collapse to 100% at settled value."""
+ data = _make_weather_data(
+ cur_temp=25.0, max_so_far=28.0,
+ local_time="2026-03-04 22:00", # Late night
+ recent_temps=[("22:00", 25.0), ("21:00", 26.0), ("20:00", 27.0)],
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+
+ assert sd["trend_info"]["is_dead_market"] is True
+ probs = sd["probabilities"]
+ assert len(probs) == 1
+ assert probs[0]["value"] == 28 # round(28.0)
+ assert probs[0]["probability"] == 1.0
+
+
+# ─── Tests: Forecast Bust Detection ───
+
+class TestForecastBust:
+ @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_heavy_forecast_bust_detected(self, _udr, _deb_acc, _dw):
+ """Heavy bust: forecast_median - max_so_far > 5.0"""
+ data = _make_weather_data(
+ cur_temp=22.0, max_so_far=23.0,
+ om_today_high=30.0, ens_median=29.0,
+ local_time="2026-03-04 16:00",
+ recent_temps=[("16:00", 22.0), ("15:00", 23.0), ("14:00", 23.0)],
+ )
+ _, ai_context, sd = analyze_weather_trend(data, "°C", "test_city")
+
+ # forecast_median=30, max_so_far=23 → miss = 7°C → heavy
+ assert "预报崩盘" in ai_context
+ assert "重" in ai_context or "级失准" in ai_context
+
+ @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_no_bust_when_on_track(self, _udr, _deb_acc, _dw):
+ """No bust when actual is close to forecast."""
+ data = _make_weather_data(
+ cur_temp=29.0, max_so_far=29.5,
+ om_today_high=30.0, ens_median=29.5,
+ local_time="2026-03-04 14:00",
+ recent_temps=[("14:00", 29.0), ("13:00", 28.5), ("12:00", 28.0)],
+ )
+ _, ai_context, _ = analyze_weather_trend(data, "°C", "test_city")
+
+ assert "预报崩盘" not in ai_context
+
+
+# ─── Tests: Trend Direction ───
+
+class TestTrendDirection:
+ @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_rising_trend(self, _udr, _deb_acc, _dw):
+ data = _make_weather_data(
+ recent_temps=[("14:00", 28.0), ("13:00", 27.0), ("12:00", 26.0)],
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+ assert sd["trend_info"]["direction"] == "rising"
+
+ @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_falling_trend(self, _udr, _deb_acc, _dw):
+ data = _make_weather_data(
+ recent_temps=[("16:00", 25.0), ("15:00", 26.0), ("14:00", 27.0)],
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+ assert sd["trend_info"]["direction"] == "falling"
+
+ @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_stagnant_trend(self, _udr, _deb_acc, _dw):
+ data = _make_weather_data(
+ recent_temps=[("14:00", 27.0), ("13:00", 27.0), ("12:00", 27.0)],
+ )
+ _, _, sd = analyze_weather_trend(data, "°C", "test_city")
+ assert sd["trend_info"]["direction"] == "stagnant"
diff --git a/web/app.py b/web/app.py
index 8b4f920c..e6abad17 100644
--- a/web/app.py
+++ b/web/app.py
@@ -268,126 +268,42 @@ def _analyze(city: str) -> Dict[str, Any]:
else:
peak_status = "before"
- # ── 10. Probability distribution ──
+ # ── 10. Shared analysis (probability, trend, AI) via trend_engine ──
+ # This single call replaces the duplicate probability engine, dead market
+ # detection, forecast bust grading, and AI context building.
+ from src.analysis.trend_engine import analyze_weather_trend as _trend_analyze
+ from src.analysis.ai_analyzer import get_ai_analysis
+
probabilities = []
mu = None
- forecast_miss_deg = 0 # How far actual is below forecasts
- if (
- ens_data["p10"] is not None
- and ens_data["p90"] is not None
- and ens_data["median"] is not None
- ):
- sigma = (ens_data["p90"] - ens_data["p10"]) / 2.56
- if sigma < 0.1:
- sigma = 0.1
+ ai_text = ""
+ try:
+ _, ai_context, sd = _trend_analyze(raw, sym, city)
- # Historical MAE floor
- acc = get_deb_accuracy(city)
- if acc:
- _, hist_mae, _, _ = acc
- if hist_mae > sigma:
- sigma = hist_mae
+ # Use structured data from shared engine
+ mu = sd.get("mu")
+ probabilities = sd.get("probabilities", [])
+ trend_info["is_dead_market"] = sd.get("trend_info", {}).get("is_dead_market", False)
+ trend_info["direction"] = sd.get("trend_info", {}).get("direction", trend_info.get("direction", "unknown"))
+ trend_info["is_cooling"] = sd.get("trend_info", {}).get("is_cooling", False)
+ peak_status = sd.get("peak_status", peak_status)
- # Shock score
- recent_obs = metar.get("recent_obs", []) if metar else []
- shock = 0.0
- if len(recent_obs) >= 2:
- o_obs, n_obs = recent_obs[-1], recent_obs[0]
- wd_o, wd_n = _sf(o_obs.get("wdir")), _sf(n_obs.get("wdir"))
- ws_n = _sf(n_obs.get("wspd")) or 0
- if wd_o is not None and wd_n is not None:
- ad = abs(wd_n - wd_o)
- if ad > 180:
- ad = 360 - ad
- shock += min(ad / 90, 1) * min(ws_n / 15, 1) * 0.4
- cr_o = o_obs.get("cloud_rank", 0)
- cr_n = n_obs.get("cloud_rank", 0)
- shock += min(abs(cr_n - cr_o) / 3, 1) * 0.35
- ap_o, ap_n = _sf(o_obs.get("altim")), _sf(n_obs.get("altim"))
- if ap_o is not None and ap_n is not None:
- shock += min(abs(ap_n - ap_o) / 4, 1) * 0.25
- if shock > 0.05:
- sigma *= 1 + 0.5 * shock
+ # Use shared DEB if not already set
+ if deb_val is None and sd.get("deb_prediction") is not None:
+ deb_val = sd["deb_prediction"]
+ deb_weights = sd.get("deb_weights", "")
- # Time-based sigma adjustment
- if local_hour_frac > last_peak_h:
- sigma *= 0.3
- elif first_peak_h <= local_hour_frac <= last_peak_h:
- sigma *= 0.7
-
- # Mu calculation — reality-anchored
- forecast_highs = [h for h in current_forecasts.values() if h is not None]
- forecast_median = (
- sorted(forecast_highs)[len(forecast_highs) // 2]
- if forecast_highs
- else ens_data["median"]
- )
-
- # Compute forecast miss magnitude
- if max_so_far is not None and forecast_median is not None:
- forecast_miss_deg = round(forecast_median - max_so_far, 1)
-
- # --- Key fix: Reality-anchored μ ---
- # If we are past or in the peak window AND actual max is significantly
- # below forecasts, anchor μ on max_so_far, not on forecast_median.
- if (
- max_so_far is not None
- and forecast_median is not None
- and peak_status in ("past", "in_window")
- and max_so_far < forecast_median - 2.0
- ):
- # Forecast bust: μ anchors on actual max, not failed predictions
- # Allow small upward margin only if still warming
- if trend_info["is_cooling"] or peak_status == "past":
- mu = max_so_far
- else:
- # Still in window and warming — small margin
- mu = max_so_far + 0.5
- else:
- # Normal case: blend forecast and ensemble
- mu = (
- forecast_median * 0.7 + ens_data["median"] * 0.3
- if forecast_median is not None
- else ens_data["median"]
+ # Append multi-model divergence for AI
+ if current_forecasts and ai_context:
+ mm_str = " | ".join(
+ [f"{k}:{v}{sym}" for k, v in current_forecasts.items() if v]
)
- if max_so_far is not None and max_so_far > mu:
- mu = max_so_far + (0.3 if not trend_info["is_cooling"] else 0.0)
+ ai_context += f"\n模型分歧: {mm_str}"
- def _norm_cdf(x, m, s):
- return 0.5 * (1 + math.erf((x - m) / (s * math.sqrt(2))))
-
- min_wu = round(max_so_far) if max_so_far is not None else -999
- probs = {}
- for n in range(round(mu) - 2, round(mu) + 3):
- if n < min_wu:
- continue
- p = _norm_cdf(n + 0.5, mu, sigma) - _norm_cdf(n - 0.5, mu, sigma)
- if p > 0.01:
- probs[n] = p
- total = sum(probs.values())
- if total > 0:
- probs = {k: v / total for k, v in probs.items()}
- for t, p in sorted(probs.items(), key=lambda x: x[1], reverse=True)[:4]:
- probabilities.append(
- {"value": t, "range": f"[{t-0.5}~{t+0.5})", "probability": round(p, 3)}
- )
-
- # ── 11. Dead market detection ──
- is_dead = False
- if max_so_far is not None and cur_temp is not None:
- if local_hour >= 21 and max_so_far - cur_temp >= 3.0:
- is_dead = True
- elif local_hour > last_peak_h and max_so_far - cur_temp >= 1.5:
- is_dead = True
- trend_info["is_dead_market"] = is_dead
-
- # Override probabilities when dead market confirmed
- if is_dead and max_so_far is not None:
- settled = round(max_so_far)
- mu = max_so_far
- probabilities = [
- {"value": settled, "range": f"[{settled-0.5}~{settled+0.5})", "probability": 1.0}
- ]
+ if ai_context:
+ ai_text = get_ai_analysis(ai_context, city, sym)
+ except Exception as e:
+ logger.warning(f"Analysis/AI skipped for {city}: {e}")
# ── 12. Hourly data (today only, for chart) ──
today_hourly: Dict[str, list] = {"times": [], "temps": [], "radiation": []}
@@ -442,28 +358,6 @@ def _analyze(city: str) -> Dict[str, Any]:
"rain_24h": _sf(mgc.get("rain_24h")),
}
- # ── 15. AI Analysis (reuse bot_listener's analyze_weather_trend) ──
- ai_text = ""
- try:
- from src.analysis.ai_analyzer import get_ai_analysis
- from bot_listener import analyze_weather_trend
-
- # analyze_weather_trend returns (display_str, ai_context)
- # It already handles: DEB, probability, dead market, forecast bust,
- # METAR trend, peak window, wind, cloud, humidity, etc.
- _, ai_context = analyze_weather_trend(raw, sym, city)
-
- # Append multi-model divergence if not already included
- if current_forecasts and ai_context:
- mm_str = " | ".join(
- [f"{k}:{v}{sym}" for k, v in current_forecasts.items() if v]
- )
- ai_context += f"\n模型分歧: {mm_str}"
-
- if ai_context:
- ai_text = get_ai_analysis(ai_context, city, sym)
- except Exception as e:
- logger.warning(f"AI analysis skipped for {city}: {e}")
# ── Assemble result ──
result = {