""" Web search integration for fetching real-time gold market news. This module provides functionality to search for recent gold market news using various search APIs. Currently supports: - DuckDuckGo search (free, no API key required) - Extensible for Tavily, SerpAPI, or other providers """ import httpx import json from typing import List, Dict, Optional from datetime import datetime, timedelta class NewsSearchService: """Service for fetching recent gold market news from the web.""" def __init__(self): self.timeout = 10.0 async def search_gold_news(self, query: str = "gold price XAU/USD", max_results: int = 5) -> List[Dict]: """ Search for recent gold market news. Args: query: Search query (default: "gold price XAU/USD") max_results: Maximum number of results to return Returns: List of news articles with title, snippet, url, and date """ try: # Use DuckDuckGo Instant Answer API (free, no key required) async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.get( "https://api.duckduckgo.com/", params={ "q": query, "format": "json", "no_html": 1, "skip_disambig": 1, } ) if response.status_code == 200: data = response.json() results = [] # Extract related topics (news items) related_topics = data.get("RelatedTopics", []) for topic in related_topics[:max_results]: if isinstance(topic, dict) and "Text" in topic: results.append({ "title": topic.get("Text", "")[:100], "snippet": topic.get("Text", ""), "url": topic.get("FirstURL", ""), "source": "DuckDuckGo", "date": datetime.now().isoformat() }) return results except Exception as e: print(f"News search error: {e}") # Return fallback generic news context return self._get_fallback_news() def _get_fallback_news(self) -> List[Dict]: """Return generic gold market context when search fails.""" return [ { "title": "Gold Market Overview", "snippet": "Gold prices influenced by USD strength, inflation expectations, and geopolitical events", "url": "", "source": "General Context", "date": datetime.now().isoformat() }, { "title": "Key Gold Drivers", "snippet": "Federal Reserve policy, US Dollar Index (DXY), real yields, and global risk sentiment", "url": "", "source": "General Context", "date": datetime.now().isoformat() } ] async def get_news_summary(self, max_items: int = 3) -> str: """ Get a formatted summary of recent gold news for AI prompts. Args: max_items: Maximum number of news items to include Returns: Formatted string with news headlines and snippets """ news_items = await self.search_gold_news(max_results=max_items) if not news_items: return "📰 Recent News: No recent news available. Analysis based on technical factors only." summary = "📰 RECENT MARKET NEWS:\n" for i, item in enumerate(news_items, 1): summary += f"{i}. {item['title']}\n" if item['snippet'] and item['snippet'] != item['title']: summary += f" {item['snippet'][:150]}...\n" return summary # Global service instance news_search_service = NewsSearchService()