refactor: extract trend_engine, unify probability engine across bot and web, add unit tests

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):
-    """根据实测与预测分析气温态势，增加峰值时刻预测"""
+    """Thin wrapper — delegates to shared trend_engine module."""
-    insights: List[str] = []
+    from src.analysis.trend_engine import analyze_weather_trend as _analyze
-    ai_features: List[str] = []
+    display_str, ai_context, _structured = _analyze(weather_data, temp_symbol, city_name)
-    mu = None
+    return display_str, ai_context
-    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"🧬 <b>DEB 融合预测</b>：<b>{blended_high}{temp_symbol}</b> ({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"📊 <b>集合预报</b>：中位数 {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"🎲 <b>结算概率</b> (μ={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"🎲 <b>结算预测</b>：已锁定 {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"🚨 <b>实测已超预报</b>：{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"⚖️ <b>结算边界</b>：当前最高 {max_so_far}{temp_symbol} → WU 结算 <b>{settled}{temp_symbol}</b>，但只差 <b>{0.5 - fractional:.1f}°</b> 就会进位到 {settled + 1}{temp_symbol}！"
-            else:
-                msg = f"⚖️ <b>结算边界</b>：当前最高 {max_so_far}{temp_symbol} → WU 结算 <b>{settled}{temp_symbol}</b>，刚刚越过进位线，再降 <b>{fractional - 0.5:.1f}°</b> 就会回落到 {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(
-                    "⚠️ <b>提示</b>：预测最热在凌晨，后续气温可能一路走低。"
-                )
-            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"🎯 <b>关注重点</b>：看看那个时段能否涨到 {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)
 
 
 def start_bot():

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"🧬 <b>DEB 融合预测</b>：<b>{blended_high}{temp_symbol}</b> ({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"📊 <b>集合预报</b>：中位数 {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"🎲 <b>结算概率</b> (μ={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"🎲 <b>结算预测</b>：已锁定 {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"🚨 <b>实测已超预报</b>：{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"⚖️ <b>结算边界</b>：当前最高 {max_so_far}{temp_symbol} → WU 结算 "
+                    f"<b>{settled}{temp_symbol}</b>，但只差 <b>{0.5 - fractional:.1f}°</b> "
+                    f"就会进位到 {settled + 1}{temp_symbol}！"
+                )
+            else:
+                msg = (
+                    f"⚖️ <b>结算边界</b>：当前最高 {max_so_far}{temp_symbol} → WU 结算 "
+                    f"<b>{settled}{temp_symbol}</b>，刚刚越过进位线，再降 "
+                    f"<b>{fractional - 0.5:.1f}°</b> 就会回落到 {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("⚠️ <b>提示</b>：预测最热在凌晨，后续气温可能一路走低。")
+            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"🎯 <b>关注重点</b>：看看那个时段能否涨到 {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

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"

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
+    ai_text = ""
-    if (
+    try:
-        ens_data["p10"] is not None
+        _, ai_context, sd = _trend_analyze(raw, sym, city)
-        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
 
-        # Historical MAE floor
+        # Use structured data from shared engine
-        acc = get_deb_accuracy(city)
+        mu = sd.get("mu")
-        if acc:
+        probabilities = sd.get("probabilities", [])
-            _, hist_mae, _, _ = acc
+        trend_info["is_dead_market"] = sd.get("trend_info", {}).get("is_dead_market", False)
-            if hist_mae > sigma:
+        trend_info["direction"] = sd.get("trend_info", {}).get("direction", trend_info.get("direction", "unknown"))
-                sigma = hist_mae
+        trend_info["is_cooling"] = sd.get("trend_info", {}).get("is_cooling", False)
+        peak_status = sd.get("peak_status", peak_status)
 
-        # Shock score
+        # Use shared DEB if not already set
-        recent_obs = metar.get("recent_obs", []) if metar else []
+        if deb_val is None and sd.get("deb_prediction") is not None:
-        shock = 0.0
+            deb_val = sd["deb_prediction"]
-        if len(recent_obs) >= 2:
+            deb_weights = sd.get("deb_weights", "")
-            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
 
-        # Time-based sigma adjustment
+        # Append multi-model divergence for AI
-        if local_hour_frac > last_peak_h:
+        if current_forecasts and ai_context:
-            sigma *= 0.3
+            mm_str = " | ".join(
-        elif first_peak_h <= local_hour_frac <= last_peak_h:
+                [f"{k}:{v}{sym}" for k, v in current_forecasts.items() if v]
-            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"]
             )
-            if max_so_far is not None and max_so_far > mu:
+            ai_context += f"\n模型分歧: {mm_str}"
-                mu = max_so_far + (0.3 if not trend_info["is_cooling"] else 0.0)
 
-        def _norm_cdf(x, m, s):
+        if ai_context:
-            return 0.5 * (1 + math.erf((x - m) / (s * math.sqrt(2))))
+            ai_text = get_ai_analysis(ai_context, city, sym)
-
+    except Exception as e:
-        min_wu = round(max_so_far) if max_so_far is not None else -999
+        logger.warning(f"Analysis/AI skipped for {city}: {e}")
-        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}
-        ]
 
     # ── 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 = {