feat: Introduce comprehensive weather data querying, analysis, and display services, integrate Polymarket data collection, and add Telegram notification utilities.

2026-03-11 10:23:33 +08:00
parent 878e3280d1
commit b3f46430ad
9 changed files with 505 additions and 63 deletions
@@ -3,7 +3,7 @@ import os
 import telebot  # type: ignore
 from loguru import logger  # type: ignore

-# 纭繚椤圭洰鏍圭洰褰曞湪 sys.path 涓?
+# 确保项目根目录在 sys.path 中
 project_root = os.path.dirname(os.path.abspath(__file__))
 if project_root not in sys.path:
    sys.path.insert(0, project_root)
@@ -339,7 +339,10 @@ def start_bot():
                return

            weather_data = weather.fetch_all_sources(
-                city_name, lat=coords["lat"], lon=coords["lon"]
+                city_name,
+                lat=coords["lat"],
+                lon=coords["lon"],
+                force_refresh=True,
            )
            city_report = build_city_query_report(
                city_name=city_name,
@@ -58,6 +58,17 @@ def _render_local_time(
    metar: Dict[str, Any],
    fallback_utc_offset: int,
 ) -> str:
+    utc_offset = open_meteo.get("utc_offset")
+    if utc_offset is None:
+        utc_offset = fallback_utc_offset
+    try:
+        local_now = datetime.now(timezone.utc).astimezone(
+            timezone(timedelta(seconds=int(utc_offset)))
+        )
+        return local_now.strftime("%H:%M")
+    except Exception:
+        pass
+
    local_time = (open_meteo.get("current") or {}).get("local_time", "")
    if " " in str(local_time):
        return str(local_time).split(" ")[1][:5]
@@ -79,13 +90,49 @@ def _render_local_time(
    if metar_obs:
        return str(metar_obs)[:5]

-    try:
-        local_now = datetime.now(timezone.utc).astimezone(
-            timezone(timedelta(seconds=int(fallback_utc_offset)))
-        )
-        return local_now.strftime("%H:%M")
-    except Exception:
-        return "N/A"
+    return "N/A"
+
+
+def _derive_mgm_daily_highs_from_hourly(
+    mgm: Dict[str, Any],
+    fallback_utc_offset: int,
+) -> Dict[str, float]:
+    if not isinstance(mgm, dict):
+        return {}
+    hourly = mgm.get("hourly")
+    if not isinstance(hourly, list) or not hourly:
+        return {}
+
+    daily_highs: Dict[str, float] = {}
+    local_tz = timezone(timedelta(seconds=int(fallback_utc_offset)))
+    for row in hourly:
+        if not isinstance(row, dict):
+            continue
+        temp = _sf(row.get("temp"))
+        raw_time = str(row.get("time") or "").strip()
+        if temp is None or not raw_time:
+            continue
+
+        date_key = None
+        if "T" in raw_time:
+            try:
+                dt = datetime.fromisoformat(raw_time.replace("Z", "+00:00"))
+                if dt.tzinfo is not None:
+                    dt = dt.astimezone(local_tz)
+                date_key = dt.strftime("%Y-%m-%d")
+            except Exception:
+                if len(raw_time) >= 10 and raw_time[4] == "-" and raw_time[7] == "-":
+                    date_key = raw_time[:10]
+        elif len(raw_time) >= 10 and raw_time[4] == "-" and raw_time[7] == "-":
+            date_key = raw_time[:10]
+
+        if not date_key:
+            continue
+
+        prev = daily_highs.get(date_key)
+        daily_highs[date_key] = temp if prev is None else max(prev, temp)
+
+    return daily_highs


 def _append_future_forecast_lines(
@@ -98,6 +145,12 @@ def _append_future_forecast_lines(
 ) -> None:
    mgm = weather_data.get("mgm") or {}
    mgm_daily = (mgm.get("daily_forecasts") or {}) if isinstance(mgm, dict) else {}
+    mgm_hourly_daily = _derive_mgm_daily_highs_from_hourly(mgm, fallback_utc_offset)
+    if not isinstance(mgm_daily, dict):
+        mgm_daily = {}
+    for date_key, day_high in mgm_hourly_daily.items():
+        if date_key not in mgm_daily:
+            mgm_daily[date_key] = day_high
    mm_raw = weather_data.get("multi_model") or {}
    mm_daily = mm_raw.get("daily_forecasts", {}) if isinstance(mm_raw, dict) else {}
    mb_daily = (weather_data.get("meteoblue") or {}).get("daily_highs", []) or []
@@ -108,8 +161,9 @@ def _append_future_forecast_lines(
        for d, t in zip(dates[1:], max_temps[1:]):
            mgm_value = mgm_daily.get(d) if isinstance(mgm_daily, dict) else None
            if mgm_value is not None:
+                mgm_display = f"{float(mgm_value):.1f}"
                future_forecasts.append(
-                    f"{d[5:]}: {t}{temp_symbol} | 🇺🇸 <b>MGM: {mgm_value}{temp_symbol}</b>"
+                    f"{d[5:]}: {t}{temp_symbol} | 🇺🇸 <b>MGM: {mgm_display}{temp_symbol}</b>"
                )
            else:
                future_forecasts.append(f"{d[5:]}: {t}{temp_symbol}")
@@ -473,6 +473,18 @@ def _bucket_label(bucket: Any) -> Optional[str]:
    return None


+def _row_yes_buy_prob(row: Dict[str, Any]) -> Optional[float]:
+    if not isinstance(row, dict):
+        return None
+    return _norm_probability(row.get("yes_buy"))
+
+
+def _has_actionable_yes_buy_quote(row: Dict[str, Any]) -> bool:
+    quote = _row_yes_buy_prob(row)
+    # 0 usually means no actionable orderbook bid, not a tradable quote.
+    return quote is not None and quote > 0.0
+
+
 def _to_celsius(temp: Optional[float], temp_symbol: str) -> Optional[float]:
    if temp is None:
        return None
@@ -555,6 +567,7 @@ def _pick_bucket_for_forecast(

    best_row: Optional[Dict[str, Any]] = None
    best_distance: Optional[float] = None
+    best_has_quote = False
    best_probability = -1.0
    best_rank = 10**9

@@ -564,12 +577,14 @@ def _pick_bucket_for_forecast(
            continue

        distance = _distance_to_bucket(target, bounds)
+        has_quote = _has_actionable_yes_buy_quote(row)
        probability = _norm_probability(row.get("probability"))
        probability_rank = probability if probability is not None else -1.0

        if best_row is None:
            best_row = row
            best_distance = distance
+            best_has_quote = has_quote
            best_probability = probability_rank
            best_rank = idx
            continue
@@ -578,19 +593,32 @@ def _pick_bucket_for_forecast(
        if distance < best_distance:
            best_row = row
            best_distance = distance
+            best_has_quote = has_quote
            best_probability = probability_rank
            best_rank = idx
            continue

        if abs(distance - best_distance) <= 1e-9:
-            if probability_rank > best_probability:
+            if has_quote and not best_has_quote:
                best_row = row
                best_distance = distance
+                best_has_quote = has_quote
                best_probability = probability_rank
                best_rank = idx
-            elif abs(probability_rank - best_probability) <= 1e-9 and idx < best_rank:
+            elif has_quote == best_has_quote and probability_rank > best_probability:
                best_row = row
                best_distance = distance
+                best_has_quote = has_quote
+                best_probability = probability_rank
+                best_rank = idx
+            elif (
+                has_quote == best_has_quote
+                and abs(probability_rank - best_probability) <= 1e-9
+                and idx < best_rank
+            ):
+                best_row = row
+                best_distance = distance
+                best_has_quote = has_quote
                best_probability = probability_rank
                best_rank = idx

@@ -972,7 +1000,12 @@ def _build_telegram_messages_mispricing(
    om_settle = snapshot.get("open_meteo_settlement")
    forecast_bucket = snapshot.get("forecast_bucket") or {}
    match_bucket_label = str(forecast_bucket.get("label") or "--").strip() or "--"
-    match_bucket_yes = _fmt_cents(forecast_bucket.get("yes_buy"))
+    match_bucket_yes_prob = _norm_probability(forecast_bucket.get("yes_buy"))
+    match_bucket_yes = (
+        _fmt_cents(match_bucket_yes_prob)
+        if match_bucket_yes_prob is not None and match_bucket_yes_prob > 0.0
+        else "--"
+    )
    market_url = str(
        snapshot.get("market_url")
        or snapshot.get("primary_market_url")
@@ -1128,6 +1128,7 @@ class PolymarketReadOnlyLayer:
                )
                orderbook_raw = self._safe_call(clob, "get_order_book", token_id)
                book, book_liquidity = self._normalize_orderbook(orderbook_raw)
+                buy, sell = self._resolve_trade_prices(buy=buy, sell=sell, book=book)
                return {
                    "buy": buy,
                    "sell": sell,
@@ -1150,6 +1151,7 @@ class PolymarketReadOnlyLayer:
        )
        orderbook_raw = self._clob_get("/book", {"token_id": token_id})
        book, book_liquidity = self._normalize_orderbook(orderbook_raw)
+        buy, sell = self._resolve_trade_prices(buy=buy, sell=sell, book=book)
        return {
            "buy": buy,
            "sell": sell,
@@ -1174,6 +1176,19 @@ class PolymarketReadOnlyLayer:
        except Exception:
            return None

+    def _resolve_trade_prices(
+        self,
+        buy: Optional[float],
+        sell: Optional[float],
+        book: Optional[Dict[str, Any]],
+    ) -> Tuple[Optional[float], Optional[float]]:
+        payload = book if isinstance(book, dict) else {}
+        best_bid = _extract_price(payload.get("best_bid"))
+        best_ask = _extract_price(payload.get("best_ask"))
+        resolved_buy = best_ask if best_ask is not None else buy
+        resolved_sell = best_bid if best_bid is not None else sell
+        return resolved_buy, resolved_sell
+
    def _normalize_orderbook(self, orderbook_raw: Any) -> Tuple[Optional[Dict[str, Any]], Optional[float]]:
        payload = _to_plain_dict(orderbook_raw)
        if not payload and isinstance(orderbook_raw, dict):
@@ -1210,6 +1225,8 @@ class PolymarketReadOnlyLayer:
        _parse_side(bids_raw, bid_levels)
        _parse_side(asks_raw, ask_levels)

+        bid_levels.sort(key=lambda level: level[0], reverse=True)
+        ask_levels.sort(key=lambda level: level[0])
        best_bid = bid_levels[0][0] if bid_levels else None
        best_ask = ask_levels[0][0] if ask_levels else None
        normalized = {
@@ -1797,8 +1797,46 @@ class WeatherDataCollector:

        return None

+    def _evict_city_caches(
+        self,
+        city: str,
+        lat: Optional[float],
+        lon: Optional[float],
+        use_fahrenheit: bool,
+    ) -> None:
+        """Drop in-memory caches for one city before a force-refresh query."""
+        if lat is not None and lon is not None:
+            base = f"{round(float(lat), 4)}:{round(float(lon), 4)}"
+            unit = "f" if use_fahrenheit else "c"
+            open_meteo_key = f"{base}:14:{unit}"
+            ensemble_key = f"{base}:{unit}"
+            meteoblue_key = ensemble_key
+            multi_model_key = ensemble_key
+
+            with self._open_meteo_cache_lock:
+                self._open_meteo_cache.pop(open_meteo_key, None)
+            with self._ensemble_cache_lock:
+                self._ensemble_cache.pop(ensemble_key, None)
+            with self._multi_model_cache_lock:
+                self._multi_model_cache.pop(multi_model_key, None)
+            with self._meteoblue_cache_lock:
+                self._meteoblue_cache.pop(meteoblue_key, None)
+
+        icao = self.get_icao_code(city)
+        if icao:
+            prefix = f"{icao}:"
+            with self._metar_cache_lock:
+                for key in list(self._metar_cache.keys()):
+                    if key.startswith(prefix):
+                        self._metar_cache.pop(key, None)
+
    def fetch_all_sources(
-        self, city: str, lat: float = None, lon: float = None, country: str = None
+        self,
+        city: str,
+        lat: float = None,
+        lon: float = None,
+        country: str = None,
+        force_refresh: bool = False,
    ) -> Dict:
        """
        Fetch weather data from all available sources
@@ -1835,9 +1873,17 @@ class WeatherDataCollector:
        # 严格判断是否为美国市场（必须完全匹配列表或缩写）
        use_fahrenheit = city_lower in us_cities

+        if force_refresh:
+            self._evict_city_caches(
+                city=city,
+                lat=lat,
+                lon=lon,
+                use_fahrenheit=use_fahrenheit,
+            )
+
        # Turkish cities: keep MGM model fallback alive when Open-Meteo is rate-limited.
        turkish_provinces = {
-            "ankara": ("17128", "Ankara"),  # settlement reference: Esenboğa airport
+            "ankara": ("17130", "Ankara"),  # MGM center station
            "istanbul": ("17060", "Istanbul"),
        }

@@ -1862,23 +1908,14 @@ class WeatherDataCollector:

                # 对土耳其城市，额外获取 MGM 官方数据与周边测站
                turkish_provinces = {
-                    "ankara": ("17128", "Ankara"),  # 使用机场站 (Esenboğa Havalimanı) 作为结算参考主站
+                    "ankara": ("17130", "Ankara"),  # use one MGM station consistently
                    "istanbul": ("17060", "Istanbul"),
                }
                if city_lower in turkish_provinces:
                    istno, province = turkish_provinces[city_lower]
-                    # 核心逻辑：实测用 istno (17128), 预报强制去 17130 拿
+                    # Use one station for both current conditions and forecasts.
                    mgm_data = self.fetch_from_mgm(istno)
-                    
-                    # 如果当前是机场站 (17128)，我们额外去 17130 拿一次预报
-                    if istno == "17128":
-                        mgm_city_center = self.fetch_from_mgm("17130")
-                        if mgm_city_center and mgm_data:
-                            # 用市中心的预报覆盖机场可能缺失的预报
-                            mgm_data["today_high"] = mgm_city_center.get("today_high")
-                            mgm_data["daily_forecasts"] = mgm_city_center.get("daily_forecasts")
-                            logger.info("⚡ 已同步 MGM 安卡拉总部 (17130) 的官方最高温预报")
-                    
+
                    if mgm_data:
                        results["mgm"] = mgm_data
                        nearby = self.fetch_mgm_nearby_stations(province, root_ist_no=istno)
@@ -1943,13 +1980,6 @@ class WeatherDataCollector:
                if city_lower in turkish_provinces:
                    istno, province = turkish_provinces[city_lower]
                    mgm_data = self.fetch_from_mgm(istno)
-                    if istno == "17128":
-                        mgm_city_center = self.fetch_from_mgm("17130")
-                        if mgm_city_center and mgm_data:
-                            mgm_data["today_high"] = mgm_city_center.get("today_high")
-                            mgm_data["daily_forecasts"] = mgm_city_center.get(
-                                "daily_forecasts"
-                            )
                    if mgm_data:
                        results["mgm"] = mgm_data
                        nearby = self.fetch_mgm_nearby_stations(
@@ -237,6 +237,109 @@ def _diff_positions(
    return changes


+def _merge_pending_update(
+    pending_updates: Dict[str, Dict[str, Any]],
+    pos_key: str,
+    pos: Dict[str, Any],
+    now_ts: int,
+) -> None:
+    entry = pending_updates.get(pos_key)
+    size_delta = _safe_float(pos.get("size_delta"))
+    old_size = _safe_float(pos.get("old_size"))
+    new_size = _safe_float(pos.get("size"))
+    old_avg = _safe_float(pos.get("old_avg_price"))
+    new_avg = _safe_float(pos.get("avg_price"))
+
+    if entry is None:
+        pending_updates[pos_key] = {
+            "count": 1,
+            "first_ts": now_ts,
+            "last_ts": now_ts,
+            "title": pos.get("title"),
+            "slug": pos.get("slug"),
+            "event_slug": pos.get("event_slug"),
+            "outcome": pos.get("outcome"),
+            "asset": pos.get("asset"),
+            "condition_id": pos.get("condition_id"),
+            "old_size": old_size,
+            "size": new_size,
+            "size_delta": size_delta,
+            "old_avg_price": old_avg,
+            "avg_price": new_avg,
+            "position_value": _safe_float(pos.get("position_value")),
+            "cash_pnl": _safe_float(pos.get("cash_pnl")),
+            "percent_pnl": _safe_float(pos.get("percent_pnl")),
+        }
+        return
+
+    entry["count"] = int(entry.get("count", 1)) + 1
+    entry["last_ts"] = now_ts
+    entry["size_delta"] = _safe_float(entry.get("size_delta")) + size_delta
+    entry["size"] = new_size
+    entry["avg_price"] = new_avg
+    entry["position_value"] = _safe_float(pos.get("position_value"))
+    entry["cash_pnl"] = _safe_float(pos.get("cash_pnl"))
+    entry["percent_pnl"] = _safe_float(pos.get("percent_pnl"))
+    pending_updates[pos_key] = entry
+
+
+def _finalize_pending_update(
+    pending_entry: Dict[str, Any],
+    now_ts: int,
+) -> Dict[str, Any]:
+    first_ts = int(pending_entry.get("first_ts") or now_ts)
+    last_ts = int(pending_entry.get("last_ts") or now_ts)
+    return {
+        "title": pending_entry.get("title") or "",
+        "slug": pending_entry.get("slug") or "",
+        "event_slug": pending_entry.get("event_slug") or "",
+        "outcome": pending_entry.get("outcome") or "",
+        "asset": pending_entry.get("asset") or "",
+        "condition_id": pending_entry.get("condition_id") or "",
+        "old_size": _safe_float(pending_entry.get("old_size")),
+        "size": _safe_float(pending_entry.get("size")),
+        "size_delta": _safe_float(pending_entry.get("size_delta")),
+        "old_avg_price": _safe_float(pending_entry.get("old_avg_price")),
+        "avg_price": _safe_float(pending_entry.get("avg_price")),
+        "position_value": _safe_float(pending_entry.get("position_value")),
+        "cash_pnl": _safe_float(pending_entry.get("cash_pnl")),
+        "percent_pnl": _safe_float(pending_entry.get("percent_pnl")),
+        "agg_count": int(pending_entry.get("count") or 1),
+        "agg_span_sec": max(0, last_ts - first_ts),
+    }
+
+
+def _flush_ready_pending_updates(
+    pending_updates: Dict[str, Dict[str, Any]],
+    now_ts: int,
+    debounce_sec: int,
+    max_hold_sec: int,
+    force_keys: Optional[set] = None,
+) -> List[Tuple[str, Dict[str, Any]]]:
+    out: List[Tuple[str, Dict[str, Any]]] = []
+    keys = list(pending_updates.keys())
+    for key in keys:
+        entry = pending_updates.get(key)
+        if not isinstance(entry, dict):
+            pending_updates.pop(key, None)
+            continue
+
+        if force_keys and key in force_keys:
+            out.append(("update", _finalize_pending_update(entry, now_ts)))
+            pending_updates.pop(key, None)
+            continue
+
+        first_ts = int(entry.get("first_ts") or now_ts)
+        last_ts = int(entry.get("last_ts") or now_ts)
+        quiet_enough = (now_ts - last_ts) >= debounce_sec
+        held_too_long = (now_ts - first_ts) >= max_hold_sec
+        if quiet_enough or held_too_long:
+            out.append(("update", _finalize_pending_update(entry, now_ts)))
+            pending_updates.pop(key, None)
+
+    return out
+
+
 def _fmt_pct(value: float) -> str:
    # Data API may return either ratio (0.12) or percent (12.0).
    display = value * 100.0 if abs(value) <= 1.5 else value
@@ -279,8 +382,14 @@ def _format_change_block(
    lines: List[str] = []
    if change_type == "new":
        lines.append("🆕 新开仓位")
+    elif change_type == "closed":
+        lines.append("❌ 仓位关闭")
    else:
-        lines.append("🔄 仓位更新")
+        agg_count = int(pos.get("agg_count") or 1)
+        if agg_count > 1:
+            lines.append("🔁 连续仓位变动汇总")
+        else:
+            lines.append("🔄 仓位更新")

    lines.append(f"钱包: {_short(wallet)}")
    if market_url:
@@ -295,11 +404,20 @@ def _format_change_block(
        now_size = _safe_float(pos.get("size"))
        delta = _safe_float(pos.get("size_delta"))
        lines.append(f"持有数量: {old_size:.3f} -> {now_size:.3f} (Δ {delta:+.3f})")
+        agg_count = int(pos.get("agg_count") or 1)
+        if agg_count > 1:
+            span_sec = int(_safe_float(pos.get("agg_span_sec")))
+            lines.append(f"变动次数: {agg_count} 次 | 聚合窗口: {span_sec}s")
    else:
        lines.append(f"持有数量: {_safe_float(pos.get('size')):.3f}")

    avg_price = _safe_float(pos.get("avg_price"))
-    if _should_show_avg_price(avg_price):
+    old_avg_price = _safe_float(pos.get("old_avg_price"))
+    agg_count = int(pos.get("agg_count") or 1)
+    if change_type == "update" and agg_count > 1:
+        if _should_show_avg_price(old_avg_price) or _should_show_avg_price(avg_price):
+            lines.append(f"建仓均价: {_fmt_price(old_avg_price)} -> {_fmt_price(avg_price)}")
+    elif _should_show_avg_price(avg_price):
        lines.append(f"建仓均价: {_fmt_price(avg_price)}")
    lines.append(f"当前价值: {_fmt_usd(_safe_float(pos.get('position_value')))}")

@@ -360,6 +478,14 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread
    max_changes = max(1, _env_int("POLYMARKET_WALLET_ACTIVITY_MAX_CHANGES_PER_MSG", 5))
    notify_closed = _env_bool("POLYMARKET_WALLET_ACTIVITY_NOTIFY_CLOSED", False)
    bootstrap_alert = _env_bool("POLYMARKET_WALLET_ACTIVITY_BOOTSTRAP_ALERT", False)
+    update_debounce_sec = max(
+        poll_sec,
+        _env_int("POLYMARKET_WALLET_ACTIVITY_UPDATE_DEBOUNCE_SEC", 90),
+    )
+    update_max_hold_sec = max(
+        update_debounce_sec,
+        _env_int("POLYMARKET_WALLET_ACTIVITY_UPDATE_MAX_HOLD_SEC", 240),
+    )

    # 价格过滤范围配置
    min_price = _env_float("POLYMARKET_WALLET_ACTIVITY_AVG_PRICE_SHOW_MIN", 0.0)
@@ -374,13 +500,15 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread

        logger.info(
            f"polymarket wallet activity watcher started users={len(users)} "
-            f"poll={poll_sec}s data_api={data_api_url} price_filter={min_price}-{max_price}"
+            f"poll={poll_sec}s data_api={data_api_url} price_filter={min_price}-{max_price} "
+            f"update_debounce={update_debounce_sec}s update_max_hold={update_max_hold_sec}s"
        )

        while True:
            touched = False
            for user in users:
                try:
+                    now_ts = int(time.time())
                    rows = _fetch_positions(
                        session=session,
                        base_url=data_api_url,
@@ -388,19 +516,30 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread
                        timeout_sec=timeout_sec,
                    )
                    current = _build_snapshot(rows, min_size_abs=min_size_abs)
-                    prev = (
-                        (users_state.get(user) or {}).get("positions")
-                        if isinstance(users_state.get(user), dict)
-                        else {}
-                    ) or {}

-                    if not prev and not bootstrap_alert:
-                        users_state[user] = {
-                            "positions": current,
-                            "updated_at": int(time.time()),
-                        }
-                        touched = True
-                        continue
+                    user_state = users_state.get(user) if isinstance(users_state.get(user), dict) else {}
+                    prev = (user_state.get("positions") if isinstance(user_state, dict) else {}) or {}
+                    if not isinstance(prev, dict):
+                        prev = {}
+                    pending_updates = (
+                        user_state.get("pending_updates") if isinstance(user_state, dict) else {}
+                    ) or {}
+                    if not isinstance(pending_updates, dict):
+                        pending_updates = {}
+                    initialized = bool(user_state.get("initialized")) if isinstance(user_state, dict) else False
+
+                    # First cycle for each wallet only initializes baseline unless bootstrap alert is enabled.
+                    if not initialized:
+                        if not bootstrap_alert:
+                            users_state[user] = {
+                                "positions": current,
+                                "pending_updates": {},
+                                "initialized": True,
+                                "updated_at": now_ts,
+                            }
+                            touched = True
+                            continue
+                        prev = {}

                    changes = _diff_positions(
                        previous=prev,
@@ -411,16 +550,63 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread
                        max_price=max_price,
                    )

-                    if changes:
-                        msg = _build_message(user, changes, max_changes=max_changes)
+                    outgoing: List[Tuple[str, Dict[str, Any]]] = []
+                    for change_type, pos in changes:
+                        pos_key = _position_key(pos)
+                        if change_type == "update":
+                            _merge_pending_update(
+                                pending_updates=pending_updates,
+                                pos_key=pos_key,
+                                pos=pos,
+                                now_ts=now_ts,
+                            )
+                            continue
+
+                        outgoing.extend(
+                            _flush_ready_pending_updates(
+                                pending_updates=pending_updates,
+                                now_ts=now_ts,
+                                debounce_sec=update_debounce_sec,
+                                max_hold_sec=update_max_hold_sec,
+                                force_keys={pos_key},
+                            )
+                        )
+                        outgoing.append((change_type, pos))
+
+                    # If a key disappeared from snapshot, flush pending summary now.
+                    missing_keys = {k for k in pending_updates.keys() if k not in current}
+                    if missing_keys:
+                        outgoing.extend(
+                            _flush_ready_pending_updates(
+                                pending_updates=pending_updates,
+                                now_ts=now_ts,
+                                debounce_sec=update_debounce_sec,
+                                max_hold_sec=update_max_hold_sec,
+                                force_keys=missing_keys,
+                            )
+                        )
+
+                    outgoing.extend(
+                        _flush_ready_pending_updates(
+                            pending_updates=pending_updates,
+                            now_ts=now_ts,
+                            debounce_sec=update_debounce_sec,
+                            max_hold_sec=update_max_hold_sec,
+                        )
+                    )
+
+                    if outgoing:
+                        msg = _build_message(user, outgoing, max_changes=max_changes)
                        bot.send_message(chat_id, msg, disable_web_page_preview=True)
                        logger.info(
-                            f"wallet activity pushed user={user} changes={len(changes)}"
+                            f"wallet activity pushed user={user} changes={len(outgoing)}"
                        )

                    users_state[user] = {
                        "positions": current,
-                        "updated_at": int(time.time()),
+                        "pending_updates": pending_updates,
+                        "initialized": True,
+                        "updated_at": now_ts,
                    }
                    touched = True
                except Exception:
@@ -444,7 +630,3 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread



-
-
-
-
@@ -133,12 +133,23 @@ def _severity_ok(alert_payload: Dict[str, Any], min_severity: str, min_trigger_c
    return SEVERITY_RANK.get(severity, 0) >= SEVERITY_RANK.get(min_severity, 0)


-def _market_price_cap_ok(alert_payload: Dict[str, Any], max_yes_buy: float) -> bool:
+def _market_price_cap_ok(
+    alert_payload: Dict[str, Any],
+    max_yes_buy: float,
+    require_actionable_quote: bool = False,
+) -> bool:
    if max_yes_buy >= 1.0:
        return True

    market = alert_payload.get("market_snapshot") or {}
    if not isinstance(market, dict) or not market.get("available"):
+        if require_actionable_quote:
+            logger.info(
+                "trade alert skipped: market snapshot unavailable city={}".format(
+                    alert_payload.get("city"),
+                )
+            )
+            return False
        return True

    # Strict rule: use the bucket mapped from Open-Meteo settlement.
@@ -149,10 +160,11 @@ def _market_price_cap_ok(alert_payload: Dict[str, Any], max_yes_buy: float) -> b
        yes_buy = _norm_prob(forecast_bucket.get("yes_buy"))
        bucket_label = str(forecast_bucket.get("label") or "").strip() or None

-    if yes_buy is None:
+    if yes_buy is None or yes_buy <= 0.0:
        logger.info(
-            "trade alert skipped: no mapped forecast bucket city={} om_settle={}".format(
+            "trade alert skipped: no actionable mapped bucket quote city={} bucket={} om_settle={}".format(
                alert_payload.get("city"),
+                bucket_label or "--",
                market.get("open_meteo_settlement"),
            )
        )
@@ -321,6 +333,7 @@ def _maybe_send_alert(
    cooldown_sec: int,
    min_severity: str,
    min_trigger_count: int,
+    mispricing_only: bool,
 ) -> bool:
    now_ts = int(time.time())
    last_by_city = state.setdefault("last_by_city", {})
@@ -330,7 +343,11 @@ def _maybe_send_alert(
        0.0,
        min(1.0, _env_float("TELEGRAM_ALERT_MISPRICING_MAX_YES_BUY", 0.10)),
    )
-    if not _market_price_cap_ok(alert_payload, max_yes_buy):
+    if not _market_price_cap_ok(
+        alert_payload,
+        max_yes_buy,
+        require_actionable_quote=mispricing_only,
+    ):
        is_active = False
    message = ((alert_payload.get("telegram") or {}).get("zh") or "").strip()

@@ -430,6 +447,7 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th
                        cooldown_sec=cooldown_sec,
                        min_severity=min_severity,
                        min_trigger_count=min_trigger_count,
+                        mispricing_only=mispricing_only,
                    ):
                        try:
                            _save_state(state_path, state)
@@ -0,0 +1,62 @@
+from src.data_collection.polymarket_readonly import PolymarketReadOnlyLayer
+
+
+def test_normalize_orderbook_uses_sorted_best_prices():
+    layer = PolymarketReadOnlyLayer()
+    raw = {
+        "bids": [
+            {"price": "0.24", "size": "10"},
+            {"price": "0.31", "size": "5"},
+            {"price": "0.27", "size": "8"},
+        ],
+        "asks": [
+            {"price": "0.44", "size": "9"},
+            {"price": "0.39", "size": "6"},
+            {"price": "0.42", "size": "4"},
+        ],
+    }
+
+    book, _liquidity = layer._normalize_orderbook(raw)
+
+    assert book is not None
+    assert book["best_bid"] == 0.31
+    assert book["best_ask"] == 0.39
+    assert book["bid_levels"][0][0] == 0.31
+    assert book["ask_levels"][0][0] == 0.39
+
+
+def test_fetch_token_market_data_prefers_orderbook_executable_prices():
+    class FakeClob:
+        @staticmethod
+        def get_price(_token_id: str, side: str):
+            if side == "BUY":
+                return {"price": "0.11"}
+            return {"price": "0.88"}
+
+        @staticmethod
+        def get_midpoint(_token_id: str):
+            return {"midpoint": "0.50"}
+
+        @staticmethod
+        def get_last_trade_price(_token_id: str):
+            return {"price": "0.49"}
+
+        @staticmethod
+        def get_order_book(_token_id: str):
+            return {
+                "bids": [{"price": "0.24", "size": "10"}],
+                "asks": [{"price": "0.26", "size": "12"}],
+            }
+
+    layer = PolymarketReadOnlyLayer()
+    layer._get_clob_client = lambda: FakeClob()
+
+    data = layer._fetch_token_market_data("token-1")
+
+    # Executable BUY should match best ask from the book.
+    assert data["buy"] == 0.26
+    # Executable SELL should match best bid from the book.
+    assert data["sell"] == 0.24
+    assert data["midpoint"] == 0.5
+    assert data["last_trade_price"] == 0.49
+
@@ -144,7 +144,12 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]:
    sym = "°F" if is_f else "°C"

    # ── 1. Fetch raw data ──
-    raw = _weather.fetch_all_sources(city, lat=lat, lon=lon)
+    raw = _weather.fetch_all_sources(
+        city,
+        lat=lat,
+        lon=lon,
+        force_refresh=force_refresh,
+    )
    om = raw.get("open-meteo", {})
    metar = raw.get("metar", {})
    mgm = raw.get("mgm") or {}
@@ -786,7 +791,45 @@ def _build_city_detail_payload(
    market_slug: Optional[str] = None,
 ) -> Dict[str, Any]:
    distribution = data.get("probabilities", {}).get("distribution", []) or []
-    primary_bucket = distribution[0] if distribution else None
+    city_name = str(data.get("name") or "").strip().lower()
+    model_map = data.get("multi_model") or {}
+    if not isinstance(model_map, dict):
+        model_map = {}
+
+    # Mispricing anchor temperature:
+    # - Ankara: use MGM today-high forecast
+    # - Others: use Open-Meteo today-high forecast
+    anchor_temp = None
+    if city_name == "ankara":
+        anchor_temp = _sf(model_map.get("MGM"))
+    else:
+        anchor_temp = _sf(model_map.get("Open-Meteo"))
+
+    if anchor_temp is None and city_name == "ankara":
+        # Keep radar available when MGM is missing unexpectedly.
+        anchor_temp = _sf(model_map.get("Open-Meteo"))
+
+    primary_bucket = None
+    if isinstance(distribution, list) and distribution:
+        if anchor_temp is None:
+            primary_bucket = distribution[0]
+        else:
+            ranked_buckets = []
+            for idx, row in enumerate(distribution):
+                if not isinstance(row, dict):
+                    continue
+                bucket_temp = _sf(row.get("value"))
+                bucket_prob = _sf(row.get("probability"))
+                if bucket_temp is None:
+                    continue
+                prob_rank = bucket_prob if bucket_prob is not None else -1.0
+                ranked_buckets.append((abs(bucket_temp - anchor_temp), -prob_rank, idx, row))
+            if ranked_buckets:
+                ranked_buckets.sort(key=lambda x: (x[0], x[1], x[2]))
+                primary_bucket = ranked_buckets[0][3]
+            else:
+                primary_bucket = distribution[0]
+
    model_probability = None
    if isinstance(primary_bucket, dict) and primary_bucket.get("probability") is not None:
        try: