import os import re import numpy as np import torch import yfinance as yf import gradio as gr import spaces from dataclasses import dataclass, field from typing import Optional from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig from peft import PeftModel # ============================================================ # CONFIG # ============================================================ HF_TOKEN = os.environ.get("HF_TOKEN", "") MODEL_REPO = "shravankotagi/finbot-sarvam-hinglish" BASE_MODEL = "sarvamai/sarvam-1" # ============================================================ # LAZY MODEL LOAD - loads on first GPU call # ============================================================ model = None tokenizer = None def load_model(): global model, tokenizer if model is not None: return print("Loading tokenizer...") tokenizer = AutoTokenizer.from_pretrained(MODEL_REPO, token=HF_TOKEN) tokenizer.pad_token = tokenizer.eos_token print("Tokenizer loaded ✓") print("Loading base model...") bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.float16, bnb_4bit_use_double_quant=True, ) base = AutoModelForCausalLM.from_pretrained( BASE_MODEL, quantization_config=bnb_config, device_map="auto", trust_remote_code=True, token=HF_TOKEN, ) base.resize_token_embeddings(len(tokenizer)) print("Base model loaded ✓") print("Loading LoRA adapter...") model = PeftModel.from_pretrained( base, MODEL_REPO, token=HF_TOKEN, ignore_mismatched_sizes=True, ) model.eval() print("FinBot model ready ✓") # ============================================================ # LAYER 1 - Intent + Entity + Router # ============================================================ TICKER_MAP = { "reliance": ("RELIANCE", "Reliance Industries"), "ril": ("RELIANCE", "Reliance Industries"), "tcs": ("TCS", "Tata Consultancy Services"), "tata consultancy": ("TCS", "Tata Consultancy Services"), "infosys": ("INFY", "Infosys"), "infy": ("INFY", "Infosys"), "wipro": ("WIPRO", "Wipro"), "hdfc bank": ("HDFCBANK", "HDFC Bank"), "hdfc": ("HDFCBANK", "HDFC Bank"), "icici bank": ("ICICIBANK", "ICICI Bank"), "icici": ("ICICIBANK", "ICICI Bank"), "sbi": ("SBIN", "State Bank of India"), "state bank": ("SBIN", "State Bank of India"), "axis bank": ("AXISBANK", "Axis Bank"), "kotak": ("KOTAKBANK", "Kotak Mahindra Bank"), "itc": ("ITC", "ITC Ltd"), "hindustan unilever": ("HINDUNILVR", "Hindustan Unilever"), "hul": ("HINDUNILVR", "Hindustan Unilever"), "airtel": ("BHARTIARTL", "Bharti Airtel"), "bharti airtel": ("BHARTIARTL", "Bharti Airtel"), "maruti": ("MARUTI", "Maruti Suzuki"), "maruti suzuki": ("MARUTI", "Maruti Suzuki"), "tata motors": ("TATAMOTORS", "Tata Motors"), "sun pharma": ("SUNPHARMA", "Sun Pharmaceutical"), "bajaj finance": ("BAJFINANCE", "Bajaj Finance"), "bajaj housing finance": ("BAJAJHFL", "Bajaj Housing Finance"), "bajaj housing": ("BAJAJHFL", "Bajaj Housing Finance"), "bajajhfl": ("BAJAJHFL", "Bajaj Housing Finance"), "adani ports": ("ADANIPORTS", "Adani Ports"), "adani green": ("ADANIGREEN", "Adani Green Energy"), "adani": ("ADANIENT", "Adani Enterprises"), "ntpc": ("NTPC", "NTPC Ltd"), "ongc": ("ONGC", "ONGC"), "power grid": ("POWERGRID", "Power Grid Corp"), "l&t": ("LT", "Larsen & Toubro"), "larsen": ("LT", "Larsen & Toubro"), "zomato": ("ZOMATO", "Zomato"), "paytm": ("PAYTM", "Paytm"), "nykaa": ("NYKAA", "Nykaa"), "dmart": ("DMART", "Avenue Supermarts"), "avenue supermarts": ("DMART", "Avenue Supermarts"), "asian paints": ("ASIANPAINT", "Asian Paints"), "titan": ("TITAN", "Titan Company"), "tata steel": ("TATASTEEL", "Tata Steel"), "jsw steel": ("JSWSTEEL", "JSW Steel"), "ultratech": ("ULTRACEMCO", "UltraTech Cement"), "dr reddy": ("DRREDDY", "Dr Reddy's Laboratories"), "cipla": ("CIPLA", "Cipla"), "divis": ("DIVISLAB", "Divi's Laboratories"), "nifty 50": (None, "Nifty 50 Index"), "nifty": (None, "Nifty 50 Index"), "sensex": (None, "BSE Sensex Index"), "bank nifty": (None, "Bank Nifty Index"), "asianpaints": ("ASIANPAINT", "Asian Paints"), "bajajfinance": ("BAJFINANCE", "Bajaj Finance"), "tatamotors": ("TATAMOTORS", "Tata Motors"), "tatasteel": ("TATASTEEL", "Tata Steel"), "hdfcbank": ("HDFCBANK", "HDFC Bank"), "icicibank": ("ICICIBANK", "ICICI Bank"), "sunpharma": ("SUNPHARMA", "Sun Pharmaceutical"), "powergrid": ("POWERGRID", "Power Grid Corp"), # Single word variations for voice queries "asianpaints": ("ASIANPAINT", "Asian Paints"), "asian paint": ("ASIANPAINT", "Asian Paints"), "vedanta": ("VEDL", "Vedanta"), "tatapower": ("TATAPOWER", "Tata Power"), "tata power": ("TATAPOWER", "Tata Power"), "swiggy": ("SWIGGY", "Swiggy"), "bajajfinance": ("BAJFINANCE", "Bajaj Finance"), "tatamotors": ("TATAMOTORS", "Tata Motors"), "tatasteel": ("TATASTEEL", "Tata Steel"), "hdfcbank": ("HDFCBANK", "HDFC Bank"), "icicibank": ("ICICIBANK", "ICICI Bank"), "sunpharma": ("SUNPHARMA", "Sun Pharmaceutical"), "powergrid": ("POWERGRID", "Power Grid Corp"), "jswsteel": ("JSWSTEEL", "JSW Steel"), "ultracemco": ("ULTRACEMCO", "UltraTech Cement"), "ultratech cement": ("ULTRACEMCO", "UltraTech Cement"), "hindalco": ("HINDALCO", "Hindalco Industries"), "indusind": ("INDUSINDBK", "IndusInd Bank"), "indusind bank": ("INDUSINDBK", "IndusInd Bank"), "bajaj auto": ("BAJAJ-AUTO", "Bajaj Auto"), "bajajauto": ("BAJAJ-AUTO", "Bajaj Auto"), "hero motocorp": ("HEROMOTOCO", "Hero MotoCorp"), "hero motor": ("HEROMOTOCO", "Hero MotoCorp"), "eicher": ("EICHERMOT", "Eicher Motors"), "britannia": ("BRITANNIA", "Britannia Industries"), "nestlé": ("NESTLEIND", "Nestle India"), "nestle": ("NESTLEIND", "Nestle India"), "havells": ("HAVELLS", "Havells India"), "pidilite": ("PIDILITIND", "Pidilite Industries"), "dabur": ("DABUR", "Dabur India"), "godrej": ("GODREJCP", "Godrej Consumer Products"), "tata consumer": ("TATACONSUM", "Tata Consumer Products"), "tataconsumer": ("TATACONSUM", "Tata Consumer Products"), "delhivery": ("DELHIVERY", "Delhivery"), "irctc": ("IRCTC", "IRCTC"), "irfc": ("IRFC", "Indian Railway Finance"), "bhel": ("BHEL", "Bharat Heavy Electricals"), "coal india": ("COALINDIA", "Coal India"), "coalindia": ("COALINDIA", "Coal India"), "gail": ("GAIL", "GAIL India"), "bpcl": ("BPCL", "Bharat Petroleum"), "hpcl": ("HPCL", "Hindustan Petroleum"), "ioc": ("IOC", "Indian Oil Corporation"), "indian oil": ("IOC", "Indian Oil Corporation"), "sbi life": ("SBILIFE", "SBI Life Insurance"), "hdfc life": ("HDFCLIFE", "HDFC Life Insurance"), "icici pru": ("ICICIPRULI", "ICICI Prudential Life"), "bajaj allianz": ("BAJAJFINSV", "Bajaj Finserv"), "bajaj finserv": ("BAJAJFINSV", "Bajaj Finserv"), "muthoot": ("MUTHOOTFIN", "Muthoot Finance"), "cholamandalam": ("CHOLAFIN", "Cholamandalam Finance"), "manappuram": ("MANAPPURAM", "Manappuram Finance"), "polycab": ("POLYCAB", "Polycab India"), "ab capital": ("ABCAPITAL", "Aditya Birla Capital"), "abcapital": ("ABCAPITAL", "Aditya Birla Capital"), "trent": ("TRENT", "Trent Ltd"), "varun beverages": ("VBL", "Varun Beverages"), "dixon": ("DIXON", "Dixon Technologies"), "amber": ("AMBER", "Amber Enterprises"), "voltas": ("VOLTAS", "Voltas"), "blue star": ("BLUESTARCO", "Blue Star"), "crompton": ("CROMPTON", "Crompton Greaves"), "cg power": ("CGPOWER", "CG Power"), "abb": ("ABB", "ABB India"), "siemens": ("SIEMENS", "Siemens India"), "kpit": ("KPITTECH", "KPIT Technologies"), "persistent": ("PERSISTENT", "Persistent Systems"), "mphasis": ("MPHASIS", "Mphasis"), "ltimindtree": ("LTIM", "LTIMindtree"), "lti mindtree": ("LTIM", "LTIMindtree"), "tech mahindra": ("TECHM", "Tech Mahindra"), "techmahindra": ("TECHM", "Tech Mahindra"), "hcl tech": ("HCLTECH", "HCL Technologies"), "hcltech": ("HCLTECH", "HCL Technologies"), "oracle": ("OFSS", "Oracle Financial Services"), "mrf": ("MRF", "MRF Ltd"), "apollo tyres": ("APOLLOTYRE", "Apollo Tyres"), "ceat": ("CEATLTD", "CEAT Ltd"), "balkrishna": ("BALKRISIND", "Balkrishna Industries"), "bki": ("BALKRISIND", "Balkrishna Industries"), } INTENT_PATTERNS = { "price_query": [ r'\b(bhav|price|keemat|rate|kitne ka|kitna hai|share price|' r'stock price|aaj ka|today price|current price|pe chal raha|' r'what is the price|price of|stock price of|share price of|' r'ka bhav|ka price|ka rate|abhi kitna|kya chal raha)\b', ], "tax_query": [ r'\b(tax|ltcg|stcg|capital gain|itr|tds|80c|elss|' r'tax lagta|kitna tax|tax kaise|tax bachao|tax dena|' r'tax saving|tax free|taxable|deduction)\b', ], "portfolio_advice": [ r'\b(portfolio|sip|invest|kya karein|loss ho raha|' r'khareedun|bechun|hold|diversif|allocation|' r'sip shuroo|sip start|paisa dalu|paisa lagaun|kahan lagaun|' r'lumpsum|lump sum|monthly invest|paisa kahan|' r'kab khareedein|kab bechein|buy karna|sell karna|' r'retirement|bachpan|future ke liye|long term invest|' r'short term invest|paisa banana|wealth|savings)\b', ], "education": [ r'\b(kya hota hai|kya hai|explain|samjhao|batao|matlab|' r'definition|pe ratio|eps|roe|roa|what is|how does|' r'kaise kaam|kya matlab|sikhaao|bताओ|meaning|' r'difference between|fark kya|concept|basics|' r'kaise hota|kyun hota|kab hota)\b', ], "comparison": [ r'\b(vs|versus|better|behtar|konsa|compare|ya mutual|ya equity|' r'ya gold|ya fd|ya sip|ya nifty|fund ya|invest karein ya|' r'konsa better|kaun sa better|kya better|ya lumpsum|lumpsum ya|' r'sip ya|lump sum ya|direct ya|regular ya|growth ya|' r'dividend ya|active ya|passive ya|kaun sa|which is better|' r'ya phir|aur ya|vs kya)\b', ], "risk_query": [ r'\b(risk|safe|surakshit|volatile|volatility|secure|' r'guaranteed|pakka|kitna safe|risky hai|safe hai|' r'market girne|crash|correction|nuksaan|loss hoga|' r'kitna nuksaan|protect|bachao paisa)\b', ], "news_sentiment": [ r'\b(news|khabar|results|quarterly|earnings|q[1-4]|' r'announcement|merger|acquisition|split|bonus|' r'dividend announce|result kab|quarterly result|' r'ipo kab|listing|demerger)\b', ], } OUT_OF_DOMAIN = [ r'\b(crypto|bitcoin|ethereum|dogecoin|nft|web3|forex|' r'currency trading|sports betting|lottery|cricket bet|' r'cricket betting|satta|gambling|arbitrage betting|' r'sports arbitrage|commodit|crude oil)\b', r'\b(personal loan|home loan rate|emi calculator|insurance premium)\b', ] AMOUNT_PATTERN = re.compile(r'₹?\s*(\d+(?:\.\d+)?)\s*(lakh|crore|k|thousand)?', re.I) TIME_PATTERN = re.compile( r'(\d+)\s*(saal|year|mahine|month|din|day|week|hafta)|' r'(long[\s-]term|short[\s-]term|medium[\s-]term)', re.I ) SYSTEM = """You are FinBot, an AI assistant for Indian retail investors. You understand Hinglish and answer finance questions clearly and factually. RULES: 1. For price queries: Say "Live price NSE/BSE pe check karein." 2. Always end with: "Yeh sirf educational info hai. SEBI registered advisor se zaroor milein." 3. Never make up specific numbers, returns, or percentages. 4. Keep answers friendly, simple, 2-4 sentences maximum. 5. Always use Indian units - crore, lakh, rupees.""" INTENT_FACTS = { "tax_query": "[FACT: LTCG on equity = flat 12.5% above Rs 1.25 lakh after 1 year. No income slabs. Budget 2024.]", "price_query": "[FACT: Never give specific price. Redirect to NSE/BSE or broker app.]", "education": "[CONTEXT: Use Indian units - crore, lakh, rupees. Never use $ or billion.]", } HARDCODED_RESPONSES = { "__HARDCODED__tax_query_ltcg": ( "LTCG yaani Long Term Capital Gains tab lagta hai jab equity shares " "ya equity mutual funds 1 saal se zyada hold karke becho. " "Rs 1.25 lakh tak ka profit tax free hai, uske upar flat 12.5% tax " "lagta hai - koi income slab nahi, koi indexation nahi. " "STCG yaani 1 saal se pehle becho toh flat 20% tax lagta hai. " "Yeh sirf educational info hai. CA ya SEBI registered advisor se zaroor milein." ), "__HARDCODED__risk_query_bear": ( "Bear market mein ghabrana zaroori nahi. Historically markets hamesha " "recover karte hain - long term investors ke liye yeh buying opportunity hoti hai. " "SIP jaari rakho, quality stocks hold karo, panic mein mat becho. " "Yeh sirf educational info hai. SEBI registered advisor se zaroor milein." ), } GARBAGE = re.compile( r' str: ctx = [] if r.entities.company: ctx.append(f"[Company: {r.entities.company}]") if r.entities.amount: ctx.append(f"[Amount: {r.entities.amount}]") if r.entities.timeframe: ctx.append(f"[Timeframe: {r.entities.timeframe}]") ctx.append(f"[Intent: {r.intent}]") if r.intent in INTENT_FACTS: ctx.append(INTENT_FACTS[r.intent]) return f"[INST] {SYSTEM}\n\n{chr(10).join(ctx)}\n\nUser query: {r.resolved_query} [/INST]" def layer1_process(user_query: str) -> Layer1Output: result = Layer1Output(original_query=user_query) q_lower = user_query.lower() # LTCG shortcut if any(x in q_lower for x in ['ltcg', 'stcg', 'long term capital gain', 'short term capital gain', 'capital gain']): result.intent = "tax_query" result.resolved_query = user_query result.enriched_prompt = "__HARDCODED__tax_query_ltcg" return result # Bear market shortcut if any(x in q_lower for x in ['bear market', 'market crash', 'market gir', 'market down', 'bazar gir']): result.intent = "risk_query" result.resolved_query = user_query result.enriched_prompt = "__HARDCODED__risk_query_bear" return result # OOD check for pat in OUT_OF_DOMAIN: if re.search(pat, q_lower, re.I): result.is_out_of_domain = True result.enriched_prompt = "" return result # Ticker resolution resolved = user_query for alias, (ticker_sym, company_name) in sorted(TICKER_MAP.items(), key=lambda x: -len(x[0])): if alias in q_lower: resolved = re.sub(re.escape(alias), company_name, resolved, flags=re.I) result.entities.ticker = ticker_sym result.entities.company = company_name break result.resolved_query = resolved # Amount + timeframe amt = AMOUNT_PATTERN.search(q_lower) if amt: result.entities.amount = f"{amt.group(1)} {amt.group(2) or ''}".strip() tf = TIME_PATTERN.search(q_lower) if tf: result.entities.timeframe = tf.group(3) or f"{tf.group(1)} {tf.group(2)}" # Intent classification best_intent, best_score = "education", 0 for intent, patterns in INTENT_PATTERNS.items(): score = sum(1 for p in patterns if re.search(p, q_lower, re.I)) if score > best_score: best_score, best_intent = score, intent result.intent = best_intent result.confidence = min(1.0, best_score / 2) # Auto-resolve unknown tickers for price queries if best_intent == "price_query" and not result.entities.ticker: company_raw = user_query.lower() for remove in ["what is the price of", "ka aaj ka bhav kya hai", "ka bhav kya hai", "ka price kya hai", "price kya hai", "kitne ka hai", "kitna hai", "share price", "stock price", "aaj ka", "bhav", "price", "?", "kya", "hai"]: company_raw = company_raw.replace(remove, "") company_raw = company_raw.strip() if company_raw: auto_ticker = resolve_ticker(company_raw) if auto_ticker: result.entities.ticker = auto_ticker result.entities.company = company_raw.title() # Clear tax ticker if best_intent == "tax_query": result.entities.ticker = None result.entities.company = None result.resolved_query = user_query # Clear index ticker for non-price intents if best_intent in ["comparison", "education", "portfolio_advice"]: if result.entities.company and "Index" in result.entities.company: result.entities.company = None result.entities.ticker = None result.resolved_query = user_query result.enriched_prompt = _build_prompt(result) return result # ============================================================ # LIVE PRICE # ============================================================ def resolve_ticker(company_name: str) -> str: """Auto-resolve company name to NSE ticker via Yahoo Finance search.""" try: import urllib.request, json, urllib.parse query = urllib.parse.quote(f"{company_name} NSE India stock") url = f"https://query2.finance.yahoo.com/v1/finance/search?q={query}"esCount=1&newsCount=0" req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=5) as r: data = json.loads(r.read()) quotes = data.get("quotes", []) for q in quotes: symbol = q.get("symbol", "") if symbol.endswith(".NS"): return symbol.replace(".NS", "") return None except: return None def get_live_price(ticker: str): # Try NSE India public API (no auth needed) try: import urllib.request url = f"https://query1.finance.yahoo.com/v8/finance/chart/{ticker}.NS?interval=1d&range=1d" req = urllib.request.Request(url, headers={ "User-Agent": "Mozilla/5.0", "Accept": "application/json", }) with urllib.request.urlopen(req, timeout=5) as r: data = __import__("json").loads(r.read()) price = data["chart"]["result"][0]["meta"]["regularMarketPrice"] prev = data["chart"]["result"][0]["meta"]["chartPreviousClose"] chg = round((price - prev) / prev * 100, 2) return {"price": round(price, 2), "change_pct": chg, "found": True} except: return {"found": False} # ============================================================ # LAYER 3 - Trust Verifier # ============================================================ DISCLAIMER_KEYWORDS = [ "sirf educational", "sirf education", "sebi registered", "financial advisor", "CA se", "professional advice", "investment advice nahi", "yeh sirf", "disclaimer", "personal advice nahi", "apne risk par", ] HINGLISH_MARKERS = [ "karo", "karein", "hota hai", "milta hai", "chahiye", "aapko", "matlab", "kyun", "kya", "hai", "mein", ] PRICE_HALLUC_PATTERNS = [ r"\b\d{2,6}(\.\d{1,2})?\s*(rupees?|rs\.?|₹|inr)", r"₹\s*\d+", r"rs\.?\s*\d{2,6}", ] def extract_features(query, response, intent, ticker, source): resp_lower = (response or "").lower() resp_len = min(len(response or "") / 500.0, 1.0) has_disclaimer = float(any(kw in resp_lower for kw in DISCLAIMER_KEYWORDS)) has_price_halluc = 0.0 if intent == "price_query": for pat in PRICE_HALLUC_PATTERNS: if re.search(pat, resp_lower): has_price_halluc = 1.0 break has_hinglish = float(sum(1 for m in HINGLISH_MARKERS if m in resp_lower) >= 3) intents = ["price_query","education","tax_query","portfolio_advice", "comparison","risk_query","out_of_domain"] intent_feats = [float(intent == i) for i in intents] return { "response_length": resp_len, "has_disclaimer": has_disclaimer, "has_price_hallucination": has_price_halluc, "has_hinglish": has_hinglish, "intent_price_query": intent_feats[0], "intent_education": intent_feats[1], "intent_tax_query": intent_feats[2], "intent_portfolio_advice": intent_feats[3], "intent_comparison": intent_feats[4], "intent_risk_query": intent_feats[5], "intent_out_of_domain": intent_feats[6], "source_hardcoded": float(source == "hardcoded"), "source_rejected": float(source == "rejected"), "source_llm": float(source == "llm"), "trust_score": 0.5, "flag_pass": 0.0, "flag_warn": 0.0, "flag_block": 0.0, } def rule_based_trust_score(features, source, intent): if source == "hardcoded": return 1.00 if source == "rejected": return 0.25 if source == "live_price": return 0.95 score = 0.75 if features["has_disclaimer"]: score += 0.10 if features["has_price_hallucination"]: score -= 0.40 if features["has_hinglish"]: score += 0.05 if intent == "tax_query": score -= 0.10 if intent == "education": score -= 0.05 return round(min(max(score, 0.0), 1.0), 2) def layer3_verify(query, response, intent, ticker, source): features = extract_features(query, response, intent, ticker, source) rule_score = rule_based_trust_score(features, source, intent) features["trust_score"] = rule_score if rule_score >= 0.75: flag = "PASS" elif rule_score >= 0.50: flag = "WARN" else: flag = "BLOCK" return { "query": query, "response": response, "intent": intent, "ticker": ticker, "source": source, "features": features, "trust_score": rule_score, "flag": flag, } # ============================================================ # LAYER 4 - Hinglish Explainability # ============================================================ def layer4_explain(l3: dict) -> dict: trust = l3["trust_score"] flag = l3["flag"] feats = l3["features"] source = l3["source"] intent = l3["intent"] if flag == "PASS": user_message = ("Verified response hai - yeh information reliable hai" if trust >= 0.95 else "Response trustworthy lagta hai - phir bhi SEBI advisor se confirm karein") elif flag == "WARN": user_message = "Is response ko dhyan se padhen - kuch points verify karne padenge" else: user_message = "Yeh query hamare scope se bahar hai ya response reliable nahi" positives = [] if feats.get("has_disclaimer"): positives.append("Disclaimer present hai") if not feats.get("has_price_hallucination"):positives.append("Koi price hallucination nahi") if feats.get("has_hinglish"): positives.append("Hinglish mein response") if source == "hardcoded": positives.append("Verified hardcoded answer - 100% accurate") if source == "rejected": positives.append("Out-of-domain query correctly reject kiya") if source == "live_price": positives.append("Live NSE data use hua") risks = [] if feats.get("has_price_hallucination"): risks.append("Specific price mention - NSE/BSE pe verify karein") if not feats.get("has_disclaimer") and source == "llm": risks.append("Disclaimer missing - advisor se zaroor milein") if intent == "tax_query" and source == "llm": risks.append("Tax info - CA se verify karein") if flag == "WARN": risks.append("Trust score low - response carefully read karein") return { "trust_score": trust, "flag": flag, "user_message": user_message, "positives": positives, "risks": risks, } # ============================================================ # FORMAT OUTPUT # ============================================================ def _format_output(response: str, l4: dict) -> tuple: flag = l4["flag"] score = l4["trust_score"] explanation = l4["user_message"] if l4["positives"]: explanation += "\n\n" + "\n".join(f" {p}" for p in l4["positives"]) if l4["risks"]: explanation += "\n\n" + "\n".join(f" {r}" for r in l4["risks"]) return response, flag, f"{score:.2f}", explanation # ============================================================ # MAIN INFERENCE - wrapped with @spaces.GPU # ============================================================ # ============================================================ # INFERENCE — GPU only for LLM, rest runs free # ============================================================ def run_no_gpu(user_query: str): l1 = layer1_process(user_query) if l1.enriched_prompt.startswith("__HARDCODED__"): response = HARDCODED_RESPONSES.get(l1.enriched_prompt, "Official source check karein.") l3 = layer3_verify(user_query, response, l1.intent, None, "hardcoded") l4 = layer4_explain(l3) return _format_output(response, l4), None if l1.is_out_of_domain: return ( "Yeh query hamare scope se bahar hai. SmartVest sirf Indian retail " "investors ke liye equity, mutual funds, aur personal finance " "questions answer karta hai.", "BLOCK", "0.25", "Out-of-domain query - reject kiya gaya" ), None if l1.intent == "price_query" and l1.entities.ticker: pd = get_live_price(l1.entities.ticker) if pd["found"]: direction = "upar" if pd["change_pct"] > 0 else "neeche" sign = "+" if pd["change_pct"] > 0 else "" response = ( f"{l1.entities.company} ({l1.entities.ticker}) ka aaj ka live NSE price " f"hai Rs {pd['price']:,.2f} ({direction} {sign}{pd['change_pct']}% aaj). " f"Invest karne se pehle apne broker app pe confirm karein. " f"Yeh sirf informational data hai. SEBI registered advisor se zaroor milein." ) else: response = ( f"{l1.entities.company} ({l1.entities.ticker}) ka live price abhi fetch " f"nahi ho saka. Please NSE website (nseindia.com) ya apne broker app " f"(Zerodha, Groww, Upstox) pe check karein. " f"Yeh sirf educational info hai. SEBI registered advisor se zaroor milein." ) l3 = layer3_verify(user_query, response, l1.intent, l1.entities.ticker, "live_price") l4 = layer4_explain(l3) return _format_output(response, l4), None return None, l1 @spaces.GPU def run_llm(l1) -> tuple: load_model() device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model.to(device) inputs = tokenizer( l1.enriched_prompt, return_tensors="pt", max_length=512, truncation=True, ).to(device) with torch.no_grad(): out = model.generate( **inputs, max_new_tokens=150, temperature=0.5, do_sample=True, repetition_penalty=1.1, top_p=0.9, top_k=40, eos_token_id=tokenizer.eos_token_id, pad_token_id=tokenizer.eos_token_id, ) raw = tokenizer.decode(out[0], skip_special_tokens=True) answer = raw.split("[/INST]")[-1].strip() answer = GARBAGE.sub("", answer).strip() sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', answer) if s.strip()] answer = " ".join(sentences[:4]) if answer and answer[-1] not in ".!?": answer += "." l3 = layer3_verify(l1.original_query, answer, l1.intent, l1.entities.ticker, "llm") l4 = layer4_explain(l3) return _format_output(answer, l4) def run_finbot(user_query: str) -> tuple: result, l1 = run_no_gpu(user_query) if result is not None: return result return run_llm(l1) # ============================================================ # GRADIO UI # ============================================================ EXAMPLES = [ "Bajaj Housing Finance ka aaj ka bhav kya hai?", "LTCG tax kya hota hai aur kitna lagta hai?", "SIP band karni chahiye kya market girne pe?", "PE ratio kya hota hai samjhao?", "FD vs mutual fund konsa better hai long term ke liye?", "Reliance ka aaj ka price kya hai?", "Bitcoin mein invest karein kya?", "loss ho raha hai portfolio mein kya karein?", ] with gr.Blocks(title="SmartVest - Hinglish Finance Assistant") as demo: gr.HTML("""

SmartVest

Hinglish Financial Assistant for Indian Retail Investors
Koi bhi finance question Hindi-English mein poochho

""") with gr.Row(): with gr.Column(scale=3): query_box = gr.Textbox( label="Apna question yahan likho", placeholder="jaise: Bajaj Housing Finance ka bhav kya hai?", lines=2, ) submit_btn = gr.Button("Send", variant="primary") response_box = gr.Textbox( label="SmartVest ka Jawab", lines=7, interactive=False, ) with gr.Column(scale=1): flag_box = gr.Textbox( label="Trust Flag", interactive=False, ) score_box = gr.Textbox( label="Trust Score (0 - 1)", interactive=False, ) explain_box = gr.Textbox( label="Explanation", lines=8, interactive=False, ) gr.Examples( examples=[[e] for e in EXAMPLES], inputs=query_box, label="Example Queries - Click to try", ) gr.HTML("""
SmartVest sirf educational information deta hai.
Investment decisions ke liye SEBI registered advisor se zaroor milein.
""") submit_btn.click( fn=run_finbot, inputs=query_box, outputs=[response_box, flag_box, score_box, explain_box], ) query_box.submit( fn=run_finbot, inputs=query_box, outputs=[response_box, flag_box, score_box, explain_box], ) # ============================================================ # WHISPER STT — Transcribe endpoint # ============================================================ from faster_whisper import WhisperModel import tempfile, base64 print("Loading Whisper STT...") whisper_model = WhisperModel("medium", device="cpu", compute_type="int8") print("Whisper loaded ✓") def transcribe_audio(audio_base64: str) -> str: try: audio_bytes = base64.b64decode(audio_base64) with tempfile.NamedTemporaryFile(suffix=".webm", delete=False) as f: f.write(audio_bytes) tmp_path = f.name segments, _ = whisper_model.transcribe( tmp_path, language="hi", vad_filter=True ) return " ".join(s.text for s in segments).strip() except Exception as e: return "" # ============================================================ # GRADIO UI # ============================================================ EXAMPLES = [ "Bajaj Housing Finance ka aaj ka bhav kya hai?", "LTCG tax kya hota hai aur kitna lagta hai?", "SIP band karni chahiye kya market girne pe?", "PE ratio kya hota hai samjhao?", "FD vs mutual fund konsa better hai long term ke liye?", "Reliance ka aaj ka price kya hai?", "Bitcoin mein invest karein kya?", "loss ho raha hai portfolio mein kya karein?", ] with gr.Blocks(title="SmartVest - Hinglish Finance Assistant") as demo: gr.HTML("""

SmartVest

Hinglish Financial Assistant for Indian Retail Investors
Koi bhi finance question Hindi-English mein poochho

""") with gr.Row(): with gr.Column(scale=3): query_box = gr.Textbox( label="Apna question yahan likho", placeholder="jaise: Bajaj Housing Finance ka bhav kya hai?", lines=2, ) submit_btn = gr.Button("Send", variant="primary") response_box = gr.Textbox( label="SmartVest ka Jawab", lines=7, interactive=False, ) with gr.Column(scale=1): flag_box = gr.Textbox( label="Trust Flag", interactive=False, ) score_box = gr.Textbox( label="Trust Score (0 - 1)", interactive=False, ) explain_box = gr.Textbox( label="Explanation", lines=8, interactive=False, ) # Hidden transcribe interface for API access transcribe_input = gr.Textbox(visible=False) transcribe_output = gr.Textbox(visible=False) transcribe_btn = gr.Button(visible=False) transcribe_btn.click( fn=transcribe_audio, inputs=transcribe_input, outputs=transcribe_output, api_name="transcribe" ) gr.Examples( examples=[[e] for e in EXAMPLES], inputs=query_box, label="Example Queries - Click to try", ) gr.HTML("""
SmartVest sirf educational information deta hai.
Investment decisions ke liye SEBI registered advisor se zaroor milein.
""") submit_btn.click( fn=run_finbot, inputs=query_box, outputs=[response_box, flag_box, score_box, explain_box], ) query_box.submit( fn=run_finbot, inputs=query_box, outputs=[response_box, flag_box, score_box, explain_box], ) demo.launch()