Economic Calendar Integration with Ollama: FOMC and GDP Impact Analysis

# Install core dependencies
pip install requests pandas python-dotenv schedule
pip install ollama-python beautifulsoup4 pytz

# Install optional visualization tools
pip install matplotlib plotly dash

import os
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import pandas as pd

class EconomicCalendarAPI:
    """
    Handles multiple economic calendar data sources
    Supports FOMC, GDP, and other key economic indicators
    """
    
    def __init__(self):
        # Configure API endpoints
        self.endpoints = {
            'tradingeconomics': 'https://api.tradingeconomics.com/calendar',
            'fxempire': 'https://api.fxempire.com/v1/en/calendar',
            'investing': 'https://api.investing.com/api/financialdata/calendar'
        }
        
        # Load API keys from environment
        self.api_keys = {
            'tradingeconomics': os.getenv('TRADING_ECONOMICS_API_KEY'),
            'fxempire': os.getenv('FX_EMPIRE_API_KEY')
        }
    
    def fetch_fomc_events(self, start_date: str, end_date: str) -> List[Dict]:
        """
        Retrieves FOMC meeting data within specified date range
        Returns structured event data for Ollama processing
        """
        params = {
            'country': 'united states',
            'category': 'interest rate',
            'startDate': start_date,
            'endDate': end_date,
            'te': self.api_keys['tradingeconomics']
        }
        
        response = requests.get(self.endpoints['tradingeconomics'], params=params)
        
        if response.status_code == 200:
            events = response.json()
            # Filter for FOMC-specific events
            fomc_events = [
                event for event in events 
                if 'fomc' in event.get('Event', '').lower() or 
                   'federal funds rate' in event.get('Event', '').lower()
            ]
            return fomc_events
        
        return []
    
    def fetch_gdp_data(self, start_date: str, end_date: str) -> List[Dict]:
        """
        Retrieves GDP release data and preliminary estimates
        Includes quarterly and annual growth rates
        """
        params = {
            'country': 'united states',
            'category': 'gdp',
            'startDate': start_date,
            'endDate': end_date,
            'te': self.api_keys['tradingeconomics']
        }
        
        response = requests.get(self.endpoints['tradingeconomics'], params=params)
        
        if response.status_code == 200:
            return response.json()
        
        return []

import ollama
import json
from datetime import datetime

class OllamaEconomicAnalyzer:
    """
    Processes economic calendar events through Ollama
    Generates impact assessments and trading insights
    """
    
    def __init__(self, model_name: str = "llama3.1:8b"):
        self.model_name = model_name
        self.client = ollama.Client()
        
        # Initialize model if not already pulled
        try:
            self.client.chat(model=model_name, messages=[
                {'role': 'user', 'content': 'Test connection'}
            ])
        except:
            print(f"Pulling {model_name} model...")
            self.client.pull(model_name)
    
    def analyze_fomc_impact(self, fomc_event: Dict) -> Dict:
        """
        Analyzes FOMC meeting outcomes and market implications
        Returns structured impact assessment
        """
        prompt = f"""
        Analyze this FOMC meeting data and provide a comprehensive impact assessment:
        
        Event: {fomc_event.get('Event', 'N/A')}
        Date: {fomc_event.get('Date', 'N/A')}
        Previous Rate: {fomc_event.get('Previous', 'N/A')}
        Actual Rate: {fomc_event.get('Actual', 'N/A')}
        Forecast: {fomc_event.get('Forecast', 'N/A')}
        
        Please provide:
        1. Interest rate decision analysis
        2. Market impact assessment (equity, bond, currency)
        3. Economic implications
        4. Historical context comparison
        5. Forward guidance interpretation
        
        Format response as JSON with these keys:
        - rate_decision_analysis
        - market_impact
        - economic_implications
        - historical_context
        - forward_guidance
        """
        
        response = self.client.chat(
            model=self.model_name,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        try:
            # Parse JSON response
            analysis = json.loads(response['message']['content'])
            return analysis
        except json.JSONDecodeError:
            # Fallback to text analysis
            return {
                'raw_analysis': response['message']['content'],
                'timestamp': datetime.now().isoformat()
            }
    
    def analyze_gdp_impact(self, gdp_event: Dict) -> Dict:
        """
        Processes GDP release data for economic trend analysis
        Correlates with FOMC policy implications
        """
        prompt = f"""
        Analyze this GDP release and its economic implications:
        
        Event: {gdp_event.get('Event', 'N/A')}
        Date: {gdp_event.get('Date', 'N/A')}
        Previous: {gdp_event.get('Previous', 'N/A')}%
        Actual: {gdp_event.get('Actual', 'N/A')}%
        Forecast: {gdp_event.get('Forecast', 'N/A')}%
        
        Provide analysis covering:
        1. GDP growth trend assessment
        2. Economic cycle positioning
        3. Federal Reserve policy implications
        4. Sector-specific impacts
        5. Market reaction predictions
        
        Return JSON format with keys:
        - growth_trend_analysis
        - economic_cycle_position
        - fed_policy_implications
        - sector_impacts
        - market_predictions
        """
        
        response = self.client.chat(
            model=self.model_name,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        try:
            return json.loads(response['message']['content'])
        except json.JSONDecodeError:
            return {
                'raw_analysis': response['message']['content'],
                'timestamp': datetime.now().isoformat()
            }

import schedule
import time
import logging
from datetime import datetime, timedelta
import sqlite3

class EconomicAnalysisSystem:
    """
    Orchestrates economic calendar integration with Ollama analysis
    Provides automated event processing and insight generation
    """
    
    def __init__(self, db_path: str = "economic_analysis.db"):
        self.calendar_api = EconomicCalendarAPI()
        self.analyzer = OllamaEconomicAnalyzer()
        self.db_path = db_path
        self.setup_database()
        
        # Configure logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    
    def setup_database(self):
        """
        Creates database tables for storing analysis results
        Enables historical trend analysis
        """
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Create events table
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS economic_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                event_type TEXT NOT NULL,
                event_name TEXT NOT NULL,
                event_date TEXT NOT NULL,
                previous_value TEXT,
                actual_value TEXT,
                forecast_value TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # Create analyses table
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS event_analyses (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                event_id INTEGER,
                analysis_type TEXT NOT NULL,
                analysis_content TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (event_id) REFERENCES economic_events (id)
            )
        """)
        
        conn.commit()
        conn.close()
    
    def process_daily_events(self):
        """
        Daily automated processing of economic calendar events
        Fetches new events and generates Ollama analysis
        """
        self.logger.info("Starting daily economic event processing")
        
        # Calculate date range (today + next 7 days)
        today = datetime.now()
        end_date = today + timedelta(days=7)
        
        date_format = "%Y-%m-%d"
        start_str = today.strftime(date_format)
        end_str = end_date.strftime(date_format)
        
        # Fetch FOMC events
        fomc_events = self.calendar_api.fetch_fomc_events(start_str, end_str)
        self.logger.info(f"Found {len(fomc_events)} FOMC events")
        
        # Process FOMC events
        for event in fomc_events:
            try:
                # Store event in database
                event_id = self.store_event(event, 'FOMC')
                
                # Generate analysis
                analysis = self.analyzer.analyze_fomc_impact(event)
                
                # Store analysis
                self.store_analysis(event_id, 'FOMC_IMPACT', analysis)
                
                self.logger.info(f"Processed FOMC event: {event.get('Event', 'Unknown')}")
                
            except Exception as e:
                self.logger.error(f"Error processing FOMC event: {str(e)}")
        
        # Fetch and process GDP events
        gdp_events = self.calendar_api.fetch_gdp_data(start_str, end_str)
        self.logger.info(f"Found {len(gdp_events)} GDP events")
        
        for event in gdp_events:
            try:
                event_id = self.store_event(event, 'GDP')
                analysis = self.analyzer.analyze_gdp_impact(event)
                self.store_analysis(event_id, 'GDP_IMPACT', analysis)
                
                self.logger.info(f"Processed GDP event: {event.get('Event', 'Unknown')}")
                
            except Exception as e:
                self.logger.error(f"Error processing GDP event: {str(e)}")
    
    def store_event(self, event: Dict, event_type: str) -> int:
        """
        Stores economic event data in database
        Returns event ID for analysis linking
        """
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            INSERT INTO economic_events 
            (event_type, event_name, event_date, previous_value, actual_value, forecast_value)
            VALUES (?, ?, ?, ?, ?, ?)
        """, (
            event_type,
            event.get('Event', 'Unknown'),
            event.get('Date', ''),
            str(event.get('Previous', '')),
            str(event.get('Actual', '')),
            str(event.get('Forecast', ''))
        ))
        
        event_id = cursor.lastrowid
        conn.commit()
        conn.close()
        
        return event_id
    
    def store_analysis(self, event_id: int, analysis_type: str, analysis: Dict):
        """
        Stores Ollama analysis results with event correlation
        """
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            INSERT INTO event_analyses (event_id, analysis_type, analysis_content)
            VALUES (?, ?, ?)
        """, (event_id, analysis_type, json.dumps(analysis)))
        
        conn.commit()
        conn.close()
    
    def get_recent_analyses(self, days: int = 30) -> List[Dict]:
        """
        Retrieves recent economic analyses for trend identification
        """
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute("""
            SELECT e.event_name, e.event_date, e.actual_value, 
                   a.analysis_type, a.analysis_content, a.created_at
            FROM economic_events e
            JOIN event_analyses a ON e.id = a.event_id
            WHERE e.created_at >= datetime('now', '-{} days')
            ORDER BY e.event_date DESC
        """.format(days))
        
        results = cursor.fetchall()
        conn.close()
        
        return [
            {
                'event_name': row[0],
                'event_date': row[1],
                'actual_value': row[2],
                'analysis_type': row[3],
                'analysis_content': json.loads(row[4]),
                'created_at': row[5]
            }
            for row in results
        ]

class FOMCTrendAnalyzer:
    """
    Analyzes FOMC decisions within historical context
    Identifies policy patterns and market reaction trends
    """
    
    def __init__(self, ollama_analyzer: OllamaEconomicAnalyzer):
        self.analyzer = ollama_analyzer
        self.fed_funds_history = []
    
    def analyze_policy_trajectory(self, recent_meetings: List[Dict]) -> Dict:
        """
        Analyzes Fed policy direction across multiple meetings
        Provides insights into hawkish/dovish trends
        """
        prompt = f"""
        Analyze this sequence of FOMC meetings to identify Federal Reserve policy trends:
        
        Meeting Data:
        {json.dumps(recent_meetings, indent=2)}
        
        Please provide comprehensive analysis including:
        1. Policy stance evolution (hawkish to dovish spectrum)
        2. Rate trajectory prediction
        3. Economic data influence on decisions
        4. Market reaction patterns
        5. Forward guidance consistency
        
        Format as JSON with these keys:
        - policy_stance_evolution
        - rate_trajectory_prediction
        - economic_data_influence
        - market_reaction_patterns
        - forward_guidance_analysis
        """
        
        response = self.analyzer.client.chat(
            model=self.analyzer.model_name,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        try:
            return json.loads(response['message']['content'])
        except json.JSONDecodeError:
            return {'raw_analysis': response['message']['content']}
    
    def correlate_with_economic_indicators(self, fomc_data: Dict, gdp_data: Dict) -> Dict:
        """
        Correlates FOMC decisions with GDP and other economic indicators
        Identifies policy effectiveness and economic impact
        """
        prompt = f"""
        Analyze the relationship between this FOMC decision and economic indicators:
        
        FOMC Data:
        {json.dumps(fomc_data, indent=2)}
        
        GDP Data:
        {json.dumps(gdp_data, indent=2)}
        
        Provide correlation analysis covering:
        1. Policy effectiveness assessment
        2. Economic growth impact
        3. Inflation targeting success
        4. Employment implications
        5. Market stability effects
        
        Return JSON format with correlation insights.
        """
        
        response = self.analyzer.client.chat(
            model=self.analyzer.model_name,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        try:
            return json.loads(response['message']['content'])
        except json.JSONDecodeError:
            return {'raw_analysis': response['message']['content']}

class GDPCycleAnalyzer:
    """
    Analyzes GDP data within economic cycle context
    Provides recession/expansion predictions
    """
    
    def __init__(self, ollama_analyzer: OllamaEconomicAnalyzer):
        self.analyzer = ollama_analyzer
        self.gdp_history = []
    
    def analyze_economic_cycle_position(self, gdp_sequence: List[Dict]) -> Dict:
        """
        Determines current economic cycle position
        Predicts cycle transitions and Fed response
        """
        prompt = f"""
        Analyze this GDP sequence to determine economic cycle positioning:
        
        GDP Data Sequence:
        {json.dumps(gdp_sequence, indent=2)}
        
        Provide analysis including:
        1. Current cycle phase (expansion, peak, contraction, trough)
        2. Cycle transition probability
        3. Historical cycle comparison
        4. Fed policy implications
        5. Recession/expansion predictions
        
        Format as JSON with cycle analysis insights.
        """
        
        response = self.analyzer.client.chat(
            model=self.analyzer.model_name,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        try:
            return json.loads(response['message']['content'])
        except json.JSONDecodeError:
            return {'raw_analysis': response['message']['content']}
    
    def generate_sector_impact_analysis(self, gdp_data: Dict) -> Dict:
        """
        Analyzes GDP impact on specific economic sectors
        Provides sector-specific investment insights
        """
        prompt = f"""
        Analyze how this GDP data affects different economic sectors:
        
        GDP Release:
        {json.dumps(gdp_data, indent=2)}
        
        Provide sector-specific analysis for:
        1. Technology sector impact
        2. Financial services implications
        3. Consumer discretionary effects
        4. Industrial and manufacturing
        5. Real estate and construction
        
        Include investment implications and sector rotation opportunities.
        Format as JSON with sector-specific insights.
        """
        
        response = self.analyzer.client.chat(
            model=self.analyzer.model_name,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        try:
            return json.loads(response['message']['content'])
        except json.JSONDecodeError:
            return {'raw_analysis': response['message']['content']}

def main():
    """
    Main execution function for economic analysis system
    Handles scheduling and error recovery
    """
    analysis_system = EconomicAnalysisSystem()
    
    # Schedule daily processing at 6 AM EST
    schedule.every().day.at("06:00").do(analysis_system.process_daily_events)
    
    # Schedule weekly trend analysis on Sundays
    schedule.every().sunday.at("08:00").do(generate_weekly_report)
    
    print("Economic Analysis System started...")
    print("Daily processing scheduled for 6:00 AM EST")
    print("Weekly reports scheduled for Sunday 8:00 AM EST")
    
    while True:
        schedule.run_pending()
        time.sleep(60)  # Check every minute

def generate_weekly_report():
    """
    Generates comprehensive weekly economic analysis report
    Combines FOMC and GDP insights with market trends
    """
    system = EconomicAnalysisSystem()
    recent_analyses = system.get_recent_analyses(days=7)
    
    # Generate comprehensive report
    fomc_analyzer = FOMCTrendAnalyzer(system.analyzer)
    gdp_analyzer = GDPCycleAnalyzer(system.analyzer)
    
    # Process recent data
    fomc_events = [a for a in recent_analyses if a['analysis_type'] == 'FOMC_IMPACT']
    gdp_events = [a for a in recent_analyses if a['analysis_type'] == 'GDP_IMPACT']
    
    if fomc_events:
        fomc_trend = fomc_analyzer.analyze_policy_trajectory(fomc_events)
        print(f"FOMC Policy Trend: {fomc_trend}")
    
    if gdp_events:
        gdp_cycle = gdp_analyzer.analyze_economic_cycle_position(gdp_events)
        print(f"GDP Cycle Analysis: {gdp_cycle}")

if __name__ == "__main__":
    main()

# Economic Calendar API Keys
TRADING_ECONOMICS_API_KEY=your_trading_economics_key_here
FX_EMPIRE_API_KEY=your_fx_empire_key_here

# Database Configuration
DATABASE_PATH=./data/economic_analysis.db
LOG_LEVEL=INFO

# Ollama Configuration
OLLAMA_MODEL=llama3.1:8b
OLLAMA_HOST=localhost:11434

import psutil
import time
from datetime import datetime

class SystemMonitor:
    """
    Monitors economic analysis system performance
    Provides health checks and performance metrics
    """
    
    def __init__(self):
        self.start_time = datetime.now()
        self.processed_events = 0
        self.analysis_errors = 0
    
    def log_system_metrics(self):
        """
        Logs system performance metrics
        Monitors memory, CPU, and processing statistics
        """
        metrics = {
            'cpu_usage': psutil.cpu_percent(interval=1),
            'memory_usage': psutil.virtual_memory().percent,
            'disk_usage': psutil.disk_usage('/').percent,
            'uptime': str(datetime.now() - self.start_time),
            'processed_events': self.processed_events,
            'analysis_errors': self.analysis_errors,
            'error_rate': (self.analysis_errors / max(self.processed_events, 1)) * 100
        }
        
        print(f"System Metrics: {json.dumps(metrics, indent=2)}")
        return metrics
    
    def check_ollama_health(self):
        """
        Verifies Ollama service availability
        Returns health status and response time
        """
        try:
            start_time = time.time()
            client = ollama.Client()
            
            # Simple health check
            response = client.chat(
                model="llama3.1:8b",
                messages=[{'role': 'user', 'content': 'Health check'}]
            )
            
            response_time = time.time() - start_time
            
            return {
                'status': 'healthy',
                'response_time': response_time,
                'timestamp': datetime.now().isoformat()
            }
        
        except Exception as e:
            return {
                'status': 'unhealthy',
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }

import time
from functools import wraps

def rate_limit_handler(max_retries=3, delay=5):
    """
    Decorator for handling API rate limiting
    Implements exponential backoff for failed requests
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.HTTPError as e:
                    if e.response.status_code == 429:  # Rate limited
                        wait_time = delay * (2 ** attempt)
                        print(f"Rate limited. Waiting {wait_time} seconds...")
                        time.sleep(wait_time)
                    else:
                        raise
            
            raise Exception(f"Max retries ({max_retries}) exceeded")
        
        return wrapper
    return decorator

def ensure_ollama_connection(self):
    """
    Ensures Ollama service is running and responsive
    Attempts to restart if necessary
    """
    try:
        health = self.monitor.check_ollama_health()
        
        if health['status'] != 'healthy':
            self.logger.warning("Ollama service unhealthy, attempting restart...")
            
            # Restart Ollama service (Linux/macOS)
            import subprocess
            subprocess.run(['ollama', 'serve'], check=True)
            
            # Wait for service to start
            time.sleep(10)
            
            # Verify restart
            health = self.monitor.check_ollama_health()
            if health['status'] == 'healthy':
                self.logger.info("Ollama service restored")
            else:
                raise Exception("Failed to restore Ollama service")
    
    except Exception as e:
        self.logger.error(f"Ollama connection error: {str(e)}")
        raise

class MultiCurrencyAnalyzer:
    """
    Analyzes economic events across multiple currencies
    Provides cross-currency impact assessment
    """
    
    def __init__(self, ollama_analyzer: OllamaEconomicAnalyzer):
        self.analyzer = ollama_analyzer
        self.supported_currencies = ['USD', 'EUR', 'GBP', 'JPY', 'CHF']
    
    def analyze_cross_currency_impact(self, event_data: Dict) -> Dict:
        """
        Analyzes how US economic events affect other currencies
        Provides forex trading insights
        """
        prompt = f"""
        Analyze how this US economic event impacts other major currencies:
        
        Event Data:
        {json.dumps(event_data, indent=2)}
        
        Provide cross-currency analysis for:
        1. EUR/USD impact and direction
        2. GBP/USD implications
        3. USD/JPY movement prediction
        4. Safe haven currency effects (CHF)
        5. Emerging market currency risks
        
        Include specific trading opportunities and risk assessments.
        Format as JSON with currency-specific insights.
        """
        
        response = self.analyzer.client.chat(
            model=self.analyzer.model_name,
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        try:
            return json.loads(response['message']['content'])
        except json.JSONDecodeError:
            return {'raw_analysis': response['message']['content']}