portiq-backend / modules /sentiment.py
ramkumar-bindrix
fix: make streamlit imports conditional for backend deployment
60c2417
Raw
History Blame Contribute Delete
4.18 kB
"""
Module for sentiment analysis using FinBERT Indian news models.
First run downloads the model (~400MB). Subsequent runs use cached version.
Run on CPU β€” no GPU required for this model size.
"""
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
# Using the industry standard ProsusAI/finbert because kdave/FineTuned_Finbert
# is missing model weights on the Hugging Face hub.
MODEL_NAME = "ProsusAI/finbert"
try:
import streamlit as st
cache_decorator = st.cache_resource(show_spinner="Loading FinBERT Model...")
except ImportError:
from functools import lru_cache
cache_decorator = lru_cache(maxsize=1)
@cache_decorator
def load_sentiment_model():
"""
Loads the tokenizer and model from Hugging Face.
Caches the model after the first load to avoid reloading.
"""
print(f"Loading sentiment model '{MODEL_NAME}'... (This may take a moment on first run)")
_tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
_model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME)
_model.eval() # Set to evaluation mode
_model.to("cpu")
print("Sentiment model loaded successfully!")
return _tokenizer, _model
def score_headline(headline_text):
"""
Scores a single news headline string.
Returns dict: {"label": "positive/negative/neutral", "score": confidence_score}
Handles empty or None input gracefully.
"""
if not headline_text or not str(headline_text).strip():
return {"label": "neutral", "score": 0.0}
tokenizer, model = load_sentiment_model()
inputs = tokenizer(headline_text, return_tensors="pt", truncation=True, padding=True, max_length=512)
with torch.no_grad():
outputs = model(**inputs)
probs = torch.nn.functional.softmax(outputs.logits, dim=-1)
top_prob, top_idx = torch.max(probs, dim=-1)
top_prob = top_prob.item()
top_idx = top_idx.item()
raw_label = model.config.id2label.get(top_idx, "neutral").lower()
if "pos" in raw_label:
label = "positive"
elif "neg" in raw_label:
label = "negative"
else:
label = "neutral"
return {"label": label, "score": top_prob}
def score_portfolio_news(news_list):
"""
Scores a list of news dictionaries.
Input format: [{"symbol": "TCS", "headline": "TCS beats Q4 estimates"}]
Returns enriched list with sentiment and score added.
"""
if not news_list:
return []
enriched_news = []
for item in news_list:
headline = item.get("headline", "")
sentiment_data = score_headline(headline)
enriched_item = item.copy()
enriched_item["sentiment"] = sentiment_data["label"]
enriched_item["score"] = round(sentiment_data["score"], 4)
enriched_news.append(enriched_item)
return enriched_news
def get_sentiment_emoji(label):
"""
Returns emoji for UI based on sentiment label.
"""
lbl = str(label).lower()
if "pos" in lbl:
return "🟒"
elif "neg" in lbl:
return "πŸ”΄"
return "🟑"
if __name__ == "__main__":
import sys
if sys.platform == 'win32':
sys.stdout.reconfigure(encoding='utf-8')
sample_news = [
{"symbol": "RELIANCE", "headline": "Reliance Industries announces massive β‚Ή75,000 crore investment in green energy, stock surges."},
{"symbol": "TCS", "headline": "TCS Q4 margins contract due to wage hikes and macroeconomic headwinds in the US."},
{"symbol": "HDFCBANK", "headline": "HDFC Bank holds interest rates steady, analysts predict stable growth for the quarter."}
]
print("Testing FinBERT Sentiment Module...\n")
results = score_portfolio_news(sample_news)
for res in results:
symbol = res["symbol"]
headline = res["headline"]
sentiment = res["sentiment"]
score = res["score"]
emoji = get_sentiment_emoji(sentiment)
print(f"[{symbol}] {headline}")
print(f"--> Sentiment: {emoji} {sentiment.upper()} (Confidence: {score:.2f})\n")