fix(news): implement persistent multi-layer deduplication, word-boundary keyword matching, and alert suppression

core/bot.py

@@ -597,11 +597,21 @@ class TradingBot:
             logger.error(f"Error during scheduled model retraining: {e}")
 
     async def _process_news(self):
-        """Fetch and analyze news."""
+        """Fetch and analyze news with strict deduplication."""
         try:
             articles = await self.news_fetcher.fetch_all_news()
+            if not articles:
+                return
 
             for article in articles[:5]:   # Process top 5 most recent for fast responsiveness
+                headline = article.get("headline", "")
+                url = article.get("url", "")
+
+                # Guard: skip if already recorded in database or processed
+                if crud.is_news_already_saved(headline, url):
+                    logger.debug(f"Skipping already recorded news: {headline[:60]}")
+                    continue
+
                 analysis = await self.news_analyzer.analyze_article(article)
 
                 # Add to sentiment aggregator
@@ -614,9 +624,9 @@ class TradingBot:
                 # Save to database
                 crud.save_news_event(
                     source=article.get("source", ""),
-                    headline=article.get("headline", ""),
+                    headline=headline,
                     summary=article.get("summary", ""),
-                    url=article.get("url", ""),
+                    url=url,
                     sentiment=analysis["sentiment"],
                     sentiment_score=analysis["combined_score"],
                     finbert_score=analysis["finbert"]["score"],
@@ -626,16 +636,16 @@ class TradingBot:
                     published_at=article.get("published_at"),
                 )
 
-                # Pause trading for HIGH impact news
+                # Pause trading for HIGH impact news (only for fresh news)
                 if analysis["impact_level"] == "HIGH":
                     logger.warning(
-                        f"⚠️ HIGH IMPACT NEWS: {article['headline'][:80]}"
+                        f"⚠️ HIGH IMPACT NEWS: {headline[:80]}"
                     )
                     self.regime_detector.set_news_pause(True)
 
                     # Notify via Telegram
                     await self.telegram.send_news_alert(
-                        headline=article.get("headline", ""),
+                        headline=headline,
                         sentiment=analysis["sentiment"],
                         impact=analysis["impact_level"],
                         score=analysis["combined_score"],

dashboard/app.py

@@ -169,12 +169,23 @@ async def get_recent_news():
 
     session = get_session()
     try:
+        from news.news_utils import normalize_headline
         events = (
             session.query(NewsEvent)
             .order_by(desc(NewsEvent.fetched_at))
-            .limit(20)
+            .limit(40)
             .all()
         )
+        unique_events = []
+        seen = set()
+        for e in events:
+            norm = normalize_headline(e.headline)
+            if norm not in seen:
+                seen.add(norm)
+                unique_events.append(e)
+            if len(unique_events) >= 20:
+                break
+
         return JSONResponse([
             {
                 "headline": e.headline,
@@ -185,7 +196,7 @@ async def get_recent_news():
                 "analysis": e.gemini_analysis,
                 "time": e.fetched_at.isoformat() if e.fetched_at else None,
             }
-            for e in events
+            for e in unique_events
         ])
     finally:
         session.close()
@@ -422,12 +433,16 @@ async def refresh_news_feed():
 
         count = 0
         for article in articles[:5]:
+            headline = article.get("headline", "")
+            url = article.get("url", "")
+            if crud.is_news_already_saved(headline, url):
+                continue
             analysis = await analyzer.analyze_article(article)
-            crud.save_news_event(
+            saved = crud.save_news_event(
                 source=article.get("source", ""),
-                headline=article.get("headline", ""),
+                headline=headline,
                 summary=article.get("summary", ""),
-                url=article.get("url", ""),
+                url=url,
                 sentiment=analysis["sentiment"],
                 sentiment_score=analysis["combined_score"],
                 finbert_score=analysis["finbert"]["score"],
@@ -436,7 +451,8 @@ async def refresh_news_feed():
                 impact_level=analysis["impact_level"],
                 published_at=article.get("published_at"),
             )
-            count += 1
+            if saved:
+                count += 1
         return JSONResponse({"success": True, "articles_processed": count})
     except Exception as e:
         logger.error(f"News refresh error: {e}")
@@ -502,13 +518,13 @@ async def start_realtime_streamer():
     """Background task to broadcast real-time price updates to WebSocket subscribers."""
     import asyncio
     async def stream_loop():
-        global connected_clients
         from core.mt5_connector import MT5Connector
         logger.info("📡 Starting real-time WebSocket market streamer")
         mt5 = MT5Connector()
         while True:
             try:
-                if connected_clients:
+                clients = globals().get("connected_clients", set())
+                if clients:
                     if mt5.connect(max_retries=1, retry_delay=0.1):
                         tick = mt5.get_current_tick()
                         if tick and "bid" in tick:
@@ -548,6 +564,7 @@ async def websocket_endpoint(websocket: WebSocket):
 
 async def broadcast(data: dict):
     """Broadcast data to all connected WebSocket clients."""
+    global connected_clients
     if not connected_clients:
         return
 

database/crud.py

@@ -328,6 +328,40 @@ def create_signal(
 # ── News Operations ──────────────────────────────────────────────────────
 
 
+def is_news_already_saved(headline: str, url: str = "", max_age_days: int = 7) -> bool:
+    """Check if an article with matching normalized headline or URL has already been recorded."""
+    session = get_session()
+    try:
+        from news.news_utils import normalize_headline
+        cutoff = datetime.now(timezone.utc) - timedelta(days=max_age_days)
+        # Check by url if present
+        if url and url.strip():
+            existing = session.query(NewsEvent.id).filter(
+                NewsEvent.url == url.strip(),
+                NewsEvent.fetched_at >= cutoff,
+            ).first()
+            if existing:
+                return True
+
+        norm_target = normalize_headline(headline)
+        if not norm_target:
+            return False
+
+        # Query recent headlines within time window
+        recent_headlines = session.query(NewsEvent.headline).filter(
+            NewsEvent.fetched_at >= cutoff
+        ).all()
+        for (h,) in recent_headlines:
+            if normalize_headline(h) == norm_target:
+                return True
+        return False
+    except Exception as e:
+        logger.error(f"Error checking if news already saved: {e}")
+        return False
+    finally:
+        session.close()
+
+
 def save_news_event(
     source: str,
     headline: str,
@@ -341,8 +375,12 @@ def save_news_event(
     gemini_analysis: str = "",
     impact_level: str = "LOW",
     published_at: datetime | None = None,
-) -> NewsEvent:
-    """Save a news event with sentiment analysis."""
+) -> NewsEvent | None:
+    """Save a news event with sentiment analysis, strictly deduplicated."""
+    if is_news_already_saved(headline, url):
+        logger.debug(f"Skipping duplicate news event save: {headline[:60]}")
+        return None
+
     session = get_session()
     try:
         event = NewsEvent(
@@ -370,6 +408,40 @@ def save_news_event(
         session.close()
 
 
+def cleanup_duplicate_news_events() -> int:
+    """Remove existing duplicate news events from database, keeping the latest entry."""
+    session = get_session()
+    try:
+        from news.news_utils import normalize_headline
+        events = session.query(NewsEvent).order_by(NewsEvent.id.desc()).all()
+        seen = set()
+        to_delete = []
+        for ev in events:
+            h_norm = normalize_headline(ev.headline)
+            u_str = (ev.url or "").strip().lower()
+            key = (h_norm, u_str)
+            if (h_norm and h_norm in seen) or (u_str and u_str in seen) or key in seen:
+                to_delete.append(ev.id)
+            else:
+                if h_norm:
+                    seen.add(h_norm)
+                if u_str:
+                    seen.add(u_str)
+                seen.add(key)
+
+        if to_delete:
+            session.query(NewsEvent).filter(NewsEvent.id.in_(to_delete)).delete(synchronize_session=False)
+            session.commit()
+            logger.info(f"Purged {len(to_delete)} duplicate news rows from database.")
+        return len(to_delete)
+    except Exception as e:
+        session.rollback()
+        logger.error(f"Error purging duplicate news events: {e}")
+        return 0
+    finally:
+        session.close()
+
+
 def get_recent_sentiment(hours: int = 6) -> float:
     """Get average sentiment score from the last N hours."""
     session = get_session()

news/news_fetcher.py

@@ -15,7 +15,6 @@ from typing import Optional
 import httpx
 import re
 
-
 def _sanitize_log(msg: str) -> str:
     """Mask API tokens from URLs and error tracebacks."""
     return re.sub(r'([?&](?:token|apikey|api_key)=)[^&\s]+', r'\1***REDACTED***', str(msg))
@@ -23,6 +22,7 @@ def _sanitize_log(msg: str) -> str:
 from loguru import logger
 
 from config.settings import get_settings
+from news.news_utils import compute_news_hash, is_headline_relevant, normalize_headline
 
 
 class NewsFetcher:
@@ -41,19 +41,42 @@ class NewsFetcher:
         self.keywords = [kw.lower() for kw in news_params.get("keywords", ["gold", "XAUUSD"])]
         self.check_interval = news_params.get("check_interval_minutes", 5)
 
-        # Cache to avoid duplicate processing
-        self._seen_headlines: set[str] = set()
+        # Cache to avoid duplicate processing (persisted from DB across restarts)
+        self._seen_hashes: set[str] = set()
+        self._seen_headlines = self._seen_hashes  # Alias for backward compatibility
         self._last_fetch: Optional[datetime] = None
         self._alpha_vantage_cooldown_until: Optional[datetime] = None
 
-    def _headline_hash(self, headline: str) -> str:
+        self._load_seen_from_db()
+
+    def _load_seen_from_db(self):
+        """Preload recent news hashes from database so restarts never re-process news."""
+        try:
+            from database.models import get_session, NewsEvent
+            session = get_session()
+            try:
+                cutoff = datetime.now(timezone.utc) - timedelta(days=7)
+                events = session.query(NewsEvent.headline, NewsEvent.url).filter(
+                    NewsEvent.fetched_at >= cutoff
+                ).all()
+                for h_text, u_text in events:
+                    if h_text:
+                        self._seen_hashes.add(compute_news_hash(h_text, u_text or ""))
+                    if u_text:
+                        self._seen_hashes.add(hashlib.sha256(u_text.strip().lower().encode("utf-8")).hexdigest())
+                logger.info(f"📰 Preloaded {len(self._seen_hashes)} seen news hashes from database.")
+            finally:
+                session.close()
+        except Exception as e:
+            logger.debug(f"Could not preload seen news from DB: {e}")
+
+    def _headline_hash(self, headline: str, url: str = "") -> str:
         """Create a hash of a headline for deduplication."""
-        return hashlib.md5(headline.lower().strip().encode()).hexdigest()
+        return compute_news_hash(headline, url)
 
     def _is_relevant(self, headline: str, summary: str = "") -> bool:
-        """Check if an article is relevant to XAUUSD trading."""
-        text = (headline + " " + summary).lower()
-        return any(kw in text for kw in self.keywords)
+        """Check if an article is relevant to XAUUSD trading using regex word boundaries."""
+        return is_headline_relevant(headline, summary, self.keywords)
 
     async def fetch_finnhub_news(self, category: str = "general") -> list[dict]:
         """
@@ -83,22 +106,23 @@ class NewsFetcher:
             for article in articles:
                 headline = article.get("headline", "")
                 summary = article.get("summary", "")
+                art_url = article.get("url", "")
 
                 # Skip if already seen
-                h = self._headline_hash(headline)
-                if h in self._seen_headlines:
+                h = compute_news_hash(headline, art_url)
+                if h in self._seen_hashes:
                     continue
 
                 # Filter for relevance
                 if not self._is_relevant(headline, summary):
                     continue
 
-                self._seen_headlines.add(h)
+                self._seen_hashes.add(h)
                 relevant.append({
                     "source": "finnhub",
                     "headline": headline,
                     "summary": summary,
-                    "url": article.get("url", ""),
+                    "url": art_url,
                     "category": article.get("category", category),
                     "published_at": datetime.fromtimestamp(
                         article.get("datetime", 0), tz=timezone.utc
@@ -119,25 +143,26 @@ class NewsFetcher:
 
     async def fetch_alpha_vantage_news(
         self,
-        tickers: Optional[str] = None,
-        topics: str = "economy_monetary,economy_fiscal,financial_markets",
+        topics: str = "economy_macro,monetary_policy",
+        tickers: str = "FOREX:XAU",
     ) -> list[dict]:
         """
-        Fetch news with sentiment from Alpha Vantage.
+        Fetch news sentiment from Alpha Vantage.
 
         Args:
-            tickers: Comma-separated tickers (e.g., "FOREX:XAU")
             topics: Comma-separated topics
+            tickers: Comma-separated tickers
 
         Returns:
-            List of news article dicts with pre-computed sentiment.
+            List of news articles with pre-computed sentiment.
         """
         if not self.alpha_vantage_key:
             logger.warning("Alpha Vantage API key not configured")
             return []
 
         if self._alpha_vantage_cooldown_until and datetime.now(timezone.utc) < self._alpha_vantage_cooldown_until:
-            logger.debug("Alpha Vantage in cooldown window, skipping fetch.")
+            wait_m = max(1, int((self._alpha_vantage_cooldown_until - datetime.now(timezone.utc)).total_seconds() / 60))
+            logger.debug(f"Alpha Vantage in cooldown, skipping fetch (resumes in {wait_m}m)")
             return []
 
         url = "https://www.alphavantage.co/query"
@@ -171,15 +196,16 @@ class NewsFetcher:
             for article in feed:
                 headline = article.get("title", "")
                 summary = article.get("summary", "")
+                art_url = article.get("url", "")
 
-                h = self._headline_hash(headline)
-                if h in self._seen_headlines:
+                h = compute_news_hash(headline, art_url)
+                if h in self._seen_hashes:
                     continue
 
                 if not self._is_relevant(headline, summary):
                     continue
 
-                self._seen_headlines.add(h)
+                self._seen_hashes.add(h)
 
                 # Alpha Vantage provides pre-computed sentiment
                 overall_sentiment = article.get("overall_sentiment_score", 0)
@@ -197,7 +223,7 @@ class NewsFetcher:
                     "source": "alpha_vantage",
                     "headline": headline,
                     "summary": summary,
-                    "url": article.get("url", ""),
+                    "url": art_url,
                     "category": ",".join([t.get("topic", "") if isinstance(t, dict) else str(t) for t in article.get("topics", [])[:3]]),
                     "published_at": pub_time,
                     "av_sentiment_score": float(overall_sentiment),
@@ -254,5 +280,5 @@ class NewsFetcher:
 
     def clear_cache(self):
         """Clear the seen headlines cache."""
-        self._seen_headlines.clear()
+        self._seen_hashes.clear()
         logger.debug("News cache cleared")

news/news_utils.py

@@ -0,0 +1,59 @@
+"""
+Utilities for news normalization, hashing, and deduplication.
+"""
+
+import re
+import hashlib
+
+
+def normalize_headline(headline: str) -> str:
+    """
+    Normalize headline by stripping source attribution tags, lowercasing,
+    and removing non-alphanumeric punctuation.
+    """
+    if not headline:
+        return ""
+
+    # Strip trailing news agency / publisher indicators like " - Reuters", " | Bloomberg", " - CNBC", etc.
+    cleaned = re.sub(
+        r"\s*[-|–—:]\s*(Reuters|Bloomberg|CNBC|FXStreet|Investing\.com|MarketWatch|Associated Press|AP|Financial Times|FT|Yahoo Finance|WSJ|BenZinga|Zacks|StreetInsider|Seeking Alpha).*$",
+        "",
+        headline,
+        flags=re.IGNORECASE,
+    )
+    # Lowercase & collapse whitespace
+    cleaned = re.sub(r"\s+", " ", cleaned).strip().lower()
+    # Strip punctuation characters while preserving spaces and alphanumerics
+    cleaned = re.sub(r"[^\w\s]", "", cleaned)
+    return cleaned.strip()
+
+
+def compute_news_hash(headline: str, url: str = "") -> str:
+    """
+    Compute a deterministic SHA-256 hash for deduplicating news articles.
+    Prefers normalized headline, falls back to normalized URL if headline is empty.
+    """
+    norm = normalize_headline(headline)
+    if not norm and url:
+        return hashlib.sha256(url.strip().lower().encode("utf-8")).hexdigest()
+    return hashlib.sha256(norm.encode("utf-8")).hexdigest()
+
+
+def is_headline_relevant(headline: str, summary: str = "", keywords: list[str] | None = None) -> bool:
+    """
+    Check if an article text is relevant using regex word boundaries to avoid
+    false-positive partial matches (e.g. 'war' in 'warms', 'CPI' in 'recipe').
+    """
+    if not keywords:
+        return False
+
+    text = f"{headline} {summary}".lower()
+    for kw in keywords:
+        kw = kw.strip().lower()
+        if not kw:
+            continue
+        # Support optional trailing 's' for plurals (e.g., war -> wars, tariff -> tariffs)
+        pattern = r"\b" + re.escape(kw) + r"s?\b"
+        if re.search(pattern, text):
+            return True
+    return False

notifications/telegram_bot.py

@@ -27,7 +27,7 @@ Features:
 """
 
 import asyncio
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, List, Optional
 from loguru import logger
 
@@ -71,9 +71,32 @@ class TelegramNotifier:
         self._is_polling = False
         self._enabled = bool(self.bot_token and self.chat_id and self.chat_id != "0")
 
+        # News alert deduplication cache
+        self._sent_news_alert_hashes: set[str] = set()
+        self._preload_alerted_news()
+
         if not self._enabled:
             logger.warning("Telegram bot not configured or chat ID missing — remote control disabled")
 
+    def _preload_alerted_news(self):
+        """Preload hashes of recent news from DB so restarts don't re-alert old high-impact items."""
+        try:
+            from database.models import get_session, NewsEvent
+            from news.news_utils import compute_news_hash
+            session = get_session()
+            try:
+                cutoff = datetime.now(timezone.utc) - timedelta(hours=48)
+                events = session.query(NewsEvent.headline, NewsEvent.url).filter(
+                    NewsEvent.fetched_at >= cutoff
+                ).all()
+                for h_text, u_text in events:
+                    if h_text:
+                        self._sent_news_alert_hashes.add(compute_news_hash(h_text, u_text or ""))
+            finally:
+                session.close()
+        except Exception as e:
+            logger.debug(f"Could not preload alerted news: {e}")
+
     def set_bot_instance(self, bot_instance: Any):
         """Link the parent TradingBot instance for live data and control."""
         self.bot_instance = bot_instance
@@ -439,14 +462,25 @@ class TelegramNotifier:
             f"<b>AI Engine:</b> Gemini 3.6 Flash + Bullion Sentiment\n\n"
         )
 
-        # Pull top news events from DB
+        # Pull top distinct news events from DB
         session = crud.get_session()
         try:
             from database.models import NewsEvent
-            events = session.query(NewsEvent).order_by(NewsEvent.id.desc()).limit(4).all()
-            if events:
+            from news.news_utils import normalize_headline
+            events = session.query(NewsEvent).order_by(NewsEvent.id.desc()).limit(20).all()
+            unique_events = []
+            seen_headlines = set()
+            for ev in events:
+                norm = normalize_headline(ev.headline)
+                if norm not in seen_headlines:
+                    seen_headlines.add(norm)
+                    unique_events.append(ev)
+                if len(unique_events) >= 4:
+                    break
+
+            if unique_events:
                 msg += "<b>Latest Macro Analysis:</b>\n"
-                for ev in events:
+                for ev in unique_events:
                     impact_emoji = "🔴" if ev.impact_level == "HIGH" else ("🟡" if ev.impact_level == "MEDIUM" else "🟢")
                     msg += (
                         f"{impact_emoji} <b>{ev.headline[:75]}...</b>\n"
@@ -487,12 +521,16 @@ class TelegramNotifier:
             articles = await fetcher.fetch_all_news()
 
             for article in articles[:5]:
+                headline = article.get("headline", "")
+                url = article.get("url", "")
+                if crud.is_news_already_saved(headline, url):
+                    continue
                 analysis = await analyzer.analyze_article(article)
                 crud.save_news_event(
                     source=article.get("source", ""),
-                    headline=article.get("headline", ""),
+                    headline=headline,
                     summary=article.get("summary", ""),
-                    url=article.get("url", ""),
+                    url=url,
                     sentiment=analysis["sentiment"],
                     sentiment_score=analysis["combined_score"],
                     finbert_score=analysis["finbert"]["score"],
@@ -1157,7 +1195,14 @@ class TelegramNotifier:
         await self.send_message(msg, reply_markup=keyboard)
 
     async def send_news_alert(self, headline: str, sentiment: str, impact: str, score: float):
-        """Send high-impact news alert."""
+        """Send high-impact news alert with strict deduplication."""
+        from news.news_utils import compute_news_hash
+        h_hash = compute_news_hash(headline)
+        if h_hash in self._sent_news_alert_hashes:
+            logger.info(f"Skipping duplicate Telegram news alert for: {headline[:60]}")
+            return
+        self._sent_news_alert_hashes.add(h_hash)
+
         impact_emoji = {"HIGH": "🔴", "MEDIUM": "🟡", "LOW": "🟢"}.get(impact, "⚪")
         sentiment_emoji = {"BULLISH": "📈", "BEARISH": "📉", "NEUTRAL": "➡️"}.get(sentiment, "➡️")