Tighten fallback reads around peak-window evidence

Fallback city-card reads now distinguish three cases that previously collapsed into the generic fast-evidence copy: observed highs above the model path, observed highs still lagging after the peak window, and low observations before the peak window that should wait for confirmation rather than down-revise immediately. The same pass removes three unused private helpers from the scan terminal service.

Constraint: Fallback output must be useful before the full AI airport-bulletin read returns

Rejected: Treat any low latest METAR as a down-revision | early-day observations can be below the forecast before the peak window

Rejected: Keep unused helper wrappers | they were unreferenced and added noise to an already large module

Confidence: high

Scope-risk: moderate

Tested: pytest tests/test_web_observability.py -q

Tested: npm run build

CHANGELOG.md

@@ -6,6 +6,7 @@
 - 城市决策卡兜底文案明确标记“快速证据模式”，避免在 DeepSeek 未完整返回时误写成“AI 机场报文解读正常”
 - 城市决策卡流式 AI 解读改为只请求 METAR/官方观测核心解读与判断依据，最高温中枢、模型一致性和风险清单由后端规则补齐，减少等待时间
 - 城市决策卡兜底判断新增实测突破识别：当最新 METAR/观测已高于 DEB 中枢或模型上沿时，改为提示最高温中枢需要上修
+- 城市决策卡兜底判断补充实测偏低和峰值窗口已过分支：峰后未追上模型时提示下修压力，峰前偏低时只提示等待确认
 - 城市决策卡市场层改用完整 `all_buckets` 并严格识别 exact / range / or higher / or lower 温度桶方向，避免最高温中枢错配到不合理尾部桶
 - 温度桶标签统一规范化 `C/F/°C/°F`，修复 `31°°C` 这类重复单位展示
 - 决策卡展示文案将“概率差”收口为“模型-市场差”，明确口径为 `模型概率 - 市场隐含概率`

tests/test_web_observability.py

@@ -173,6 +173,118 @@ def test_city_ai_fallback_revises_up_when_latest_metar_breaks_above_models():
     assert "继续上修最高温中枢" in payload["risks_zh"][0]
 
 
+def test_city_ai_fallback_revises_down_after_peak_when_observed_high_lags():
+    payload = scan_terminal_service._build_city_ai_fallback(
+        {
+            "city_display_name": "London",
+            "temp_symbol": "°C",
+            "deb": {"prediction": 30.0},
+            "model_cluster": {
+                "sources": [
+                    {"value": 29.2},
+                    {"value": 30.0},
+                    {"value": 31.1},
+                ]
+            },
+            "window_phase": "post_peak",
+            "peak_window_label": "14:00-16:59",
+            "observation_anchor": {
+                "is_airport_metar": True,
+                "station_code": "EGLL",
+            },
+            "airport_current": {
+                "station_code": "EGLL",
+                "temp": 27.0,
+                "max_so_far": 27.5,
+                "report_time": "16:30Z",
+                "raw_metar": "EGLL 271630Z 22008KT 9999 SCT030 27/15 Q1012",
+            },
+        },
+        locale="zh-CN",
+        reason="stream preview",
+    )
+
+    assert payload["predicted_max"] == 27.5
+    assert "峰值窗口（14:00-16:59）已过或接近结束" in payload["final_judgment_zh"]
+    assert "最高温中枢需先下修到 27.5°C" in payload["final_judgment_zh"]
+    assert "共同支撑本轮最高温中枢" not in payload["reasoning_zh"]
+    assert "下修压力" in payload["reasoning_zh"]
+    assert "继续下修最高温中枢" in payload["risks_zh"][0]
+
+
+def test_city_ai_fallback_does_not_downrevise_before_peak_window():
+    payload = scan_terminal_service._build_city_ai_fallback(
+        {
+            "city_display_name": "Dubai",
+            "temp_symbol": "°C",
+            "deb": {"prediction": 41.0},
+            "model_cluster": {
+                "sources": [
+                    {"value": 40.5},
+                    {"value": 41.0},
+                    {"value": 41.6},
+                ]
+            },
+            "window_phase": "early_today",
+            "minutes_until_peak_start": 240,
+            "peak_window_label": "14:00-16:59",
+            "observation_anchor": {
+                "is_airport_metar": True,
+                "station_code": "OMDB",
+            },
+            "airport_current": {
+                "station_code": "OMDB",
+                "temp": 35.0,
+                "report_time": "08:00Z",
+                "raw_metar": "OMDB 270800Z 29007KT CAVOK 35/20 Q1008",
+            },
+        },
+        locale="zh-CN",
+        reason="stream preview",
+    )
+
+    assert payload["predicted_max"] == 41.0
+    assert "暂不直接下修" in payload["reasoning_zh"]
+    assert "峰值窗口尚未到来" in payload["reasoning_zh"]
+    assert "若峰值窗口前继续偏低，需要下修最高温中枢" in payload["risks_zh"][0]
+
+
+def test_city_ai_fallback_marks_peak_window_passed_without_waiting_for_warming():
+    payload = scan_terminal_service._build_city_ai_fallback(
+        {
+            "city_display_name": "Paris",
+            "temp_symbol": "°C",
+            "deb": {"prediction": 28.0},
+            "model_cluster": {
+                "sources": [
+                    {"value": 27.6},
+                    {"value": 28.0},
+                    {"value": 28.5},
+                ]
+            },
+            "window_phase": "post_peak",
+            "peak_window_label": "13:00-15:59",
+            "observation_anchor": {
+                "is_airport_metar": True,
+                "station_code": "LFPG",
+            },
+            "airport_current": {
+                "station_code": "LFPG",
+                "temp": 27.0,
+                "max_so_far": 27.2,
+                "report_time": "17:00Z",
+                "raw_metar": "LFPG 271700Z 25006KT 9999 FEW035 27/13 Q1014",
+            },
+        },
+        locale="zh-CN",
+        reason="stream preview",
+    )
+
+    assert "峰值窗口（13:00-15:59）已过" in payload["final_judgment_zh"]
+    assert "不是继续按待升温路径解读" in payload["reasoning_zh"]
+    assert "避免继续上调最高温中枢" in payload["risks_zh"][0]
+
+
 def test_city_ai_stream_request_only_asks_provider_for_observation_read():
     request_payload = scan_terminal_service._build_city_ai_stream_request(
         {

web/scan_terminal_service.py

@@ -456,11 +456,6 @@ def _extract_json_string_field_fragment(raw_text: str, field: str) -> tuple[str,
     return _decode_json_string_fragment("".join(chars)).strip(), closed
 
 
-def _extract_json_string_field_from_fragment(raw_text: str, field: str) -> str:
-    value, _closed = _extract_json_string_field_fragment(raw_text, field)
-    return value
-
-
 _CITY_AI_TEXT_FIELDS = {
     "metar_read_zh",
     "metar_read_en",
@@ -678,34 +673,107 @@ def _build_city_ai_fallback(
     current_temp = _safe_float(
         airport_current.get("temp") if is_airport_metar else current_obs.get("temp")
     )
+    current_max_so_far = _safe_float(
+        ai_input.get("current_max_so_far")
+        or current_obs.get("max_so_far")
+        or airport_current.get("max_so_far")
+    )
+    observed_high_so_far = max(
+        [value for value in (current_temp, current_max_so_far) if value is not None],
+        default=None,
+    )
     predicted = deb_value
     if predicted is None and values:
         predicted = sum(values) / len(values)
     if predicted is None:
-        predicted = current_temp
+        predicted = observed_high_so_far
     range_low = min(values) if values else predicted
     range_high = max(values) if values else predicted
+    peak = ai_input.get("peak") if isinstance(ai_input.get("peak"), dict) else {}
+    window_phase = str(
+        ai_input.get("window_phase")
+        or peak.get("window_phase")
+        or peak.get("phase")
+        or ""
+    ).strip().lower()
+    remaining_window_minutes = _safe_float(
+        ai_input.get("remaining_window_minutes")
+        if ai_input.get("remaining_window_minutes") is not None
+        else peak.get("remaining_window_minutes")
+        if peak.get("remaining_window_minutes") is not None
+        else peak.get("remaining_minutes")
+    )
+    minutes_until_peak_start = _safe_float(
+        ai_input.get("minutes_until_peak_start")
+        if ai_input.get("minutes_until_peak_start") is not None
+        else peak.get("minutes_until_peak_start")
+    )
+    minutes_until_peak_end = _safe_float(
+        ai_input.get("minutes_until_peak_end")
+        if ai_input.get("minutes_until_peak_end") is not None
+        else peak.get("minutes_until_peak_end")
+    )
+    peak_window_label = str(
+        ai_input.get("peak_window_label")
+        or peak.get("peak_window_label")
+        or peak.get("label")
+        or ""
+    ).strip()
+    peak_has_passed = (
+        window_phase in {"post_peak", "past"}
+        or (minutes_until_peak_end is not None and minutes_until_peak_end < 0)
+    )
+    peak_is_closing = (
+        window_phase == "active_peak"
+        and remaining_window_minutes is not None
+        and remaining_window_minutes <= 90
+    )
+    peak_not_started = (
+        window_phase in {"early_today", "setup_today", "tomorrow", "week_ahead"}
+        or (minutes_until_peak_start is not None and minutes_until_peak_start > 0)
+    )
     model_range_high = range_high
+    model_range_low = range_low
     current_above_predicted = (
-        current_temp is not None
+        observed_high_so_far is not None
         and predicted is not None
-        and current_temp > predicted + 0.2
+        and observed_high_so_far > predicted + 0.2
     )
     current_above_model_range = (
-        current_temp is not None
+        observed_high_so_far is not None
         and model_range_high is not None
-        and current_temp > model_range_high + 0.2
+        and observed_high_so_far > model_range_high + 0.2
     )
     observed_high_break = bool(current_above_predicted or current_above_model_range)
+    current_below_predicted = (
+        observed_high_so_far is not None
+        and predicted is not None
+        and observed_high_so_far < predicted - 1.5
+    )
+    current_below_model_range = (
+        observed_high_so_far is not None
+        and model_range_low is not None
+        and observed_high_so_far < model_range_low - 0.2
+    )
+    observed_low_break = bool(current_below_predicted and (peak_has_passed or peak_is_closing))
+    observed_low_lag = bool(current_below_predicted and not observed_low_break)
     original_predicted = predicted
     if observed_high_break:
         predicted = max(
             value
-            for value in (predicted, current_temp)
+            for value in (predicted, observed_high_so_far)
             if value is not None
         )
-        if range_high is not None and current_temp is not None:
+        if range_high is not None and observed_high_so_far is not None:
-            range_high = max(range_high, current_temp)
+            range_high = max(range_high, observed_high_so_far)
+    elif observed_low_break:
+        predicted = min(
+            value
+            for value in (predicted, observed_high_so_far)
+            if value is not None
+        )
+        if range_low is not None and observed_high_so_far is not None:
+            range_low = min(range_low, observed_high_so_far)
     city = str(ai_input.get("city_display_name") or ai_input.get("city") or "this city")
     station = str((airport_current.get("station_code") if is_airport_metar else None) or observation_anchor.get("station_code") or current_obs.get("station_code") or "")
     raw_metar = str(airport_current.get("raw_metar") or "").strip() if is_airport_metar else ""
@@ -739,9 +807,12 @@ def _build_city_ai_fallback(
         metar_zh = f"当前没有可用的{source_name_zh}正文，暂以 DEB、多模型路径与最新实测为主。"
         metar_en = f"No raw {source_name_en} text is available, so DEB, latest observations and the model cluster carry the read."
     predicted_text = _format_ai_temperature(predicted, unit) or "--"
-    current_text = _format_ai_temperature(current_temp, unit) or "--"
+    current_text = _format_ai_temperature(observed_high_so_far, unit) or "--"
     original_predicted_text = _format_ai_temperature(original_predicted, unit) or "--"
     model_range_high_text = _format_ai_temperature(model_range_high, unit) or "--"
+    model_range_low_text = _format_ai_temperature(model_range_low, unit) or "--"
+    peak_label_text_zh = f"（{peak_window_label}）" if peak_window_label else ""
+    peak_label_text_en = f" ({peak_window_label})" if peak_window_label else ""
     if partial_ai.get("final_judgment_zh") or partial_ai.get("final_judgment_en"):
         final_zh = str(partial_ai.get("final_judgment_zh") or partial_ai.get("final_judgment_en") or "").strip()
         final_en = str(partial_ai.get("final_judgment_en") or partial_ai.get("final_judgment_zh") or "").strip()
@@ -751,36 +822,53 @@ def _build_city_ai_fallback(
     elif observed_high_break:
         final_zh = f"{city} 最新实测已达 {current_text}，高于原先 {original_predicted_text} 中枢；最高温中枢需先上修到至少 {predicted_text} 附近。"
         final_en = f"{city} latest observation has reached {current_text}, above the prior {original_predicted_text} center; the daily-high center should be revised up to at least near {predicted_text}."
+    elif observed_low_break:
+        final_zh = f"{city} 峰值窗口{peak_label_text_zh}已过或接近结束，实测最高仍约 {current_text}，低于原先 {original_predicted_text} 中枢；最高温中枢需先下修到 {predicted_text} 附近。"
+        final_en = f"{city} peak window{peak_label_text_en} has passed or is nearly over, and observed high is still near {current_text}, below the prior {original_predicted_text} center; revise the daily-high center down toward {predicted_text}."
+    elif peak_has_passed:
+        final_zh = f"{city} 峰值窗口{peak_label_text_zh}已过；最高温暂以 {predicted_text} 附近为中枢，并以已观测到的高点为主要校准。"
+        final_en = f"{city} peak window{peak_label_text_en} has passed; the daily high stays centered near {predicted_text}, calibrated mainly against the observed high so far."
     elif timed_out:
         final_zh = f"{city} 预计最高温暂以 {predicted_text} 附近为中枢；当前已先用 DEB、多模型和{source_name_zh}快速证据模式判断。"
         final_en = f"{city} daily high is centered near {predicted_text}; the current read uses the fast DEB/model/{source_name_en} evidence mode."
     else:
         final_zh = f"{city} 预计最高温暂以 {predicted_text} 附近为中枢；当前已先用 DEB、多模型和{source_name_zh}快速证据模式判断。"
         final_en = f"{city} daily high is centered near {predicted_text}; the current read uses the fast DEB/model/{source_name_en} evidence mode."
-    reasoning_zh = str(partial_ai.get("reasoning_zh") or "").strip() or (
+    if partial_ai:
-        f"AI {bulletin_zh}解读已用于校准日内节奏；DEB 与多模型集合继续约束最高温中枢，后续{source_name_zh}用于确认是否需要上调或下修。"
+        fallback_reasoning_zh = f"AI {bulletin_zh}解读已用于校准日内节奏；DEB 与多模型集合继续约束最高温中枢，后续{source_name_zh}用于确认是否需要上调或下修。"
-        if partial_ai
+        fallback_reasoning_en = f"The AI {bulletin_en} read is already used to calibrate the intraday pace; DEB and the model cluster still constrain the high-temperature center, while later {source_name_en} updates confirm whether to revise it."
-        else f"当前为快速证据模式；最新{source_name_zh}已高于原先 {original_predicted_text} 中枢{('，并超过模型上沿 ' + model_range_high_text) if current_above_model_range else ''}，本轮最高温判断应优先承认实测突破并等待完整 AI {bulletin_zh}解读合并。"
+    elif observed_high_break:
-        if observed_high_break
+        fallback_reasoning_zh = f"当前为快速证据模式；最新{source_name_zh}已高于原先 {original_predicted_text} 中枢{('，并超过模型上沿 ' + model_range_high_text) if current_above_model_range else ''}，本轮最高温判断应优先承认实测突破并等待完整 AI {bulletin_zh}解读合并。"
-        else f"当前为快速证据模式；DEB、多模型集合和最新{source_name_zh}共同支撑本轮最高温中枢，完整 AI {bulletin_zh}解读返回后再合并。"
+        fallback_reasoning_en = f"This is the fast evidence mode; latest {source_name_en} is above the prior {original_predicted_text} center{(' and above the model upper edge ' + model_range_high_text) if current_above_model_range else ''}, so the high-temperature read should first acknowledge the observed break and merge the full AI {bulletin_en} read when available."
-    )
+    elif observed_low_break:
-    reasoning_en = str(partial_ai.get("reasoning_en") or "").strip() or (
+        fallback_reasoning_zh = f"当前为快速证据模式；峰值窗口已过或接近结束，最新实测高点仍低于原先 {original_predicted_text} 中枢{('，并低于模型下沿 ' + model_range_low_text) if current_below_model_range else ''}，本轮最高温判断应优先承认下修压力并等待完整 AI {bulletin_zh}解读合并。"
-        f"The AI {bulletin_en} read is already used to calibrate the intraday pace; DEB and the model cluster still constrain the high-temperature center, while later {source_name_en} updates confirm whether to revise it."
+        fallback_reasoning_en = f"This is the fast evidence mode; the peak window has passed or is nearly over, and the observed high remains below the prior {original_predicted_text} center{(' and below the model lower edge ' + model_range_low_text) if current_below_model_range else ''}, so the high-temperature read should first acknowledge downward revision pressure and merge the full AI {bulletin_en} read when available."
-        if partial_ai
+    elif observed_low_lag and peak_not_started:
-        else f"This is the fast evidence mode; latest {source_name_en} is above the prior {original_predicted_text} center{(' and above the model upper edge ' + model_range_high_text) if current_above_model_range else ''}, so the high-temperature read should first acknowledge the observed break and merge the full AI {bulletin_en} read when available."
+        fallback_reasoning_zh = f"当前为快速证据模式；最新{source_name_zh}仍低于原先 {original_predicted_text} 中枢，但峰值窗口尚未到来，暂不直接下修，只把后续升温是否追上模型路径作为关键确认。"
-        if observed_high_break
+        fallback_reasoning_en = f"This is the fast evidence mode; latest {source_name_en} remains below the prior {original_predicted_text} center, but the peak window has not arrived, so do not revise down yet and use later warming as the key confirmation."
-        else f"This is the fast evidence mode; DEB, the model cluster and latest {source_name_en} jointly support the current daily-high center, and the full AI {bulletin_en} read will be merged when available."
+    elif peak_has_passed:
-    )
+        fallback_reasoning_zh = f"当前为快速证据模式；峰值窗口已过，后续{source_name_zh}主要用于确认是否已形成日内高点，而不是继续按待升温路径解读。"
-    risks_zh = (
+        fallback_reasoning_en = f"This is the fast evidence mode; the peak window has passed, so later {source_name_en} updates mainly confirm whether the daily high is already set rather than assuming further warming."
-        [f"最新{source_name_zh}已突破原模型路径，若后续报文继续持平或升温，需要继续上修最高温中枢。"]
+    else:
-        if observed_high_break
+        fallback_reasoning_zh = f"当前为快速证据模式；DEB、多模型集合和最新{source_name_zh}共同支撑本轮最高温中枢，完整 AI {bulletin_zh}解读返回后再合并。"
-        else [f"后续{source_name_zh}若明显偏离模型路径，需及时修正最高温中枢。"]
+        fallback_reasoning_en = f"This is the fast evidence mode; DEB, the model cluster and latest {source_name_en} jointly support the current daily-high center, and the full AI {bulletin_en} read will be merged when available."
-    )
+    reasoning_zh = str(partial_ai.get("reasoning_zh") or "").strip() or fallback_reasoning_zh
-    risks_en = (
+    reasoning_en = str(partial_ai.get("reasoning_en") or "").strip() or fallback_reasoning_en
-        [f"Latest {source_name_en} has already broken above the prior model path; if later reports hold steady or warm further, keep revising the daily-high center upward."]
+    if observed_high_break:
-        if observed_high_break
+        risks_zh = [f"最新{source_name_zh}已突破原模型路径，若后续报文继续持平或升温，需要继续上修最高温中枢。"]
-        else [f"If later {source_name_en} updates diverge from the model path, revise the daily-high center promptly."]
+        risks_en = [f"Latest {source_name_en} has already broken above the prior model path; if later reports hold steady or warm further, keep revising the daily-high center upward."]
-    )
+    elif observed_low_break:
+        risks_zh = [f"峰值窗口已过或接近结束且实测仍偏低，若后续{source_name_zh}没有反弹，需要继续下修最高温中枢。"]
+        risks_en = [f"The peak window has passed or is nearly over while observations remain low; if later {source_name_en} does not rebound, keep revising the daily-high center lower."]
+    elif observed_low_lag:
+        risks_zh = [f"最新{source_name_zh}仍未追上原模型路径；若峰值窗口前继续偏低，需要下修最高温中枢。"]
+        risks_en = [f"Latest {source_name_en} has not caught up with the prior model path; if it stays low before the peak window, revise the daily-high center lower."]
+    elif peak_has_passed:
+        risks_zh = [f"峰值窗口已过，后续{source_name_zh}若未再创新高，应避免继续上调最高温中枢。"]
+        risks_en = [f"The peak window has passed; avoid raising the daily-high center unless later {source_name_en} sets a new high."]
+    else:
+        risks_zh = [f"后续{source_name_zh}若明显偏离模型路径，需及时修正最高温中枢。"]
+        risks_en = [f"If later {source_name_en} updates diverge from the model path, revise the daily-high center promptly."]
     return {
         "predicted_max": partial_ai.get("predicted_max", predicted),
         "range_low": partial_ai.get("range_low", range_low),
@@ -966,27 +1054,6 @@ def _normalize_ai_city_key(value: Any) -> str:
     return str(value or "").strip().lower().replace(" ", "").replace("-", "").replace("_", "")
 
 
-def _compact_ai_distribution(row: Dict[str, Any]) -> List[Dict[str, Any]]:
-    raw_items = row.get("distribution_full") or row.get("distribution_preview") or []
-    if not isinstance(raw_items, list):
-        return []
-    out: List[Dict[str, Any]] = []
-    for item in raw_items:
-        if not isinstance(item, dict):
-            continue
-        out.append(
-            {
-                "label": item.get("label"),
-                "value": item.get("value"),
-                "unit": item.get("unit") or row.get("target_unit") or row.get("temp_symbol"),
-                "model_probability": item.get("model_probability"),
-                "market_probability": item.get("market_probability"),
-                "highlighted": item.get("highlighted"),
-            }
-        )
-    return out
-
-
 def _compact_ai_model_sources(row: Dict[str, Any]) -> List[Dict[str, Any]]:
     raw_sources = row.get("model_cluster_sources")
     if not isinstance(raw_sources, dict):
@@ -1945,10 +2012,6 @@ def _scan_city_ai_cache_key(ai_input: Dict[str, Any]) -> str:
     return "city-ai:" + hashlib.sha256(raw.encode("utf-8")).hexdigest()
 
 
-def _city_forecast_cache_key(city_name: str) -> str:
-    return f"city_forecast:{SCAN_CITY_AI_PROMPT_VERSION}:{city_name.lower()}"
-
-
 def _sse_event(event: str, payload: Dict[str, Any]) -> str:
     return (
         f"event: {event}\n"