Stock Market News Impact Analysis: Ollama Real-Time Event Detection System

# Install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Pull the recommended model for financial analysis
ollama pull llama2:13b

# Install Python dependencies
pip install ollama requests pandas beautifulsoup4 yfinance schedule

import requests
from bs4 import BeautifulSoup
import json
from datetime import datetime
import time

class NewsCollector:
    def __init__(self):
        self.sources = [
            'https://finance.yahoo.com/news/',
            'https://www.marketwatch.com/latest-news',
            'https://www.cnbc.com/finance/'
        ]
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
    
    def fetch_articles(self, max_articles=50):
        """Fetch recent financial news articles"""
        articles = []
        
        for source in self.sources:
            try:
                response = requests.get(source, headers=self.headers, timeout=10)
                soup = BeautifulSoup(response.content, 'html.parser')
                
                # Extract article headlines and links
                # This is simplified - real implementation needs source-specific parsing
                headlines = soup.find_all('h3', class_='article-title')[:max_articles//len(self.sources)]
                
                for headline in headlines:
                    article_data = {
                        'title': headline.get_text().strip(),
                        'url': headline.find('a')['href'] if headline.find('a') else '',
                        'timestamp': datetime.now().isoformat(),
                        'source': source
                    }
                    articles.append(article_data)
                    
            except Exception as e:
                print(f"Error fetching from {source}: {e}")
                continue
                
        return articles
    
    def extract_full_content(self, article_url):
        """Extract full article content for detailed analysis"""
        try:
            response = requests.get(article_url, headers=self.headers, timeout=10)
            soup = BeautifulSoup(response.content, 'html.parser')
            
            # Remove ads, navigation, and other noise
            for element in soup(['nav', 'footer', 'aside', 'script', 'style']):
                element.decompose()
            
            # Extract main content (adapt based on news source structure)
            content = soup.find('div', class_='article-content') or soup.find('div', class_='story-body')
            return content.get_text().strip() if content else ""
            
        except Exception as e:
            print(f"Error extracting content from {article_url}: {e}")
            return ""

import ollama
import json
from typing import Dict, List

class MarketSentimentAnalyzer:
    def __init__(self, model_name='llama2:13b'):
        self.model = model_name
        self.client = ollama.Client()
        
    def analyze_market_impact(self, article_text: str, company_symbols: List[str] = None) -> Dict:
        """Analyze article for market sentiment and potential stock impact"""
        
        # Create focused prompt for financial analysis
        prompt = f"""
        Analyze this financial news article for market impact:
        
        Article: {article_text[:2000]}  # Limit for context window
        
        Provide analysis in JSON format:
        {{
            "overall_sentiment": "positive/negative/neutral",
            "confidence_score": 0.0-1.0,
            "market_impact": "high/medium/low",
            "affected_sectors": ["sector1", "sector2"],
            "key_events": ["event1", "event2"],
            "price_direction": "up/down/stable",
            "urgency": "immediate/short-term/long-term"
        }}
        
        Focus on actionable insights for traders.
        """
        
        try:
            response = self.client.generate(
                model=self.model,
                prompt=prompt,
                stream=False
            )
            
            # Parse the JSON response
            result = json.loads(response['response'])
            result['analysis_timestamp'] = datetime.now().isoformat()
            
            return result
            
        except Exception as e:
            print(f"Error analyzing sentiment: {e}")
            return self._default_analysis()
    
    def _default_analysis(self) -> Dict:
        """Return default analysis when processing fails"""
        return {
            "overall_sentiment": "neutral",
            "confidence_score": 0.0,
            "market_impact": "low",
            "affected_sectors": [],
            "key_events": [],
            "price_direction": "stable",
            "urgency": "long-term",
            "analysis_timestamp": datetime.now().isoformat()
        }
    
    def batch_analyze(self, articles: List[Dict]) -> List[Dict]:
        """Analyze multiple articles efficiently"""
        results = []
        
        for article in articles:
            # Combine title and content for analysis
            full_text = f"{article['title']} {article.get('content', '')}"
            
            analysis = self.analyze_market_impact(full_text)
            
            # Merge article data with analysis
            article_result = {**article, **analysis}
            results.append(article_result)
            
            # Rate limiting to prevent overload
            time.sleep(0.5)
            
        return results

import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta

class MarketCorrelator:
    def __init__(self):
        self.major_indices = ['^GSPC', '^DJI', '^IXIC', '^VIX']  # S&P 500, Dow, NASDAQ, VIX
        
    def get_current_market_state(self) -> Dict:
        """Get current market conditions for context"""
        market_data = {}
        
        for symbol in self.major_indices:
            try:
                ticker = yf.Ticker(symbol)
                hist = ticker.history(period='1d', interval='1m')
                
                if not hist.empty:
                    current_price = hist['Close'].iloc[-1]
                    open_price = hist['Open'].iloc[0]
                    change_pct = ((current_price - open_price) / open_price) * 100
                    
                    market_data[symbol] = {
                        'current_price': current_price,
                        'change_percent': change_pct,
                        'volume': hist['Volume'].sum(),
                        'volatility': hist['Close'].std()
                    }
                    
            except Exception as e:
                print(f"Error fetching data for {symbol}: {e}")
                
        return market_data
    
    def correlate_news_with_price_movement(self, analysis_results: List[Dict]) -> List[Dict]:
        """Correlate sentiment analysis with actual price movements"""
        
        current_market = self.get_current_market_state()
        
        for result in analysis_results:
            # Add market context to each analysis
            result['market_context'] = current_market
            
            # Calculate correlation score based on sentiment vs actual movement
            sentiment_score = self._sentiment_to_score(result['overall_sentiment'])
            market_movement = self._calculate_market_movement(current_market)
            
            correlation = self._calculate_correlation(sentiment_score, market_movement)
            result['correlation_score'] = correlation
            
        return analysis_results
    
    def _sentiment_to_score(self, sentiment: str) -> float:
        """Convert sentiment to numerical score"""
        sentiment_map = {
            'positive': 1.0,
            'neutral': 0.0,
            'negative': -1.0
        }
        return sentiment_map.get(sentiment, 0.0)
    
    def _calculate_market_movement(self, market_data: Dict) -> float:
        """Calculate overall market movement score"""
        if not market_data:
            return 0.0
            
        movements = [data['change_percent'] for data in market_data.values()]
        return sum(movements) / len(movements) / 100  # Normalize to -1 to 1 range
    
    def _calculate_correlation(self, sentiment_score: float, market_movement: float) -> float:
        """Calculate correlation between sentiment and market movement"""
        # Simple correlation: positive when sentiment and movement align
        return sentiment_score * market_movement

import schedule
import time
from datetime import datetime

class RealTimeMarketMonitor:
    def __init__(self):
        self.collector = NewsCollector()
        self.analyzer = MarketSentimentAnalyzer()
        self.correlator = MarketCorrelator()
        self.alert_threshold = 0.7  # High confidence alerts only
        
    def run_analysis_cycle(self):
        """Execute one complete analysis cycle"""
        print(f"Starting analysis cycle at {datetime.now()}")
        
        # Step 1: Collect latest news
        articles = self.collector.fetch_articles(max_articles=20)
        print(f"Collected {len(articles)} articles")
        
        # Step 2: Analyze sentiment
        analyzed_articles = self.analyzer.batch_analyze(articles)
        
        # Step 3: Correlate with market data
        correlated_results = self.correlator.correlate_news_with_price_movement(analyzed_articles)
        
        # Step 4: Generate alerts
        alerts = self.generate_alerts(correlated_results)
        
        # Step 5: Log results
        self.log_results(correlated_results, alerts)
        
        return correlated_results
    
    def generate_alerts(self, results: List[Dict]) -> List[Dict]:
        """Generate high-priority alerts based on analysis"""
        alerts = []
        
        for result in results:
            confidence = result.get('confidence_score', 0)
            impact = result.get('market_impact', 'low')
            urgency = result.get('urgency', 'long-term')
            
            # Alert criteria: high confidence + significant impact + immediate urgency
            if (confidence >= self.alert_threshold and 
                impact in ['high', 'medium'] and 
                urgency == 'immediate'):
                
                alert = {
                    'timestamp': datetime.now().isoformat(),
                    'title': result['title'],
                    'sentiment': result['overall_sentiment'],
                    'confidence': confidence,
                    'impact': impact,
                    'urgency': urgency,
                    'correlation': result.get('correlation_score', 0),
                    'recommendation': self._generate_recommendation(result)
                }
                alerts.append(alert)
                
        return alerts
    
    def _generate_recommendation(self, result: Dict) -> str:
        """Generate trading recommendation based on analysis"""
        sentiment = result['overall_sentiment']
        confidence = result['confidence_score']
        impact = result['market_impact']
        
        if confidence < 0.5:
            return "Monitor closely - low confidence signal"
        
        if sentiment == 'positive' and impact == 'high':
            return "Consider long positions - strong positive signal"
        elif sentiment == 'negative' and impact == 'high':
            return "Consider defensive positions - strong negative signal"
        else:
            return "Neutral - maintain current positions"
    
    def log_results(self, results: List[Dict], alerts: List[Dict]):
        """Log analysis results and alerts"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        # Save detailed results
        with open(f'analysis_results_{timestamp}.json', 'w') as f:
            json.dump(results, f, indent=2)
        
        # Save alerts separately
        if alerts:
            with open(f'alerts_{timestamp}.json', 'w') as f:
                json.dump(alerts, f, indent=2)
            
            print(f"Generated {len(alerts)} high-priority alerts")
            for alert in alerts:
                print(f"ALERT: {alert['title']} - {alert['recommendation']}")
    
    def start_monitoring(self):
        """Start the real-time monitoring system"""
        print("Starting real-time market monitoring...")
        
        # Schedule analysis every 5 minutes during market hours
        schedule.every(5).minutes.do(self.run_analysis_cycle)
        
        # Run initial analysis
        self.run_analysis_cycle()
        
        # Keep the monitor running
        while True:
            schedule.run_pending()
            time.sleep(60)  # Check every minute

class SectorAnalyzer:
    def __init__(self):
        self.sector_keywords = {
            'technology': ['AI', 'software', 'cloud', 'cybersecurity', 'semiconductor'],
            'healthcare': ['FDA', 'drug', 'vaccine', 'clinical trial', 'biotech'],
            'energy': ['oil', 'gas', 'renewable', 'pipeline', 'refinery'],
            'financial': ['Fed', 'interest rate', 'bank', 'credit', 'lending'],
            'retail': ['consumer', 'sales', 'holiday shopping', 'e-commerce']
        }
    
    def identify_affected_sectors(self, article_text: str) -> List[str]:
        """Identify which sectors are most likely affected by the news"""
        affected_sectors = []
        text_lower = article_text.lower()
        
        for sector, keywords in self.sector_keywords.items():
            keyword_count = sum(1 for keyword in keywords if keyword.lower() in text_lower)
            
            # Sector is affected if multiple keywords are present
            if keyword_count >= 2:
                affected_sectors.append(sector)
                
        return affected_sectors

def predict_volatility_impact(sentiment_score: float, current_vix: float, article_impact: str) -> Dict:
    """Predict potential volatility impact based on sentiment and current market conditions"""
    
    # Base volatility multiplier based on sentiment strength
    base_multiplier = abs(sentiment_score) * 1.5
    
    # Adjust based on current VIX level
    if current_vix > 30:  # High fear market
        vix_multiplier = 1.3
    elif current_vix < 15:  # Complacent market
        vix_multiplier = 1.1
    else:
        vix_multiplier = 1.0
    
    # Impact level adjustment
    impact_multipliers = {'high': 2.0, 'medium': 1.5, 'low': 1.0}
    impact_mult = impact_multipliers.get(article_impact, 1.0)
    
    predicted_volatility_change = base_multiplier * vix_multiplier * impact_mult
    
    return {
        'predicted_vix_change': predicted_volatility_change,
        'confidence': min(abs(sentiment_score) * 100, 95),
        'risk_level': 'high' if predicted_volatility_change > 2.0 else 'medium' if predicted_volatility_change > 1.0 else 'low'
    }

import gc
from concurrent.futures import ThreadPoolExecutor, as_completed

class OptimizedAnalyzer:
    def __init__(self, max_workers=4):
        self.max_workers = max_workers
        
    def parallel_analysis(self, articles: List[Dict]) -> List[Dict]:
        """Process articles in parallel for better performance"""
        results = []
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # Submit analysis tasks
            future_to_article = {
                executor.submit(self.analyzer.analyze_market_impact, article['content']): article 
                for article in articles
            }
            
            # Collect results as they complete
            for future in as_completed(future_to_article):
                article = future_to_article[future]
                try:
                    analysis = future.result()
                    results.append({**article, **analysis})
                except Exception as e:
                    print(f"Analysis failed for article: {e}")
                    
        # Clean up memory
        gc.collect()
        return results

class TradingIntegration:
    def __init__(self, platform='alpaca'):  # Alpaca, Interactive Brokers, etc.
        self.platform = platform
        self.api_key = os.getenv('TRADING_API_KEY')
        
    def send_signal(self, symbol: str, action: str, confidence: float):
        """Send trading signal based on analysis results"""
        if confidence < 0.8:
            return  # Only act on high-confidence signals
            
        signal_data = {
            'symbol': symbol,
            'action': action,  # 'buy', 'sell', 'hold'
            'confidence': confidence,
            'timestamp': datetime.now().isoformat(),
            'source': 'ollama_news_analysis'
        }
        
        # Platform-specific implementation
        if self.platform == 'alpaca':
            self._send_alpaca_signal(signal_data)
        elif self.platform == 'ib':
            self._send_ib_signal(signal_data)

from sklearn.ensemble import RandomForestClassifier
import numpy as np

class HybridPredictor:
    def __init__(self):
        self.sentiment_model = MarketSentimentAnalyzer()
        self.ml_model = RandomForestClassifier(n_estimators=100)
        
    def train_price_predictor(self, historical_data: List[Dict]):
        """Train ML model on historical sentiment vs price movement data"""
        features = []
        targets = []
        
        for data_point in historical_data:
            # Extract features from sentiment analysis
            feature_vector = [
                data_point['confidence_score'],
                1 if data_point['overall_sentiment'] == 'positive' else -1 if data_point['overall_sentiment'] == 'negative' else 0,
                data_point.get('correlation_score', 0),
                len(data_point.get('affected_sectors', [])),
                1 if data_point['urgency'] == 'immediate' else 0
            ]
            features.append(feature_vector)
            
            # Target: actual price movement direction
            targets.append(data_point['actual_price_movement'])
            
        self.ml_model.fit(np.array(features), np.array(targets))
    
    def predict_with_confidence(self, sentiment_analysis: Dict) -> Dict:
        """Combine Ollama sentiment with ML prediction"""
        feature_vector = [
            sentiment_analysis['confidence_score'],
            1 if sentiment_analysis['overall_sentiment'] == 'positive' else -1 if sentiment_analysis['overall_sentiment'] == 'negative' else 0,
            sentiment_analysis.get('correlation_score', 0),
            len(sentiment_analysis.get('affected_sectors', [])),
            1 if sentiment_analysis['urgency'] == 'immediate' else 0
        ]
        
        ml_prediction = self.ml_model.predict([feature_vector])[0]
        ml_confidence = self.ml_model.predict_proba([feature_vector]).max()
        
        return {
            'hybrid_prediction': ml_prediction,
            'ml_confidence': ml_confidence,
            'sentiment_confidence': sentiment_analysis['confidence_score'],
            'combined_confidence': (ml_confidence + sentiment_analysis['confidence_score']) / 2
        }

class GlobalNewsAnalyzer:
    def __init__(self):
        self.language_models = {
            'english': 'llama2:13b',
            'spanish': 'llama2:13b',  # Many models support multiple languages
            'chinese': 'qwen:14b',    # Specialized Chinese model
            'japanese': 'elyza:7b'    # Specialized Japanese model
        }
    
    def detect_language(self, text: str) -> str:
        """Detect article language for appropriate model selection"""
        # Implement language detection logic
        # Could use langdetect library or Ollama itself
        pass
    
    def analyze_multilingual_sentiment(self, article: Dict) -> Dict:
        """Analyze sentiment using appropriate language model"""
        language = self.detect_language(article['content'])
        model = self.language_models.get(language, 'llama2:13b')
        
        # Use appropriate model for analysis
        analyzer = MarketSentimentAnalyzer(model_name=model)
        return analyzer.analyze_market_impact(article['content'])