#!/usr/bin/env python3 # =========================================== # Alpha Sentiment Engine — Live Scraper V5 # =========================================== # THREE-PHASE HYBRID ARCHITECTURE: # Phase A — Sector news scan (20 sectors) # Phase B — Direct company news (top 200) # Phase C — Hybrid merge (1000+ companies) # # FEATURES: # - Entity validation on direct news # - Headline deduplication # - Stock price overlay via yfinance # - Confidence levels (HIGH / MEDIUM / LOW) # - Score types (DIRECT / HYBRID / SECTOR) # - Multi-source: Google News + Yahoo Finance # =========================================== import time import json import hashlib import feedparser import requests import yfinance as yf from urllib.parse import quote from datetime import datetime, timezone from concurrent.futures import ThreadPoolExecutor, as_completed from colorama import Fore, Style, init from services.sentiment_service import SentimentService from services.database import save_score, save_average, get_stats from domain.models import NewsItem init(autoreset=True) # ─────────────────────────────────────────── # Settings # ─────────────────────────────────────────── SCRAPE_INTERVAL: int = 600 # 10 minutes (Improved from 30 mins) HEADLINES_PER_SEARCH: int = 8 STOCKS_FILE = "data/indian_stocks.json" # How many top companies get direct news search DIRECT_SCAN_COUNT = 200 # How many total companies to score (via hybrid merge) TOTAL_COMPANY_TARGET = 1000 # Sector news search queries SECTOR_QUERIES = { "Banking": "Indian banking sector news stock market", "IT": "Indian IT sector technology stocks news", "Pharma": "Indian pharma healthcare stocks news", "Auto": "Indian auto sector automobile stocks news", "FMCG": "Indian FMCG consumer goods stocks news", "Energy": "Indian energy power oil gas stocks news", "Metal": "Indian metal steel mining stocks news", "Telecom": "Indian telecom sector stocks news", "Real Estate": "Indian real estate construction cement stocks news", "Chemicals": "Indian chemicals paints sector stocks news", "Textiles": "Indian textile apparel sector stocks news", "Media": "Indian media entertainment sector stocks news", "Insurance": "Indian insurance sector stocks news", "Agriculture": "Indian agriculture sugar fertilizer stocks news", "Logistics": "Indian logistics transport aviation stocks news", "Retail": "Indian retail e-commerce stocks news", "Defence": "Indian defence aerospace shipbuilding stocks news", "Consumer Durables": "Indian consumer durables electronics stocks news", "Infrastructure": "Indian infrastructure stocks news", "Hotels": "Indian hotel hospitality tourism stocks news", } # Top 200 companies for direct news scanning (Nifty 50 + Next 50 + Midcap 100) TOP_COMPANIES = [ # Nifty 50 "RELIANCE", "TCS", "HDFCBANK", "ICICIBANK", "INFY", "HINDUNILVR", "ITC", "SBIN", "BHARTIARTL", "KOTAKBANK", "LT", "AXISBANK", "BAJFINANCE", "MARUTI", "HCLTECH", "ASIANPAINT", "SUNPHARMA", "TITAN", "WIPRO", "ONGC", "BAJAJFINSV", "ULTRACEMCO", "NTPC", "TATAMOTORS", "POWERGRID", "NESTLEIND", "JSWSTEEL", "TATASTEEL", "ADANIPORTS", "ADANIENT", "TECHM", "M&M", "COALINDIA", "INDUSINDBK", "HINDALCO", "SBILIFE", "GRASIM", "HDFCLIFE", "CIPLA", "DIVISLAB", "DRREDDY", "EICHERMOT", "BPCL", "APOLLOHOSP", "HEROMOTOCO", "BAJAJ-AUTO", "TATACONSUM", "BRITANNIA", "SHRIRAMFIN", "LTIM", # Nifty Next 50 "ADANIGREEN", "ADANIPOWER", "AMBUJACEM", "BANKBARODA", "BERGEPAINT", "BOSCHLTD", "CANBK", "CHOLAFIN", "COLPAL", "DABUR", "DLF", "GAIL", "GODREJCP", "HAVELLS", "ICICIPRULI", "IDFCFIRSTB", "INDIGO", "IOC", "IRCTC", "JINDALSTEL", "LUPIN", "MAXHEALTH", "MOTHERSON", "NAUKRI", "OBEROIRLTY", "OFSS", "PAGEIND", "PEL", "PIDILITIND", "PNB", "SAIL", "SRF", "TATAPOWER", "TORNTPHARM", "TRENT", "UNIONBANK", "VEDL", "YESBANK", "ZOMATO", "PAYTM", # Midcap 100 extras "DMART", "POLICYBZR", "NYKAA", "PB", "IRFC", "IDEA", "RECLTD", "NHPC", "HAL", "BEL", "TATAELXSI", "PERSISTENT", "COFORGE", "MPHASIS", "LTTS", "PVRINOX", "PATANJALI", "MAZDOCK", "COCHINSHIP", "GRSE", "BIOCON", "AUROPHARMA", "ALKEM", "LALPATHLAB", "MARICO", "ASHOKLEY", "TVSMOTORS", "EXIDEIND", "AMARARAJA", "MUTHOOTFIN", "MANAPPURAM", "VOLTAS", "CROMPTON", "DIXON", "NMDC", "NATIONALUM", "LICHSGFIN", "CANFINHOME", "SHREECEM", "ACC", "GODREJPROP", "EMAMILTD", "HAPPSTMNDS", "MASTEK", "SONATA", "ABCAPITAL", "IIFL", "FEDERALBNK", "RBLBANK", "BANDHANBNK", "JUBLFOOD", "MCDOWELL-N", "CONCOR", "SUNTV", "NETWORK18", "DEEPAKNTR", "ATUL", "PIIND", "ASTRAL", "SUPREMEIND", "APLAPOLLO", "KENNAMET", "CUMMINSIND", "SIEMENS", "ABB", "HONAUT", "BHARATFORG", "SUNDRMFAST", "MFSL", "NAM-INDIA", "IPCALAB", "NATCOPHARMA", "GLENMARK", "LAURUSLABS", "SYNGENE", "GODREJIND", "UBL", "TATACOMM", "STARHEALTH", "KPITTECH", ] # ─────────────────────────────────────────── # Load stock universe # ─────────────────────────────────────────── def load_stocks(): print("📥 Loading Indian stock database...") try: with open(STOCKS_FILE, "r") as f: data = json.load(f) return data["stocks"], data.get("sectors", {}) except Exception as e: print(f"❌ Could not load {STOCKS_FILE}. {e}") return [], {} # ─────────────────────────────────────────── # News Sources # ─────────────────────────────────────────── def fetch_google_news(query: str, count: int = 8) -> list[dict]: """Fetch headlines from Google News RSS.""" encoded_query = quote(query) url = f"https://news.google.com/rss/search?q={encoded_query}&hl=en-IN&gl=IN&ceid=IN:en" try: feed = feedparser.parse(url) results = [] for entry in feed.entries[:count]: title = entry.get("title", "").split(" - ")[0] source = entry.get("source", {}).get("title", "News") if hasattr(entry, "source") else "News" if title: results.append({"title": title, "source": f"📰 {source}"}) return results except Exception: return [] def fetch_yahoo_finance_news(query: str, count: int = 5) -> list[dict]: """Fetch headlines from Yahoo Finance RSS as a secondary source.""" encoded_query = quote(query) url = f"https://news.google.com/rss/search?q={encoded_query}+share+price+NSE+BSE&hl=en-IN&gl=IN&ceid=IN:en" try: feed = feedparser.parse(url) results = [] for entry in feed.entries[:count]: title = entry.get("title", "").split(" - ")[0] if title: results.append({"title": title, "source": "📈 Finance"}) return results except Exception: return [] # ─────────────────────────────────────────── # Entity Validation # ─────────────────────────────────────────── def validate_headline(headline: str, company_name: str, ticker: str) -> bool: """Check if a headline is ACTUALLY about this specific company.""" headline_lower = headline.lower() clean_name = company_name.lower() for suffix in [" limited", " ltd", " ltd.", " corporation", " corp", " inc"]: clean_name = clean_name.replace(suffix, "").strip() name_parts = clean_name.split() if ticker.lower() in headline_lower: return True if len(clean_name) > 3 and clean_name in headline_lower: return True if len(name_parts) >= 1 and len(name_parts[0]) > 4: if name_parts[0] in headline_lower: return True if len(name_parts) >= 2: two_word = f"{name_parts[0]} {name_parts[1]}" if two_word in headline_lower: return True return False # ─────────────────────────────────────────── # Headline Deduplication # ─────────────────────────────────────────── def headline_hash(text: str) -> str: """Create a short hash of a headline for dedup.""" return hashlib.md5(text.lower().strip().encode()).hexdigest()[:12] # ─────────────────────────────────────────── # Stock Price Fetcher (yfinance) # ─────────────────────────────────────────── def get_price_change(yahoo_ticker: str) -> float | None: """Get today's stock price % change using yfinance.""" try: # Final Polish: Rate limiting to prevent IP blocks time.sleep(0.7) stock = yf.Ticker(yahoo_ticker) hist = stock.history(period="2d") if len(hist) >= 2: prev_close = hist["Close"].iloc[-2] curr_close = hist["Close"].iloc[-1] pct = round(((curr_close - prev_close) / prev_close) * 100, 2) return pct except Exception: pass return None # ─────────────────────────────────────────── # Scoring Helpers # ─────────────────────────────────────────── def score_headline(service: SentimentService, headline: str) -> float: item = NewsItem(ticker="DUMMY", headline=headline) result = service.analyze(item) return result.sentiment_score def get_sentiment_color(score: float) -> str: if score > 0.3: return Fore.GREEN elif score < -0.3: return Fore.RED else: return Fore.YELLOW def get_sentiment_label(score: float) -> str: if score > 0.5: return "🟢 V.Bullish" elif score > 0.3: return "🟢 Bullish" elif score > -0.3: return "🟡 Neutral" elif score > -0.5: return "🔴 Bearish" else: return "🔴 V.Bearish" def get_confidence(num_headlines: int, score_type: str) -> str: if score_type == "SECTOR": return "LOW" if num_headlines >= 3: return "HIGH" elif num_headlines >= 1: return "MEDIUM" return "LOW" # ─────────────────────────────────────────── # MAIN: Three-Phase Scrape Cycle (V5) # ─────────────────────────────────────────── def run_scrape_cycle(service: SentimentService, stocks: list[dict], sectors_map: dict) -> None: now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") start_time = time.time() # Build lookup dictionaries stock_by_symbol = {s["symbol"]: s for s in stocks} seen_headlines = set() # For deduplication print("\n" + "=" * 80) print(f" 📡 LIVE SCRAPE V5 — {now}") print(f" 🧠 Three-Phase Hybrid Architecture") print(f" 📊 Phase A: {len(SECTOR_QUERIES)} sectors | Phase B: {len(TOP_COMPANIES)} direct | Phase C: {TOTAL_COMPANY_TARGET}+ hybrid") print("=" * 80) # ══════════════════════════════════════════ # PHASE A: Sector News Scan # ══════════════════════════════════════════ print(f"\n ═══ PHASE A: Scanning {len(SECTOR_QUERIES)} Sector News Feeds ═══") sector_scores = {} for sector_name, query in SECTOR_QUERIES.items(): headlines = fetch_google_news(query, HEADLINES_PER_SEARCH) if not headlines: continue scores = [] for article in headlines: h = headline_hash(article["title"]) if h in seen_headlines: continue seen_headlines.add(h) score = score_headline(service, article["title"]) scores.append(score) save_score(f"SECTOR_{sector_name}", article["title"], score, article["source"]) if scores: avg = sum(scores) / len(scores) sector_scores[sector_name] = {"score": avg, "headlines": len(scores)} color = get_sentiment_color(avg) print(f" {sector_name:<22} {color}{avg:+.4f}{Style.RESET_ALL} ({len(scores)} headlines) {get_sentiment_label(avg)}") print(f" ✅ Phase A complete: {len(sector_scores)} sectors scored") # ══════════════════════════════════════════ # PHASE B: Direct Company News Scan # ══════════════════════════════════════════ print(f"\n ═══ PHASE B: Direct News for {len(TOP_COMPANIES)} Companies ═══") direct_scores = {} total_headlines_found = 0 total_validated = 0 total_rejected = 0 for i, symbol in enumerate(TOP_COMPANIES): stock_obj = stock_by_symbol.get(symbol) if not stock_obj: continue company_name = stock_obj["name"] # Multi-source: Google News + Yahoo Finance queries headlines = fetch_google_news(company_name, HEADLINES_PER_SEARCH) if not headlines: continue validated_scores = [] for article in headlines: total_headlines_found += 1 title = article["title"] # Dedup check h = headline_hash(title) if h in seen_headlines: continue seen_headlines.add(h) # Entity validation is_valid = validate_headline(title, company_name, symbol) if is_valid: score = score_headline(service, title) validated_scores.append(score) save_score(symbol, title, score, article["source"], validated=True) total_validated += 1 else: total_rejected += 1 if validated_scores: avg_score = sum(validated_scores) / len(validated_scores) direct_scores[symbol] = { "score": avg_score, "headlines": len(validated_scores), "name": company_name, } color = get_sentiment_color(avg_score) print(f" {symbol:<15} {color}{avg_score:+.4f}{Style.RESET_ALL} " f"({len(validated_scores)}/{len(headlines)} validated) " f"{get_sentiment_label(avg_score)}") # Rate limiting if (i + 1) % 25 == 0: print(f" ... {i + 1}/{len(TOP_COMPANIES)} companies scanned ...") time.sleep(1) print(f" ✅ Phase B complete: {len(direct_scores)} companies with direct news") print(f" Headlines: {total_headlines_found} found | {total_validated} validated | {total_rejected} rejected") # ══════════════════════════════════════════ # PHASE C: Hybrid Merge + Price Overlay # ══════════════════════════════════════════ print(f"\n ═══ PHASE C: Hybrid Merge + Price Overlay ═══") # Determine which stocks to score (up to TOTAL_COMPANY_TARGET) all_symbols = list(stock_by_symbol.keys())[:TOTAL_COMPANY_TARGET] # Fetch prices for direct-scored companies in parallel print(f" 📈 Fetching stock prices for {len(direct_scores)} companies...") price_data = {} def fetch_price(symbol, yahoo_ticker): return symbol, get_price_change(yahoo_ticker) with ThreadPoolExecutor(max_workers=5) as executor: futures = {} for sym in direct_scores: stock_obj = stock_by_symbol.get(sym) if stock_obj: yahoo_ticker = stock_obj.get("yahoo_ticker", f"{sym}.NS") futures[executor.submit(fetch_price, sym, yahoo_ticker)] = sym for future in as_completed(futures): try: sym, pct = future.result() if pct is not None: price_data[sym] = pct except Exception: pass print(f" ✅ Got price data for {len(price_data)} companies") # Merge and save final_scored = {} direct_count = 0 hybrid_count = 0 sector_only_count = 0 for symbol in all_symbols: stock_obj = stock_by_symbol.get(symbol) if not stock_obj: continue sector = stock_obj.get("sector", "General") sector_score_data = sector_scores.get(sector) direct_data = direct_scores.get(symbol) price_pct = price_data.get(symbol) if direct_data and sector_score_data: # HYBRID: 70% direct + 30% sector final_score = 0.7 * direct_data["score"] + 0.3 * sector_score_data["score"] num_headlines = direct_data["headlines"] score_type = "HYBRID" confidence = get_confidence(num_headlines, score_type) hybrid_count += 1 elif direct_data: # DIRECT only (sector has no score) final_score = direct_data["score"] num_headlines = direct_data["headlines"] score_type = "DIRECT" confidence = get_confidence(num_headlines, score_type) direct_count += 1 elif sector_score_data: # SECTOR only (no direct news for this company) final_score = sector_score_data["score"] num_headlines = sector_score_data["headlines"] score_type = "SECTOR" confidence = "LOW" sector_only_count += 1 else: # No data at all (sector is "General" with no news) continue save_average( ticker=symbol, average_score=round(final_score, 6), num_headlines=num_headlines, confidence=confidence, price_change=price_pct, score_type=score_type, ) final_scored[symbol] = { "score": final_score, "type": score_type, "confidence": confidence, "price": price_pct, "name": stock_obj["name"], } elapsed = time.time() - start_time # ─── SUMMARY ─── print("\n" + "=" * 80) print(" 📊 V5 CYCLE COMPLETE: Three-Phase Hybrid Scoring") print("=" * 80) print(f" ✅ Total companies scored: {len(final_scored)}") print(f" 🎯 DIRECT: {direct_count} (company-specific news)") print(f" 📊 HYBRID: {hybrid_count} (direct + sector combined)") print(f" 🏷️ SECTOR: {sector_only_count} (sector news only)") print(f" 📰 Total unique headlines: {len(seen_headlines)}") print(f" 📈 Price data: {len(price_data)} companies") # Top movers top_bullish = sorted( [(s, d) for s, d in final_scored.items() if d["score"] > 0.3], key=lambda x: -x[1]["score"] )[:10] top_bearish = sorted( [(s, d) for s, d in final_scored.items() if d["score"] < -0.3], key=lambda x: x[1]["score"] )[:10] print("\n 🟢 TOP BULLISH:") for symbol, data in top_bullish: price_str = f" 📈 {data['price']:+.2f}%" if data["price"] is not None else "" type_icon = "🎯" if data["type"] == "DIRECT" else ("📊" if data["type"] == "HYBRID" else "🏷️") print(f" {symbol:<12} {Fore.GREEN}{data['score']:+.4f}{Style.RESET_ALL} " f"[{data['confidence']}] {type_icon}{price_str} {data['name'][:30]}") print("\n 🔴 TOP BEARISH:") for symbol, data in top_bearish: price_str = f" 📉 {data['price']:+.2f}%" if data["price"] is not None else "" type_icon = "🎯" if data["type"] == "DIRECT" else ("📊" if data["type"] == "HYBRID" else "🏷️") print(f" {symbol:<12} {Fore.RED}{data['score']:+.4f}{Style.RESET_ALL} " f"[{data['confidence']}] {type_icon}{price_str} {data['name'][:30]}") print("\n" + "=" * 80) stats = get_stats() print(f" 💾 DB Totals: {stats['total_scores']} headlines | {stats['total_averages']} averages") print(f" ⏱️ Cycle completed in {elapsed:.1f} seconds") print("=" * 80) def run_sync_loop(service=None): """ Run the scraper in a continuous loop. Can be called from a background thread in app.py. """ print("\n" + "=" * 80) print(" 🚀 SENTIX BACKGROUND SYNC ENGINE — ACTIVE") print("=" * 80) stocks, sectors = load_stocks() if not stocks: return if service is None: print("⏳ Loading FinBERT model for background worker...") service = SentimentService() print("✅ AI loaded and ready!") while True: try: print(f"\n[{datetime.now().strftime('%H:%M:%S')}] 🚀 Scrape Cycle Started ({len(stocks)} symbols)") run_scrape_cycle(service, stocks, sectors) print(f"\n💤 Cycle complete. Sleeping for {SCRAPE_INTERVAL // 60} minutes...") except Exception as e: print(f"❌ FATAL ERROR IN SYNC CYCLE: {str(e)}") import traceback traceback.print_exc() time.sleep(SCRAPE_INTERVAL) def main(): print("=" * 80) print(" 🚀 ALPHA SENTIMENT ENGINE — V5 (1000+ Companies, Hybrid Scoring)") print("=" * 80) service = SentimentService() run_sync_loop(service) if __name__ == "__main__": main()