Add news_engine.py
This commit is contained in:
+486
@@ -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, [])
|
||||
Reference in New Issue
Block a user