Fix news engine (keywords, calendar fallback, RSS URL), remove Telegram guards

News engine: expand sentiment keywords (18%→60% headline coverage), add
MyFXBook RSS as fallback calendar source when Forex Factory returns 429,
add retry/backoff on calendar API rate limits, update stale ForexLive URL
to investinglive.com, add date formats without seconds for MyFXBook RSS.

Telegram bot: remove flood control, rate limiter, and dashboard failure
tracking — all sends go directly through bot.send_message/edit_message_text.
This commit is contained in:
harwinder
2026-05-14 13:42:32 -04:00
parent 39c91c1aa4
commit 31ecdf4292
3 changed files with 167 additions and 124 deletions
+1 -1
View File
@@ -123,7 +123,7 @@ class NewsConfig:
post_event_buffer_minutes: int = 15 post_event_buffer_minutes: int = 15
rss_feeds: List[str] = field(default_factory=lambda: [ rss_feeds: List[str] = field(default_factory=lambda: [
"https://www.forexlive.com/feed/", "https://investinglive.com/feed/",
"https://www.fxstreet.com/rss", "https://www.fxstreet.com/rss",
]) ])
+145 -23
View File
@@ -9,6 +9,8 @@ A perfect technical setup can be destroyed by an unexpected NFP print.
""" """
import logging import logging
import re
import time
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
@@ -81,17 +83,26 @@ class NewsEngine:
# Keywords that indicate directional sentiment # Keywords that indicate directional sentiment
BULLISH_KEYWORDS = { BULLISH_KEYWORDS = {
"rate hike", "hawkish", "strong growth", "beat expectations", "rate hike", "hawkish", "strong growth", "beat expectations",
"above forecast", "surplus", "expansion", "employment gains", "above forecast", "above estimate", "above expectation",
"surplus", "expansion", "employment gains",
"strong jobs", "gdp growth", "inflation rising", "tightening", "strong jobs", "gdp growth", "inflation rising", "tightening",
"positive surprise", "record high", "bullish", "rally", "positive surprise", "record high", "bullish", "rally",
"stronger than expected", "upbeat", "stronger than expected", "upbeat",
"beats", "surges", "soars", "jumps", "climbs", "rises",
"higher than expected", "better than expected",
"strong data", "robust", "resilient", "accelerates",
} }
BEARISH_KEYWORDS = { BEARISH_KEYWORDS = {
"rate cut", "dovish", "recession", "slowdown", "contraction", "rate cut", "dovish", "recession", "slowdown", "contraction",
"below forecast", "deficit", "miss expectations", "weak data", "below forecast", "below estimate", "below expectation",
"deficit", "miss expectations", "weak data",
"unemployment rise", "negative surprise", "bearish", "selloff", "unemployment rise", "negative surprise", "bearish", "selloff",
"weaker than expected", "downbeat", "easing", "stimulus", "weaker than expected", "downbeat", "easing", "stimulus",
"misses", "slides", "drops", "falls", "plunges", "tumbles",
"weakens", "slumps", "declines", "retreats",
"lower than expected", "worse than expected",
"disappointing", "soft data",
} }
# Currency-specific keywords for headline parsing # Currency-specific keywords for headline parsing
@@ -115,6 +126,8 @@ class NewsEngine:
"boe rate", "rba rate", "snb rate", "boc rate", "boe rate", "rba rate", "snb rate", "boc rate",
} }
_calendar_backoff_until: float = 0.0
def __init__(self, config: NewsConfig): def __init__(self, config: NewsConfig):
self.config = config self.config = config
self._session = requests.Session() self._session = requests.Session()
@@ -252,18 +265,27 @@ class NewsEngine:
# ── Private Methods ──────────────────────────────────────────── # ── Private Methods ────────────────────────────────────────────
# Maps country names in MyFXBook titles to currency codes
_COUNTRY_TO_CURRENCY = {
"united states": "USD", "euro area": "EUR", "united kingdom": "GBP",
"japan": "JPY", "switzerland": "CHF", "australia": "AUD",
"canada": "CAD", "new zealand": "NZD",
}
def _fetch_economic_calendar(self, currencies: List[str]) -> List[EconomicEvent]: def _fetch_economic_calendar(self, currencies: List[str]) -> List[EconomicEvent]:
""" """
Fetch economic calendar events from a free API. Fetch economic calendar events from a free API.
Tries Forex Factory first, falls back to MyFXBook RSS.
Returns an empty list (with a log warning) if unavailable — Returns an empty list (with a log warning) if unavailable —
we never generate fake events since that causes constant false alarms. we never generate fake events since that causes constant false alarms.
""" """
try: for fetcher in (self._try_fetch_calendar_api, self._try_fetch_myfxbook_rss):
events = self._try_fetch_calendar_api() try:
if events: events = fetcher()
return [e for e in events if e.currency in currencies] if events:
except Exception as e: return [e for e in events if e.currency in currencies]
logger.debug(f"Calendar API fetch exception: {e}") except Exception as e:
logger.debug(f"Calendar fetch exception ({fetcher.__name__}): {e}")
logger.info( logger.info(
"Economic calendar data unavailable. " "Economic calendar data unavailable. "
@@ -273,36 +295,134 @@ class NewsEngine:
def _try_fetch_calendar_api(self) -> List[EconomicEvent]: def _try_fetch_calendar_api(self) -> List[EconomicEvent]:
"""Attempt to fetch from the Forex Factory free JSON calendar.""" """Attempt to fetch from the Forex Factory free JSON calendar."""
events = [] now = time.time()
if now < NewsEngine._calendar_backoff_until:
wait = NewsEngine._calendar_backoff_until - now
logger.debug(f"Calendar API in backoff for {wait:.0f}s more")
return []
url = "https://nfs.faireconomy.media/ff_calendar_thisweek.json" url = "https://nfs.faireconomy.media/ff_calendar_thisweek.json"
resp = self._session.get(url, timeout=10)
for attempt in range(3):
try:
resp = self._session.get(url, timeout=10)
except requests.RequestException as e:
logger.debug(f"Calendar request failed (attempt {attempt + 1}): {e}")
if attempt < 2:
time.sleep(2 ** attempt)
continue
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 300))
retry_after = min(retry_after, 600)
NewsEngine._calendar_backoff_until = time.time() + retry_after
logger.warning(
f"Calendar API rate-limited (429), backing off {retry_after}s"
)
return []
if resp.status_code != 200:
logger.debug(f"Calendar API returned {resp.status_code}")
return []
data = resp.json()
events = []
impact_map = {"High": "high", "Medium": "medium", "Low": "low"}
for item in data:
try:
impact = impact_map.get(item.get("impact", ""), "low")
date_str = item.get("date", "")
if not date_str:
continue
event_time = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
events.append(EconomicEvent(
time=event_time,
currency=item.get("country", ""),
impact=impact,
event=item.get("title", ""),
forecast=item.get("forecast") or None,
previous=item.get("previous") or None,
actual=item.get("actual") or None,
))
except Exception:
continue
return events
return []
def _try_fetch_myfxbook_rss(self) -> List[EconomicEvent]:
"""Fallback calendar source: MyFXBook economic calendar RSS."""
url = "https://www.myfxbook.com/rss/forex-economic-calendar-events"
try:
resp = self._session.get(url, timeout=10)
except requests.RequestException as e:
logger.debug(f"MyFXBook calendar request failed: {e}")
return []
if resp.status_code != 200: if resp.status_code != 200:
return [] return []
data = resp.json() try:
impact_map = {"High": "high", "Medium": "medium", "Low": "low"} root = ET.fromstring(resp.content)
except ET.ParseError:
return []
for item in data: events = []
for item in root.iter("item"):
try: try:
impact = impact_map.get(item.get("impact", ""), "low") title_el = item.find("title")
pub_el = item.find("pubDate")
date_str = item.get("date", "") desc_el = item.find("description")
if not date_str: if title_el is None or title_el.text is None:
continue continue
event_time = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
title = title_el.text.strip()
currency = ""
title_lower = title.lower()
for country, cur in self._COUNTRY_TO_CURRENCY.items():
if title_lower.startswith(country):
currency = cur
title = title[len(country):].strip()
break
if not currency:
continue
event_time = self._parse_rss_date(
pub_el.text if pub_el is not None else None
)
if event_time is None:
continue
impact = "low"
desc_text = desc_el.text if desc_el is not None and desc_el.text else ""
impact_match = re.search(r"sprite-(high|medium|low)-impact", desc_text)
if impact_match:
impact = impact_match.group(1)
cells = re.findall(r"<td[^>]*>(.*?)</td>", desc_text, re.DOTALL)
cells = [re.sub(r"<[^>]+>", "", c).strip() for c in cells]
previous = cells[2] if len(cells) > 2 and cells[2] else None
forecast = cells[3] if len(cells) > 3 and cells[3] else None
actual = cells[4] if len(cells) > 4 and cells[4] else None
events.append(EconomicEvent( events.append(EconomicEvent(
time=event_time, time=event_time,
currency=item.get("country", ""), currency=currency,
impact=impact, impact=impact,
event=item.get("title", ""), event=title,
forecast=item.get("forecast") or None, forecast=forecast,
previous=item.get("previous") or None, previous=previous,
actual=item.get("actual") or None, actual=actual,
)) ))
except Exception: except Exception:
continue continue
logger.debug(f"MyFXBook calendar: parsed {len(events)} events")
return events return events
def _fetch_news_sentiment(self) -> List[NewsSentiment]: def _fetch_news_sentiment(self) -> List[NewsSentiment]:
@@ -401,6 +521,8 @@ class NewsEngine:
formats = [ formats = [
"%a, %d %b %Y %H:%M:%S %z", # RFC 822 with numeric tz offset "%a, %d %b %Y %H:%M:%S %z", # RFC 822 with numeric tz offset
"%a, %d %b %Y %H:%M:%S %Z", # RFC 822 with tz abbreviation (GMT/UTC) "%a, %d %b %Y %H:%M:%S %Z", # RFC 822 with tz abbreviation (GMT/UTC)
"%a, %d %b %Y %H:%M %Z", # RFC 822 without seconds (MyFXBook)
"%a, %d %b %Y %H:%M %z", # RFC 822 without seconds, numeric offset
"%Y-%m-%dT%H:%M:%S%z", # ISO 8601 with offset "%Y-%m-%dT%H:%M:%S%z", # ISO 8601 with offset
"%Y-%m-%d %H:%M:%S", # ISO 8601 naive (assume UTC) "%Y-%m-%d %H:%M:%S", # ISO 8601 naive (assume UTC)
] ]
+21 -100
View File
@@ -131,77 +131,6 @@ AUTOSCAN_CONFIRMATIONS = {
_executor = ThreadPoolExecutor(max_workers=4) _executor = ThreadPoolExecutor(max_workers=4)
# ── Telegram rate limiter with flood control ─────────────────────────
_send_lock = asyncio.Lock()
_last_send_per_chat: Dict[int, float] = {}
_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)
wait = _TELEGRAM_MIN_INTERVAL - (now - last)
if wait > 0:
await asyncio.sleep(wait)
try:
result = await asyncio.wait_for(
bot.send_message(chat_id, text, **kwargs),
timeout=_TELEGRAM_SEND_TIMEOUT,
)
_last_send_per_chat[chat_id] = time.time()
return result
except asyncio.TimeoutError:
logger.error(f"Telegram send_message timed out for chat {chat_id}")
return None
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),
timeout=_TELEGRAM_SEND_TIMEOUT,
)
return True
except asyncio.TimeoutError:
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
# ── Markdown escaping ───────────────────────────────────────────────── # ── Markdown escaping ─────────────────────────────────────────────────
def _escape_md(text: str) -> str: def _escape_md(text: str) -> str:
@@ -470,30 +399,24 @@ def _format_stats_dashboard() -> str:
_dashboard_msg_ids: Dict[int, int] = {} _dashboard_msg_ids: Dict[int, int] = {}
_dashboard_task: Optional[asyncio.Task] = None _dashboard_task: Optional[asyncio.Task] = None
_last_dashboard_refresh: float = 0 _last_dashboard_refresh: float = 0
_DASHBOARD_THROTTLE = 30 _DASHBOARD_REFRESH_INTERVAL = 30
_DASHBOARD_LOOP_INTERVAL = 30
_dashboard_lock = asyncio.Lock() _dashboard_lock = asyncio.Lock()
_dashboard_edit_failures: Dict[int, int] = {}
_DASHBOARD_MAX_EDIT_FAILURES = 3
async def _send_dashboard(bot, chat_id: int) -> None: async def _send_dashboard(bot, chat_id: int) -> None:
if _is_flood_blocked():
return
text = _format_stats_dashboard() text = _format_stats_dashboard()
msg = await _rate_limited_send(bot, chat_id, text, parse_mode=ParseMode.MARKDOWN) try:
if msg: msg = await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
_dashboard_msg_ids[chat_id] = msg.message_id _dashboard_msg_ids[chat_id] = msg.message_id
_dashboard_edit_failures[chat_id] = 0 except Exception as exc:
logger.warning(f"Dashboard send failed for chat {chat_id}: {exc}")
async def _refresh_dashboard(bot, force: bool = False) -> None: async def _refresh_dashboard(bot, force: bool = False) -> None:
global _last_dashboard_refresh global _last_dashboard_refresh
if _is_flood_blocked():
return
async with _dashboard_lock: async with _dashboard_lock:
now = time.time() now = time.time()
if not force and (now - _last_dashboard_refresh) < _DASHBOARD_THROTTLE: if not force and (now - _last_dashboard_refresh) < _DASHBOARD_REFRESH_INTERVAL:
return return
_last_dashboard_refresh = now _last_dashboard_refresh = now
@@ -503,25 +426,20 @@ async def _refresh_dashboard(bot, force: bool = False) -> None:
if not msg_id: if not msg_id:
await _send_dashboard(bot, chat_id) await _send_dashboard(bot, chat_id)
continue continue
ok = await _rate_limited_edit( try:
bot, text, chat_id=chat_id, message_id=msg_id, await bot.edit_message_text(
parse_mode=ParseMode.MARKDOWN, text, chat_id=chat_id, message_id=msg_id,
) parse_mode=ParseMode.MARKDOWN,
if not ok: )
failures = _dashboard_edit_failures.get(chat_id, 0) + 1 except Exception as exc:
_dashboard_edit_failures[chat_id] = failures if "message is not modified" not in str(exc).lower():
if failures >= _DASHBOARD_MAX_EDIT_FAILURES: logger.warning(f"Dashboard edit failed for chat {chat_id}: {exc}")
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: async def _dashboard_loop(bot) -> None:
try: try:
while _autoscan_chats: while _autoscan_chats:
await asyncio.sleep(_DASHBOARD_LOOP_INTERVAL) await asyncio.sleep(_DASHBOARD_REFRESH_INTERVAL)
if not _autoscan_chats: if not _autoscan_chats:
break break
await _refresh_dashboard(bot) await _refresh_dashboard(bot)
@@ -623,7 +541,10 @@ async def _scan_timeframe(
for chat_id in list(_autoscan_chats): for chat_id in list(_autoscan_chats):
for report in actionable: for report in actionable:
detail = format_signal_message(report) detail = format_signal_message(report)
await _rate_limited_send(bot, chat_id, detail, parse_mode=ParseMode.MARKDOWN) try:
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
except Exception as exc:
logger.warning(f"Failed to send signal for chat {chat_id}: {exc}")
# ── WebSocket level-breach monitor ────────────────────────────────── # ── WebSocket level-breach monitor ──────────────────────────────────
@@ -1065,8 +986,8 @@ async def _post_init(app: Application) -> None:
) )
try: try:
await _rate_limited_send( await app.bot.send_message(
app.bot, chat_id, chat_id,
f"🤖 *ForexBot2 started*\n\n" f"🤖 *ForexBot2 started*\n\n"
f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n" f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n"
f"📊 Live dashboard refreshes every 60s.\n" f"📊 Live dashboard refreshes every 60s.\n"