修复

.env.example

@@ -179,6 +179,11 @@ STREAM_MAX_TRADES_PER_WALLET=20
 # true → 每 30s 输出一次 poll 统计；false → 每 10 轮才输出一次
 STREAM_VERBOSE_LOGGING=false
 
+# 单个钱包 /activity 轮询超时（秒）
+# /activity 比 /positions 更重（返回完整交易历史），需要更长超时
+# 如果太短，活跃钱包会全部超时 → _last_seen 永远为空 → 无信号
+STREAM_POLL_TIMEOUT=15
+
 
 # ╔══════════════════════════════════════════════════════════════╗
 # ║               共识信号参数                                     ║

src/config/settings.py

@@ -54,6 +54,11 @@ class Settings(BaseSettings):
     stream_warmup_seconds: int = Field(default=30, alias="STREAM_WARMUP_SECONDS")
     stream_max_trades_per_wallet: int = Field(default=20, alias="STREAM_MAX_TRADES_PER_WALLET")
     stream_verbose_logging: bool = Field(default=False, alias="STREAM_VERBOSE_LOGGING")
+    # Per-wallet /activity poll timeout. /activity is heavier than /positions
+    # (returns full trade history), so it needs a longer timeout than the
+    # wallet-pool builder's 5s default. Without this, active wallets with
+    # many trades consistently time out → _last_seen never set → no signals.
+    stream_poll_timeout: float = Field(default=15.0, alias="STREAM_POLL_TIMEOUT")
 
     # Aggregator (Phase 2)
     consensus_window_seconds: int = Field(default=600, alias="CONSENSUS_WINDOW_SECONDS")
@@ -101,7 +106,12 @@ class Settings(BaseSettings):
     wallet_pool_backoff_emails: int = Field(default=30, alias="WALLET_POOL_BACKOFF_EMA")  # consecutive empty responses → sleep
 
     # Bayesian credibility updater concurrency
-    bayes_concurrency: int = Field(default=10, alias="BAYES_CONCURRENCY")
+    # Keep low — bayes fetches closed_positions for each wallet, which is
+    # API-heavy (up to N pages per wallet). Combined with the 5-thread
+    # parallel pagination inside get_closed_positions_since, the effective
+    # request burst = bayes_concurrency × 5. At 10 that's 50 simultaneous
+    # requests → 429 Too Many Requests from Polymarket API.
+    bayes_concurrency: int = Field(default=3, alias="BAYES_CONCURRENCY")
 
     # Telegram notifications
     telegram_enabled: bool = Field(default=False, alias="TELEGRAM_ENABLED")

src/main.py

@@ -356,9 +356,15 @@ _watcher: Optional[CopyTrader] = None
 
 
 def signal_handler(signum, frame):
+    """Set running=False so the asyncio loops exit cleanly.
+
+    Do NOT call sys.exit() here — it raises SystemExit inside the signal
+    handler, which fires during threading._shutdown and prints an ugly
+    traceback. Instead, let the event loop drain naturally so _teardown()
+    can flush state and close resources.
+    """
     if _watcher:
         _watcher.stop()
-    sys.exit(0)
 
 
 @app.command()

src/services/aggregator.py

@@ -55,6 +55,7 @@ from src.services.kelly import KellySizer
 logger = logging.getLogger(__name__)
 
 _DEBOUNCE_STATE_KEY = "wallet_debounce"
+_RECENTLY_EMITTED_KEY = "recently_emitted"
 _PERSIST_DEBOUNCE_SECONDS = 5.0  # throttle accumulator snapshots
 
 
@@ -180,8 +181,10 @@ class SignalAggregator:
         self.accumulators: Dict[str, MarketAccumulator] = {}
         self.wallet_credibility: Dict[str, float] = {}
         self.wallet_debounce: Dict[str, float] = {}
-        # Same-market signal cooldown: cid → timestamp of last emit.
-        # Prevents spamming signals on one market within cooldown window.
+        # Same-market signal cooldown: market_slug → timestamp of last emit.
+        # Keyed by market_slug (not condition_id) so that all markets within
+        # a multi-market event share the same cooldown. Persisted to DB so a
+        # restart doesn't let the same market re-fire.
         self.recently_emitted: Dict[str, float] = {}
         self._signals_emitted = 0
         self._dirty = False
@@ -192,6 +195,7 @@ class SignalAggregator:
         self._market_meta_cache: Dict[str, Tuple[str, str, float, bool]] = {}
         self._load_accumulators()
         self._load_debounce()
+        self._load_recently_emitted()
 
     def _load_accumulators(self) -> None:
         """Restore in-flight consensus accumulators from DB (survives restart)."""
@@ -271,6 +275,32 @@ class SignalAggregator:
         except Exception as e:
             logger.debug(f"[agg] persist debounce failed: {e}")
 
+    def _load_recently_emitted(self) -> None:
+        """Restore recently_emitted from DB; drop entries past cooldown."""
+        raw = self.db.get_pool_state_json(_RECENTLY_EMITTED_KEY)
+        if not raw or not isinstance(raw, dict):
+            return
+        now = time.time()
+        cooldown = self.settings.signal_cooldown_seconds
+        loaded = 0
+        for slug, ts_val in raw.items():
+            try:
+                ts = float(ts_val)
+            except (TypeError, ValueError):
+                continue
+            if now - ts < cooldown:
+                self.recently_emitted[slug] = ts
+                loaded += 1
+        if loaded:
+            logger.info(f"[agg] restored {loaded} signal cooldown entries from DB")
+
+    def _save_recently_emitted(self) -> None:
+        """Persist recently_emitted snapshot to DB."""
+        try:
+            self.db.set_pool_state_json(_RECENTLY_EMITTED_KEY, self.recently_emitted)
+        except Exception as e:
+            logger.debug(f"[agg] persist recently_emitted failed: {e}")
+
     def _mark_dirty(self) -> None:
         self._dirty = True
 
@@ -429,25 +459,28 @@ class SignalAggregator:
         if side not in ("BUY", "SELL"):
             return None
 
-        # P1-7: same-market cooldown. Once a signal has fired for this
-        # condition_id, ignore all further trades on it until cooldown
-        # expires. Prevents spamming signals on the same market.
-        last_emit = self.recently_emitted.get(cid, 0)
-        if time.time() - last_emit < self.settings.signal_cooldown_seconds:
-            logger.debug(
-                f"[agg] {cid[:10]} in signal cooldown "
-                f"({int(self.settings.signal_cooldown_seconds - (time.time() - last_emit))}s left), skipping"
-            )
-            return None
-
         # P2-10 + 附加: fetch market meta early. We need endDate for the
         # min-hours-to-resolution filter, and closed/active flags to skip
         # resolved markets. Doing this before accumulator creation ensures
-        # we never open an accumulator on a dead market.
+        # we never open an accumulator on a dead market. We also need
+        # market_slug for the cooldown key (see below).
         market_q, market_slug, end_ts, is_closed = await self._lazy_market_meta(cid)
         if is_closed:
             logger.debug(f"[agg] {cid[:10]} is closed/inactive, skipping")
             return None
+
+        # P1-7: same-market cooldown. Keyed by market_slug (not condition_id)
+        # so that all markets within a multi-market event share the same
+        # cooldown. Prevents spamming signals on the same event within the
+        # cooldown window. Persisted to DB so a restart doesn't re-fire.
+        cooldown_key = market_slug or cid
+        last_emit = self.recently_emitted.get(cooldown_key, 0)
+        if time.time() - last_emit < self.settings.signal_cooldown_seconds:
+            logger.debug(
+                f"[agg] {cooldown_key[:30]} in signal cooldown "
+                f"({int(self.settings.signal_cooldown_seconds - (time.time() - last_emit))}s left), skipping"
+            )
+            return None
         if end_ts > 0:
             hours_left = (end_ts - time.time()) / 3600.0
             if hours_left < self.settings.min_hours_to_resolution:
@@ -615,8 +648,12 @@ class SignalAggregator:
             "n_contributors": n,
         }
         self._signals_emitted += 1
-        # P1-7: record cooldown so we don't re-emit on this market.
-        self.recently_emitted[cid] = time.time()
+        # P1-7: record cooldown keyed by market_slug so we don't re-emit on
+        # this market (or any sibling condition_id within the same event).
+        # Persist immediately so a restart doesn't let it re-fire.
+        cooldown_key = acc.market_slug or cid
+        self.recently_emitted[cooldown_key] = time.time()
+        await asyncio.to_thread(self._save_recently_emitted)
         # P1-6: invalidate risk caches so the next can_emit() sees the new
         # open-position count.
         if self.risk is not None:

src/services/bayesian.py

@@ -55,10 +55,9 @@ class BayesianUpdater:
                 try:
                     # get_closed_positions_since already filters by cutoff_ts,
                     # so all returned items are within the decay window.
-                    # Use 3x the pool-building max_pages since bayesian only
-                    # processes ~100 pool wallets and PnL accuracy directly
-                    # affects credibility scores.
-                    bayes_max_pages = self.settings.closed_positions_max_pages * 3
+                    # Use the configured max_pages (default 20). Previously
+                    # used 3x which caused massive API bursts → 429 errors.
+                    bayes_max_pages = self.settings.closed_positions_max_pages
                     closed = await asyncio.to_thread(
                         self.data.get_closed_positions_since, addr, cutoff,
                         bayes_max_pages,

src/services/data_api.py

@@ -75,23 +75,45 @@ class DataAPIClient:
           - sortBy default REALIZEDPNL → returns most-profitable first, which
             hides recent-but-small positions. Default to TIMESTAMP so callers
             can filter by time correctly.
+
+        Retries on 429 Too Many Requests with exponential backoff.
         """
+        import time as _time
+
         kwargs: Dict[str, Any] = {}
         if timeout is not None:
             kwargs["timeout"] = timeout
-        r = self._client.get(
-            f"{DATA_BASE}/closed-positions",
-            params={
-                "user": address,
-                "limit": max(0, min(limit, 50)),
-                "offset": max(0, offset),
-                "sortBy": sort_by,
-                "sortDirection": sort_direction,
-            },
-            **kwargs,
-        )
-        r.raise_for_status()
-        return r.json()
+
+        max_retries = 3
+        for attempt in range(max_retries):
+            r = self._client.get(
+                f"{DATA_BASE}/closed-positions",
+                params={
+                    "user": address,
+                    "limit": max(0, min(limit, 50)),
+                    "offset": max(0, offset),
+                    "sortBy": sort_by,
+                    "sortDirection": sort_direction,
+                },
+                **kwargs,
+            )
+            if r.status_code == 429:
+                if attempt < max_retries - 1:
+                    wait = 2 ** attempt  # 1s, 2s, 4s
+                    logger.debug(
+                        f"[data_api] 429 for {address[:10]} offset={offset}, "
+                        f"retry {attempt+1}/{max_retries} after {wait}s"
+                    )
+                    _time.sleep(wait)
+                    continue
+                else:
+                    logger.warning(
+                        f"[data_api] 429 exhausted retries for {address[:10]} "
+                        f"offset={offset}"
+                    )
+            r.raise_for_status()
+            return r.json()
+        return []
 
     def get_closed_positions_since(
         self,

src/services/telegram.py

@@ -22,19 +22,42 @@ class TelegramNotifier:
         if not self.settings.telegram_bot_token or not self.settings.telegram_chat_id:
             logger.warning("Telegram credentials missing; notifier disabled")
             return
-        try:
-            from telegram import Bot
-            from telegram.request import HTTPXRequest
-            proxy = self.settings.http_proxy
-            kwargs = {}
-            if proxy:
-                kwargs["request"] = HTTPXRequest(proxy=proxy)
-            self._bot = Bot(token=self.settings.telegram_bot_token, **kwargs)
-            me = await self._bot.get_me()
-            logger.info(f"Telegram bot started: @{me.username} (proxy={proxy})")
-        except Exception as e:
-            logger.error(f"Failed to start Telegram bot: {e}")
-            self._bot = None
+
+        from telegram import Bot
+        from telegram.request import HTTPXRequest
+
+        proxy = self.settings.http_proxy
+
+        # Try with proxy first, then fall back to direct connection.
+        # Each attempt has explicit connect/read timeouts so we don't hang
+        # the whole startup if the proxy is down.
+        attempts = []
+        if proxy:
+            attempts.append(("proxy", HTTPXRequest(
+                proxy=proxy, connect_timeout=10, read_timeout=15,
+            )))
+        attempts.append(("direct", HTTPXRequest(
+            connect_timeout=10, read_timeout=15,
+        )))
+
+        for label, request in attempts:
+            try:
+                self._bot = Bot(
+                    token=self.settings.telegram_bot_token,
+                    request=request,
+                )
+                me = await asyncio.wait_for(self._bot.get_me(), timeout=20)
+                logger.info(
+                    f"Telegram bot started: @{me.username} (mode={label})"
+                )
+                return
+            except Exception as e:
+                logger.warning(
+                    f"Telegram start via {label} failed: {type(e).__name__}: {e}"
+                )
+                self._bot = None
+
+        logger.error("Failed to start Telegram bot after all attempts")
 
     async def send_signal(self, market: str, side: str, strength: float,
                          size_usd: float, n_wallets: int,
@@ -45,12 +68,20 @@ class TelegramNotifier:
             return
         ref = market_slug or condition_id
         link = f"https://polymarket.com/market/{ref}" if ref else ""
+
+        # Build a clear action description.
+        # For binary markets outcome is "Yes"/"No"; for multi-outcome markets
+        # it's the actual option name (e.g. "Portland Trail Blazers").
+        # "BUY Yes" → "买入 Yes"; "BUY No" → "买入 No";
+        # "BUY <option>" → "买入 <option>"
+        action = f"买入 {outcome}" if side == "BUY" else f"卖出 {outcome}"
+
         text = (
             f"🎯 *跟单信号*\n"
             f"市场：{market[:80]}\n"
             f"{'链接：' + link + '\n' if link else ''}"
-            f"方向：{side} {outcome}\n"
-            f"入场价：${entry_price:.4f}\n"
+            f"操作：{action}\n"
+            f"入场价：{entry_price:.4f} ({entry_price*100:.1f}%)\n"
             f"信号强度：{strength:.2f}\n"
             f"建议仓位：${size_usd:,.0f}\n"
             f"来源钱包：{n_wallets}\n"

src/services/user_stream.py

@@ -102,8 +102,7 @@ class UserTradeStream:
 
                 # Parallel wallet polling (semaphore limits concurrency)
                 sem = asyncio.Semaphore(self.settings.wallet_pool_concurrency)
-                poll_sem = self.settings.wallet_pool_concurrency
-                poll_timeout = self.settings.wallet_pool_request_timeout
+                poll_timeout = self.settings.stream_poll_timeout
                 now_ts = time.time()
                 address_list = [
                     a for a in addresses
@@ -113,6 +112,13 @@ class UserTradeStream:
                 if skipped and self._poll_count % 10 == 0:
                     logger.debug(f"[stream] skipped {skipped} wallets in backoff")
 
+                # Log first poll specifically so we can diagnose empty results
+                if self._poll_count == 1:
+                    logger.info(
+                        f"[stream] first poll: {len(address_list)} wallets, "
+                        f"timeout={poll_timeout}s, warmup={'yes' if in_warmup else 'no'}"
+                    )
+
                 async def poll_one(addr: str) -> None:
                     async with sem:
                         try:
@@ -137,17 +143,23 @@ class UserTradeStream:
                                     f"({n_fail}x, backoff {backoff}s): {type(e).__name__}"
                                 )
 
-                await asyncio.gather(*[poll_one(addr) for addr in addresses])
+                await asyncio.gather(*[poll_one(addr) for addr in address_list])
 
-                if self._poll_count % 10 == 0 or self.settings.stream_verbose_logging:
+                # Track how many wallets now have a baseline (non-None _last_seen)
+                tracked = sum(1 for v in self._last_seen.values() if v is not None)
+
+                if self._poll_count % 10 == 0 or self.settings.stream_verbose_logging or self._poll_count == 1:
                     logger.info(
                         f"[stream] poll #{self._poll_count} complete "
-                        f"({len(addresses)} wallets, {self._trades_emitted} total trades emitted)"
+                        f"({len(address_list)} polled, {tracked} tracked, "
+                        f"{self._trades_emitted} total trades emitted)"
                     )
 
-                # Persist last_seen every ~5 min (each poll is ~30s).
+                # Persist last_seen every ~2 min (each poll is ~30s).
+                # Frequent enough to survive restarts without losing much
+                # progress, but not so frequent as to hammer the DB.
                 now = time.time()
-                if now - last_save > 300:
+                if now - last_save > 120:
                     self._save_last_seen()
                     last_save = now