gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Switch market digests to interval-based pushes

Browse Source
This commit is contained in:
2569718930@qq.com
2026-04-01 19:23:35 +08:00
parent 9e9e5a56ba
commit 149452dcc0
4 changed files with 96 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+1 -2
View File
@@ -79,9 +79,8 @@ TELEGRAM_ALERT_MIN_SEVERITY=medium
TELEGRAM_ALERT_MISPRICING_ONLY=true
TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC=7200
TELEGRAM_MARKET_FOCUS_DIGEST_ENABLED=true
TELEGRAM_MARKET_FOCUS_DIGEST_HOURS=11,18
TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC=1800
TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N=5
TELEGRAM_MARKET_FOCUS_DIGEST_GRACE_MINUTES=180
TELEGRAM_ALERT_CITIES=ankara,london,paris,seoul,hong kong,shanghai,singapore,tokyo,tel aviv,toronto,buenos aires,wellington,new york,chicago,dallas,miami,atlanta,seattle,lucknow,sao paulo,munich
POLYWEATHER_MONITORING_ALERT_CHAT_IDS=
docs/CONFIGURATION_ZH.md
+6 -10
View File
@@ -125,9 +125,8 @@ PolyWeather 的环境变量很多，但不是所有变量都属于同一层级
- `TELEGRAM_ALERT_MIN_SEVERITY`
- `TELEGRAM_ALERT_MISPRICING_ONLY`
- `TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC`
- `TELEGRAM_MARKET_FOCUS_DIGEST_HOURS`
- `TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC`
- `TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N`
- `TELEGRAM_MARKET_FOCUS_DIGEST_GRACE_MINUTES`
- `POLYWEATHER_PAYMENT_RPC_URLS`
- `TAF_CACHE_TTL_SEC`
@@ -232,9 +231,8 @@ TELEGRAM_ALERT_MIN_SEVERITY=medium
TELEGRAM_ALERT_MISPRICING_ONLY=true
TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC=7200
TELEGRAM_MARKET_FOCUS_DIGEST_ENABLED=true
TELEGRAM_MARKET_FOCUS_DIGEST_HOURS=11,18
TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC=1800
TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N=5
TELEGRAM_MARKET_FOCUS_DIGEST_GRACE_MINUTES=180
POLYMARKET_WALLET_ACTIVITY_ENABLED=false
```
@@ -246,8 +244,8 @@ POLYMARKET_WALLET_ACTIVITY_ENABLED=false
- `docker-compose.yml` 会把这个目录同时挂载到容器内的 `/var/lib/polyweather` 和 `/app/data`，兼容现有缓存与 SQLite 路径。
- `POLYWEATHER_STATE_STORAGE_MODE` 当前线上推荐直接使用 `sqlite`。
- `POLYWEATHER_PAYMENT_RPC_URLS` 支持逗号分隔多个 RPC；如果暂时只用单 RPC，也可以继续只配 `POLYWEATHER_PAYMENT_RPC_URL`。
- 机器人市场监控当前分成两类消息：`关键提醒` 与 `关注清单`。
- `TELEGRAM_MARKET_FOCUS_DIGEST_HOURS` 直接按服务器本地时间解释，不再单独配置时区。
- 机器人市场监控当前以 `关注清单` 为主，按固定间隔主动推送。
- `TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC` 表示主动推送间隔，默认 `1800` 秒（30 分钟）。
- `POLYMARKET_WALLET_ACTIVITY_ENABLED` 已退役，保留为 `false` 即可，不建议再启用钱包异动监听。
### 6.3 机器人市场监控建议配置
@@ -268,18 +266,16 @@ TELEGRAM_ALERT_MIN_SEVERITY=medium
TELEGRAM_ALERT_MISPRICING_ONLY=true
TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC=7200
TELEGRAM_MARKET_FOCUS_DIGEST_ENABLED=true
TELEGRAM_MARKET_FOCUS_DIGEST_HOURS=11,18
TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC=1800
TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N=5
TELEGRAM_MARKET_FOCUS_DIGEST_GRACE_MINUTES=180
POLYMARKET_WALLET_ACTIVITY_ENABLED=false
```
说明：
- `TELEGRAM_ALERT_MISPRICING_ONLY=true` 表示关键提醒优先围绕错价/市场触发，不把机器人做成泛通知器。
- `TELEGRAM_MARKET_FOCUS_DIGEST_HOURS=11,18` 按服务器本地时间解释，通常可以对应白天与傍晚两个摘要窗口。
- `TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC=1800` 表示频道每 30 分钟主动推送一轮机会清单。
- `TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N=5` 建议先保持较小，避免机器人一次推太多城市。
- `TELEGRAM_MARKET_FOCUS_DIGEST_GRACE_MINUTES=180` 用于容忍进程重启/部署后的补发窗口。
- `POLYMARKET_WALLET_ACTIVITY_ENABLED=false` 表示停用旧的钱包异动监听，统一收敛到市场监控。
## 7. 当前建议的运维规则
src/bot/handlers/basic.py
+14 -5
View File
@@ -257,22 +257,31 @@ class BasicCommandHandler:
)
trace.set_status("blocked", f"unsupported_chat_type:{chat_type}")
return
self.bot.reply_to(message, "⏳ 正在生成当前市场概览，请稍候...")
chat_id = getattr(getattr(message, "chat", None), "id", None)
from src.utils.telegram_push import (
build_market_monitor_digest,
load_cached_market_monitor_digest,
)
cached_summary = load_cached_market_monitor_digest()
if cached_summary:
self.bot.reply_to(message, cached_summary)
else:
self.bot.reply_to(message, "⏳ 正在生成当前市场概览，请稍候...")
def _worker() -> None:
try:
from src.utils.telegram_push import build_market_monitor_digest
summary = build_market_monitor_digest(
self.config,
slot_label="当前市场概览",
force_refresh=False,
)
self.bot.send_message(chat_id, summary)
if not cached_summary or summary.strip() != cached_summary.strip():
self.bot.send_message(chat_id, summary)
except Exception:
self.bot.send_message(chat_id, "❌ 当前市场概览生成失败，请稍后重试。")
if not cached_summary:
self.bot.send_message(chat_id, "❌ 当前市场概览生成失败，请稍后重试。")
threading.Thread(
target=_worker,
src/utils/telegram_push.py
+75 -53
View File
@@ -262,24 +262,26 @@ def _cleanup_state(state: Dict[str, Any], now_ts: int, keep_sec: int = 7 * 86400
last_by_city.pop(city, None)
def _parse_hour_list(raw: Optional[str], default: List[int]) -> List[int]:
if not raw:
return default
out: List[int] = []
seen: set[int] = set()
for part in str(raw).replace(";", ",").split(","):
token = str(part).strip()
if not token:
continue
try:
hour = int(token)
except Exception:
continue
if not (0 <= hour <= 23) or hour in seen:
continue
seen.add(hour)
out.append(hour)
return sorted(out) or default
def _cache_market_monitor_digest(
state: Dict[str, Any],
*,
message: str,
slot_label: str,
generated_at_ts: Optional[int] = None,
) -> None:
state["last_market_monitor_digest"] = {
"message": str(message or "").strip(),
"slot_label": str(slot_label or "").strip(),
"generated_at_ts": int(generated_at_ts or time.time()),
}
def load_cached_market_monitor_digest() -> str:
state = _load_state(_state_file())
cached = state.get("last_market_monitor_digest") or {}
if not isinstance(cached, dict):
return ""
return str(cached.get("message") or "").strip()
def _minute_of_day(text: Optional[str]) -> Optional[int]:
@@ -310,6 +312,17 @@ def _format_minutes_window(delta_minutes: int) -> str:
return text
def _format_interval_brief(seconds: int) -> str:
total = max(1, int(seconds))
if total % 3600 == 0:
hours = total // 3600
return f"{hours}小时"
if total % 60 == 0:
minutes = total // 60
return f"{minutes}分钟"
return f"{total}秒"
def _local_peak_context(alert_payload: Dict[str, Any]) -> Dict[str, Any]:
evidence = alert_payload.get("evidence") or {}
generated_local_time = str(evidence.get("generated_local_time") or "").strip()
@@ -437,6 +450,10 @@ def _build_focus_digest_message(
slot_label: str,
top_n: int,
) -> str:
scan_interval = _format_interval_brief(_env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 300))
digest_interval = _format_interval_brief(
_env_int("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", 1800),
)
ranked = sorted(
payloads,
key=lambda item: _market_monitor_score(item),
@@ -502,7 +519,11 @@ def _build_focus_digest_message(
lines.append(f" 链接：{market_url}")
lines.append("")
lines.append("用途：先筛今晚值得盯的市场，真正进入关键窗口时仍会继续推送。")
frequency_parts = [
f"后台扫描：约每{scan_interval}一次",
f"主动推送：约每{digest_interval}一次",
]
lines.append("更新频率：" + "；".join(frequency_parts))
return "\n".join(lines).strip()
@@ -512,32 +533,25 @@ def _maybe_send_focus_digest(
payloads: List[Dict[str, Any]],
state: Dict[str, Any],
*,
digest_hours: List[int],
digest_interval_sec: int,
top_n: int,
grace_minutes: int,
) -> bool:
if not chat_ids or not payloads or not digest_hours:
if not chat_ids or not payloads or digest_interval_sec <= 0:
return False
now_ts = int(time.time())
last_digest_ts = int(state.get("last_focus_digest_ts") or 0)
if last_digest_ts and now_ts - last_digest_ts < digest_interval_sec:
return False
local_now = datetime.now().astimezone()
eligible_slot: Optional[datetime] = None
for hour in sorted(digest_hours, reverse=True):
slot_dt = local_now.replace(hour=hour, minute=0, second=0, microsecond=0)
delta_minutes = int((local_now - slot_dt).total_seconds() // 60)
if 0 <= delta_minutes <= grace_minutes:
eligible_slot = slot_dt
break
if eligible_slot is None:
return False
slot_key = eligible_slot.strftime("%Y-%m-%d@%H")
digest_slots = state.setdefault("focus_digest_slots", {})
if digest_slots.get(slot_key):
return False
slot_label = "白天关注" if eligible_slot.hour < 15 else "今晚关注"
hour = local_now.hour
if 6 <= hour < 15:
slot_label = "白天关注"
elif 15 <= hour < 23:
slot_label = "今晚关注"
else:
slot_label = "夜间关注"
message = _build_focus_digest_message(
payloads,
slot_label=slot_label,
@@ -546,6 +560,12 @@ def _maybe_send_focus_digest(
if not message:
return False
_cache_market_monitor_digest(
state,
message=message,
slot_label=slot_label,
)
sent_count = 0
for chat_id in chat_ids:
try:
@@ -561,10 +581,10 @@ def _maybe_send_focus_digest(
if sent_count <= 0:
return False
digest_slots[slot_key] = int(time.time())
state["last_focus_digest_ts"] = now_ts
logger.info(
"market focus digest pushed slot={} items={} chat_targets={}",
slot_key,
"market focus digest pushed interval_sec={} items={} chat_targets={}",
digest_interval_sec,
min(top_n, len(payloads)),
sent_count,
)
@@ -598,6 +618,13 @@ def build_market_monitor_digest(
top_n=digest_top_n,
)
if message:
state = _load_state(_state_file())
_cache_market_monitor_digest(
state,
message=message,
slot_label=slot_label,
)
_save_state(_state_file(), state)
return message
return "ℹ️ 当前没有可用的市场监控摘要。"
@@ -965,14 +992,10 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th
cities = _parse_city_list(os.getenv("TELEGRAM_ALERT_CITIES"))
state_path = _state_file()
focus_digest_enabled = _env_bool("TELEGRAM_MARKET_FOCUS_DIGEST_ENABLED", True)
focus_digest_hours = _parse_hour_list(
os.getenv("TELEGRAM_MARKET_FOCUS_DIGEST_HOURS"),
[11, 18],
)
focus_digest_top_n = max(3, min(8, _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N", 5)))
focus_digest_grace_minutes = max(
30,
min(240, _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_GRACE_MINUTES", 180)),
focus_digest_interval_sec = max(
300,
_env_int("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", 1800),
)
def _runner() -> None:
try:
@@ -982,7 +1005,7 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th
logger.info(
f"telegram market monitor loop started mode=focus-digest-only "
f"cities={len(cities)} interval={interval_sec}s chat_targets={len(chat_ids)} "
f"focus_digest_enabled={focus_digest_enabled} focus_hours={focus_digest_hours} "
f"focus_digest_enabled={focus_digest_enabled} focus_interval={focus_digest_interval_sec}s "
f"state_path={state_path}"
)
while True:
@@ -1006,9 +1029,8 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th
chat_ids=chat_ids,
payloads=cycle_payloads,
state=state,
digest_hours=focus_digest_hours,
digest_interval_sec=focus_digest_interval_sec,
top_n=focus_digest_top_n,
grace_minutes=focus_digest_grace_minutes,
):
_save_state(state_path, state)
except Exception: