From e08b593a86eff340a7134b7dde2ea3e7de4519e1 Mon Sep 17 00:00:00 2001 From: harry Date: Mon, 4 May 2026 04:09:32 +0000 Subject: [PATCH] Add news_engine.py --- news_engine.py | 486 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 486 insertions(+) create mode 100644 news_engine.py diff --git a/news_engine.py b/news_engine.py new file mode 100644 index 0000000..b254a3c --- /dev/null +++ b/news_engine.py @@ -0,0 +1,486 @@ +""" +Forex Signal Intelligence Platform - News & Sentiment Engine +============================================================= +Monitors economic calendar events and news sentiment to provide +fundamental context for technical signals. + +Key principle: High-impact news events override technical signals. +A perfect technical setup can be destroyed by an unexpected NFP print. +""" + +import logging +import re +import xml.etree.ElementTree as ET +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional, Tuple + +import requests + +from config import NewsConfig + +logger = logging.getLogger(__name__) + + +@dataclass +class EconomicEvent: + """Single economic calendar event.""" + time: datetime + currency: str + impact: str # "low", "medium", "high" + event: str + forecast: Optional[str] = None + previous: Optional[str] = None + actual: Optional[str] = None + surprise: Optional[str] = None # "positive", "negative", "inline", None + + @property + def is_upcoming(self) -> bool: + return self.time > datetime.now(timezone.utc) + + @property + def minutes_until(self) -> float: + delta = self.time - datetime.now(timezone.utc) + return delta.total_seconds() / 60 + + @property + def is_high_impact(self) -> bool: + return self.impact == "high" + + +@dataclass +class NewsSentiment: + """News sentiment analysis result.""" + headline: str + source: str + timestamp: datetime + currencies_mentioned: List[str] + sentiment_score: float # -1.0 (bearish) to +1.0 (bullish) + keywords: List[str] + + +@dataclass +class FundamentalContext: + """Complete fundamental/news context for signal decisions.""" + events_today: List[EconomicEvent] = field(default_factory=list) + events_upcoming_1h: List[EconomicEvent] = field(default_factory=list) + high_impact_warning: bool = False + warning_message: str = "" + news_sentiment: Dict[str, float] = field(default_factory=dict) # currency -> sentiment + news_headlines: List[NewsSentiment] = field(default_factory=list) + event_impact_on_pairs: Dict[str, str] = field(default_factory=dict) + + +class NewsEngine: + """ + Economic calendar and news sentiment analyzer. + + Provides fundamental context that modifies technical signal confidence. + """ + + # Keywords that indicate directional sentiment + BULLISH_KEYWORDS = { + "rate hike", "hawkish", "strong growth", "beat expectations", + "above forecast", "surplus", "expansion", "employment gains", + "strong jobs", "gdp growth", "inflation rising", "tightening", + "positive surprise", "record high", "bullish", "rally", + "stronger than expected", "upbeat", + } + + BEARISH_KEYWORDS = { + "rate cut", "dovish", "recession", "slowdown", "contraction", + "below forecast", "deficit", "miss expectations", "weak data", + "unemployment rise", "negative surprise", "bearish", "selloff", + "weaker than expected", "downbeat", "easing", "stimulus", + } + + # Currency-specific keywords for headline parsing + CURRENCY_KEYWORDS = { + "USD": ["dollar", "fed", "fomc", "treasury", "us ", "american", "nonfarm", "nfp"], + "EUR": ["euro", "ecb", "eurozone", "eu ", "german", "french"], + "GBP": ["pound", "sterling", "boe", "bank of england", "uk ", "british"], + "JPY": ["yen", "boj", "bank of japan", "japanese"], + "CHF": ["franc", "snb", "swiss"], + "AUD": ["aussie", "rba", "australian", "australia"], + "CAD": ["loonie", "boc", "canadian", "canada"], + "NZD": ["kiwi", "rbnz", "new zealand"], + } + + # High-impact event types (these move markets significantly) + HIGH_IMPACT_EVENTS = { + "interest rate decision", "nonfarm payrolls", "non-farm", + "gdp", "cpi", "inflation", "employment change", + "retail sales", "pmi", "fomc", "ecb press conference", + "boe rate", "rba rate", "snb rate", "boc rate", + } + + def __init__(self, config: NewsConfig): + self.config = config + self._session = requests.Session() + self._session.headers.update({ + "User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)" + }) + + def get_fundamental_context( + self, currencies: Optional[List[str]] = None + ) -> FundamentalContext: + """ + Get complete fundamental context for signal decisions. + + Combines economic calendar + news sentiment into actionable context. + """ + if currencies is None: + currencies = self.config.tracked_currencies + + context = FundamentalContext() + + # 1. Economic calendar events + try: + events = self._fetch_economic_calendar(currencies) + now = datetime.now(timezone.utc) + + context.events_today = [e for e in events if e.time.date() == now.date()] + context.events_upcoming_1h = [ + e for e in events + if 0 < e.minutes_until <= 60 + ] + + # Check for high-impact warnings + for event in events: + if event.is_high_impact and 0 < event.minutes_until <= self.config.pre_event_buffer_minutes: + context.high_impact_warning = True + context.warning_message = ( + f"HIGH IMPACT: {event.currency} {event.event} " + f"in {event.minutes_until:.0f} min — signals may be unreliable" + ) + break + + # Map events to affected pairs + for event in context.events_upcoming_1h: + if event.is_high_impact: + affected = self._get_affected_pairs(event.currency) + for pair in affected: + context.event_impact_on_pairs[pair] = ( + f"{event.currency} {event.event} in {event.minutes_until:.0f}min" + ) + + except Exception as e: + logger.warning(f"Economic calendar fetch failed: {e}") + + # 2. News sentiment + try: + headlines = self._fetch_news_sentiment() + context.news_headlines = headlines + + # Aggregate sentiment per currency + for currency in currencies: + scores = [ + h.sentiment_score for h in headlines + if currency in h.currencies_mentioned + ] + if scores: + context.news_sentiment[currency] = sum(scores) / len(scores) + + except Exception as e: + logger.warning(f"News sentiment fetch failed: {e}") + + return context + + def get_event_risk_level(self, pair_symbol: str, context: FundamentalContext) -> Tuple[str, str]: + """ + Get event risk level for a specific pair. + Returns (risk_level, description). + risk_level: "clear", "caution", "danger" + """ + # Extract currencies from pair + clean = pair_symbol.replace("=X", "") + base = clean[:3] + quote = clean[3:] + + # Check upcoming high-impact events + for event in context.events_upcoming_1h: + if event.is_high_impact and event.currency in (base, quote): + mins = event.minutes_until + if mins <= self.config.pre_event_buffer_minutes: + return "danger", f"{event.currency} {event.event} in {mins:.0f}min" + else: + return "caution", f"{event.currency} {event.event} in {mins:.0f}min" + + # Check for recent high-impact releases (within post-event buffer) + for event in context.events_today: + if event.is_high_impact and event.currency in (base, quote): + if not event.is_upcoming and event.minutes_until > -self.config.post_event_buffer_minutes: + return "caution", f"{event.currency} {event.event} just released" + + return "clear", "No significant events" + + def get_pair_sentiment( + self, pair_symbol: str, context: FundamentalContext + ) -> Tuple[float, str]: + """ + Get net sentiment for a pair based on news. + Returns (score, description) where score is -1 to +1. + Positive = bullish for the pair, negative = bearish. + """ + clean = pair_symbol.replace("=X", "") + base = clean[:3] + quote = clean[3:] + + base_sent = context.news_sentiment.get(base, 0.0) + quote_sent = context.news_sentiment.get(quote, 0.0) + + # Net sentiment: positive base or negative quote = bullish for pair + net = base_sent - quote_sent + + if abs(net) < 0.1: + desc = "Neutral news sentiment" + elif net > 0.3: + desc = f"Bullish sentiment ({base} positive / {quote} negative)" + elif net > 0: + desc = f"Slightly bullish sentiment" + elif net < -0.3: + desc = f"Bearish sentiment ({base} negative / {quote} positive)" + else: + desc = f"Slightly bearish sentiment" + + return net, desc + + # ── Private Methods ──────────────────────────────────────────── + + def _fetch_economic_calendar(self, currencies: List[str]) -> List[EconomicEvent]: + """ + Fetch economic calendar events. + Uses a lightweight approach since most free APIs require auth. + Falls back to manually constructed high-impact schedule. + """ + events = [] + + # Try fetching from a free calendar source + try: + events = self._try_fetch_calendar_api() + if events: + return [e for e in events if e.currency in currencies] + except Exception: + pass + + # Fallback: generate known recurring high-impact events + events = self._generate_known_events(currencies) + return events + + def _try_fetch_calendar_api(self) -> List[EconomicEvent]: + """Attempt to fetch from a free economic calendar API.""" + # Using a simple RSS/JSON approach + events = [] + try: + # Try Forex Factory RSS (most widely available) + url = "https://nfs.faireconomy.media/ff_calendar_thisweek.json" + resp = self._session.get(url, timeout=10) + if resp.status_code == 200: + data = resp.json() + for item in data: + try: + impact_map = {"High": "high", "Medium": "medium", "Low": "low"} + impact = impact_map.get(item.get("impact", ""), "low") + + # Parse date + date_str = item.get("date", "") + if date_str: + event_time = datetime.fromisoformat( + date_str.replace("Z", "+00:00") + ) + else: + continue + + events.append(EconomicEvent( + time=event_time, + currency=item.get("country", ""), + impact=impact, + event=item.get("title", ""), + forecast=item.get("forecast", None), + previous=item.get("previous", None), + actual=item.get("actual") if item.get("actual") else None, + )) + except Exception: + continue + except Exception as e: + logger.debug(f"Calendar API unavailable: {e}") + + return events + + def _generate_known_events(self, currencies: List[str]) -> List[EconomicEvent]: + """ + Generate a schedule of known recurring high-impact events. + This is the fallback when no API is available. + """ + events = [] + now = datetime.now(timezone.utc) + today = now.date() + + # Known high-impact regular events (approximate times) + recurring = [ + ("USD", "FOMC Rate Decision", 18, 0), # 2pm ET + ("USD", "Non-Farm Payrolls", 12, 30), # 8:30am ET (first Friday) + ("USD", "CPI Release", 12, 30), + ("EUR", "ECB Rate Decision", 12, 45), + ("GBP", "BOE Rate Decision", 11, 0), + ("AUD", "RBA Rate Decision", 3, 30), + ("CAD", "BOC Rate Decision", 14, 0), + ("JPY", "BOJ Rate Decision", 3, 0), + ("USD", "Initial Jobless Claims", 12, 30), # Weekly Thursday + ("USD", "Retail Sales", 12, 30), + ("EUR", "German ZEW", 9, 0), + ] + + # Only include if they'd be today or tomorrow + for currency, event_name, hour, minute in recurring: + if currency not in currencies: + continue + for day_offset in range(2): + event_date = today + timedelta(days=day_offset) + event_time = datetime( + event_date.year, event_date.month, event_date.day, + hour, minute, tzinfo=timezone.utc, + ) + events.append(EconomicEvent( + time=event_time, + currency=currency, + impact="high", + event=event_name, + )) + + return events + + def _fetch_news_sentiment(self) -> List[NewsSentiment]: + """Fetch and analyze news headlines for sentiment.""" + headlines = [] + + for feed_url in self.config.rss_feeds: + try: + resp = self._session.get(feed_url, timeout=10) + if resp.status_code != 200: + continue + + root = ET.fromstring(resp.content) + + for item in root.iter("item"): + title_el = item.find("title") + pub_date_el = item.find("pubDate") + + if title_el is None: + continue + + title = title_el.text or "" + pub_date = self._parse_rss_date( + pub_date_el.text if pub_date_el is not None else None + ) + + # Only recent headlines (last 24h) + if pub_date and (datetime.now(timezone.utc) - pub_date).total_seconds() > 86400: + continue + + sentiment = self._analyze_headline_sentiment(title) + currencies = self._extract_currencies(title) + + if currencies: + headlines.append(NewsSentiment( + headline=title, + source=feed_url, + timestamp=pub_date or datetime.now(timezone.utc), + currencies_mentioned=currencies, + sentiment_score=sentiment, + keywords=self._extract_keywords(title), + )) + + except Exception as e: + logger.debug(f"Failed to fetch RSS from {feed_url}: {e}") + + return headlines + + def _analyze_headline_sentiment(self, headline: str) -> float: + """ + Simple but effective keyword-based sentiment analysis. + + Returns -1.0 (very bearish) to +1.0 (very bullish). + + Note: This is intentionally simple. NLP models like FinBERT + would be better but require large dependencies. This catches + the most impactful sentiment signals from headlines. + """ + headline_lower = headline.lower() + bullish_score = 0 + bearish_score = 0 + + for keyword in self.BULLISH_KEYWORDS: + if keyword in headline_lower: + bullish_score += 1 + + for keyword in self.BEARISH_KEYWORDS: + if keyword in headline_lower: + bearish_score += 1 + + total = bullish_score + bearish_score + if total == 0: + return 0.0 + + return (bullish_score - bearish_score) / total + + def _extract_currencies(self, headline: str) -> List[str]: + """Extract mentioned currencies from a headline.""" + headline_lower = headline.lower() + found = [] + + for currency, keywords in self.CURRENCY_KEYWORDS.items(): + if currency.lower() in headline_lower: + found.append(currency) + continue + for kw in keywords: + if kw in headline_lower: + found.append(currency) + break + + return list(set(found)) + + def _extract_keywords(self, headline: str) -> List[str]: + """Extract relevant trading keywords from headline.""" + headline_lower = headline.lower() + found = [] + for kw in self.BULLISH_KEYWORDS | self.BEARISH_KEYWORDS: + if kw in headline_lower: + found.append(kw) + return found + + @staticmethod + def _parse_rss_date(date_str: Optional[str]) -> Optional[datetime]: + """Parse various RSS date formats.""" + if not date_str: + return None + + formats = [ + "%a, %d %b %Y %H:%M:%S %z", + "%a, %d %b %Y %H:%M:%S %Z", + "%Y-%m-%dT%H:%M:%S%z", + "%Y-%m-%d %H:%M:%S", + ] + for fmt in formats: + try: + return datetime.strptime(date_str.strip(), fmt).replace( + tzinfo=timezone.utc if fmt[-2:] != "%z" else None + ) + except ValueError: + continue + return None + + @staticmethod + def _get_affected_pairs(currency: str) -> List[str]: + """Get all major pairs affected by a currency event.""" + pairs = { + "USD": ["EURUSD=X", "GBPUSD=X", "USDJPY=X", "USDCHF=X", "AUDUSD=X", "USDCAD=X", "NZDUSD=X"], + "EUR": ["EURUSD=X", "EURGBP=X", "EURJPY=X", "EURCHF=X"], + "GBP": ["GBPUSD=X", "EURGBP=X", "GBPJPY=X"], + "JPY": ["USDJPY=X", "EURJPY=X", "GBPJPY=X", "AUDJPY=X", "CADJPY=X"], + "CHF": ["USDCHF=X", "EURCHF=X"], + "AUD": ["AUDUSD=X", "AUDJPY=X"], + "CAD": ["USDCAD=X", "CADJPY=X"], + "NZD": ["NZDUSD=X"], + } + return pairs.get(currency, []) \ No newline at end of file