diff --git a/config.py b/config.py index 6ce1cdd..14ae41c 100755 --- a/config.py +++ b/config.py @@ -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", ]) diff --git a/news_engine.py b/news_engine.py index e9266ec..28155f6 100755 --- a/news_engine.py +++ b/news_engine.py @@ -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"]*>(.*?)", 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) ] diff --git a/telegram_bot.py b/telegram_bot.py index 8746b5d..50f78e4 100755 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -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"