from typing import Dict, Any from datetime import datetime from src.modules.utils import create_meeting_list, load_meetings_data TOOLS = [ { "type": "function", "function": { "name": "search_meetings", "description": "Search across all FOMC meeting data for relevant information using keywords or phrases. Searches across date, title, action, rate, magnitude, forward_guidance, economic_outlook, market_impact, and key_economic_factors fields.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "Search term or phrase to look for in meetings. Examples: 'inflation expectations', 'employment concerns', 'rate cuts', 'economic uncertainty'" }, "limit": { "type": "integer", "description": "Maximum number of meetings to return. Default is 3, recommend 3-5 for most queries.", "default": 3, "minimum": 1, "maximum": 10 } }, "required": ["query"] } } }, { "type": "function", "function": { "name": "get_latest_meeting", "description": "Get details of the last N FOMC meetings. Use for quick updates or general information.", "parameters": { "type": "object", "properties": { "last_n": { "type": "integer", "description": "Number of meetings to return. Default is 1 to get the latest meeting.", "default": 1, } } } } }, { "type": "function", "function": { "name": "get_rate_decision", "description": "Get details about a specific FOMC meeting's rate decision by date. Finds exact match or closest meeting within 30 days. Use for date-specific queries or historical lookups.", "parameters": { "type": "object", "properties": { "date": { "type": "string", "description": "Meeting date in YYYY-MM-DD format (e.g., '2024-12-18'). For approximate dates, use typical FOMC meeting dates (usually 3rd Wednesday of month).", "pattern": "^\\d{4}-\\d{2}-\\d{2}$" } }, "required": ["date"] } } }, { "type": "function", "function": { "name": "compare_meetings", "description": "Compare two FOMC meetings side by side with detailed analysis of differences, similarities, and factor changes. Use for trend analysis or when user asks about changes over time.", "parameters": { "type": "object", "properties": { "date1": { "type": "string", "description": "First meeting date in YYYY-MM-DD format (e.g., '2024-06-12')", "pattern": "^\\d{4}-\\d{2}-\\d{2}$" }, "date2": { "type": "string", "description": "Second meeting date in YYYY-MM-DD format (e.g., '2024-12-18')", "pattern": "^\\d{4}-\\d{2}-\\d{2}$" } }, "required": ["date1", "date2"] } } }, { "type": "function", "function": { "name": "search_meetings_by_date", "description": "Search for meetings by date, date can be a string containing YYYY, YYYY-MM, or YYYY-MM-DD or a string containing month name and year (e.g., 'June 2024').", "parameters": { "type": "object", "properties": { "date_str": { "type": "string", "description": "Date search string in various formats" } }, "required": ["date_str"] } } } ] def search_meetings_by_date(date_str: str) -> Dict[str, Any]: """ Search for meetings by date, date can be a string containing YYYY, YYYY-MM, or YYYY-MM-DD or a string containing month name and year (e.g., "June 2024"). Args: date_str (str): Date search string in various formats Returns: Dict containing: - success (bool): Whether the search succeeded - query (str): The original search query - results (List[Dict]): List of matching meetings - count (int): Number of results returned - search_type (str): Type of search performed """ meetings_data = load_meetings_data() if not meetings_data: return { "success": False, "error": "No meetings data available", "results": [], "count": 0 } if not date_str.strip(): transformed_results = create_meeting_list(meetings_data) return { "success": True, "query": date_str, "results": transformed_results, "count": len(transformed_results), "search_type": "all" } date_str = date_str.strip() matched_meetings = [] search_type = "unknown" # Try different date formats try: # Check for YYYY format (e.g., "2024") if len(date_str) == 4 and date_str.isdigit(): search_type = "year" for meeting in meetings_data: meeting_date = meeting.get('date', '') if meeting_date.startswith(date_str): matched_meetings.append(meeting) # Check for YYYY-MM format (e.g., "2024-07", "2025-12") elif len(date_str) == 7 and date_str[4] == '-': search_type = "year_month" for meeting in meetings_data: meeting_date = meeting.get('date', '') if meeting_date.startswith(date_str): matched_meetings.append(meeting) # Check for YYYY-MM-DD format (e.g., "2024-07-31") elif len(date_str) == 10 and date_str[4] == '-' and date_str[7] == '-': search_type = "full_date" for meeting in meetings_data: meeting_date = meeting.get('date', '') if meeting_date == date_str: matched_meetings.append(meeting) # Check for month name and year (e.g., "June 2024", "July 2025") else: search_type = "month_year" date_str_lower = date_str.lower() # Month name mappings month_names = { 'january': '01', 'jan': '01', 'february': '02', 'feb': '02', 'march': '03', 'mar': '03', 'april': '04', 'apr': '04', 'may': '05', 'june': '06', 'jun': '06', 'july': '07', 'jul': '07', 'august': '08', 'aug': '08', 'september': '09', 'sep': '09', 'sept': '09', 'october': '10', 'oct': '10', 'november': '11', 'nov': '11', 'december': '12', 'dec': '12' } # Try to parse month name + year parts = date_str_lower.split() if len(parts) == 2: month_part, year_part = parts if month_part in month_names and year_part.isdigit() and len(year_part) == 4: month_num = month_names[month_part] search_pattern = f"{year_part}-{month_num}" for meeting in meetings_data: meeting_date = meeting.get('date', '') if meeting_date.startswith(search_pattern): matched_meetings.append(meeting) except Exception as e: print(f"Error occurred during date parsing: {str(e)}") search_type = "fallback" for meeting in meetings_data: meeting_date = meeting.get('date', '') if date_str.lower() in meeting_date.lower(): matched_meetings.append(meeting) # Transform results using create_meeting_list if matched_meetings: transformed_results = create_meeting_list(matched_meetings) return { "success": True, "query": date_str, "results": transformed_results, "count": len(transformed_results), "search_type": search_type } else: return { "success": True, "query": date_str, "results": [], "count": 0, "search_type": search_type } def search_meetings(query: str, limit: int = 10) -> Dict[str, Any]: """ Search across all FOMC meeting data for relevant information. Args: query (str): Search term or phrase to look for in meetings limit (int): Maximum number of meetings to return (default: 3) Returns: Dict containing: - success (bool): Whether the search succeeded - query (str): The original search query - results (List[Dict]): List of matching meetings - count (int): Number of results returned - total_matches (int): Total number of meetings that matched """ meetings_data = load_meetings_data() if not meetings_data: return { "success": False, "error": "No meetings data available", "results": [], "count": 0 } if not query.strip(): transformed_results = create_meeting_list(meetings_data) return { "success": True, "query": query, "results": transformed_results, "match_details": [{"score": 0, "matched_fields": []} for _ in meetings_data], "count": len(transformed_results), "total_matches": len(meetings_data) } query_lower = query.lower() scored_meetings = [] for meeting in meetings_data: score = 0 matched_fields = [] search_fields = { 'date': 2, 'title': 1, 'action': 3, 'rate': 3, 'magnitude': 2, 'forward_guidance': 2, 'economic_outlook': 1, 'market_impact': 1, 'key_economic_factors': 1 } for field, weight in search_fields.items(): field_value = meeting.get(field, '') if isinstance(field_value, list): field_value = ' '.join(field_value) if field_value and query_lower in str(field_value).lower(): score += weight matched_fields.append(field) if score > 0: scored_meetings.append({ 'meeting': meeting, 'score': score, 'matched_fields': matched_fields }) scored_meetings.sort(key=lambda x: x['score'], reverse=True) _limit = min(limit, len(scored_meetings)) top_results = scored_meetings[:_limit] transformed_results = create_meeting_list(top_results) return { "success": True, "query": query, "results": transformed_results, "match_details": [{'score': r['score'], 'matched_fields': r['matched_fields']} for r in top_results], "count": len(top_results), "total_matches": len(scored_meetings) } def get_rate_decision(date: str) -> Dict[str, Any]: """ Get details about a specific FOMC meeting's rate decision. Args: date (str): Meeting date in YYYY-MM-DD format (e.g., "2025-06-18") Returns: Dict containing: - success (bool): Whether the meeting was found - meeting (Dict): Complete meeting data if found - exact_match (bool): True if exact date match, False if closest match - requested_date (str): The date that was requested """ meetings_data = load_meetings_data() if not meetings_data: return { "success": False, "error": "No meetings data available" } target_meeting = None for meeting in meetings_data: if meeting.get('date') == date: target_meeting = meeting break if not target_meeting and date: try: target_date = datetime.strptime(date, '%Y-%m-%d') closest_meeting = None min_diff = float('inf') for meeting in meetings_data: try: meeting_date = datetime.strptime(meeting.get('date', ''), '%Y-%m-%d') diff = abs((meeting_date - target_date).days) if diff < min_diff and diff <= 30: min_diff = diff closest_meeting = meeting except ValueError: continue target_meeting = closest_meeting except ValueError: pass if target_meeting: return { "success": True, "requested_date": date, "meeting": target_meeting, "exact_match": target_meeting.get('date') == date } else: return { "success": False, "error": f"No meeting found for date {date}", "requested_date": date } def compare_meetings(date1: str, date2: str) -> Dict[str, Any]: """ Compare two FOMC meetings side by side to analyze differences and similarities. Args: date1 (str): First meeting date in YYYY-MM-DD format date2 (str): Second meeting date in YYYY-MM-DD format Returns: Dict containing: - success (bool): Whether both meetings were found - meeting1/meeting2 (Dict): Data for each meeting - differences (Dict): Fields that differ between meetings - similarities (Dict): Fields that are the same - factor_analysis (Dict): Analysis of key economic factors """ meeting1_result = get_rate_decision(date1) meeting2_result = get_rate_decision(date2) if not meeting1_result["success"]: return { "success": False, "error": f"Could not find meeting for {date1}: {meeting1_result.get('error')}" } if not meeting2_result["success"]: return { "success": False, "error": f"Could not find meeting for {date2}: {meeting2_result.get('error')}" } meeting1 = meeting1_result["meeting"] meeting2 = meeting2_result["meeting"] # Compare key fields comparison = { "success": True, "meeting1": { "date": meeting1.get('date'), "title": meeting1.get('title'), "data": meeting1 }, "meeting2": { "date": meeting2.get('date'), "title": meeting2.get('title'), "data": meeting2 }, "differences": {}, "similarities": {} } # Compare specific fields compare_fields = ['action', 'rate', 'magnitude', 'forward_guidance', 'economic_outlook', 'market_impact'] for field in compare_fields: val1 = meeting1.get(field, '') val2 = meeting2.get(field, '') if val1 != val2: comparison["differences"][field] = { "meeting1_value": val1, "meeting2_value": val2 } else: comparison["similarities"][field] = val1 # Compare key economic factors factors1 = set(meeting1.get('key_economic_factors', [])) factors2 = set(meeting2.get('key_economic_factors', [])) comparison["factor_analysis"] = { "common_factors": list(factors1.intersection(factors2)), "unique_to_meeting1": list(factors1 - factors2), "unique_to_meeting2": list(factors2 - factors1) } return comparison def get_latest_meeting(last_n=1) -> Dict[str, Any]: """ Get the most recent FOMC meeting data. Returns: Dict containing: - success (bool): Whether data was retrieved successfully - meeting (Dict): The most recent meeting data - total_meetings (int): Total number of meetings in database """ meetings_data = load_meetings_data() if not meetings_data: return { "success": False, "error": "No meetings data available" } # Data is already sorted by date (newest first) latest_meetings = meetings_data[:last_n] return { "success": True, "meeting": latest_meetings, "total_meetings": len(meetings_data) } FED_TOOLS= { "search_meetings": search_meetings, "get_latest_meeting": get_latest_meeting, "get_rate_decision": get_rate_decision, "compare_meetings": compare_meetings }