Fix duplicate-index crash in data validation, add Telegram flood control
- Move index dedup before OHLC consistency fix in _validate_and_clean to prevent crash when df.loc[dup_idx] returns Series instead of scalar - Add global flood backoff: detect RetryAfter, pause all sends/edits until ban expires instead of retrying and escalating the ban - Stop dashboard edit-fail → new-message spam cascade: track consecutive failures, only resend after 3 failed edits - Increase dashboard refresh from 5s to 30s, min send interval to 1.5s - Add comprehensive QA test suite (140 tests across all 7 modules) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
+47
-7
@@ -131,15 +131,32 @@ AUTOSCAN_CONFIRMATIONS = {
|
||||
|
||||
_executor = ThreadPoolExecutor(max_workers=4)
|
||||
|
||||
# ── Telegram rate limiter (1 msg/sec per chat) ────────────────────────
|
||||
# ── Telegram rate limiter with flood control ─────────────────────────
|
||||
|
||||
_send_lock = asyncio.Lock()
|
||||
_last_send_per_chat: Dict[int, float] = {}
|
||||
_TELEGRAM_MIN_INTERVAL = 1.0
|
||||
_TELEGRAM_MIN_INTERVAL = 1.5
|
||||
_TELEGRAM_SEND_TIMEOUT = 30
|
||||
_flood_until: float = 0.0
|
||||
|
||||
|
||||
def _is_flood_blocked() -> bool:
|
||||
return time.time() < _flood_until
|
||||
|
||||
|
||||
def _handle_retry_after(retry_seconds: int) -> None:
|
||||
global _flood_until
|
||||
_flood_until = time.time() + retry_seconds
|
||||
logger.warning(
|
||||
f"Telegram flood control: pausing ALL sends for {retry_seconds}s "
|
||||
f"(until {datetime.fromtimestamp(_flood_until, tz=timezone.utc).strftime('%H:%M:%S UTC')})"
|
||||
)
|
||||
|
||||
|
||||
async def _rate_limited_send(bot, chat_id: int, text: str, **kwargs) -> Optional[object]:
|
||||
if _is_flood_blocked():
|
||||
logger.debug(f"Send skipped — flood backoff active ({_flood_until - time.time():.0f}s left)")
|
||||
return None
|
||||
async with _send_lock:
|
||||
now = time.time()
|
||||
last = _last_send_per_chat.get(chat_id, 0)
|
||||
@@ -156,12 +173,17 @@ async def _rate_limited_send(bot, chat_id: int, text: str, **kwargs) -> Optional
|
||||
except asyncio.TimeoutError:
|
||||
logger.error(f"Telegram send_message timed out for chat {chat_id}")
|
||||
return None
|
||||
except Exception:
|
||||
logger.exception(f"Telegram send_message failed for chat {chat_id}")
|
||||
except Exception as exc:
|
||||
if "retry" in str(exc).lower() and hasattr(exc, "retry_after"):
|
||||
_handle_retry_after(int(exc.retry_after))
|
||||
else:
|
||||
logger.exception(f"Telegram send_message failed for chat {chat_id}")
|
||||
return None
|
||||
|
||||
|
||||
async def _rate_limited_edit(bot, text: str, chat_id: int, message_id: int, **kwargs) -> bool:
|
||||
if _is_flood_blocked():
|
||||
return False
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
bot.edit_message_text(text, chat_id=chat_id, message_id=message_id, **kwargs),
|
||||
@@ -172,6 +194,9 @@ async def _rate_limited_edit(bot, text: str, chat_id: int, message_id: int, **kw
|
||||
logger.error(f"Telegram edit_message timed out for chat {chat_id}")
|
||||
return False
|
||||
except Exception as exc:
|
||||
if "retry" in str(exc).lower() and hasattr(exc, "retry_after"):
|
||||
_handle_retry_after(int(exc.retry_after))
|
||||
return False
|
||||
if "message is not modified" not in str(exc).lower():
|
||||
logger.warning(f"Telegram edit_message failed for chat {chat_id}: {exc}")
|
||||
return False
|
||||
@@ -445,19 +470,27 @@ def _format_stats_dashboard() -> str:
|
||||
_dashboard_msg_ids: Dict[int, int] = {}
|
||||
_dashboard_task: Optional[asyncio.Task] = None
|
||||
_last_dashboard_refresh: float = 0
|
||||
_DASHBOARD_THROTTLE = 5
|
||||
_DASHBOARD_THROTTLE = 30
|
||||
_DASHBOARD_LOOP_INTERVAL = 30
|
||||
_dashboard_lock = asyncio.Lock()
|
||||
_dashboard_edit_failures: Dict[int, int] = {}
|
||||
_DASHBOARD_MAX_EDIT_FAILURES = 3
|
||||
|
||||
|
||||
async def _send_dashboard(bot, chat_id: int) -> None:
|
||||
if _is_flood_blocked():
|
||||
return
|
||||
text = _format_stats_dashboard()
|
||||
msg = await _rate_limited_send(bot, chat_id, text, parse_mode=ParseMode.MARKDOWN)
|
||||
if msg:
|
||||
_dashboard_msg_ids[chat_id] = msg.message_id
|
||||
_dashboard_edit_failures[chat_id] = 0
|
||||
|
||||
|
||||
async def _refresh_dashboard(bot, force: bool = False) -> None:
|
||||
global _last_dashboard_refresh
|
||||
if _is_flood_blocked():
|
||||
return
|
||||
async with _dashboard_lock:
|
||||
now = time.time()
|
||||
if not force and (now - _last_dashboard_refresh) < _DASHBOARD_THROTTLE:
|
||||
@@ -475,13 +508,20 @@ async def _refresh_dashboard(bot, force: bool = False) -> None:
|
||||
parse_mode=ParseMode.MARKDOWN,
|
||||
)
|
||||
if not ok:
|
||||
await _send_dashboard(bot, chat_id)
|
||||
failures = _dashboard_edit_failures.get(chat_id, 0) + 1
|
||||
_dashboard_edit_failures[chat_id] = failures
|
||||
if failures >= _DASHBOARD_MAX_EDIT_FAILURES:
|
||||
logger.info(f"Dashboard edit failed {failures}x for chat {chat_id}, sending new message")
|
||||
_dashboard_edit_failures[chat_id] = 0
|
||||
await _send_dashboard(bot, chat_id)
|
||||
else:
|
||||
_dashboard_edit_failures[chat_id] = 0
|
||||
|
||||
|
||||
async def _dashboard_loop(bot) -> None:
|
||||
try:
|
||||
while _autoscan_chats:
|
||||
await asyncio.sleep(5)
|
||||
await asyncio.sleep(_DASHBOARD_LOOP_INTERVAL)
|
||||
if not _autoscan_chats:
|
||||
break
|
||||
await _refresh_dashboard(bot)
|
||||
|
||||
Reference in New Issue
Block a user