Upload sentiment_model.py

sentiment_model.py

@@ -0,0 +1,197 @@
+"""News + Sentiment Alpha Model using FinBERT."""
+import numpy as np
+import pandas as pd
+import torch
+from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline
+from typing import List, Dict, Optional
+import warnings
+warnings.filterwarnings('ignore')
+
+
+class SentimentAlphaModel:
+    """Financial sentiment analysis using FinBERT"""
+    
+    def __init__(self, model_name: str = "ProsusAI/finbert", 
+                 device: str = 'cpu', max_length: int = 512):
+        self.model_name = model_name
+        self.device = device
+        self.max_length = max_length
+        
+        print(f"Loading FinBERT model: {model_name}")
+        try:
+            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
+            self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
+            self.model.to(device)
+            self.model.eval()
+            self.pipeline = pipeline(
+                "sentiment-analysis",
+                model=self.model,
+                tokenizer=self.tokenizer,
+                device=0 if device == 'cuda' else -1
+            )
+            self.is_loaded = True
+        except Exception as e:
+            print(f"Error loading FinBERT: {e}")
+            self.is_loaded = False
+    
+    def analyze_text(self, text: str) -> Dict:
+        """Analyze sentiment of a single text"""
+        if not self.is_loaded:
+            return {'label': 'neutral', 'score': 0.5, 'sentiment_score': 0.0}
+        
+        try:
+            result = self.pipeline(text[:self.max_length])[0]
+            label = result['label'].lower()
+            score = result['score']
+            
+            # Convert to numeric sentiment score (-1 to 1)
+            if label == 'positive':
+                sentiment_score = score
+            elif label == 'negative':
+                sentiment_score = -score
+            else:
+                sentiment_score = 0.0
+            
+            return {
+                'label': label,
+                'score': score,
+                'sentiment_score': sentiment_score
+            }
+        except Exception as e:
+            print(f"Error analyzing text: {e}")
+            return {'label': 'neutral', 'score': 0.5, 'sentiment_score': 0.0}
+    
+    def analyze_batch(self, texts: List[str], batch_size: int = 32) -> List[Dict]:
+        """Analyze sentiment for a batch of texts"""
+        if not self.is_loaded:
+            return [{'label': 'neutral', 'score': 0.5, 'sentiment_score': 0.0} for _ in texts]
+        
+        results = []
+        for i in range(0, len(texts), batch_size):
+            batch = texts[i:i+batch_size]
+            try:
+                batch_results = self.pipeline(batch)
+                for res in batch_results:
+                    label = res['label'].lower()
+                    score = res['score']
+                    if label == 'positive':
+                        sentiment_score = score
+                    elif label == 'negative':
+                        sentiment_score = -score
+                    else:
+                        sentiment_score = 0.0
+                    results.append({
+                        'label': label,
+                        'score': score,
+                        'sentiment_score': sentiment_score
+                    })
+            except Exception as e:
+                print(f"Error in batch: {e}")
+                for _ in batch:
+                    results.append({'label': 'neutral', 'score': 0.5, 'sentiment_score': 0.0})
+        
+        return results
+    
+    def generate_sentiment_alpha(self, news_data: pd.DataFrame, 
+                                  ticker_col: str = 'ticker',
+                                  text_col: str = 'text',
+                                  date_col: str = 'date',
+                                  window: int = 5) -> pd.DataFrame:
+        """
+        Generate daily sentiment alpha scores per asset
+        
+        news_data: DataFrame with columns [date, ticker, text]
+        Returns: DataFrame with [date, ticker, sentiment_alpha]
+        """
+        if not self.is_loaded:
+            print("FinBERT not loaded, returning zeros")
+            return pd.DataFrame({
+                'date': news_data[date_col].unique(),
+                'sentiment_alpha': 0.0
+            })
+        
+        print(f"Analyzing sentiment for {len(news_data)} news items...")
+        
+        # Analyze all texts
+        texts = news_data[text_col].tolist()
+        sentiments = self.analyze_batch(texts)
+        
+        news_data = news_data.copy()
+        news_data['sentiment_score'] = [s['sentiment_score'] for s in sentiments]
+        news_data['sentiment_magnitude'] = [abs(s['sentiment_score']) for s in sentiments]
+        
+        # Aggregate by ticker and date
+        daily_sentiment = news_data.groupby([date_col, ticker_col]).agg({
+            'sentiment_score': ['mean', 'std', 'count'],
+            'sentiment_magnitude': 'mean'
+        }).reset_index()
+        
+        daily_sentiment.columns = [date_col, ticker_col, 'sentiment_mean', 
+                                    'sentiment_std', 'sentiment_count', 'sentiment_magnitude']
+        
+        # Apply confidence weighting (more articles = more confident)
+        daily_sentiment['confidence'] = np.minimum(daily_sentiment['sentiment_count'] / 5, 1.0)
+        daily_sentiment['sentiment_alpha'] = (
+            daily_sentiment['sentiment_mean'] * daily_sentiment['confidence']
+        )
+        
+        # Rolling window smoothing
+        daily_sentiment = daily_sentiment.sort_values([ticker_col, date_col])
+        daily_sentiment['sentiment_alpha_smooth'] = daily_sentiment.groupby(ticker_col)[
+            'sentiment_alpha'
+        ].transform(lambda x: x.rolling(window, min_periods=1).mean())
+        
+        return daily_sentiment[[date_col, ticker_col, 'sentiment_alpha_smooth', 
+                                 'sentiment_count', 'confidence']]
+    
+    def generate_synthetic_news(self, tickers: List[str], 
+                                 dates: pd.DatetimeIndex,
+                                 n_news_per_day: int = 3) -> pd.DataFrame:
+        """Generate synthetic financial news for testing"""
+        np.random.seed(42)
+        
+        templates_positive = [
+            "{ticker} reports strong quarterly earnings, beating analyst expectations",
+            "{ticker} announces new product launch, stock rises in pre-market",
+            "Analysts upgrade {ticker} to buy rating, price target raised",
+            "{ticker} secures major contract, revenue outlook improved",
+            "{ticker} demonstrates strong growth in emerging markets"
+        ]
+        
+        templates_negative = [
+            "{ticker} misses earnings expectations, stock falls sharply",
+            "{ticker} faces regulatory scrutiny, shares decline",
+            "Analysts downgrade {ticker} amid slowing growth concerns",
+            "{ticker} announces layoffs as part of restructuring plan",
+            "Supply chain issues impact {ticker} quarterly guidance"
+        ]
+        
+        templates_neutral = [
+            "{ticker} maintains dividend policy, no changes expected",
+            "{ticker} announces board restructuring, effective next quarter",
+            "Market awaits {ticker} earnings report due next week",
+            "{ticker} trading volume remains within normal range",
+            "Analysts maintain hold rating on {ticker}"
+        ]
+        
+        news_items = []
+        for date in dates:
+            for ticker in tickers:
+                for _ in range(n_news_per_day):
+                    sentiment_type = np.random.choice(['pos', 'neg', 'neu'], 
+                                                       p=[0.35, 0.35, 0.3])
+                    if sentiment_type == 'pos':
+                        text = np.random.choice(templates_positive).format(ticker=ticker)
+                    elif sentiment_type == 'neg':
+                        text = np.random.choice(templates_negative).format(ticker=ticker)
+                    else:
+                        text = np.random.choice(templates_neutral).format(ticker=ticker)
+                    
+                    news_items.append({
+                        'date': date,
+                        'ticker': ticker,
+                        'text': text,
+                        'source': 'synthetic'
+                    })
+        
+        return pd.DataFrame(news_items)