harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
harwinder
2026-05-14 13:42:32 -04:00
parent bf9609708c
commit 57769967d1
3 changed files with 167 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.py
+1 -1
View File
@@ -123,7 +123,7 @@ class NewsConfig:
post_event_buffer_minutes: int = 15
rss_feeds: List[str] = field(default_factory=lambda: [
"https://www.forexlive.com/feed/",
"https://investinglive.com/feed/",
"https://www.fxstreet.com/rss",
])
news_engine.py
+145 -23
View File
@@ -9,6 +9,8 @@ A perfect technical setup can be destroyed by an unexpected NFP print.
"""
import logging
import re
import time
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
@@ -81,17 +83,26 @@ class NewsEngine:
# Keywords that indicate directional sentiment
BULLISH_KEYWORDS = {
"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",
"positive surprise", "record high", "bullish", "rally",
"stronger than expected", "upbeat",
"beats", "surges", "soars", "jumps", "climbs", "rises",
"higher than expected", "better than expected",
"strong data", "robust", "resilient", "accelerates",
}
BEARISH_KEYWORDS = {
"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",
"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
@@ -115,6 +126,8 @@ class NewsEngine:
"boe rate", "rba rate", "snb rate", "boc rate",
}
_calendar_backoff_until: float = 0.0
def __init__(self, config: NewsConfig):
self.config = config
self._session = requests.Session()
@@ -252,18 +265,27 @@ class NewsEngine:
# ── 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]:
"""
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 —
we never generate fake events since that causes constant false alarms.
"""
try:
events = self._try_fetch_calendar_api()
if events:
return [e for e in events if e.currency in currencies]
except Exception as e:
logger.debug(f"Calendar API fetch exception: {e}")
for fetcher in (self._try_fetch_calendar_api, self._try_fetch_myfxbook_rss):
try:
events = fetcher()
if events:
return [e for e in events if e.currency in currencies]
except Exception as e:
logger.debug(f"Calendar fetch exception ({fetcher.__name__}): {e}")
logger.info(
"Economic calendar data unavailable. "
@@ -273,36 +295,134 @@ class NewsEngine:
def _try_fetch_calendar_api(self) -> List[EconomicEvent]:
"""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"
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:
return []
data = resp.json()
impact_map = {"High": "high", "Medium": "medium", "Low": "low"}
try:
root = ET.fromstring(resp.content)
except ET.ParseError:
return []
for item in data:
events = []
for item in root.iter("item"):
try:
impact = impact_map.get(item.get("impact", ""), "low")
date_str = item.get("date", "")
if not date_str:
title_el = item.find("title")
pub_el = item.find("pubDate")
desc_el = item.find("description")
if title_el is None or title_el.text is None:
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(
time=event_time,
currency=item.get("country", ""),
currency=currency,
impact=impact,
event=item.get("title", ""),
forecast=item.get("forecast") or None,
previous=item.get("previous") or None,
actual=item.get("actual") or None,
event=title,
forecast=forecast,
previous=previous,
actual=actual,
))
except Exception:
continue
logger.debug(f"MyFXBook calendar: parsed {len(events)} events")
return events
def _fetch_news_sentiment(self) -> List[NewsSentiment]:
@@ -401,6 +521,8 @@ class NewsEngine:
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 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-%d %H:%M:%S", # ISO 8601 naive (assume UTC)
]
telegram_bot.py
+21 -100
View File
@@ -131,77 +131,6 @@ AUTOSCAN_CONFIRMATIONS = {
_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 ─────────────────────────────────────────────────
def _escape_md(text: str) -> str:
@@ -470,30 +399,24 @@ def _format_stats_dashboard() -> str:
_dashboard_msg_ids: Dict[int, int] = {}
_dashboard_task: Optional[asyncio.Task] = None
_last_dashboard_refresh: float = 0
_DASHBOARD_THROTTLE = 30
_DASHBOARD_LOOP_INTERVAL = 30
_DASHBOARD_REFRESH_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:
try:
msg = await bot.send_message(chat_id, text, parse_mode=ParseMode.MARKDOWN)
_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:
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:
if not force and (now - _last_dashboard_refresh) < _DASHBOARD_REFRESH_INTERVAL:
return
_last_dashboard_refresh = now
@@ -503,25 +426,20 @@ async def _refresh_dashboard(bot, force: bool = False) -> None:
if not msg_id:
await _send_dashboard(bot, chat_id)
continue
ok = await _rate_limited_edit(
bot, 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
_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
try:
await bot.edit_message_text(
text, chat_id=chat_id, message_id=msg_id,
parse_mode=ParseMode.MARKDOWN,
)
except Exception as exc:
if "message is not modified" not in str(exc).lower():
logger.warning(f"Dashboard edit failed for chat {chat_id}: {exc}")
async def _dashboard_loop(bot) -> None:
try:
while _autoscan_chats:
await asyncio.sleep(_DASHBOARD_LOOP_INTERVAL)
await asyncio.sleep(_DASHBOARD_REFRESH_INTERVAL)
if not _autoscan_chats:
break
await _refresh_dashboard(bot)
@@ -623,7 +541,10 @@ async def _scan_timeframe(
for chat_id in list(_autoscan_chats):
for report in actionable:
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 ──────────────────────────────────
@@ -1065,8 +986,8 @@ async def _post_init(app: Application) -> None:
)
try:
await _rate_limited_send(
app.bot, chat_id,
await app.bot.send_message(
chat_id,
f"🤖 *ForexBot2 started*\n\n"
f"Auto-scanning {len(SCAN_PAIRS)} pairs on M5/M15/H1/H4.\n"
f"📊 Live dashboard refreshes every 60s.\n"