feat: implement PolyWeather dashboard with map UI, data collection, analysis, and comprehensive documentation.

src/analysis/market_alert_engine.py

@@ -421,6 +421,111 @@ def _join_trigger_types_cn(rules: Dict[str, Dict[str, Any]]) -> str:
     return " + ".join(parts)
 
 
+def _norm_probability(v: Any) -> Optional[float]:
+    n = _sf(v)
+    if n is None:
+        return None
+    if n > 1.0:
+        n = n / 100.0
+    return max(0.0, min(1.0, n))
+
+
+def _fmt_percent(v: Any) -> str:
+    n = _norm_probability(v)
+    if n is None:
+        return "--"
+    return f"{n * 100:.1f}%"
+
+
+def _fmt_cents(v: Any) -> str:
+    n = _norm_probability(v)
+    if n is None:
+        return "--"
+    cents = n * 100.0
+    return f"{cents:.1f}c"
+
+
+def _bucket_label(bucket: Any) -> Optional[str]:
+    if not isinstance(bucket, dict):
+        return None
+    direct = (
+        str(bucket.get("label") or "").strip()
+        or str(bucket.get("bucket") or "").strip()
+        or str(bucket.get("range") or "").strip()
+    )
+    if direct:
+        return direct
+    value = _sf(bucket.get("value"))
+    if value is not None:
+        return f"{round(value)}C"
+    temp = _sf(bucket.get("temp"))
+    if temp is not None:
+        return f"{round(temp)}C"
+    return None
+
+
+def _extract_market_snapshot(city_weather: Dict[str, Any]) -> Dict[str, Any]:
+    scan = city_weather.get("market_scan") or {}
+    if not isinstance(scan, dict):
+        return {"available": False}
+    if not scan.get("available"):
+        return {"available": False}
+
+    yes_buy = _norm_probability(scan.get("yes_buy"))
+    yes_sell = _norm_probability(scan.get("yes_sell"))
+    market_prob = _norm_probability(
+        scan.get("market_price")
+        or ((scan.get("yes_token") or {}).get("implied_probability"))
+    )
+    model_prob = _norm_probability(scan.get("model_probability"))
+    spread = None
+    if yes_buy is not None and yes_sell is not None:
+        spread = abs(yes_sell - yes_buy)
+
+    top_bucket = None
+    top_bucket_rows: List[Dict[str, Any]] = []
+    top_buckets = scan.get("top_buckets") or []
+    if isinstance(top_buckets, list):
+        normalized = []
+        for row in top_buckets:
+            if not isinstance(row, dict):
+                continue
+            p = _norm_probability(row.get("probability"))
+            if p is None:
+                continue
+            normalized.append((p, row))
+        if normalized:
+            normalized.sort(key=lambda x: x[0], reverse=True)
+            top_bucket = normalized[0][1]
+            for p, row in normalized[:4]:
+                top_bucket_rows.append(
+                    {
+                        "label": _bucket_label(row),
+                        "probability": p,
+                        "yes_buy": _norm_probability(row.get("yes_buy")),
+                        "yes_sell": _norm_probability(row.get("yes_sell")),
+                    }
+                )
+
+    return {
+        "available": True,
+        "selected_bucket": _bucket_label(scan.get("temperature_bucket")),
+        "top_bucket": _bucket_label(top_bucket) if isinstance(top_bucket, dict) else None,
+        "top_bucket_prob": _norm_probability(
+            top_bucket.get("probability") if isinstance(top_bucket, dict) else None
+        ),
+        "market_prob": market_prob,
+        "model_prob": model_prob,
+        "yes_buy": yes_buy,
+        "yes_sell": yes_sell,
+        "spread": spread,
+        "edge_percent": _sf(scan.get("edge_percent")),
+        "signal_label": scan.get("signal_label"),
+        "confidence": scan.get("confidence"),
+        "top_bucket_rows": top_bucket_rows,
+    }
+
+
 def _build_advice_cn(
     rules: Dict[str, Dict[str, Any]],
     temp_symbol: str,
@@ -472,6 +577,7 @@ def _build_telegram_messages(
     city_weather: Dict[str, Any],
     rules: Dict[str, Dict[str, Any]],
     map_url: Optional[str],
+    market_snapshot: Optional[Dict[str, Any]] = None,
     suppression: Optional[Dict[str, Any]] = None,
 ) -> Dict[str, str]:
     temp_symbol = city_weather.get("temp_symbol", "°C")
@@ -482,6 +588,7 @@ def _build_telegram_messages(
     center_deb = rules.get("ankara_center_deb_hit", {})
     momentum = rules.get("momentum_spike", {})
     advection = rules.get("advection", {})
+    market_snapshot = market_snapshot or _extract_market_snapshot(city_weather)
 
     if current_temp is None:
         return {"zh": "", "en": ""}
@@ -559,6 +666,30 @@ def _build_telegram_messages(
         lines_zh.append(peak_line)
     if lead_line:
         lines_zh.append(lead_line)
+    if market_snapshot.get("available") and market_snapshot.get("top_bucket_rows"):
+        lines_zh.append("市场结算概率分布（Top4）：")
+        for row in (market_snapshot.get("top_bucket_rows") or [])[:4]:
+            label = row.get("label") or "--"
+            prob_text = _fmt_percent(row.get("probability"))
+            yes_buy_text = _fmt_cents(row.get("yes_buy"))
+            lines_zh.append(f"{label} {prob_text} | 买Yes: {yes_buy_text}")
+    if market_snapshot.get("available") and not market_snapshot.get("top_bucket_rows"):
+        market_edge = _sf(market_snapshot.get("edge_percent"))
+        market_edge_text = f"{market_edge:+.1f}%" if market_edge is not None else "--"
+        lines_zh.append(
+            "市场联动：同桶 "
+            f"模型 {_fmt_percent(market_snapshot.get('model_prob'))} vs "
+            f"市场 {_fmt_percent(market_snapshot.get('market_prob'))} | "
+            f"Yes {_fmt_cents(market_snapshot.get('yes_buy'))}/{_fmt_cents(market_snapshot.get('yes_sell'))} | "
+            f"点差 {_fmt_cents(market_snapshot.get('spread'))} | "
+            f"偏差 {market_edge_text} | "
+            f"信号 {market_snapshot.get('signal_label') or '--'}/{market_snapshot.get('confidence') or '--'}"
+        )
+        if market_snapshot.get("top_bucket"):
+            lines_zh.append(
+                f"市场最热桶：{market_snapshot.get('top_bucket')} "
+                f"({_fmt_percent(market_snapshot.get('top_bucket_prob'))})"
+            )
     lines_zh.append(f"AI 建议：{advice}")
     lines_zh.append(f"点击查看实时地图：{final_map}")
 
@@ -602,6 +733,30 @@ def _build_telegram_messages(
                 f"Peak state: intraday high {max_so_far:.1f}{temp_symbol} at {max_temp_time}, "
                 f"now off by {rollback:.1f}{temp_symbol}"
             )
+    if market_snapshot.get("available") and market_snapshot.get("top_bucket_rows"):
+        lines_en.append("Settlement distribution (Top4):")
+        for row in (market_snapshot.get("top_bucket_rows") or [])[:4]:
+            label = row.get("label") or "--"
+            prob_text = _fmt_percent(row.get("probability"))
+            yes_buy_text = _fmt_cents(row.get("yes_buy"))
+            lines_en.append(f"{label} {prob_text} | Buy Yes: {yes_buy_text}")
+    if market_snapshot.get("available") and not market_snapshot.get("top_bucket_rows"):
+        market_edge = _sf(market_snapshot.get("edge_percent"))
+        market_edge_text = f"{market_edge:+.1f}%" if market_edge is not None else "--"
+        lines_en.append(
+            "Market: same-bucket "
+            f"model {_fmt_percent(market_snapshot.get('model_prob'))} vs "
+            f"market {_fmt_percent(market_snapshot.get('market_prob'))} | "
+            f"Yes {_fmt_cents(market_snapshot.get('yes_buy'))}/{_fmt_cents(market_snapshot.get('yes_sell'))} | "
+            f"spread {_fmt_cents(market_snapshot.get('spread'))} | "
+            f"edge {market_edge_text} | "
+            f"signal {market_snapshot.get('signal_label') or '--'}/{market_snapshot.get('confidence') or '--'}"
+        )
+        if market_snapshot.get("top_bucket"):
+            lines_en.append(
+                f"Top market bucket: {market_snapshot.get('top_bucket')} "
+                f"({_fmt_percent(market_snapshot.get('top_bucket_prob'))})"
+            )
     lines_en.append(f"Action: {advice}")
     lines_en.append(f"Map: {final_map}")
 
@@ -618,6 +773,7 @@ def build_trading_alerts(
     temp_symbol = city_weather.get("temp_symbol", "°C")
     city = city_weather.get("name", "")
     now = datetime.now(timezone.utc).isoformat()
+    market_snapshot = _extract_market_snapshot(city_weather)
 
     rules: Dict[str, Dict[str, Any]] = {
         "ankara_center_deb_hit": _calc_ankara_center_deb_alert(city_weather, temp_symbol),
@@ -658,6 +814,7 @@ def build_trading_alerts(
         city_weather=city_weather,
         rules=rules,
         map_url=map_url,
+        market_snapshot=market_snapshot,
         suppression=suppression,
     )
 
@@ -668,6 +825,7 @@ def build_trading_alerts(
         "severity": severity,
         "trigger_count": len(triggered),
         "rules": rules,
+        "market_snapshot": market_snapshot,
         "suppression": suppression,
         "triggered_alerts": triggered,
         "telegram": telegram,

src/data_collection/polymarket_readonly.py

@@ -394,6 +394,7 @@ class PolymarketReadOnlyLayer:
             "liquidity": None,
             "volume": None,
             "sparkline": fallback_sparkline or [],
+            "top_buckets": [],
             "recent_trades": [],
             "websocket": {},
         }
@@ -485,6 +486,13 @@ class PolymarketReadOnlyLayer:
 
         signal_label, confidence = self._derive_signal(edge_percent, liquidity)
 
+        top_buckets = self._build_top_temperature_buckets(
+            city_key=city_key,
+            target_date=date_str,
+            primary_market=market,
+            limit=4,
+        )
+
         yes_payload = {
             "outcome": yes_token.get("outcome") or "Yes",
             "token_id": yes_token.get("token_id"),
@@ -551,6 +559,7 @@ class PolymarketReadOnlyLayer:
                 "liquidity": liquidity,
                 "volume": volume,
                 "sparkline": sparkline_values,
+                "top_buckets": top_buckets,
                 "websocket": {
                     "market_url": market_url,
                     "asset_ids": [
@@ -1174,3 +1183,232 @@ class PolymarketReadOnlyLayer:
         if slug:
             return f"https://polymarket.com/market/{slug}"
         return None
+
+    def _build_top_temperature_buckets(
+        self,
+        city_key: str,
+        target_date: str,
+        primary_market: Dict[str, Any],
+        limit: int = 4,
+    ) -> List[Dict[str, Any]]:
+        candidate_markets = self._collect_related_temperature_markets(
+            city_key=city_key,
+            target_date=target_date,
+            primary_market=primary_market,
+        )
+        if not candidate_markets:
+            return []
+
+        ranked: List[
+            Tuple[
+                float,
+                float,
+                Dict[str, Any],
+                Dict[str, Any],
+                Dict[str, Any],
+                Dict[str, Any],
+                Dict[str, Any],
+            ]
+        ] = []
+        for market in candidate_markets:
+            tokens = self._extract_market_tokens(market)
+            yes_token, no_token = self._resolve_yes_no_tokens(tokens)
+            if not yes_token or not no_token:
+                continue
+
+            yes_token_id = str(yes_token.get("token_id") or "").strip()
+            no_token_id = str(no_token.get("token_id") or "").strip()
+            yes_prices = self._get_token_market_data(yes_token_id) if yes_token_id else {}
+            no_prices = self._get_token_market_data(no_token_id) if no_token_id else {}
+
+            yes_midpoint = _extract_price(yes_prices.get("midpoint"))
+            yes_implied = _extract_price(yes_token.get("implied_probability"))
+            no_implied = _extract_price(no_token.get("implied_probability"))
+            market_prob = (
+                yes_midpoint
+                if yes_midpoint is not None
+                else (
+                    yes_implied
+                    if yes_implied is not None
+                    else (1.0 - no_implied if no_implied is not None else None)
+                )
+            )
+            if market_prob is None:
+                continue
+
+            market_prob = max(0.0, min(1.0, float(market_prob)))
+            volume = (
+                _extract_price(
+                    market.get("volumeNum")
+                    or market.get("volume")
+                    or market.get("volume24hr")
+                )
+                or 0.0
+            )
+            ranked.append(
+                (
+                    market_prob,
+                    volume,
+                    market,
+                    yes_token,
+                    no_token,
+                    yes_prices,
+                    no_prices,
+                )
+            )
+
+        if not ranked:
+            return []
+
+        ranked.sort(key=lambda item: (item[0], item[1]), reverse=True)
+        top_rows: List[Dict[str, Any]] = []
+        max_items = max(1, int(limit or 4))
+        primary_slug = str(primary_market.get("slug") or "").strip().lower()
+
+        for market_prob, _volume, market, yes_token, no_token, yes_prices, no_prices in ranked[
+            :max_items
+        ]:
+            yes_buy = _extract_price(yes_prices.get("buy"))
+            yes_sell = _extract_price(yes_prices.get("sell"))
+            yes_midpoint = _extract_price(yes_prices.get("midpoint")) or market_prob
+            no_buy = _extract_price(no_prices.get("buy"))
+            no_sell = _extract_price(no_prices.get("sell"))
+
+            if no_buy is None and yes_buy is not None:
+                no_buy = max(0.0, min(1.0, 1.0 - yes_buy))
+            if no_sell is None and yes_sell is not None:
+                no_sell = max(0.0, min(1.0, 1.0 - yes_sell))
+
+            bucket_temp = self._extract_market_bucket_temp(market)
+            market_slug = str(market.get("slug") or "").strip()
+
+            top_rows.append(
+                {
+                    "label": self._extract_market_bucket_label(market, bucket_temp),
+                    "value": bucket_temp,
+                    "temp": bucket_temp,
+                    "probability": market_prob,
+                    "market_price": yes_midpoint,
+                    "yes_buy": yes_buy,
+                    "yes_sell": yes_sell,
+                    "no_buy": no_buy,
+                    "no_sell": no_sell,
+                    "slug": market_slug or None,
+                    "question": market.get("question") or market.get("title"),
+                    "is_primary": bool(
+                        primary_slug
+                        and market_slug
+                        and primary_slug == market_slug.strip().lower()
+                    ),
+                }
+            )
+
+        return top_rows
+
+    def _collect_related_temperature_markets(
+        self,
+        city_key: str,
+        target_date: str,
+        primary_market: Dict[str, Any],
+    ) -> List[Dict[str, Any]]:
+        related: List[Dict[str, Any]] = []
+        canonical_event_slug = self._build_weather_event_slug(city_key, target_date)
+        if canonical_event_slug:
+            related.extend(self._load_event_markets(canonical_event_slug))
+
+        event_slug = self._extract_event_slug(primary_market)
+        if event_slug and event_slug != canonical_event_slug:
+            related.extend(self._load_event_markets(event_slug))
+
+        if not related:
+            for market in self._load_markets(active_only=True):
+                if self._score_market(city_key, target_date, market) <= 0:
+                    continue
+                if self._extract_market_bucket_temp(market) is None:
+                    continue
+                related.append(market)
+
+        related.append(primary_market)
+
+        unique: List[Dict[str, Any]] = []
+        seen = set()
+        for market in related:
+            if not isinstance(market, dict):
+                continue
+            dedupe_key = str(
+                market.get("id")
+                or market.get("slug")
+                or market.get("conditionId")
+                or ""
+            ).strip()
+            if not dedupe_key:
+                continue
+            if dedupe_key in seen:
+                continue
+            seen.add(dedupe_key)
+            unique.append(market)
+        return unique
+
+    def _extract_event_slug(self, market: Dict[str, Any]) -> Optional[str]:
+        event_slug = str(market.get("eventSlug") or "").strip().lower()
+        if event_slug:
+            return event_slug
+
+        slug = str(market.get("slug") or "").strip().lower()
+        if not slug:
+            return None
+
+        trimmed = re.sub(
+            r"-(?:m)?\d+(?:-\d+)?c(?:-or-(?:higher|lower|above|below))?$",
+            "",
+            slug,
+        )
+        trimmed = trimmed.strip("-")
+        return trimmed or None
+
+    def _load_event_markets(self, event_slug: str) -> List[Dict[str, Any]]:
+        normalized_slug = str(event_slug or "").strip().lower()
+        if not normalized_slug:
+            return []
+
+        try:
+            resp = self._session.get(
+                f"{self.gamma_url}/events",
+                params={"slug": normalized_slug, "limit": 5},
+                timeout=self.http_timeout,
+            )
+            resp.raise_for_status()
+            payload = resp.json()
+        except Exception:
+            return []
+
+        events = payload if isinstance(payload, list) else []
+        out: List[Dict[str, Any]] = []
+        for event in events:
+            if not isinstance(event, dict):
+                continue
+            event_item_slug = str(event.get("slug") or "").strip().lower()
+            if event_item_slug and event_item_slug != normalized_slug:
+                continue
+            for market in event.get("markets") or []:
+                if not isinstance(market, dict):
+                    continue
+                market["eventSlug"] = market.get("eventSlug") or event_item_slug
+                market["eventTitle"] = market.get("eventTitle") or event.get("title")
+                out.append(market)
+        return out
+
+    def _extract_market_bucket_label(
+        self,
+        market: Dict[str, Any],
+        bucket_temp: Optional[float],
+    ) -> str:
+        question = str(market.get("question") or market.get("title") or "").strip()
+        text = question.lower()
+        if bucket_temp is not None:
+            if "or higher" in text or "or above" in text or "and above" in text:
+                return f"{bucket_temp:g}C+"
+            if "or lower" in text or "or below" in text or "and below" in text:
+                return f"<={bucket_temp:g}C"
+            return f"{bucket_temp:g}C"
+        return question or str(market.get("slug") or "")

src/utils/telegram_push.py

@@ -117,6 +117,12 @@ def _trigger_type_key(alert_payload: Dict[str, Any]) -> str:
         for alert in (alert_payload.get("triggered_alerts") or [])
         if alert.get("type")
     )
+    market = alert_payload.get("market_snapshot") or {}
+    if isinstance(market, dict) and market.get("available"):
+        signal = str(market.get("signal_label") or "").strip()
+        bucket = str(market.get("selected_bucket") or "").strip()
+        if signal:
+            trigger_types.append(f"mkt:{signal}:{bucket}")
     return "|".join(trigger_types)
 
 
@@ -127,6 +133,7 @@ def _alert_signature(alert_payload: Dict[str, Any]) -> str:
     breakthrough = rules.get("forecast_breakthrough") or {}
     advection = rules.get("advection") or {}
     suppression = alert_payload.get("suppression") or {}
+    market = alert_payload.get("market_snapshot") or {}
 
     signature_payload = {
         "city": alert_payload.get("city"),
@@ -149,6 +156,18 @@ def _alert_signature(alert_payload: Dict[str, Any]) -> str:
         "suppression_reason": suppression.get("reason"),
         "suppression_peak_time": suppression.get("max_temp_time"),
         "suppression_rollback": round(float(suppression.get("rollback") or 0.0), 1),
+        "market_available": bool(market.get("available")),
+        "market_bucket": market.get("selected_bucket"),
+        "market_top_bucket": market.get("top_bucket"),
+        "market_top_bucket_prob": round(float(market.get("top_bucket_prob") or 0.0), 3),
+        "market_prob": round(float(market.get("market_prob") or 0.0), 3),
+        "model_prob": round(float(market.get("model_prob") or 0.0), 3),
+        "market_yes_buy": round(float(market.get("yes_buy") or 0.0), 3),
+        "market_yes_sell": round(float(market.get("yes_sell") or 0.0), 3),
+        "market_spread": round(float(market.get("spread") or 0.0), 3),
+        "market_edge_percent": round(float(market.get("edge_percent") or 0.0), 2),
+        "market_signal": market.get("signal_label"),
+        "market_confidence": market.get("confidence"),
     }
     raw = json.dumps(signature_payload, sort_keys=True, ensure_ascii=True)
     return hashlib.sha1(raw.encode("utf-8")).hexdigest()
@@ -160,10 +179,18 @@ def build_trade_alert_for_city(
     force_refresh: bool = False,
     target_date: Optional[str] = None,
 ) -> Dict[str, Any]:
-    from web.app import _analyze
+    from web.app import _analyze, _build_city_detail_payload
     from src.analysis.market_alert_engine import build_trading_alerts
 
     city_weather = _analyze(city, force_refresh=force_refresh)
+    try:
+        aggregate_detail = _build_city_detail_payload(city_weather)
+        market_scan = aggregate_detail.get("market_scan")
+        if isinstance(market_scan, dict):
+            city_weather = {**city_weather, "market_scan": market_scan}
+    except Exception as exc:
+        logger.debug(f"market scan attach skipped city={city}: {exc}")
+
     resolved_target_date = target_date or city_weather.get("local_date")
     if resolved_target_date:
         datetime.strptime(resolved_target_date, "%Y-%m-%d")
@@ -212,7 +239,7 @@ def _maybe_send_alert(
     last_sig_ts = int((state.get("by_signature") or {}).get(signature) or 0)
     last_city_active = bool(last_city.get("active"))
 
-    if last_city_active and last_city_key == trigger_key:
+    if last_city_active and last_city_key == trigger_key and last_city_sig == signature:
         return False
 
     if last_city_ts and now_ts - last_city_ts < cooldown_sec: