File: //opt/af/src/ai/market_intelligence_agents.py
"""
Market Intelligence Agents module for real-time financial market news and alerts.
"""
import logging
from typing import Dict, List, Optional, Any
import os
from datetime import datetime, timedelta
import pandas as pd
from agno.agent import Agent as AgnoAgent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from src.utils.config import OPENAI_CONFIG
from src.ai.data_sources.financial_news import FinancialDataManager
from src.ai.data_sources.market_data import MarketDataManager
from src.ai.market_alerts import MarketMovementDetector, calculate_key_levels
log = logging.getLogger(__name__)
class MarketNewsAgent:
"""Agent specialized in retrieving and analyzing financial market news."""
def __init__(self):
"""Initialize a Market News agent with financial news data sources."""
self.agent = None
self.data_store = {} # Shared data store for this agent
try:
# Log info about dependencies
log.info("Initializing MarketNewsAgent with FinancialDataManager")
# Initialize news manager with error handling
try:
self.news_manager = FinancialDataManager()
log.info("FinancialDataManager initialized")
except Exception as news_manager_error:
log.error(f"Failed to initialize FinancialDataManager: {news_manager_error}", exc_info=True)
self.news_manager = None
# Check API key and log information
openai_api_key = OPENAI_CONFIG.get('api_key')
if not openai_api_key:
log.error("OpenAI API key not found in environment or config. OPENAI_CONFIG contents: %s", OPENAI_CONFIG)
raise ValueError("OpenAI API key not found in environment or config.")
else:
log.info("OpenAI API key found (truncated): %s...", openai_api_key[:5] if openai_api_key else "None")
# Log Agno dependency information
log.info("Creating AgnoAgent with DuckDuckGoTools")
# Use both Agno's DuckDuckGoTools and our news data manager
self.agent = AgnoAgent(
name="Market News Intelligence",
role="Research and provide financial market news and insights",
model=OpenAIChat(
api_key=openai_api_key,
id="gpt-4o"
),
tools=[
DuckDuckGoTools()
],
markdown=True,
instructions="""You are a financial market news intelligence expert providing analysis within ActionForex.com.
Your goal is to find, summarize, and analyze the most relevant and recent financial market news.
Always:
1. Prioritize recent news (within the last 48 hours when possible)
2. Organize news by market segment (forex, stocks, commodities, etc.)
3. Highlight major market-moving events and their impacts
4. Cite your sources clearly with URLs and timestamps
5. Provide objective analysis backed by factual information
6. Identify patterns across multiple news sources when relevant
7. Remember users are already on ActionForex.com - DO NOT refer them to other platforms
8. Focus on actionable insights traders can use
When responding to user queries:
- For general market updates, provide a concise overview of the most important developments
- For specific currency pairs or assets, focus your analysis on those particular markets
- Always include both facts (what happened) and potential implications (what it might mean)
""",
show_tool_calls=True
)
log.info("Market News Agent successfully initialized")
except ImportError as import_error:
log.error(f"Import error during Market News Agent initialization: {import_error}", exc_info=True)
self.agent = None
except ValueError as value_error:
log.error(f"Value error during Market News Agent initialization: {value_error}", exc_info=True)
self.agent = None
except Exception as e:
log.error(f"Error during Market News Agent initialization: {e}", exc_info=True)
self.agent = None
def get_forex_news(self, currency_pairs: Optional[List[str]] = None,
limit: int = 10) -> List[Dict]:
"""
Get recent forex news for specific currency pairs.
Args:
currency_pairs: List of currency pairs to focus on
limit: Maximum number of news items to return
Returns:
List of news articles as dictionaries
"""
try:
news_items = self.news_manager.get_forex_news(
currency_pairs=currency_pairs,
limit_per_source=limit
)
log.info(f"Retrieved {len(news_items)} forex news items")
return news_items[:limit] # Ensure we don't exceed the limit
except Exception as e:
log.error(f"Error retrieving forex news: {e}", exc_info=True)
return []
def get_economic_news(self, days_back: int = 2,
limit: int = 10) -> List[Dict]:
"""
Get recent economic news.
Args:
days_back: How many days back to fetch news
limit: Maximum number of news items to return
Returns:
List of news articles as dictionaries
"""
try:
news_items = self.news_manager.get_economic_news(
limit_per_source=limit,
days_back=days_back
)
log.info(f"Retrieved {len(news_items)} economic news items")
return news_items[:limit] # Ensure we don't exceed the limit
except Exception as e:
log.error(f"Error retrieving economic news: {e}", exc_info=True)
return []
def search_news(self, query: str, limit: int = 10) -> List[Dict]:
"""
Search for specific financial news.
Args:
query: Search query
limit: Maximum number of news items to return
Returns:
List of news articles as dictionaries
"""
try:
news_items = self.news_manager.search_financial_news(
query=query,
limit_per_source=limit
)
log.info(f"Retrieved {len(news_items)} news items for query '{query}'")
return news_items[:limit] # Ensure we don't exceed the limit
except Exception as e:
log.error(f"Error searching news for '{query}': {e}", exc_info=True)
return []
def process_query(self, query: str, context: dict = None) -> str:
"""Process a user query and return the agent's response."""
if not self.agent:
return f"Error: The Market News Agent could not be initialized."
log.info(f"Processing query with Market News Agent: '{query}'")
# Pre-fetch relevant news based on query keywords
news_context = {}
# Check for currency pair mentions
currency_pairs = ['EUR/USD', 'USD/JPY', 'GBP/USD', 'AUD/USD', 'USD/CHF', 'USD/CAD']
mentioned_pairs = [pair for pair in currency_pairs if pair in query]
if mentioned_pairs:
forex_news = self.get_forex_news(currency_pairs=mentioned_pairs)
if forex_news:
news_titles = [f"- {item.get('title', 'Untitled')} ({item.get('source', 'Unknown')})"
for item in forex_news[:5]]
news_context['recent_forex_news'] = "\n".join(news_titles)
# For general market queries, add economic news
general_terms = ['market', 'economy', 'economic', 'financial']
if any(term in query.lower() for term in general_terms) or not mentioned_pairs:
eco_news = self.get_economic_news()
if eco_news:
news_titles = [f"- {item.get('title', 'Untitled')} ({item.get('source', 'Unknown')})"
for item in eco_news[:5]]
news_context['recent_economic_news'] = "\n".join(news_titles)
# Combine with existing context
combined_context = context or {}
combined_context.update(news_context)
try:
response = self.agent.run(query, context=combined_context)
log.info(f"Market News Agent response received")
# Check if response is a RunResponse object and extract content
if hasattr(response, 'content'):
log.info("Response is a RunResponse object, extracting content")
response_content = str(response.content)
return response_content
else:
log.info("Response is not a RunResponse object")
return response
except Exception as e:
log.error(f"Error during Market News Agent processing: {e}", exc_info=True)
return f"An error occurred while processing your request: {str(e)}"
class MarketAlertsAgent:
"""Agent specialized in detecting and notifying of significant market movements."""
def __init__(self):
"""Initialize a Market Alerts agent with market movement detector."""
self.agent = None
self.data_store = {} # Shared data store for this agent
self.movement_detector = MarketMovementDetector()
self.recent_alerts = []
# Initialize the market data manager for free data sources
try:
self.market_data_manager = MarketDataManager(data_dir="data/market")
log.info("Market Data Manager successfully initialized for alerts")
except Exception as data_manager_error:
log.error(f"Failed to initialize Market Data Manager: {data_manager_error}", exc_info=True)
self.market_data_manager = None
try:
openai_api_key = OPENAI_CONFIG.get('api_key')
if not openai_api_key:
raise ValueError("OpenAI API key not found in environment or config.")
self.agent = AgnoAgent(
name="Market Alerts Intelligence",
role="Monitor and alert about significant market movements",
model=OpenAIChat(
api_key=openai_api_key,
id="gpt-4o"
),
tools=[
DuckDuckGoTools()
],
markdown=True,
instructions="""You are a financial market alerts specialist focusing on significant price movements and patterns.
Your goal is to identify, explain, and contextualize important market events.
Always:
1. Prioritize the most significant market movements
2. Explain both what happened and potential reasons why
3. Organize alerts by urgency and potential impact
4. Use technical terminology appropriately but explain complex concepts
5. Indicate the timeframe of the movement and its significance
6. Compare unusual movements with normal market behavior
7. Highlight correlations between different markets when relevant
8. When analyzing market events, leverage web search to find up-to-date information
9. For specific currency pairs, research recent news that might explain price movements
When presenting alerts:
- Use clear, direct language that emphasizes the key facts
- Format price changes clearly showing direction and magnitude
- Include timestamps to indicate when events occurred
- Group related alerts together when they suggest a broader pattern
- Include relevant recent news that might explain the market movements
""",
show_tool_calls=True
)
log.info("Market Alerts Agent successfully initialized")
# Configure alert callback
self.movement_detector.set_notification_callback(self.handle_alert)
# Load initial market data for common pairs if available
if self.market_data_manager:
self._preload_market_data()
except Exception as e:
log.error(f"Error during Market Alerts Agent initialization: {e}", exc_info=True)
self.agent = None
def _preload_market_data(self) -> None:
"""Preload market data for common currency pairs."""
common_pairs = ['EUR/USD', 'USD/JPY', 'GBP/USD', 'USD/CHF', 'AUD/USD']
timeframes = ['daily']
for pair in common_pairs:
for timeframe in timeframes:
try:
data = self.market_data_manager.get_market_data(
pair=pair,
timeframe=timeframe,
# Get last 30 days by default
start_date=(datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
)
if not data.empty:
self.update_market_data(pair, data)
log.info(f"Preloaded {len(data)} records of {timeframe} data for {pair}")
except Exception as e:
log.error(f"Error preloading {timeframe} data for {pair}: {e}", exc_info=True)
def handle_alert(self, alert_type: str, alert_data: Dict) -> None:
"""
Handle an alert from the movement detector.
Args:
alert_type: Type of alert
alert_data: Alert data dictionary
"""
# Store recent alerts with timestamp
alert_data['received_at'] = datetime.now()
self.recent_alerts.append(alert_data)
# Keep only the most recent 50 alerts
if len(self.recent_alerts) > 50:
self.recent_alerts = self.recent_alerts[-50:]
log.info(f"Received {alert_type} alert for {alert_data.get('pair')}")
def get_recent_alerts(self, hours_back: int = 24, pair: Optional[str] = None) -> List[Dict]:
"""
Get recent alerts, optionally filtered by currency pair.
Args:
hours_back: How many hours back to retrieve alerts
pair: Specific currency pair to filter by
Returns:
List of alert dictionaries
"""
now = datetime.now()
cutoff = now - timedelta(hours=hours_back)
# Filter alerts by time and optionally by pair
filtered_alerts = [
alert for alert in self.recent_alerts
if alert.get('received_at', now) >= cutoff and
(pair is None or alert.get('pair') == pair)
]
# Sort by timestamp, most recent first
sorted_alerts = sorted(
filtered_alerts,
key=lambda x: x.get('received_at', now),
reverse=True
)
return sorted_alerts
def update_market_data(self, pair: str, data: pd.DataFrame) -> None:
"""
Update market data for a currency pair.
Args:
pair: Currency pair symbol
data: DataFrame with market data
"""
try:
# Check if this is synthetic data
is_synthetic = False
data_source = "Unknown"
# Check for synthetic data markers
if 'data_source' in data.columns and data['data_source'].iloc[0]:
data_source = data['data_source'].iloc[0]
if '[SYNTHETIC]' in str(data_source):
is_synthetic = True
# Check dataframe attributes
if hasattr(data, 'attrs') and data.attrs.get('is_synthetic'):
is_synthetic = True
# Store data source info
if 'data_sources_info' not in self.data_store:
self.data_store['data_sources_info'] = []
# Add or update source info
source_info = {
'pair': pair,
'data_source': data_source,
'is_synthetic': is_synthetic,
'updated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'record_count': len(data)
}
# Update existing entry or add new one
updated = False
for i, info in enumerate(self.data_store.get('data_sources_info', [])):
if info.get('pair') == pair:
self.data_store['data_sources_info'][i] = source_info
updated = True
break
if not updated and 'data_sources_info' in self.data_store:
self.data_store['data_sources_info'].append(source_info)
# Update data in the movement detector
self.movement_detector.update_data(pair, data)
# Calculate and update key levels
levels = calculate_key_levels(data)
self.movement_detector.update_key_levels(pair, levels)
log.info(f"Updated market data for {pair} (synthetic: {is_synthetic})")
except Exception as e:
log.error(f"Error updating market data for {pair}: {e}", exc_info=True)
def load_market_data(self, pair: str, timeframe: str = "daily",
days_back: int = 30) -> bool:
"""
Load market data for a currency pair from free sources.
Args:
pair: Currency pair symbol
timeframe: Data timeframe ('daily', 'hourly', etc.)
days_back: Number of days of data to load
Returns:
True if data was loaded successfully, False otherwise
"""
if not self.market_data_manager:
log.error(f"Cannot load market data: Market Data Manager not initialized")
return False
try:
# Calculate start date based on days_back
start_date = (datetime.now() - timedelta(days=days_back)).strftime('%Y-%m-%d')
# Get data from the manager (will try CSV files first, then ActionForex)
data = self.market_data_manager.get_market_data(
pair=pair,
timeframe=timeframe,
start_date=start_date
)
if data.empty:
log.warning(f"No {timeframe} data found for {pair} from any source")
return False
# Update the movement detector with this data
self.update_market_data(pair, data)
log.info(f"Loaded {len(data)} records of {timeframe} data for {pair}")
return True
except Exception as e:
log.error(f"Error loading {timeframe} data for {pair}: {e}", exc_info=True)
return False
def start_monitoring(self, interval_seconds: int = 60) -> None:
"""
Start background monitoring of market movements.
Args:
interval_seconds: Seconds between checks
"""
# Ensure we have data before starting monitoring
if not any(self.movement_detector.data.values()):
log.warning("No market data loaded. Loading default data before starting monitoring.")
self._preload_market_data()
try:
self.movement_detector.start_monitoring(interval_seconds)
log.info(f"Started market monitoring with {interval_seconds}s interval")
except Exception as e:
log.error(f"Error starting market monitoring: {e}", exc_info=True)
def stop_monitoring(self) -> None:
"""Stop background monitoring of market movements."""
try:
self.movement_detector.stop_monitoring_thread()
log.info("Stopped market monitoring")
except Exception as e:
log.error(f"Error stopping market monitoring: {e}", exc_info=True)
def search_market_news(self, query: str, limit: int = 3) -> List[Dict]:
"""
Search the web for market-related news and information.
Args:
query: Search query related to market information
limit: Maximum number of results to return
Returns:
List of search result dictionaries
"""
if not self.agent:
log.error("Cannot search web: Agent not initialized")
return []
try:
# Formulate a search query that focuses on recent market information
# Enhance query to focus on recent financial news
enhanced_query = f"latest {query} market news today financial update"
# Use the DuckDuckGo tool from the agent to perform the search
# Note: we're working with the raw tool call here rather than higher-level interfaces
duckduckgo_tool = None
for tool in self.agent.tools:
if isinstance(tool, DuckDuckGoTools):
duckduckgo_tool = tool
break
if not duckduckgo_tool:
log.error("DuckDuckGo search tool not available")
return []
# Perform the search
log.info(f"Searching web for market information: {enhanced_query}")
search_results = duckduckgo_tool.search(enhanced_query)
# Process the results into a more usable format
processed_results = []
for result in search_results[:limit]:
processed_results.append({
'title': result.get('title', 'No title'),
'snippet': result.get('snippet', result.get('body', 'No snippet')),
'url': result.get('href', ''),
'source': result.get('source', 'Web Search')
})
log.info(f"Found {len(processed_results)} web search results for {query}")
return processed_results
except Exception as e:
log.error(f"Error searching web for market information: {e}", exc_info=True)
return []
def process_query(self, query: str, context: dict = None) -> str:
"""Process a user query and return the agent's response."""
if not self.agent:
return f"Error: The Market Alerts Agent could not be initialized."
log.info(f"Processing query with Market Alerts Agent: '{query}'")
# Pre-fetch relevant alerts based on query keywords
alerts_context = {}
# Check for currency pair mentions
currency_pairs = ['EUR/USD', 'USD/JPY', 'GBP/USD', 'AUD/USD', 'USD/CHF', 'USD/CAD']
mentioned_pairs = [pair for pair in currency_pairs if pair in query]
# Determine time period
time_periods = {
'today': 24,
'yesterday': 48,
'week': 168,
'hour': 1,
'recent': 12
}
hours_back = 24 # default
for period, hours in time_periods.items():
if period in query.lower():
hours_back = hours
break
# Check if the query is asking for loading data
load_data_terms = ['load data', 'update data', 'refresh data', 'get data', 'fetch data']
if any(term in query.lower() for term in load_data_terms) and mentioned_pairs:
load_results = []
for pair in mentioned_pairs:
result = self.load_market_data(pair)
load_results.append(f"{pair}: {'Success' if result else 'Failed'}")
alerts_context['data_loading'] = "Market data loading results:\n" + "\n".join(load_results)
# Get alerts
if mentioned_pairs:
# Get alerts for specific currency pairs
for pair in mentioned_pairs:
alerts = self.get_recent_alerts(hours_back=hours_back, pair=pair)
if alerts:
alerts_context[f'{pair}_alerts'] = f"{len(alerts)} alerts for {pair} in the past {hours_back} hours"
# Search for market news for each mentioned pair
market_news = self.search_market_news(f"{pair} forex")
if market_news:
news_items = [f"- {item['title']} ({item['source']})" for item in market_news]
alerts_context[f'{pair}_market_news'] = f"Recent market news for {pair}:\n" + "\n".join(news_items)
else:
# Get all alerts
all_alerts = self.get_recent_alerts(hours_back=hours_back)
if all_alerts:
# Group by pair
pair_counts = {}
for alert in all_alerts:
pair = alert.get('pair', 'unknown')
pair_counts[pair] = pair_counts.get(pair, 0) + 1
alert_summary = []
for pair, count in pair_counts.items():
alert_summary.append(f"{pair}: {count} alerts")
alerts_context['recent_alerts'] = f"Recent alerts in the past {hours_back} hours:\n" + "\n".join(alert_summary)
# For general market queries, search for overall market news
market_terms = ['market', 'forex', 'currencies', 'financial', 'economic']
if any(term in query.lower() for term in market_terms):
market_news = self.search_market_news("forex market trends today")
if market_news:
news_items = [f"- {item['title']} ({item['source']})" for item in market_news]
alerts_context['general_market_news'] = f"Recent market trends:\n" + "\n".join(news_items)
# Check for specific market events that might require further research
market_events = ['interest rate', 'fed', 'central bank', 'inflation', 'GDP', 'unemployment', 'recession', 'rally', 'crash']
for event in market_events:
if event in query.lower():
event_news = self.search_market_news(f"{event} forex impact")
if event_news:
news_items = [f"- {item['title']} ({item['source']})" for item in event_news]
alerts_context[f'{event}_news'] = f"Recent news about {event}:\n" + "\n".join(news_items)
# Include synthetic data warning if applicable
has_synthetic_data = False
if 'data_sources_info' in self.data_store:
synthetic_sources = [s for s in self.data_store['data_sources_info'] if s.get('is_synthetic', False)]
if synthetic_sources:
has_synthetic_data = True
synthetic_pairs = [s.get('pair') for s in synthetic_sources]
synthetic_warning = f"\n\n**[SYNTHETIC DATA]** This response uses synthesized data for the following pairs: {', '.join(synthetic_pairs)}. This data does not represent real market movements and should not be used for financial decisions."
alerts_context['synthetic_warning'] = synthetic_warning
# Combine with existing context
combined_context = context or {}
combined_context.update(alerts_context)
try:
response = self.agent.run(query, context=combined_context)
log.info(f"Market Alerts Agent response received")
# Check if response is a RunResponse object and extract content
if hasattr(response, 'content'):
log.info("Response is a RunResponse object, extracting content")
response_content = str(response.content)
else:
log.info("Response is not a RunResponse object")
response_content = str(response)
# If we have synthetic data but the warning isn't in the response, add it
if has_synthetic_data and '[SYNTHETIC DATA]' not in response_content:
response_content += "\n\n" + alerts_context.get('synthetic_warning', '')
# If the original response was a RunResponse object, return the content
# Otherwise return the modified response
if hasattr(response, 'content'):
return response_content
else:
return response_content
except Exception as e:
log.error(f"Error during Market Alerts Agent processing: {e}", exc_info=True)
return f"An error occurred while processing your request: {str(e)}"