From f4fea03f3552d9c388748bba6ff0f676dc2ee836 Mon Sep 17 00:00:00 2001
From: "2569718930@qq.com" <2569718930@qq.com>
Date: Thu, 12 Mar 2026 11:44:52 +0800
Subject: [PATCH] feat: Implement new bot architecture including handlers,
services, analysis modules, and comprehensive tests.
---
.env.example | 2 +
bot_listener.py | 407 +---------------------
src/analysis/market_alert_engine.py | 41 ++-
src/analysis/trend_engine.py | 76 ++--
src/bot/__init__.py | 2 +
src/bot/analysis/__init__.py | 2 +
src/bot/analysis/city_analysis_service.py | 36 ++
src/bot/analysis/deb_analysis_service.py | 220 ++++++++++++
src/bot/analysis_layer.py | 31 ++
src/bot/command_guard.py | 47 +++
src/bot/handlers/__init__.py | 2 +
src/bot/handlers/activity.py | 20 ++
src/bot/handlers/basic.py | 75 ++++
src/bot/handlers/city.py | 76 ++++
src/bot/handlers/deb.py | 71 ++++
src/bot/io_layer.py | 113 ++++++
src/bot/observability.py | 43 +++
src/bot/orchestrator.py | 93 +++++
src/bot/runtime_coordinator.py | 236 +++++++++++++
src/bot/services/__init__.py | 2 +
src/bot/services/city_command_service.py | 38 ++
src/bot/services/deb_command_service.py | 31 ++
src/bot/services/entitlement_service.py | 54 +++
src/bot/settings.py | 7 +
tests/test_bot_basic_handler.py | 55 +++
tests/test_bot_command_guard.py | 50 +++
tests/test_bot_handlers.py | 98 ++++++
tests/test_bot_runtime_coordinator.py | 42 +++
28 files changed, 1525 insertions(+), 445 deletions(-)
create mode 100644 src/bot/__init__.py
create mode 100644 src/bot/analysis/__init__.py
create mode 100644 src/bot/analysis/city_analysis_service.py
create mode 100644 src/bot/analysis/deb_analysis_service.py
create mode 100644 src/bot/analysis_layer.py
create mode 100644 src/bot/command_guard.py
create mode 100644 src/bot/handlers/__init__.py
create mode 100644 src/bot/handlers/activity.py
create mode 100644 src/bot/handlers/basic.py
create mode 100644 src/bot/handlers/city.py
create mode 100644 src/bot/handlers/deb.py
create mode 100644 src/bot/io_layer.py
create mode 100644 src/bot/observability.py
create mode 100644 src/bot/orchestrator.py
create mode 100644 src/bot/runtime_coordinator.py
create mode 100644 src/bot/services/__init__.py
create mode 100644 src/bot/services/city_command_service.py
create mode 100644 src/bot/services/deb_command_service.py
create mode 100644 src/bot/services/entitlement_service.py
create mode 100644 src/bot/settings.py
create mode 100644 tests/test_bot_basic_handler.py
create mode 100644 tests/test_bot_command_guard.py
create mode 100644 tests/test_bot_handlers.py
create mode 100644 tests/test_bot_runtime_coordinator.py
diff --git a/.env.example b/.env.example
index 2af711d4..29275f83 100644
--- a/.env.example
+++ b/.env.example
@@ -26,6 +26,8 @@ HTTP_PROXY=http://127.0.0.1:7890
LOG_LEVEL=INFO
ENV=production
POLYWEATHER_MAP_URL=https://polyweather-pro.vercel.app/
+# Bot command entitlement guard (/city, /deb). Off by default until payment API is wired.
+POLYWEATHER_BOT_REQUIRE_ENTITLEMENT=false
# Backend entitlement guard (for /api/cities, /api/city/*, /api/history/*)
POLYWEATHER_REQUIRE_ENTITLEMENT=false
POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN=
diff --git a/bot_listener.py b/bot_listener.py
index 2e8aa275..cbd8270d 100644
--- a/bot_listener.py
+++ b/bot_listener.py
@@ -1,416 +1,13 @@
-import sys
import os
-import telebot # type: ignore
-from loguru import logger # type: ignore
+import sys
# 确保项目根目录在 sys.path 中
project_root = os.path.dirname(os.path.abspath(__file__))
if project_root not in sys.path:
sys.path.insert(0, project_root)
-from src.utils.config_loader import load_config # type: ignore # noqa: E402
-from src.utils.telegram_push import start_trade_alert_push_loop # type: ignore # noqa: E402
-from src.onchain.polygon_wallet_watcher import start_polygon_wallet_watch_loop # type: ignore # noqa: E402
-from src.onchain.polymarket_wallet_activity_watcher import start_polymarket_wallet_activity_loop # type: ignore # noqa: E402
-from src.data_collection.weather_sources import WeatherDataCollector # type: ignore # noqa: E402
-from src.database.db_manager import DBManager
-from src.analysis.city_query_service import (
- resolve_city_name,
- build_city_query_report,
-)
-
-MESSAGE_POINTS = 4
-MESSAGE_DAILY_CAP = 50
-MESSAGE_MIN_LENGTH = 2
-MESSAGE_COOLDOWN_SEC = 30
-CITY_QUERY_COST = 1
-DEB_QUERY_COST = 1
-
-
-def start_bot():
- config = load_config()
- token = os.getenv("TELEGRAM_BOT_TOKEN")
- if not token:
- logger.error("未找到 TELEGRAM_BOT_TOKEN 环境变量")
- return
-
- bot = telebot.TeleBot(token)
- db = DBManager()
- weather = WeatherDataCollector(config)
- start_trade_alert_push_loop(bot, config)
- start_polygon_wallet_watch_loop(bot)
- start_polymarket_wallet_activity_loop(bot)
-
- def _display_name(user) -> str:
- return user.username or user.first_name or f"User_{user.id}"
-
- def _ensure_query_points(message, cost: int, label: str) -> bool:
- user = message.from_user
- db.upsert_user(user.id, _display_name(user))
- result = db.spend_points(user.id, cost)
- if result.get("ok"):
- return True
-
- balance = int(result.get("balance") or 0)
- required = int(result.get("required") or cost)
- missing = max(0, required - balance)
- bot.reply_to(
- message,
- (
- f"❌ 积分不足，无法执行 {label}\n"
- f"当前积分: {balance}\n"
- f"需要积分: {required}\n"
- f"还差积分: {missing}\n\n"
- f"积分规则：每日签到（有效发言满 {MESSAGE_MIN_LENGTH} 字）获得 {MESSAGE_POINTS} 积分，"
- f"每日上限 {MESSAGE_DAILY_CAP} 分。"
- ),
- parse_mode="HTML",
- )
- return False
-
- @bot.message_handler(commands=["start", "help"])
- def send_welcome(message):
- welcome_text = (
- "🚀 PolyWeather 天气查询机器人\n\n"
- "可用指令:\n"
- f"/city [城市名] - 查询城市天气预测与实测 (消耗 {CITY_QUERY_COST} 积分)\n"
- f"/deb [城市名] - 查看 DEB 融合预测准确率 (消耗 {DEB_QUERY_COST} 积分)\n"
- "/top - 查看积分排行榜\n"
- "/id - 获取当前聊天的 Chat ID\n\n"
- "示例: /city 伦敦\n"
- f"💡 提示: 每日签到(有效发言满 {MESSAGE_MIN_LENGTH} 字)获得 {MESSAGE_POINTS} 积分，"
- f"每日上限 {MESSAGE_DAILY_CAP} 分。"
- )
- bot.reply_to(message, welcome_text, parse_mode="HTML")
-
- @bot.message_handler(commands=["id"])
- def get_chat_id(message):
- bot.reply_to(
- message,
- f"🎯 当前聊天的 Chat ID 是: {message.chat.id}",
- parse_mode="HTML",
- )
-
- @bot.message_handler(commands=["top"])
- def show_points(message):
- """显示当前用户的积分及排行榜"""
- user = message.from_user
- db.upsert_user(user.id, _display_name(user))
- user_info = db.get_user(user.id)
-
- leaderboard = db.get_leaderboard(limit=5)
- rank_text = "🏆 PolyWeather 活跃度排行榜\n"
- rank_text += "────────────────────\n"
- for i, entry in enumerate(leaderboard):
- medal = ["🥇", "🥈", "🥉", " ", " "][i] if i < 5 else " "
- rank_text += f"{medal} {entry['username'][:12]}: {entry['points']} 点\n"
-
- if user_info:
- rank_text += "────────────────────\n"
- rank_text += (
- f"👤 我的状态：\n"
- f"┣ 积分: {user_info['points']}\n"
- f"┣ 发言: {user_info['message_count']} 次\n"
- f"┣ 今日发言积分: {user_info.get('daily_points') or 0}/{MESSAGE_DAILY_CAP}\n"
- f"┗ /city 消耗: {CITY_QUERY_COST} | /deb 消耗: {DEB_QUERY_COST}"
- )
-
- bot.send_message(message.chat.id, rank_text, parse_mode="HTML")
- @bot.message_handler(commands=["deb"])
- def deb_accuracy(message):
- """查询 DEB 融合预测的近 7 天准确率。"""
- try:
- parts = message.text.split(maxsplit=1)
- if len(parts) < 2:
- bot.reply_to(
- message,
- "❌ 用法: /deb ankara",
- parse_mode="HTML",
- )
- return
-
- from datetime import datetime as _dt, timedelta as _td
- import os as _os
-
- from src.analysis.deb_algorithm import (
- load_history,
- _is_excluded_model_name,
- reconcile_recent_actual_highs,
- )
- from src.data_collection.city_registry import ALIASES
-
- city_input = parts[1].strip().lower()
- city_name = ALIASES.get(city_input, city_input)
-
- project_root = _os.path.dirname(_os.path.abspath(__file__))
- history_file = _os.path.join(project_root, "data", "daily_records.json")
- data = load_history(history_file)
-
- if city_name not in data or not data[city_name]:
- bot.reply_to(
- message,
- f"❌ 暂无 {city_name} 的历史数据。",
- parse_mode="HTML",
- )
- return
-
- if not _ensure_query_points(message, DEB_QUERY_COST, "/deb"):
- return
-
- reconcile_info = reconcile_recent_actual_highs(city_name, lookback_days=7)
- # Reload in case reconciliation updated the file
- data = load_history(history_file)
- city_data = data[city_name]
- today = _dt.now().date()
- today_str = today.strftime("%Y-%m-%d")
- cutoff_date = today - _td(days=6)
-
- recent_items = []
- for date_str, record in city_data.items():
- try:
- row_date = _dt.strptime(date_str, "%Y-%m-%d").date()
- except Exception:
- continue
- if row_date >= cutoff_date:
- recent_items.append((date_str, record, row_date))
-
- recent_items.sort(key=lambda item: item[0])
-
- lines = [
- f"📊 DEB 准确率报告 - {city_name.title()}",
- "",
- "📅 近日记录：",
- ]
- if (
- isinstance(reconcile_info, dict)
- and reconcile_info.get("ok")
- and int(reconcile_info.get("updated") or 0) > 0
- ):
- lines.extend(
- [
- f"🔁 已用 METAR 历史回填修正 {int(reconcile_info.get('updated'))} 天实测最高温",
- "",
- ]
- )
- total_days = 0
- hits = 0
- deb_errors = []
- signed_errors = []
- model_errors = {}
-
- for date_str, record, _row_date in recent_items:
- actual = record.get("actual_high")
- deb_pred = record.get("deb_prediction")
- forecasts = record.get("forecasts", {}) or {}
-
- if actual is None:
- continue
-
- try:
- actual = float(actual)
- if deb_pred is not None:
- deb_pred = float(deb_pred)
- except Exception:
- continue
-
- if deb_pred is None and forecasts:
- valid_preds = [
- float(v)
- for k, v in forecasts.items()
- if v is not None and not _is_excluded_model_name(k)
- ]
- if valid_preds:
- deb_pred = round(sum(valid_preds) / len(valid_preds), 1)
-
- actual_wu = round(actual)
-
- if date_str == today_str:
- lines.append(f" {date_str}: 📍 今天进行中 (实测暂 {actual:.1f})")
- elif deb_pred is not None:
- total_days += 1
- deb_wu = round(deb_pred)
- hit = deb_wu == actual_wu
- if hit:
- hits += 1
- err = deb_pred - actual
- deb_errors.append(abs(err))
- signed_errors.append(err)
-
- if hit:
- result_icon = "✅"
- err_text = f"偏差{abs(err):.1f}°"
- elif err < 0:
- result_icon = "❌"
- err_text = f"低估{abs(err):.1f}°"
- else:
- result_icon = "❌"
- err_text = f"高估{abs(err):.1f}°"
-
- retro = "≈" if "deb_prediction" not in record else ""
- lines.append(
- f" {date_str}: DEB {retro}{deb_pred:.1f}→{deb_wu} vs 实测 {actual:.1f}→{actual_wu} "
- f"{result_icon} {err_text}"
- )
-
- if date_str != today_str and actual is not None:
- for model, pred in forecasts.items():
- if _is_excluded_model_name(model):
- continue
- if pred is None:
- continue
- try:
- model_errors.setdefault(model, []).append(abs(float(pred) - actual))
- except Exception:
- continue
-
- if total_days > 0:
- hit_rate = hits / total_days * 100
- deb_mae = sum(deb_errors) / len(deb_errors)
- lines.append("")
- lines.append(
- f"🏁 DEB 总战绩：WU命中 {hits}/{total_days} ({hit_rate:.0f}%) | MAE: {deb_mae:.1f}°"
- )
-
- if model_errors:
- lines.append("")
- lines.append("📈 模型 MAE 对比：")
- model_maes = {m: sum(e) / len(e) for m, e in model_errors.items() if e}
- sorted_models = sorted(model_maes.items(), key=lambda item: item[1])
- for model, mae in sorted_models:
- tag = " ⭐" if mae <= deb_mae else ""
- lines.append(f" {model}: {mae:.1f}°{tag}")
- lines.append(f" DEB融合: {deb_mae:.1f}°")
-
- mean_bias = sum(signed_errors) / len(signed_errors)
- underest = sum(1 for e in signed_errors if e < -0.3)
- overest = sum(1 for e in signed_errors if e > 0.3)
- accurate = total_days - underest - overest
-
- lines.append("")
- lines.append("🔍 偏差分析：")
- if abs(mean_bias) > 0.3:
- bias_label = "系统性低估" if mean_bias < 0 else "系统性高估"
- lines.append(f" ⚠️ {bias_label}：平均偏差 {mean_bias:+.1f}°")
- else:
- lines.append(f" ✅ 整体无明显系统偏差：平均偏差 {mean_bias:+.1f}°")
- lines.append(f" (低估 {underest} 次 | 高估 {overest} 次 | 准确 {accurate} 次)")
-
- lines.append("")
- lines.append("💡 建议：")
- if underest > overest and abs(mean_bias) > 0.5:
- lines.append(
- f" 该城市模型集体低估趋势明显（{mean_bias:+.1f}°），实际最高温可能比 DEB 融合值高 "
- f"{abs(mean_bias):.0f}-{abs(mean_bias) + 0.5:.0f}°。交易时建议适当看高。"
- )
- elif overest > underest and abs(mean_bias) > 0.5:
- lines.append(
- f" 该城市模型集体高估趋势明显（{mean_bias:+.1f}°），实际最高温可能低于 DEB 融合值。交易时注意追高风险。"
- )
- elif deb_mae > 1.5:
- lines.append(f" 近期模型波动较大（MAE {deb_mae:.1f}°），建议降低对单一日预测的信任度。")
- elif hit_rate >= 60:
- lines.append(" DEB 近期表现稳定，可继续作为主要参考。")
- else:
- lines.append(" 近期准确率一般，建议结合主站实测与周边站点共同判断。")
-
- lines.append("")
- lines.append("📝 MAE = 平均绝对误差，越小越准。⭐ = 优于 DEB 融合。")
- lines.append("📅 统计窗口：近7天滚动样本。")
- else:
- lines.append("")
- lines.append("🔔 近 7 天尚无完整的 DEB 预测记录。")
-
- lines.append("")
- lines.append(f"💸 本次消耗 {DEB_QUERY_COST} 积分。")
- bot.reply_to(message, "\n".join(lines), parse_mode="HTML")
- except Exception as e:
- bot.reply_to(message, f"❌ 查询失败: {e}")
-
- @bot.message_handler(commands=["city"])
- def get_city_info(message):
- """查询指定城市的天气详情"""
- try:
- parts = message.text.split(maxsplit=1)
- if len(parts) < 2:
- bot.reply_to(
- message,
- "❌ 请输入城市名称\n\n用法: /city chicago",
- parse_mode="HTML",
- )
- return
-
- city_input = parts[1].strip().lower()
- city_name, supported_cities = resolve_city_name(city_input)
- if not city_name:
- city_list = ", ".join(supported_cities)
- bot.reply_to(
- message,
- f"❌ 未找到城市: {city_input}\n\n支持的城市: {city_list}",
- parse_mode="HTML",
- )
- return
-
- if not _ensure_query_points(message, CITY_QUERY_COST, "/city"):
- return
-
- bot.send_message(
- message.chat.id, f"🔍 正在查询 {city_name.title()} 的天气数据..."
- )
-
- coords = weather.get_coordinates(city_name)
- if not coords:
- bot.reply_to(message, f"❌ 未找到城市坐标: {city_name}")
- return
-
- weather_data = weather.fetch_all_sources(
- city_name,
- lat=coords["lat"],
- lon=coords["lon"],
- force_refresh=True,
- )
- city_report = build_city_query_report(
- city_name=city_name,
- weather_data=weather_data,
- city_query_cost=CITY_QUERY_COST,
- )
- bot.send_message(message.chat.id, city_report, parse_mode="HTML")
- except Exception as e:
- import traceback
-
- logger.error(f"查询失败: {e}\n{traceback.format_exc()}")
- bot.reply_to(message, f"❌ 查询失败: {e}")
-
- @bot.message_handler(func=lambda message: True, content_types=['text'])
- def track_activity(message):
- """全量监听消息，用于记录群内发言积分(非指令消息)"""
- if message.text.startswith('/'):
- return
- if message.chat.type not in ("group", "supergroup"):
- return
-
- user = message.from_user
- username = _display_name(user)
- db.upsert_user(user.id, username)
-
- result = db.add_message_activity(
- user.id,
- text=message.text,
- points_to_add=MESSAGE_POINTS,
- cooldown_sec=MESSAGE_COOLDOWN_SEC,
- daily_cap=MESSAGE_DAILY_CAP,
- min_text_length=MESSAGE_MIN_LENGTH,
- )
- if result.get("awarded"):
- logger.info(
- f"message points awarded user={user.id} points=+{MESSAGE_POINTS} "
- f"daily_points={result.get('daily_points')}/{MESSAGE_DAILY_CAP}"
- )
-
- logger.info("🤖 Bot 启动中...")
- bot.infinity_polling()
+from src.bot.orchestrator import start_bot # noqa: E402
if __name__ == "__main__":
start_bot()
-
-
diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py
index 166d3991..8962940c 100644
--- a/src/analysis/market_alert_engine.py
+++ b/src/analysis/market_alert_engine.py
@@ -1009,6 +1009,8 @@ def _build_telegram_messages_mispricing(
city_weather: Dict[str, Any],
rules: Dict[str, Dict[str, Any]],
market_snapshot: Optional[Dict[str, Any]] = None,
+ suppression: Optional[Dict[str, Any]] = None,
+ map_url: Optional[str] = None,
) -> Dict[str, str]:
temp_symbol = str(city_weather.get("temp_symbol") or "°C")
city_name = city_weather.get("display_name") or city_weather.get("name", "").title()
@@ -1019,8 +1021,12 @@ def _build_telegram_messages_mispricing(
snapshot = market_snapshot or _extract_market_snapshot(city_weather)
momentum = rules.get("momentum_spike", {})
+ center_deb = rules.get("ankara_center_deb_hit", {})
local_time = str(city_weather.get("local_time") or "").strip()
obs_time = str(current.get("obs_time") or "").strip()
+ suppressed = bool((suppression or {}).get("suppressed"))
+ has_active_trigger = any(rule.get("triggered") for rule in rules.values())
+ types_cn = "高温已过（暂停推送）" if suppressed else (_join_trigger_types_cn(rules) or "天气状态快照")
delta_temp = _sf(momentum.get("delta_temp"))
delta_min = momentum.get("delta_minutes")
@@ -1055,9 +1061,34 @@ def _build_telegram_messages_mispricing(
or snapshot.get("primary_market_url")
or ""
).strip()
+ final_map = map_url or "https://polyweather-pro.vercel.app/"
+ advice = _build_advice_cn(rules, temp_symbol, suppression=suppression)
- lines_zh = [f"🚨 PolyWeather 错价雷达 [{city_name}]"]
+ center_line = ""
+ if center_deb.get("triggered"):
+ center_station = center_deb.get("center_station") or {}
+ center_name = center_station.get("name") or "Ankara Center"
+ center_temp = _sf(center_station.get("temp"))
+ deb_prediction = _sf(center_deb.get("deb_prediction"))
+ airport_temp = _sf(center_deb.get("airport_temp"))
+ lead_gap = _sf(center_deb.get("center_lead_vs_airport"))
+ if center_temp is not None and deb_prediction is not None:
+ center_line = (
+ f"Center信号：{center_name} {center_temp:.1f}{temp_symbol} 已达到 DEB {deb_prediction:.1f}{temp_symbol}"
+ )
+ if airport_temp is not None:
+ center_line += f" | 机场 {airport_temp:.1f}{temp_symbol}"
+ if lead_gap is not None:
+ center_line += f" | 领先 {lead_gap:+.1f}{temp_symbol}"
+
+ if snapshot.get("available"):
+ title_zh = "🚨 PolyWeather 错价雷达"
+ else:
+ title_zh = "🚨 PolyWeather 异动预警" if (has_active_trigger or suppressed) else "📍 PolyWeather 状态快照"
+
+ lines_zh = [f"{title_zh} [{city_name}]"]
lines_zh.append("")
+ lines_zh.append(f"类型：{types_cn}")
if anchor_high_c is not None and anchor_settle is not None:
if anchor_model:
lines_zh.append(
@@ -1071,6 +1102,8 @@ def _build_telegram_messages_mispricing(
lines_zh.append("基准：多模型最高温 --（结算参考 --）")
lines_zh.append(f"命中桶：{match_bucket_label} | Yes: {match_bucket_yes}")
lines_zh.append("触发：该桶 Yes 价格 < 10c，疑似低估")
+ if center_line:
+ lines_zh.append(center_line)
lines_zh.append("")
lines_zh.append(f"动态：{dynamic_text}")
if local_time or obs_time:
@@ -1080,17 +1113,21 @@ def _build_telegram_messages_mispricing(
lines_zh.append(f"时间：当地 {local_time}")
else:
lines_zh.append(f"时间：观测 {obs_time}")
+ lines_zh.append(f"建议：{advice}")
lines_zh.append("")
if market_url:
lines_zh.append(f"市场链接：{market_url}")
+ lines_zh.append(f"地图：{final_map}")
lines_en = [
f"🚨 PolyWeather Mispricing Radar [{city_name}]",
"",
+ f"Type: {types_cn}",
f"Now: {dynamic_text}",
]
if market_url:
lines_en.append(f"Market link: {market_url}")
+ lines_en.append(f"Map: {final_map}")
return {"zh": "\n".join(lines_zh), "en": "\n".join(lines_en)}
@@ -1297,6 +1334,8 @@ def build_trading_alerts(
city_weather=city_weather,
rules=rules,
market_snapshot=market_snapshot,
+ suppression=suppression,
+ map_url=map_url,
)
evidence = _build_alert_evidence(
city_weather=city_weather,
diff --git a/src/analysis/trend_engine.py b/src/analysis/trend_engine.py
index 98e8033f..13406842 100644
--- a/src/analysis/trend_engine.py
+++ b/src/analysis/trend_engine.py
@@ -385,34 +385,43 @@ def analyze_weather_trend(
probabilities: List[Dict[str, Any]] = []
forecast_miss_deg = 0.0
- if (ens_p10 is not None and ens_p90 is not None) or fallback_sigma:
- if not is_dead_market:
- # Forecast miss magnitude
- if max_so_far is not None and forecast_median is not None:
- forecast_miss_deg = round(forecast_median - max_so_far, 1)
-
- fallback_center = forecast_median if forecast_median is not None else (forecast_high if forecast_high is not None else cur_temp)
- center = ens_median if ens_median is not None else fallback_center
-
- # Reality-anchored μ
- if (
- max_so_far is not None
- and forecast_median is not None
- and peak_status in ("past", "in_window")
- and max_so_far < forecast_median - 2.0
- ):
- if is_cooling or peak_status == "past":
- mu = max_so_far
- else:
- mu = max_so_far + 0.5
+ if is_dead_market:
+ settled_wu = wu_round(max_so_far) if max_so_far is not None else 0
+ dead_msg = f"🎲 结算预测：已锁定 {settled_wu}{temp_symbol} (死盘确认)"
+ insights.append(dead_msg)
+ ai_features.append("🎲 状态: 确认死盘，结算已无悬念。")
+ if max_so_far is not None:
+ mu = max_so_far
+ probabilities = [
+ {"value": settled_wu, "range": f"[{settled_wu-0.5}~{settled_wu+0.5})", "probability": 1.0}
+ ]
+ elif (ens_p10 is not None and ens_p90 is not None) or fallback_sigma:
+ # Forecast miss magnitude
+ if max_so_far is not None and forecast_median is not None:
+ forecast_miss_deg = round(forecast_median - max_so_far, 1)
+
+ fallback_center = forecast_median if forecast_median is not None else (forecast_high if forecast_high is not None else cur_temp)
+ center = ens_median if ens_median is not None else fallback_center
+
+ # Reality-anchored μ
+ if (
+ max_so_far is not None
+ and forecast_median is not None
+ and peak_status in ("past", "in_window")
+ and max_so_far < forecast_median - 2.0
+ ):
+ if is_cooling or peak_status == "past":
+ mu = max_so_far
else:
- mu = (
- forecast_median * 0.7 + center * 0.3
- if forecast_median is not None and center is not None
- else center
- )
- if max_so_far is not None and mu is not None and max_so_far > mu:
- mu = max_so_far + (0.3 if not is_cooling else 0.0)
+ mu = max_so_far + 0.5
+ else:
+ mu = (
+ forecast_median * 0.7 + center * 0.3
+ if forecast_median is not None and center is not None
+ else center
+ )
+ if max_so_far is not None and mu is not None and max_so_far > mu:
+ mu = max_so_far + (0.3 if not is_cooling else 0.0)
# Forecast miss severity for AI
if forecast_miss_deg > 2.0 and peak_status in ("past", "in_window"):
@@ -431,7 +440,7 @@ def analyze_weather_trend(
mu = probs_result.get("mu", mu)
probabilities = probs_result.get("probabilities", [])
sorted_probs = probs_result.get("sorted_probs", [])
-
+
if sorted_probs:
prob_parts = [
f"{int(t)}{temp_symbol} [{t - 0.5}~{t + 0.5}) {p * 100:.0f}%"
@@ -442,17 +451,6 @@ def analyze_weather_trend(
insights.append(f"🎲 结算概率 (μ={mu:.1f})：{prob_str}")
ai_features.append(f"🎲 数学概率分布：{prob_str}")
- elif is_dead_market:
- settled_wu = wu_round(max_so_far) if max_so_far is not None else 0
- dead_msg = f"🎲 结算预测：已锁定 {settled_wu}{temp_symbol} (死盘确认)"
- insights.append(dead_msg)
- ai_features.append("🎲 状态: 确认死盘，结算已无悬念。")
- if max_so_far is not None:
- mu = max_so_far
- probabilities = [
- {"value": settled_wu, "range": f"[{settled_wu-0.5}~{settled_wu+0.5})", "probability": 1.0}
- ]
-
# === Actual exceeds forecast ===
if max_so_far is not None and forecast_high is not None:
if max_so_far > forecast_high + 0.5:
diff --git a/src/bot/__init__.py b/src/bot/__init__.py
new file mode 100644
index 00000000..46ce5319
--- /dev/null
+++ b/src/bot/__init__.py
@@ -0,0 +1,2 @@
+"""Telegram bot composition layers."""
+
diff --git a/src/bot/analysis/__init__.py b/src/bot/analysis/__init__.py
new file mode 100644
index 00000000..881e175d
--- /dev/null
+++ b/src/bot/analysis/__init__.py
@@ -0,0 +1,2 @@
+"""Bot analysis services."""
+
diff --git a/src/bot/analysis/city_analysis_service.py b/src/bot/analysis/city_analysis_service.py
new file mode 100644
index 00000000..06b51699
--- /dev/null
+++ b/src/bot/analysis/city_analysis_service.py
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class CityAnalysisService:
+ """City analysis adapter with lazy imports to trim cold startup."""
+
+ def __init__(self, weather: Any):
+ self.weather = weather
+
+ def resolve_city(self, city_input: str):
+ from src.analysis.city_query_service import resolve_city_name
+
+ return resolve_city_name(city_input)
+
+ def build_city_report(self, city_name: str, city_query_cost: int) -> str:
+ coords = self.weather.get_coordinates(city_name)
+ if not coords:
+ raise ValueError(f"未找到城市坐标: {city_name}")
+
+ weather_data = self.weather.fetch_all_sources(
+ city_name,
+ lat=coords["lat"],
+ lon=coords["lon"],
+ force_refresh=True,
+ )
+
+ from src.analysis.city_query_service import build_city_query_report
+
+ return build_city_query_report(
+ city_name=city_name,
+ weather_data=weather_data,
+ city_query_cost=city_query_cost,
+ )
+
diff --git a/src/bot/analysis/deb_analysis_service.py b/src/bot/analysis/deb_analysis_service.py
new file mode 100644
index 00000000..d588a0f8
--- /dev/null
+++ b/src/bot/analysis/deb_analysis_service.py
@@ -0,0 +1,220 @@
+from __future__ import annotations
+
+import os
+from datetime import datetime as _dt
+from datetime import timedelta as _td
+
+
+class DebAnalysisService:
+ """DEB analytics adapter with lazy imports to trim cold startup."""
+
+ def __init__(self, project_root: str):
+ self.project_root = project_root
+ self.history_file = os.path.join(project_root, "data", "daily_records.json")
+
+ @staticmethod
+ def _load_aliases() -> dict[str, str]:
+ from src.data_collection.city_registry import ALIASES
+
+ return ALIASES
+
+ @staticmethod
+ def _load_deb_module_api():
+ from src.analysis.deb_algorithm import (
+ _is_excluded_model_name,
+ load_history,
+ reconcile_recent_actual_highs,
+ )
+
+ return _is_excluded_model_name, load_history, reconcile_recent_actual_highs
+
+ def resolve_city(self, city_input: str) -> str:
+ aliases = self._load_aliases()
+ city_input_norm = city_input.strip().lower()
+ return aliases.get(city_input_norm, city_input_norm)
+
+ def has_history(self, city_name: str) -> bool:
+ _is_excluded_model_name, load_history, reconcile_recent_actual_highs = (
+ self._load_deb_module_api()
+ )
+ del _is_excluded_model_name, reconcile_recent_actual_highs
+ data = load_history(self.history_file)
+ city_data = data.get(city_name)
+ return isinstance(city_data, dict) and bool(city_data)
+
+ def build_deb_accuracy_report(self, city_name: str, deb_query_cost: int) -> str:
+ _is_excluded_model_name, load_history, reconcile_recent_actual_highs = (
+ self._load_deb_module_api()
+ )
+ data = load_history(self.history_file)
+ if city_name not in data or not data[city_name]:
+ raise ValueError(f"暂无 {city_name} 的历史数据。")
+
+ reconcile_info = reconcile_recent_actual_highs(city_name, lookback_days=7)
+ data = load_history(self.history_file)
+ city_data = data[city_name]
+ today = _dt.now().date()
+ today_str = today.strftime("%Y-%m-%d")
+ cutoff_date = today - _td(days=6)
+
+ recent_items = []
+ for date_str, record in city_data.items():
+ try:
+ row_date = _dt.strptime(date_str, "%Y-%m-%d").date()
+ except Exception:
+ continue
+ if row_date >= cutoff_date:
+ recent_items.append((date_str, record, row_date))
+
+ recent_items.sort(key=lambda item: item[0])
+
+ lines = [
+ f"📊 DEB 准确率报告 - {city_name.title()}",
+ "",
+ "📅 近日记录：",
+ ]
+ if (
+ isinstance(reconcile_info, dict)
+ and reconcile_info.get("ok")
+ and int(reconcile_info.get("updated") or 0) > 0
+ ):
+ lines.extend(
+ [
+ f"🔁 已用 METAR 历史回填修正 {int(reconcile_info.get('updated'))} 天实测最高温",
+ "",
+ ]
+ )
+ total_days = 0
+ hits = 0
+ deb_errors = []
+ signed_errors = []
+ model_errors: dict[str, list[float]] = {}
+
+ for date_str, record, _row_date in recent_items:
+ actual = record.get("actual_high")
+ deb_pred = record.get("deb_prediction")
+ forecasts = record.get("forecasts", {}) or {}
+
+ if actual is None:
+ continue
+
+ try:
+ actual = float(actual)
+ if deb_pred is not None:
+ deb_pred = float(deb_pred)
+ except Exception:
+ continue
+
+ if deb_pred is None and forecasts:
+ valid_preds = [
+ float(v)
+ for k, v in forecasts.items()
+ if v is not None and not _is_excluded_model_name(k)
+ ]
+ if valid_preds:
+ deb_pred = round(sum(valid_preds) / len(valid_preds), 1)
+
+ actual_wu = round(actual)
+
+ if date_str == today_str:
+ lines.append(f" {date_str}: 📍 今天进行中 (实测暂 {actual:.1f})")
+ elif deb_pred is not None:
+ total_days += 1
+ deb_wu = round(deb_pred)
+ hit = deb_wu == actual_wu
+ if hit:
+ hits += 1
+ err = deb_pred - actual
+ deb_errors.append(abs(err))
+ signed_errors.append(err)
+
+ if hit:
+ result_icon = "✅"
+ err_text = f"偏差{abs(err):.1f}°"
+ elif err < 0:
+ result_icon = "❌"
+ err_text = f"低估{abs(err):.1f}°"
+ else:
+ result_icon = "❌"
+ err_text = f"高估{abs(err):.1f}°"
+
+ retro = "≈" if "deb_prediction" not in record else ""
+ lines.append(
+ f" {date_str}: DEB {retro}{deb_pred:.1f}→{deb_wu} vs 实测 {actual:.1f}→{actual_wu} "
+ f"{result_icon} {err_text}"
+ )
+
+ if date_str != today_str and actual is not None:
+ for model, pred in forecasts.items():
+ if _is_excluded_model_name(model):
+ continue
+ if pred is None:
+ continue
+ try:
+ model_errors.setdefault(model, []).append(abs(float(pred) - actual))
+ except Exception:
+ continue
+
+ if total_days > 0:
+ hit_rate = hits / total_days * 100
+ deb_mae = sum(deb_errors) / len(deb_errors)
+ lines.append("")
+ lines.append(
+ f"🏁 DEB 总战绩：WU命中 {hits}/{total_days} ({hit_rate:.0f}%) | MAE: {deb_mae:.1f}°"
+ )
+
+ if model_errors:
+ lines.append("")
+ lines.append("📈 模型 MAE 对比：")
+ model_maes = {m: sum(e) / len(e) for m, e in model_errors.items() if e}
+ sorted_models = sorted(model_maes.items(), key=lambda item: item[1])
+ for model, mae in sorted_models:
+ tag = " ⭐" if mae <= deb_mae else ""
+ lines.append(f" {model}: {mae:.1f}°{tag}")
+ lines.append(f" DEB融合: {deb_mae:.1f}°")
+
+ mean_bias = sum(signed_errors) / len(signed_errors)
+ underest = sum(1 for e in signed_errors if e < -0.3)
+ overest = sum(1 for e in signed_errors if e > 0.3)
+ accurate = total_days - underest - overest
+
+ lines.append("")
+ lines.append("🔍 偏差分析：")
+ if abs(mean_bias) > 0.3:
+ bias_label = "系统性低估" if mean_bias < 0 else "系统性高估"
+ lines.append(f" ⚠️ {bias_label}：平均偏差 {mean_bias:+.1f}°")
+ else:
+ lines.append(f" ✅ 整体无明显系统偏差：平均偏差 {mean_bias:+.1f}°")
+ lines.append(f" (低估 {underest} 次 | 高估 {overest} 次 | 准确 {accurate} 次)")
+
+ lines.append("")
+ lines.append("💡 建议：")
+ if underest > overest and abs(mean_bias) > 0.5:
+ lines.append(
+ f" 该城市模型集体低估趋势明显（{mean_bias:+.1f}°），实际最高温可能比 DEB 融合值高 "
+ f"{abs(mean_bias):.0f}-{abs(mean_bias) + 0.5:.0f}°。交易时建议适当看高。"
+ )
+ elif overest > underest and abs(mean_bias) > 0.5:
+ lines.append(
+ f" 该城市模型集体高估趋势明显（{mean_bias:+.1f}°），实际最高温可能低于 DEB 融合值。交易时注意追高风险。"
+ )
+ elif deb_mae > 1.5:
+ lines.append(
+ f" 近期模型波动较大（MAE {deb_mae:.1f}°），建议降低对单一日预测的信任度。"
+ )
+ elif hit_rate >= 60:
+ lines.append(" DEB 近期表现稳定，可继续作为主要参考。")
+ else:
+ lines.append(" 近期准确率一般，建议结合主站实测与周边站点共同判断。")
+
+ lines.append("")
+ lines.append("📝 MAE = 平均绝对误差，越小越准。⭐ = 优于 DEB 融合。")
+ lines.append("📅 统计窗口：近7天滚动样本。")
+ else:
+ lines.append("")
+ lines.append("🔔 近 7 天尚无完整的 DEB 预测记录。")
+
+ lines.append("")
+ lines.append(f"💸 本次消耗 {deb_query_cost} 积分。")
+ return "\n".join(lines)
+
diff --git a/src/bot/analysis_layer.py b/src/bot/analysis_layer.py
new file mode 100644
index 00000000..834f0cca
--- /dev/null
+++ b/src/bot/analysis_layer.py
@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from src.bot.analysis.city_analysis_service import CityAnalysisService
+from src.bot.analysis.deb_analysis_service import DebAnalysisService
+
+
+class BotAnalysisLayer:
+ """
+ Backward-compatible facade.
+ New code should use CityAnalysisService / DebAnalysisService directly.
+ """
+
+ def __init__(self, weather, project_root: str):
+ self._city = CityAnalysisService(weather=weather)
+ self._deb = DebAnalysisService(project_root=project_root)
+
+ def resolve_city(self, city_input: str):
+ return self._city.resolve_city(city_input)
+
+ def build_city_report(self, city_name: str, city_query_cost: int) -> str:
+ return self._city.build_city_report(city_name, city_query_cost)
+
+ def resolve_deb_city(self, city_input: str) -> str:
+ return self._deb.resolve_city(city_input)
+
+ def has_deb_history(self, city_name: str) -> bool:
+ return self._deb.has_history(city_name)
+
+ def build_deb_accuracy_report(self, city_name: str, deb_query_cost: int) -> str:
+ return self._deb.build_deb_accuracy_report(city_name, deb_query_cost)
+
diff --git a/src/bot/command_guard.py b/src/bot/command_guard.py
new file mode 100644
index 00000000..1e9f60b7
--- /dev/null
+++ b/src/bot/command_guard.py
@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from typing import Any
+
+from loguru import logger
+
+from src.bot.io_layer import BotIOLayer
+from src.bot.services.entitlement_service import BotEntitlementService
+
+
+class CommandGuard:
+ """Unified command gate: entitlement + points charge."""
+
+ def __init__(self, io_layer: BotIOLayer, entitlement_service: BotEntitlementService):
+ self.io_layer = io_layer
+ self.entitlement_service = entitlement_service
+
+ def ensure_entitled(self, message: Any, command_label: str) -> bool:
+ user = getattr(message, "from_user", None)
+ user_id = getattr(user, "id", None)
+ if user_id is None:
+ return False
+
+ decision = self.entitlement_service.check(int(user_id), command_label)
+ if decision.allowed:
+ return True
+
+ self.io_layer.bot.reply_to(
+ message,
+ (
+ "🔒 当前指令需要高级权限。\n"
+ "请先开通订阅后再使用。"
+ ),
+ )
+ logger.info(
+ "bot entitlement blocked command={} user_id={} reason={}",
+ command_label,
+ user_id,
+ decision.reason,
+ )
+ return False
+
+ def ensure_access_and_points(self, message: Any, cost: int, command_label: str) -> bool:
+ if not self.ensure_entitled(message, command_label):
+ return False
+ return self.io_layer.ensure_query_points(message, cost, command_label)
+
diff --git a/src/bot/handlers/__init__.py b/src/bot/handlers/__init__.py
new file mode 100644
index 00000000..ec2a04e3
--- /dev/null
+++ b/src/bot/handlers/__init__.py
@@ -0,0 +1,2 @@
+"""Bot command handlers."""
+
diff --git a/src/bot/handlers/activity.py b/src/bot/handlers/activity.py
new file mode 100644
index 00000000..de4b6f34
--- /dev/null
+++ b/src/bot/handlers/activity.py
@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from typing import Any
+
+from src.bot.io_layer import BotIOLayer
+
+
+class ActivityHandler:
+ def __init__(self, bot: Any, io_layer: BotIOLayer):
+ self.bot = bot
+ self.io_layer = io_layer
+
+ def register(self) -> None:
+ @self.bot.message_handler(func=lambda message: True, content_types=["text"])
+ def _activity(message):
+ self.handle(message)
+
+ def handle(self, message: Any) -> None:
+ self.io_layer.track_group_text_activity(message)
+
diff --git a/src/bot/handlers/basic.py b/src/bot/handlers/basic.py
new file mode 100644
index 00000000..e74f2c42
--- /dev/null
+++ b/src/bot/handlers/basic.py
@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from typing import Any
+from typing import Callable
+
+from src.bot.io_layer import BotIOLayer
+from src.bot.observability import CommandTrace
+from src.bot.runtime_coordinator import RuntimeStatus, render_runtime_status_html
+
+
+class BasicCommandHandler:
+ def __init__(
+ self,
+ bot: Any,
+ io_layer: BotIOLayer,
+ runtime_status_provider: Callable[[], RuntimeStatus],
+ ):
+ self.bot = bot
+ self.io_layer = io_layer
+ self.runtime_status_provider = runtime_status_provider
+
+ def register(self) -> None:
+ @self.bot.message_handler(commands=["start", "help"])
+ def _start_help(message):
+ self.handle_start_help(message)
+
+ @self.bot.message_handler(commands=["id"])
+ def _id(message):
+ self.handle_id(message)
+
+ @self.bot.message_handler(commands=["top"])
+ def _top(message):
+ self.handle_top(message)
+
+ @self.bot.message_handler(commands=["diag"])
+ def _diag(message):
+ self.handle_diag(message)
+
+ def handle_start_help(self, message: Any) -> None:
+ trace = CommandTrace("/start", message)
+ try:
+ self.bot.reply_to(message, self.io_layer.build_welcome_text(), parse_mode="HTML")
+ trace.set_status("ok")
+ finally:
+ trace.emit()
+
+ def handle_id(self, message: Any) -> None:
+ trace = CommandTrace("/id", message)
+ try:
+ self.bot.reply_to(
+ message,
+ f"🎯 当前聊天的 Chat ID 是: {message.chat.id}",
+ parse_mode="HTML",
+ )
+ trace.set_status("ok")
+ finally:
+ trace.emit()
+
+ def handle_top(self, message: Any) -> None:
+ trace = CommandTrace("/top", message)
+ try:
+ rank_text = self.io_layer.build_points_rank_text(message.from_user)
+ self.bot.send_message(message.chat.id, rank_text, parse_mode="HTML")
+ trace.set_status("ok")
+ finally:
+ trace.emit()
+
+ def handle_diag(self, message: Any) -> None:
+ trace = CommandTrace("/diag", message)
+ try:
+ status = self.runtime_status_provider()
+ self.bot.reply_to(message, render_runtime_status_html(status), parse_mode="HTML")
+ trace.set_status("ok")
+ finally:
+ trace.emit()
diff --git a/src/bot/handlers/city.py b/src/bot/handlers/city.py
new file mode 100644
index 00000000..1e429b48
--- /dev/null
+++ b/src/bot/handlers/city.py
@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from typing import Any
+
+from loguru import logger
+
+from src.bot.command_guard import CommandGuard
+from src.bot.observability import CommandTrace
+from src.bot.services.city_command_service import CityCommandService
+from src.bot.settings import CITY_QUERY_COST
+
+
+class CityCommandHandler:
+ def __init__(
+ self,
+ bot: Any,
+ guard: CommandGuard,
+ city_service: CityCommandService,
+ ):
+ self.bot = bot
+ self.guard = guard
+ self.city_service = city_service
+
+ def register(self) -> None:
+ @self.bot.message_handler(commands=["city"])
+ def _city(message):
+ self.handle(message)
+
+ def handle(self, message: Any) -> None:
+ trace = CommandTrace("/city", message)
+ try:
+ parts = (message.text or "").split(maxsplit=1)
+ if len(parts) < 2:
+ trace.set_status("bad_request", "missing_city")
+ self.bot.reply_to(
+ message,
+ "❌ 请输入城市名称\n\n用法: /city chicago",
+ parse_mode="HTML",
+ )
+ return
+
+ city_input = parts[1].strip().lower()
+ resolved = self.city_service.resolve_city(city_input)
+ if not resolved.ok:
+ city_list = ", ".join(resolved.supported_cities or [])
+ trace.set_status("bad_request", "city_not_supported")
+ self.bot.reply_to(
+ message,
+ f"❌ 未找到城市: {city_input}\n\n支持的城市: {city_list}",
+ parse_mode="HTML",
+ )
+ return
+
+ city_name = str(resolved.city_name)
+ if not self.guard.ensure_access_and_points(message, CITY_QUERY_COST, "/city"):
+ trace.set_status("blocked", "guard_rejected")
+ return
+
+ self.bot.send_message(
+ message.chat.id, f"🔍 正在查询 {city_name.title()} 的天气数据..."
+ )
+ report_result = self.city_service.build_report(city_name, CITY_QUERY_COST)
+ if not report_result.ok:
+ trace.set_status("failed", report_result.error or "city_report_failed")
+ self.bot.reply_to(message, f"❌ 查询失败: {report_result.error}")
+ return
+
+ self.bot.send_message(message.chat.id, str(report_result.report), parse_mode="HTML")
+ trace.set_status("ok", city_name)
+ except Exception as exc:
+ trace.set_status("failed", "unexpected_error")
+ logger.exception("查询 /city 失败")
+ self.bot.reply_to(message, f"❌ 查询失败: {exc}")
+ finally:
+ trace.emit()
+
diff --git a/src/bot/handlers/deb.py b/src/bot/handlers/deb.py
new file mode 100644
index 00000000..0056d3aa
--- /dev/null
+++ b/src/bot/handlers/deb.py
@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from typing import Any
+
+from loguru import logger
+
+from src.bot.command_guard import CommandGuard
+from src.bot.observability import CommandTrace
+from src.bot.services.deb_command_service import DebCommandService
+from src.bot.settings import DEB_QUERY_COST
+
+
+class DebCommandHandler:
+ def __init__(
+ self,
+ bot: Any,
+ guard: CommandGuard,
+ deb_service: DebCommandService,
+ ):
+ self.bot = bot
+ self.guard = guard
+ self.deb_service = deb_service
+
+ def register(self) -> None:
+ @self.bot.message_handler(commands=["deb"])
+ def _deb(message):
+ self.handle(message)
+
+ def handle(self, message: Any) -> None:
+ trace = CommandTrace("/deb", message)
+ try:
+ parts = (message.text or "").split(maxsplit=1)
+ if len(parts) < 2:
+ trace.set_status("bad_request", "missing_city")
+ self.bot.reply_to(
+ message,
+ "❌ 用法: /deb ankara",
+ parse_mode="HTML",
+ )
+ return
+
+ city_input = parts[1].strip().lower()
+ city_name = self.deb_service.resolve_city(city_input)
+ if not self.deb_service.has_history(city_name):
+ trace.set_status("bad_request", "history_missing")
+ self.bot.reply_to(
+ message,
+ f"❌ 暂无 {city_name} 的历史数据。",
+ parse_mode="HTML",
+ )
+ return
+
+ if not self.guard.ensure_access_and_points(message, DEB_QUERY_COST, "/deb"):
+ trace.set_status("blocked", "guard_rejected")
+ return
+
+ report_result = self.deb_service.build_report(city_name, DEB_QUERY_COST)
+ if not report_result.ok:
+ trace.set_status("failed", report_result.error or "deb_report_failed")
+ self.bot.reply_to(message, f"❌ 查询失败: {report_result.error}")
+ return
+
+ self.bot.reply_to(message, str(report_result.report), parse_mode="HTML")
+ trace.set_status("ok", city_name)
+ except Exception as exc:
+ trace.set_status("failed", "unexpected_error")
+ logger.exception("查询 /deb 失败")
+ self.bot.reply_to(message, f"❌ 查询失败: {exc}")
+ finally:
+ trace.emit()
+
diff --git a/src/bot/io_layer.py b/src/bot/io_layer.py
new file mode 100644
index 00000000..b2e7dd01
--- /dev/null
+++ b/src/bot/io_layer.py
@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from typing import Any
+
+from loguru import logger
+
+from src.bot.settings import (
+ CITY_QUERY_COST,
+ DEB_QUERY_COST,
+ MESSAGE_COOLDOWN_SEC,
+ MESSAGE_DAILY_CAP,
+ MESSAGE_MIN_LENGTH,
+ MESSAGE_POINTS,
+)
+from src.database.db_manager import DBManager
+
+
+class BotIOLayer:
+ """Telegram IO + points/account side effects."""
+
+ def __init__(self, bot: Any, db: DBManager):
+ self.bot = bot
+ self.db = db
+
+ @staticmethod
+ def display_name(user: Any) -> str:
+ return user.username or user.first_name or f"User_{user.id}"
+
+ def ensure_query_points(self, message: Any, cost: int, label: str) -> bool:
+ user = message.from_user
+ self.db.upsert_user(user.id, self.display_name(user))
+ result = self.db.spend_points(user.id, cost)
+ if result.get("ok"):
+ return True
+
+ balance = int(result.get("balance") or 0)
+ required = int(result.get("required") or cost)
+ missing = max(0, required - balance)
+ self.bot.reply_to(
+ message,
+ (
+ f"❌ 积分不足，无法执行 {label}\n"
+ f"当前积分: {balance}\n"
+ f"需要积分: {required}\n"
+ f"还差积分: {missing}\n\n"
+ f"积分规则：每日签到（有效发言满 {MESSAGE_MIN_LENGTH} 字）获得 {MESSAGE_POINTS} 积分，"
+ f"每日上限 {MESSAGE_DAILY_CAP} 分。"
+ ),
+ parse_mode="HTML",
+ )
+ return False
+
+ def build_welcome_text(self) -> str:
+ return (
+ "🚀 PolyWeather 天气查询机器人\n\n"
+ "可用指令:\n"
+ f"/city [城市名] - 查询城市天气预测与实测 (消耗 {CITY_QUERY_COST} 积分)\n"
+ f"/deb [城市名] - 查看 DEB 融合预测准确率 (消耗 {DEB_QUERY_COST} 积分)\n"
+ "/top - 查看积分排行榜\n"
+ "/id - 获取当前聊天的 Chat ID\n\n"
+ "/diag - 查看 Bot 启动诊断\n\n"
+ "示例: /city 伦敦\n"
+ f"💡 提示: 每日签到(有效发言满 {MESSAGE_MIN_LENGTH} 字)获得 {MESSAGE_POINTS} 积分，"
+ f"每日上限 {MESSAGE_DAILY_CAP} 分。"
+ )
+
+ def build_points_rank_text(self, user: Any) -> str:
+ self.db.upsert_user(user.id, self.display_name(user))
+ user_info = self.db.get_user(user.id)
+
+ leaderboard = self.db.get_leaderboard(limit=5)
+ rank_text = "🏆 PolyWeather 活跃度排行榜\n"
+ rank_text += "────────────────────\n"
+ for i, entry in enumerate(leaderboard):
+ medal = ["🥇", "🥈", "🥉", " ", " "][i] if i < 5 else " "
+ rank_text += f"{medal} {entry['username'][:12]}: {entry['points']} 点\n"
+
+ if user_info:
+ rank_text += "────────────────────\n"
+ rank_text += (
+ "👤 我的状态：\n"
+ f"┣ 积分: {user_info['points']}\n"
+ f"┣ 发言: {user_info['message_count']} 次\n"
+ f"┣ 今日发言积分: {user_info.get('daily_points') or 0}/{MESSAGE_DAILY_CAP}\n"
+ f"┗ /city 消耗: {CITY_QUERY_COST} | /deb 消耗: {DEB_QUERY_COST}"
+ )
+ return rank_text
+
+ def track_group_text_activity(self, message: Any) -> None:
+ text = str(getattr(message, "text", "") or "")
+ if text.startswith("/"):
+ return
+ chat = getattr(message, "chat", None)
+ if not chat or chat.type not in ("group", "supergroup"):
+ return
+
+ user = message.from_user
+ username = self.display_name(user)
+ self.db.upsert_user(user.id, username)
+
+ result = self.db.add_message_activity(
+ user.id,
+ text=text,
+ points_to_add=MESSAGE_POINTS,
+ cooldown_sec=MESSAGE_COOLDOWN_SEC,
+ daily_cap=MESSAGE_DAILY_CAP,
+ min_text_length=MESSAGE_MIN_LENGTH,
+ )
+ if result.get("awarded"):
+ logger.info(
+ f"message points awarded user={user.id} points=+{MESSAGE_POINTS} "
+ f"daily_points={result.get('daily_points')}/{MESSAGE_DAILY_CAP}"
+ )
diff --git a/src/bot/observability.py b/src/bot/observability.py
new file mode 100644
index 00000000..a8e277d3
--- /dev/null
+++ b/src/bot/observability.py
@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from time import perf_counter
+from typing import Any
+
+from loguru import logger
+
+
+def _safe_attr(obj: Any, attr: str, default: Any = None) -> Any:
+ try:
+ return getattr(obj, attr, default)
+ except Exception:
+ return default
+
+
+class CommandTrace:
+ """Small helper for structured command logs."""
+
+ def __init__(self, command: str, message: Any):
+ self.command = command
+ self.user_id = _safe_attr(_safe_attr(message, "from_user"), "id", "unknown")
+ self.chat_id = _safe_attr(_safe_attr(message, "chat"), "id", "unknown")
+ self.start = perf_counter()
+ self.status = "ok"
+ self.note = ""
+
+ def set_status(self, status: str, note: str = "") -> None:
+ self.status = status
+ self.note = note
+
+ def emit(self) -> None:
+ elapsed_ms = (perf_counter() - self.start) * 1000.0
+ note_part = f" note={self.note}" if self.note else ""
+ logger.info(
+ "bot command command={} status={} user_id={} chat_id={} elapsed_ms={:.1f}{}",
+ self.command,
+ self.status,
+ self.user_id,
+ self.chat_id,
+ elapsed_ms,
+ note_part,
+ )
+
diff --git a/src/bot/orchestrator.py b/src/bot/orchestrator.py
new file mode 100644
index 00000000..79bbae39
--- /dev/null
+++ b/src/bot/orchestrator.py
@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from loguru import logger # type: ignore
+
+from src.bot.analysis.city_analysis_service import CityAnalysisService
+from src.bot.analysis.deb_analysis_service import DebAnalysisService
+from src.bot.command_guard import CommandGuard
+from src.bot.handlers.activity import ActivityHandler
+from src.bot.handlers.basic import BasicCommandHandler
+from src.bot.handlers.city import CityCommandHandler
+from src.bot.handlers.deb import DebCommandHandler
+from src.bot.io_layer import BotIOLayer
+from src.bot.runtime_coordinator import StartupCoordinator
+from src.bot.services.city_command_service import CityCommandService
+from src.bot.services.deb_command_service import DebCommandService
+from src.bot.services.entitlement_service import BotEntitlementService
+
+
+def _project_root() -> str:
+ return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+
+def _register_handlers(
+ bot: Any,
+ io_layer: BotIOLayer,
+ guard: CommandGuard,
+ city_service: CityCommandService,
+ deb_service: DebCommandService,
+ startup_coordinator: StartupCoordinator,
+) -> None:
+ BasicCommandHandler(
+ bot=bot,
+ io_layer=io_layer,
+ runtime_status_provider=startup_coordinator.get_runtime_status,
+ ).register()
+ CityCommandHandler(bot=bot, guard=guard, city_service=city_service).register()
+ DebCommandHandler(bot=bot, guard=guard, deb_service=deb_service).register()
+ ActivityHandler(bot=bot, io_layer=io_layer).register()
+
+
+def start_bot() -> None:
+ import telebot # type: ignore
+
+ from src.data_collection.weather_sources import WeatherDataCollector
+ from src.database.db_manager import DBManager
+ from src.utils.config_loader import load_config
+
+ config = load_config()
+ token = os.getenv("TELEGRAM_BOT_TOKEN")
+ if not token:
+ logger.error("未找到 TELEGRAM_BOT_TOKEN 环境变量")
+ return
+
+ bot = telebot.TeleBot(token)
+ db = DBManager()
+ weather = WeatherDataCollector(config)
+
+ io_layer = BotIOLayer(bot=bot, db=db)
+ city_analysis = CityAnalysisService(weather=weather)
+ deb_analysis = DebAnalysisService(project_root=_project_root())
+ entitlement_service = BotEntitlementService(db=db)
+ guard = CommandGuard(io_layer=io_layer, entitlement_service=entitlement_service)
+ city_service = CityCommandService(analysis=city_analysis)
+ deb_service = DebCommandService(analysis=deb_analysis)
+ startup_coordinator = StartupCoordinator(
+ bot=bot,
+ config=config,
+ entitlement_enabled=entitlement_service.enabled,
+ protected_commands=sorted(entitlement_service.protected_commands),
+ )
+
+ _register_handlers(
+ bot=bot,
+ io_layer=io_layer,
+ guard=guard,
+ city_service=city_service,
+ deb_service=deb_service,
+ startup_coordinator=startup_coordinator,
+ )
+ runtime_status = startup_coordinator.start_all()
+ started_count = sum(1 for loop in runtime_status.loops if loop.started)
+
+ logger.info(
+ "🤖 Bot 启动中... entitlement_enabled={} protected_commands={} loops_started={}/{}",
+ entitlement_service.enabled,
+ ",".join(sorted(entitlement_service.protected_commands)),
+ started_count,
+ len(runtime_status.loops),
+ )
+ bot.infinity_polling()
diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py
new file mode 100644
index 00000000..27438ffb
--- /dev/null
+++ b/src/bot/runtime_coordinator.py
@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from datetime import datetime
+from importlib import import_module
+from typing import Any, Callable, Dict, List, Optional
+
+
+def _env_bool(name: str, default: bool) -> bool:
+ raw = os.getenv(name)
+ if raw is None:
+ return default
+ return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _env_int(name: str, default: int) -> int:
+ raw = os.getenv(name)
+ if raw is None:
+ return default
+ try:
+ return int(raw)
+ except Exception:
+ return default
+
+
+def _parse_csv_count(raw: Optional[str]) -> int:
+ if not raw:
+ return 0
+ return len([part for part in str(raw).split(",") if str(part).strip()])
+
+
+@dataclass
+class LoopStatus:
+ key: str
+ label: str
+ configured_enabled: bool
+ started: bool
+ reason: str
+ details: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class RuntimeStatus:
+ started_at: str
+ loops: List[LoopStatus]
+ entitlement_enabled: bool
+ protected_commands: List[str]
+
+ def loop_map(self) -> Dict[str, LoopStatus]:
+ return {loop.key: loop for loop in self.loops}
+
+
+class StartupCoordinator:
+ """Centralized startup orchestration + diagnostics snapshot."""
+
+ def __init__(
+ self,
+ bot: Any,
+ config: Dict[str, Any],
+ entitlement_enabled: bool,
+ protected_commands: List[str],
+ ):
+ self.bot = bot
+ self.config = config
+ self.entitlement_enabled = entitlement_enabled
+ self.protected_commands = protected_commands
+ self._runtime_status = RuntimeStatus(
+ started_at=datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"),
+ loops=[],
+ entitlement_enabled=entitlement_enabled,
+ protected_commands=protected_commands,
+ )
+
+ def get_runtime_status(self) -> RuntimeStatus:
+ return self._runtime_status
+
+ def start_all(self) -> RuntimeStatus:
+ loops = [
+ self._start_trade_alert_loop(),
+ self._start_polygon_wallet_loop(),
+ self._start_polymarket_wallet_activity_loop(),
+ ]
+ self._runtime_status = RuntimeStatus(
+ started_at=datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"),
+ loops=loops,
+ entitlement_enabled=self.entitlement_enabled,
+ protected_commands=self.protected_commands,
+ )
+ return self._runtime_status
+
+ def _start_with_validation(
+ self,
+ key: str,
+ label: str,
+ configured_enabled: bool,
+ details: Dict[str, Any],
+ validation_error: Optional[str],
+ starter: Callable[[], Any],
+ ) -> LoopStatus:
+ if not configured_enabled:
+ return LoopStatus(
+ key=key,
+ label=label,
+ configured_enabled=False,
+ started=False,
+ reason="disabled_by_env",
+ details=details,
+ )
+ if validation_error:
+ return LoopStatus(
+ key=key,
+ label=label,
+ configured_enabled=True,
+ started=False,
+ reason=validation_error,
+ details=details,
+ )
+ thread = starter()
+ started = thread is not None
+ reason = "started" if started else "starter_returned_none"
+ if started:
+ details = {**details, "thread": getattr(thread, "name", "")}
+ return LoopStatus(
+ key=key,
+ label=label,
+ configured_enabled=True,
+ started=started,
+ reason=reason,
+ details=details,
+ )
+
+ def _start_trade_alert_loop(self) -> LoopStatus:
+ enabled = _env_bool("TELEGRAM_ALERT_PUSH_ENABLED", True)
+ chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip()
+ mispricing_only = _env_bool("TELEGRAM_ALERT_MISPRICING_ONLY", True)
+ interval = (
+ max(300, _env_int("TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC", 7200))
+ if mispricing_only
+ else max(60, _env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 300))
+ )
+ cities_count = _parse_csv_count(os.getenv("TELEGRAM_ALERT_CITIES"))
+ details = {
+ "mode": "mispricing-only" if mispricing_only else "full",
+ "interval_sec": interval,
+ "cities_count": cities_count,
+ }
+ validation_error = None if chat_id else "missing_TELEGRAM_CHAT_ID"
+ return self._start_with_validation(
+ key="trade_alert_push",
+ label="错价雷达推送",
+ configured_enabled=enabled,
+ details=details,
+ validation_error=validation_error,
+ starter=lambda: import_module("src.utils.telegram_push").start_trade_alert_push_loop(
+ self.bot,
+ self.config,
+ ),
+ )
+
+ def _start_polygon_wallet_loop(self) -> LoopStatus:
+ enabled = _env_bool("POLYGON_WALLET_WATCH_ENABLED", False)
+ chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip()
+ rpc_url = str(os.getenv("POLYGON_RPC_URL") or "").strip()
+ wallets_count = _parse_csv_count(os.getenv("POLYGON_WALLET_WATCH_ADDRESSES"))
+ poll = max(3, _env_int("POLYGON_WALLET_WATCH_INTERVAL_SEC", 8))
+ details = {
+ "poll_sec": poll,
+ "wallets_count": wallets_count,
+ "polymarket_only": _env_bool("POLYGON_WALLET_WATCH_POLYMARKET_ONLY", True),
+ }
+ validation_error = None
+ if not chat_id:
+ validation_error = "missing_TELEGRAM_CHAT_ID"
+ elif not rpc_url:
+ validation_error = "missing_POLYGON_RPC_URL"
+ elif wallets_count == 0:
+ validation_error = "missing_POLYGON_WALLET_WATCH_ADDRESSES"
+ return self._start_with_validation(
+ key="polygon_wallet_watch",
+ label="Polygon 钱包监听",
+ configured_enabled=enabled,
+ details=details,
+ validation_error=validation_error,
+ starter=lambda: import_module(
+ "src.onchain.polygon_wallet_watcher"
+ ).start_polygon_wallet_watch_loop(self.bot),
+ )
+
+ def _start_polymarket_wallet_activity_loop(self) -> LoopStatus:
+ enabled = _env_bool("POLYMARKET_WALLET_ACTIVITY_ENABLED", False)
+ chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip()
+ users_count = _parse_csv_count(os.getenv("POLYMARKET_WALLET_ACTIVITY_USERS"))
+ poll = max(5, _env_int("POLYMARKET_WALLET_ACTIVITY_INTERVAL_SEC", 20))
+ details = {
+ "poll_sec": poll,
+ "users_count": users_count,
+ "link_preview": _env_bool("POLYMARKET_WALLET_ACTIVITY_LINK_PREVIEW", True),
+ }
+ validation_error = None
+ if not chat_id:
+ validation_error = "missing_TELEGRAM_CHAT_ID"
+ elif users_count == 0:
+ validation_error = "missing_POLYMARKET_WALLET_ACTIVITY_USERS"
+ return self._start_with_validation(
+ key="polymarket_wallet_activity",
+ label="Polymarket 钱包异动监听",
+ configured_enabled=enabled,
+ details=details,
+ validation_error=validation_error,
+ starter=lambda: import_module(
+ "src.onchain.polymarket_wallet_activity_watcher"
+ ).start_polymarket_wallet_activity_loop(self.bot),
+ )
+
+
+def render_runtime_status_html(status: RuntimeStatus) -> str:
+ lines = [
+ "🧭 Bot 启动诊断",
+ f"启动时间: {status.started_at}",
+ "",
+ f"Entitlement: {'ON' if status.entitlement_enabled else 'OFF'}",
+ f"受保护命令: {', '.join(status.protected_commands) or '--'}",
+ "",
+ "后台循环:",
+ ]
+ for loop in status.loops:
+ icon = "✅" if loop.started else ("⏸" if not loop.configured_enabled else "⚠️")
+ detail_str = ", ".join(f"{k}={v}" for k, v in sorted(loop.details.items()))
+ lines.append(
+ f"{icon} {loop.label} | enabled={str(loop.configured_enabled).lower()} | "
+ f"started={str(loop.started).lower()} | reason={loop.reason}"
+ )
+ if detail_str:
+ lines.append(f" {detail_str}")
+ return "\n".join(lines)
diff --git a/src/bot/services/__init__.py b/src/bot/services/__init__.py
new file mode 100644
index 00000000..73224098
--- /dev/null
+++ b/src/bot/services/__init__.py
@@ -0,0 +1,2 @@
+"""Bot service layer."""
+
diff --git a/src/bot/services/city_command_service.py b/src/bot/services/city_command_service.py
new file mode 100644
index 00000000..878f6b7b
--- /dev/null
+++ b/src/bot/services/city_command_service.py
@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import List, Optional
+
+from src.bot.analysis.city_analysis_service import CityAnalysisService
+
+
+@dataclass
+class CityResolveResult:
+ ok: bool
+ city_name: Optional[str] = None
+ supported_cities: List[str] | None = None
+
+
+@dataclass
+class CityReportResult:
+ ok: bool
+ report: Optional[str] = None
+ error: Optional[str] = None
+
+
+class CityCommandService:
+ def __init__(self, analysis: CityAnalysisService):
+ self.analysis = analysis
+
+ def resolve_city(self, city_input: str) -> CityResolveResult:
+ city_name, supported = self.analysis.resolve_city(city_input)
+ if not city_name:
+ return CityResolveResult(ok=False, supported_cities=supported)
+ return CityResolveResult(ok=True, city_name=city_name, supported_cities=supported)
+
+ def build_report(self, city_name: str, city_query_cost: int) -> CityReportResult:
+ try:
+ report = self.analysis.build_city_report(city_name, city_query_cost)
+ return CityReportResult(ok=True, report=report)
+ except Exception as exc:
+ return CityReportResult(ok=False, error=str(exc))
diff --git a/src/bot/services/deb_command_service.py b/src/bot/services/deb_command_service.py
new file mode 100644
index 00000000..8e9bb66d
--- /dev/null
+++ b/src/bot/services/deb_command_service.py
@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+from src.bot.analysis.deb_analysis_service import DebAnalysisService
+
+
+@dataclass
+class DebReportResult:
+ ok: bool
+ report: Optional[str] = None
+ error: Optional[str] = None
+
+
+class DebCommandService:
+ def __init__(self, analysis: DebAnalysisService):
+ self.analysis = analysis
+
+ def resolve_city(self, city_input: str) -> str:
+ return self.analysis.resolve_deb_city(city_input)
+
+ def has_history(self, city_name: str) -> bool:
+ return self.analysis.has_deb_history(city_name)
+
+ def build_report(self, city_name: str, deb_query_cost: int) -> DebReportResult:
+ try:
+ report = self.analysis.build_deb_accuracy_report(city_name, deb_query_cost)
+ return DebReportResult(ok=True, report=report)
+ except Exception as exc:
+ return DebReportResult(ok=False, error=str(exc))
diff --git a/src/bot/services/entitlement_service.py b/src/bot/services/entitlement_service.py
new file mode 100644
index 00000000..49327b20
--- /dev/null
+++ b/src/bot/services/entitlement_service.py
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Iterable, Set
+
+from src.database.db_manager import DBManager
+
+
+def _env_bool(name: str, default: bool = False) -> bool:
+ raw = str(os.getenv(name, "")).strip().lower()
+ if not raw:
+ return default
+ return raw in {"1", "true", "yes", "on"}
+
+
+@dataclass
+class EntitlementDecision:
+ allowed: bool
+ reason: str
+
+
+class BotEntitlementService:
+ """
+ Payment/entitlement pre-hook for command access.
+
+ Disabled by default. Enable with:
+ POLYWEATHER_BOT_REQUIRE_ENTITLEMENT=true
+ """
+
+ def __init__(
+ self,
+ db: DBManager,
+ enabled: bool | None = None,
+ protected_commands: Iterable[str] | None = None,
+ ):
+ self.db = db
+ self.enabled = _env_bool("POLYWEATHER_BOT_REQUIRE_ENTITLEMENT", False) if enabled is None else enabled
+ commands = protected_commands or ("/city", "/deb")
+ self.protected_commands: Set[str] = {str(c).strip().lower() for c in commands if str(c).strip()}
+
+ def check(self, user_id: int, command_label: str) -> EntitlementDecision:
+ command = str(command_label or "").strip().lower()
+ if not self.enabled:
+ return EntitlementDecision(True, "entitlement_disabled")
+ if command not in self.protected_commands:
+ return EntitlementDecision(True, "command_not_protected")
+
+ user = self.db.get_user(user_id) or {}
+ has_premium = bool(user.get("is_web_premium") or user.get("is_group_premium"))
+ if has_premium:
+ return EntitlementDecision(True, "premium_user")
+ return EntitlementDecision(False, "premium_required")
+
diff --git a/src/bot/settings.py b/src/bot/settings.py
new file mode 100644
index 00000000..c5bf5f94
--- /dev/null
+++ b/src/bot/settings.py
@@ -0,0 +1,7 @@
+MESSAGE_POINTS = 4
+MESSAGE_DAILY_CAP = 50
+MESSAGE_MIN_LENGTH = 2
+MESSAGE_COOLDOWN_SEC = 30
+CITY_QUERY_COST = 1
+DEB_QUERY_COST = 1
+
diff --git a/tests/test_bot_basic_handler.py b/tests/test_bot_basic_handler.py
new file mode 100644
index 00000000..103ca42f
--- /dev/null
+++ b/tests/test_bot_basic_handler.py
@@ -0,0 +1,55 @@
+from types import SimpleNamespace
+
+from src.bot.handlers.basic import BasicCommandHandler
+from src.bot.runtime_coordinator import RuntimeStatus
+
+
+class DummyBot:
+ def __init__(self):
+ self.replies = []
+
+ def reply_to(self, message, text, parse_mode=None):
+ self.replies.append({"text": text, "parse_mode": parse_mode, "chat_id": message.chat.id})
+
+ def send_message(self, chat_id, text, parse_mode=None): # pragma: no cover
+ pass
+
+ def message_handler(self, *args, **kwargs): # pragma: no cover - decorator stub
+ def _decorator(func):
+ return func
+
+ return _decorator
+
+
+def _message(text: str):
+ return SimpleNamespace(
+ text=text,
+ from_user=SimpleNamespace(id=1, username="u", first_name="U"),
+ chat=SimpleNamespace(id=100),
+ )
+
+
+def test_basic_handler_diag_returns_html():
+ runtime = RuntimeStatus(
+ started_at="2026-03-12 00:00:00 UTC",
+ loops=[],
+ entitlement_enabled=False,
+ protected_commands=["/city", "/deb"],
+ )
+ bot = DummyBot()
+ io_layer = SimpleNamespace(
+ build_welcome_text=lambda: "WELCOME",
+ build_points_rank_text=lambda _user: "TOP",
+ )
+ handler = BasicCommandHandler(
+ bot=bot,
+ io_layer=io_layer,
+ runtime_status_provider=lambda: runtime,
+ )
+
+ handler.handle_diag(_message("/diag"))
+
+ assert len(bot.replies) == 1
+ assert bot.replies[0]["parse_mode"] == "HTML"
+ assert "Bot 启动诊断" in bot.replies[0]["text"]
+
diff --git a/tests/test_bot_command_guard.py b/tests/test_bot_command_guard.py
new file mode 100644
index 00000000..537ca206
--- /dev/null
+++ b/tests/test_bot_command_guard.py
@@ -0,0 +1,50 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from src.bot.command_guard import CommandGuard
+from src.bot.services.entitlement_service import BotEntitlementService
+
+
+class _FakeDB:
+ def __init__(self, user):
+ self._user = user
+
+ def get_user(self, _user_id):
+ return self._user
+
+
+def _message():
+ return SimpleNamespace(
+ from_user=SimpleNamespace(id=123, username="tester", first_name="Tester"),
+ chat=SimpleNamespace(id=999),
+ )
+
+
+def test_guard_blocks_non_premium_when_entitlement_enabled():
+ fake_bot = SimpleNamespace(reply_to=Mock())
+ io_layer = SimpleNamespace(bot=fake_bot, ensure_query_points=Mock(return_value=True))
+ entitlement = BotEntitlementService(db=_FakeDB(user=None), enabled=True)
+ guard = CommandGuard(io_layer=io_layer, entitlement_service=entitlement)
+
+ ok = guard.ensure_access_and_points(_message(), 1, "/city")
+
+ assert ok is False
+ assert io_layer.ensure_query_points.call_count == 0
+ assert fake_bot.reply_to.call_count == 1
+
+
+def test_guard_allows_premium_then_charges_points():
+ fake_bot = SimpleNamespace(reply_to=Mock())
+ io_layer = SimpleNamespace(bot=fake_bot, ensure_query_points=Mock(return_value=True))
+ entitlement = BotEntitlementService(
+ db=_FakeDB(user={"is_web_premium": 1, "is_group_premium": 0}),
+ enabled=True,
+ )
+ guard = CommandGuard(io_layer=io_layer, entitlement_service=entitlement)
+
+ ok = guard.ensure_access_and_points(_message(), 1, "/city")
+
+ assert ok is True
+ assert io_layer.ensure_query_points.call_count == 1
+ assert fake_bot.reply_to.call_count == 0
+
diff --git a/tests/test_bot_handlers.py b/tests/test_bot_handlers.py
new file mode 100644
index 00000000..34a7d235
--- /dev/null
+++ b/tests/test_bot_handlers.py
@@ -0,0 +1,98 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from src.bot.handlers.city import CityCommandHandler
+from src.bot.handlers.deb import DebCommandHandler
+from src.bot.services.city_command_service import CityReportResult, CityResolveResult
+from src.bot.services.deb_command_service import DebReportResult
+
+
+class DummyBot:
+ def __init__(self):
+ self.replies = []
+ self.sent = []
+
+ def reply_to(self, message, text, parse_mode=None):
+ self.replies.append(
+ {
+ "chat_id": message.chat.id,
+ "text": text,
+ "parse_mode": parse_mode,
+ }
+ )
+
+ def send_message(self, chat_id, text, parse_mode=None):
+ self.sent.append(
+ {
+ "chat_id": chat_id,
+ "text": text,
+ "parse_mode": parse_mode,
+ }
+ )
+
+ def message_handler(self, *args, **kwargs): # pragma: no cover - decorator stub
+ def _decorator(func):
+ return func
+
+ return _decorator
+
+
+def _message(text: str):
+ return SimpleNamespace(
+ text=text,
+ from_user=SimpleNamespace(id=123, username="tester", first_name="Tester"),
+ chat=SimpleNamespace(id=999),
+ )
+
+
+def test_city_handler_missing_city_shows_usage():
+ bot = DummyBot()
+ guard = SimpleNamespace(ensure_access_and_points=Mock(return_value=True))
+ city_service = SimpleNamespace(
+ resolve_city=Mock(),
+ build_report=Mock(),
+ )
+ handler = CityCommandHandler(bot=bot, guard=guard, city_service=city_service)
+
+ handler.handle(_message("/city"))
+
+ assert len(bot.replies) == 1
+ assert "请输入城市名称" in bot.replies[0]["text"]
+ assert guard.ensure_access_and_points.call_count == 0
+
+
+def test_city_handler_happy_path_pushes_progress_and_report():
+ bot = DummyBot()
+ guard = SimpleNamespace(ensure_access_and_points=Mock(return_value=True))
+ city_service = SimpleNamespace(
+ resolve_city=Mock(
+ return_value=CityResolveResult(ok=True, city_name="london", supported_cities=["london"])
+ ),
+ build_report=Mock(return_value=CityReportResult(ok=True, report="CITY_REPORT")),
+ )
+ handler = CityCommandHandler(bot=bot, guard=guard, city_service=city_service)
+
+ handler.handle(_message("/city london"))
+
+ assert len(bot.sent) == 2
+ assert "正在查询 London 的天气数据" in bot.sent[0]["text"]
+ assert bot.sent[1]["text"] == "CITY_REPORT"
+ assert bot.sent[1]["parse_mode"] == "HTML"
+
+
+def test_deb_handler_history_missing_returns_hint():
+ bot = DummyBot()
+ guard = SimpleNamespace(ensure_access_and_points=Mock(return_value=True))
+ deb_service = SimpleNamespace(
+ resolve_city=Mock(return_value="ankara"),
+ has_history=Mock(return_value=False),
+ build_report=Mock(return_value=DebReportResult(ok=True, report="DEB_REPORT")),
+ )
+ handler = DebCommandHandler(bot=bot, guard=guard, deb_service=deb_service)
+
+ handler.handle(_message("/deb ankara"))
+
+ assert len(bot.replies) == 1
+ assert "暂无 ankara 的历史数据" in bot.replies[0]["text"]
+ assert guard.ensure_access_and_points.call_count == 0
+
diff --git a/tests/test_bot_runtime_coordinator.py b/tests/test_bot_runtime_coordinator.py
new file mode 100644
index 00000000..2f217c49
--- /dev/null
+++ b/tests/test_bot_runtime_coordinator.py
@@ -0,0 +1,42 @@
+from src.bot.runtime_coordinator import RuntimeStatus, StartupCoordinator, render_runtime_status_html
+
+
+class DummyBot:
+ pass
+
+
+def test_startup_coordinator_respects_disable_flags(monkeypatch):
+ monkeypatch.setenv("TELEGRAM_ALERT_PUSH_ENABLED", "false")
+ monkeypatch.setenv("POLYGON_WALLET_WATCH_ENABLED", "false")
+ monkeypatch.setenv("POLYMARKET_WALLET_ACTIVITY_ENABLED", "false")
+ monkeypatch.delenv("TELEGRAM_CHAT_ID", raising=False)
+
+ coordinator = StartupCoordinator(
+ bot=DummyBot(),
+ config={},
+ entitlement_enabled=False,
+ protected_commands=["/city", "/deb"],
+ )
+ runtime = coordinator.start_all()
+ loop_map = runtime.loop_map()
+
+ assert loop_map["trade_alert_push"].configured_enabled is False
+ assert loop_map["trade_alert_push"].started is False
+ assert loop_map["trade_alert_push"].reason == "disabled_by_env"
+ assert loop_map["polygon_wallet_watch"].reason == "disabled_by_env"
+ assert loop_map["polymarket_wallet_activity"].reason == "disabled_by_env"
+
+
+def test_render_runtime_status_html_contains_key_fields():
+ runtime = RuntimeStatus(
+ started_at="2026-03-12 00:00:00 UTC",
+ entitlement_enabled=True,
+ protected_commands=["/city", "/deb"],
+ loops=[],
+ )
+ html = render_runtime_status_html(runtime)
+
+ assert "Bot 启动诊断" in html
+ assert "Entitlement" in html
+ assert "/city, /deb" in html
+