"""
News injection for market briefings.
"""
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from .sentiment import get_company_news, get_insider_sentiment
NEWS_DAYS_BACK = 7
NEWS_PER_TICKER = 5
TOP_HEADLINES = 3
def fetch_ticker_news(symbol: str, days_back: int = NEWS_DAYS_BACK) -> List[Dict[str, Any]]:
"""
Fetch recent news for a ticker.
Args:
symbol: Stock ticker symbol
days_back: Number of days to look back for news
Returns:
List of news items
"""
to_date = datetime.now().strftime("%Y-%m-%d")
from_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%d")
news = get_company_news(symbol, from_date, to_date)
# Sort by datetime (most recent first) and limit
news = sorted(news, key=lambda x: x.get("datetime", 0), reverse=True)
return news[:NEWS_PER_TICKER]
def score_news_importance(news_item: Dict[str, Any]) -> int:
"""
Score news item importance (higher = more important).
Factors:
- Headline mentions earnings, merger, acquisition, FDA, etc.
- Recent publication (within last 24 hours)
- Source credibility
"""
score = 0
headline = news_item.get("headline", "").lower()
source = news_item.get("source", "").lower()
# Keywords that indicate significant news
important_keywords = [
"earnings", "beats", "misses", "guidance", "revenue",
"merger", "acquisition", "acquires", "buys",
"fda", "approval", "clinical", "trial",
"upgrade", "downgrade", "analyst", "price target",
"layoff", "layoffs", "job cuts", "restructuring",
"ceo", "cfo", "executive", "resigns", "appointed",
"dividend", "buyback", "split",
"partnership", "collaboration", "deal",
"lawsuit", "settlement", "investigation",
"ipo", "offering", "debt",
"ai", "artificial intelligence", "chatgpt",
"cyberattack", "breach", "security"
]
for keyword in important_keywords:
if keyword in headline:
score += 10
# Recency bonus
published_time = news_item.get("datetime", 0)
if published_time:
published_dt = datetime.fromtimestamp(published_time)
hours_ago = (datetime.now() - published_dt).total_seconds() / 3600
if hours_ago < 24:
score += 20
elif hours_ago < 48:
score += 10
elif hours_ago < 72:
score += 5
# Source credibility (higher scores for reputable sources)
high_credibility = ["bloomberg", "reuters", "wsj", "wall street journal",
"cnbc", "financial times", "ft", "marketwatch",
"barron's", "seeking alpha", "the street",
"techcrunch", " verge", "business insider"]
for cred_source in high_credibility:
if cred_source in source:
score += 5
break
return score
def extract_top_headlines(news_items: List[Dict[str, Any]],
count: int = TOP_HEADLINES) -> List[Dict[str, Any]]:
"""
Extract the top N most significant headlines from news items.
Args:
news_items: List of news items
count: Number of top headlines to return
Returns:
List of top headlines with metadata
"""
# Score each item
scored = [(item, score_news_importance(item)) for item in news_items]
# Sort by score (highest first)
scored.sort(key=lambda x: x[1], reverse=True)
# Return top items
return [item for item, score in scored[:count] if score > 0]
def format_news_headline(ticker: str, news_item: Dict[str, Any]) -> str:
"""Format a single news headline."""
headline = news_item.get("headline", "")
source = news_item.get("source", "Unknown")
return f"📰 {ticker}: {headline} ({source})"
def format_headline_summary(news_item: Dict[str, Any]) -> str:
"""Format a headline for the Top Stories section."""
headline = news_item.get("headline", "")
source = news_item.get("source", "Unknown")
# Clean up headline for display
if len(headline) > 100:
headline = headline[:97] + "..."
return f"• {headline} ({source})"
def get_news_for_watchlist(symbols: List[str]) -> Dict[str, List[Dict[str, Any]]]:
"""
Get news for all tickers in a watchlist.
Args:
symbols: List of ticker symbols
Returns:
Dict mapping symbol to list of news items
"""
result = {}
for symbol in symbols:
news = fetch_ticker_news(symbol)
if news:
result[symbol] = news
return result
def get_top_stories(symbols: List[str], count: int = 5) -> List[Dict[str, Any]]:
"""
Get the top stories across all watchlist tickers.
Args:
symbols: List of ticker symbols
count: Number of stories to return
Returns:
List of top stories with ticker attribution
"""
all_news = []
for symbol in symbols:
news = fetch_ticker_news(symbol)
for item in news:
item["_ticker"] = symbol # Add ticker for attribution
all_news.append(item)
# Score and extract top
top = extract_top_headlines(all_news, count)
return top
def get_insider_sentiment_summary(symbols: List[str]) -> List[str]:
"""
Get insider sentiment summary for watchlist.
Returns notable insider activity as formatted strings.
"""
notable_activity = []
for symbol in symbols[:5]: # Check top 5 to avoid rate limits
try:
from_date = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d")
to_date = datetime.now().strftime("%Y-%m-%d")
sentiment = get_insider_sentiment(symbol, from_date, to_date)
if sentiment and sentiment.get("data"):
data = sentiment["data"]
if isinstance(data, list) and len(data) > 0:
# Get most recent month
recent = data[-1]
mspr = recent.get("mspr", 0) # Monthly Share Purchase Ratio
if mspr > 0.5:
notable_activity.append(f"• {symbol}: Strong insider buying (MSPR: {mspr:.2f})")
elif mspr < -0.5:
notable_activity.append(f"• {symbol}: Insider selling pressure (MSPR: {mspr:.2f})")
except Exception:
pass # Skip if data unavailable
if not notable_activity:
notable_activity = ["• No significant insider activity detected this week"]
return notable_activity
def get_earnings_this_week(symbols: List[str]) -> List[str]:
"""
Get earnings announcements for this week from watchlist.
Returns formatted strings for tickers with earnings this week.
"""
from .sentiment import get_earnings_calendar
today = datetime.now()
week_start = today - timedelta(days=today.weekday()) # Monday
week_end = week_start + timedelta(days=6) # Sunday
from_date = week_start.strftime("%Y-%m-%d")
to_date = week_end.strftime("%Y-%m-%d")
try:
earnings = get_earnings_calendar(from_date, to_date)
watchlist_set = set(symbols)
this_week = []
for earning in earnings:
symbol = earning.get("symbol", "")
if symbol in watchlist_set:
date = earning.get("date", "")
eps_est = earning.get("epsEstimate", "TBD")
this_week.append(f"• {symbol}: {date} (Est EPS: {eps_est})")
if not this_week:
return ["• No major earnings from watchlist this week"]
return this_week[:5] # Limit to 5
except Exception:
return ["• Earnings calendar unavailable"]
if name == "main":
# Test
symbols = ["AAPL", "MSFT", "TSLA"]
print("News for watchlist:", get_news_for_watchlist(symbols))
print("Top stories:", get_top_stories(symbols))