diff --git a/bot_listener.py b/bot_listener.py
index 5c490d66..65976cb2 100644
--- a/bot_listener.py
+++ b/bot_listener.py
@@ -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,
diff --git a/src/analysis/city_query_service.py b/src/analysis/city_query_service.py
index d7f215c3..751ac789 100644
--- a/src/analysis/city_query_service.py
+++ b/src/analysis/city_query_service.py
@@ -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} | 🇺🇸 MGM: {mgm_value}{temp_symbol}"
+ f"{d[5:]}: {t}{temp_symbol} | 🇺🇸 MGM: {mgm_display}{temp_symbol}"
)
else:
future_forecasts.append(f"{d[5:]}: {t}{temp_symbol}")
diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py
index 83ec024a..4df0f810 100644
--- a/src/analysis/market_alert_engine.py
+++ b/src/analysis/market_alert_engine.py
@@ -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")
diff --git a/src/data_collection/polymarket_readonly.py b/src/data_collection/polymarket_readonly.py
index d1c87e89..089947cd 100644
--- a/src/data_collection/polymarket_readonly.py
+++ b/src/data_collection/polymarket_readonly.py
@@ -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 = {
diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py
index 8404d967..179a291d 100644
--- a/src/data_collection/weather_sources.py
+++ b/src/data_collection/weather_sources.py
@@ -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(
diff --git a/src/onchain/polymarket_wallet_activity_watcher.py b/src/onchain/polymarket_wallet_activity_watcher.py
index 02746662..93e2f5f9 100644
--- a/src/onchain/polymarket_wallet_activity_watcher.py
+++ b/src/onchain/polymarket_wallet_activity_watcher.py
@@ -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
-
-
-
-
diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py
index 237b036a..297a685e 100644
--- a/src/utils/telegram_push.py
+++ b/src/utils/telegram_push.py
@@ -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)
diff --git a/tests/test_polymarket_readonly.py b/tests/test_polymarket_readonly.py
new file mode 100644
index 00000000..fdecde32
--- /dev/null
+++ b/tests/test_polymarket_readonly.py
@@ -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
+
diff --git a/web/app.py b/web/app.py
index adc51de6..a96f89d2 100644
--- a/web/app.py
+++ b/web/app.py
@@ -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: