FinBot | Elite Finance AI

import os import uuid import time import json import urllib.request from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse from llama_cpp import Llama import uvicorn import xml.etree.ElementTree as ET from datetime import datetime # --- Config --- MODEL_PATH = "/app/model.gguf" MODEL_URL = "https://huggingface.co/prakhar146/indian-finance-llm-v1/resolve/main/qwen2.5-7b-instruct-Q4_K_M.gguf?download=true" # --- Top 1% System Prompt (Short = Smart) --- ELITE_SYSTEM_PROMPT = """You are FinBot, a world-class Indian financial advisor. - Respond only in professional English. - Use Markdown formatting (bold, lists) for perfect readability. Do not use tables. - Be concise, factual, and actionable. No fluff.""" # --- Live Micro-Stuffing Engine --- cached_news = "" last_fetch_time = None def get_market_news(): global cached_news, last_fetch_time now = datetime.now() # Fetch only once per hour to avoid rate limits if last_fetch_time and (now - last_fetch_time).seconds < 3600 and cached_news: return cached_news try: url = "https://economictimes.indiatimes.com/markets/rssfeeds/1977021501.cms" req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) response = urllib.request.urlopen(req, timeout=5) root = ET.fromstring(response.read()) headlines = [item.find('title').text for item in root.findall('.//item')[:3]] cached_news = "\n".join([f"- {h}" for h in headlines]) last_fetch_time = now return cached_news except: return cached_news if cached_news else "Market data temporarily unavailable." # --- Live Stock Price Tool --- STOCK_MAP = { "reliance": "RELIANCE.NS", "tcs": "TCS.NS", "infosys": "INFY.NS", "infy": "INFY.NS", "hdfc": "HDFCBANK.NS", "icici": "ICICIBANK.NS", "sbi": "SBIN.NS", "tata motors": "TATAMOTORS.NS", "itc": "ITC.NS", "wipro": "WIPRO.NS", "bajaj finance": "BAJFINANCE.NS", "adani": "ADANIENT.NS", "nifty": "^NSEI", "sensex": "^BSESN" } def get_live_price(query): query_lower = query.lower() ticker = None for name, tick in STOCK_MAP.items(): if name in query_lower: ticker = tick break if not ticker: return None try: url = f"https://query1.finance.yahoo.com/v8/finance/chart/{ticker}?range=1d&interval=1m" req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) response = urllib.request.urlopen(req, timeout=5) data = json.loads(response.read()) price = data['chart']['result'][0]['meta']['regularMarketPrice'] symbol = data['chart']['result'][0]['meta']['symbol'] return f"[Live Data: {symbol} is currently trading at ₹{price}]" except: return None # --- Download model if it doesn't exist --- if not os.path.exists(MODEL_PATH): print("⏳ Downloading model...") urllib.request.urlretrieve(MODEL_URL, MODEL_PATH) print("✅ Model downloaded!") # --- Load model (MAXIMUM SPEED) --- print("⏳ Loading model...") llm = Llama( model_path=MODEL_PATH, n_ctx=2048, n_batch=512, n_threads=2, n_threads_batch=2, use_mlock=True, use_mmap=True, verbose=False ) print("✅ Model loaded!") app = FastAPI(title="FinBot") # --- Top 1% Premium UI --- HTML = """ FinBot | Elite Finance AI

Welcome to FinBot

Your elite Indian financial advisor. Ask anything about finance.

""" @app.get("/", response_class=HTMLResponse) async def root(): return HTML @app.post("/v1/chat/completions") async def chat(request: Request): body = await request.json() messages = body.get("messages", []) use_stream = body.get("stream", False) # ReAct Tool Router: Check if user wants a live price live_price_data = get_live_price(messages[-1].get("content", "")) if live_price_data: messages[-1]["content"] = f"{live_price_data}\nUser Query: {messages[-1]['content']}" if not messages or messages[0].get("role") != "system": live_news = get_market_news() dynamic_prompt = ELITE_SYSTEM_PROMPT + f"\n\n**Latest Indian Market Headlines (Live):**\n{live_news}" messages.insert(0, {"role": "system", "content": dynamic_prompt}) if use_stream: def generate(): response = llm.create_chat_completion( messages=messages, max_tokens=1024, temperature=0.7, top_p=0.9, stream=True ) for chunk in response: # The correct format for create_chat_completion stream if "content" in chunk["choices"][0]["delta"]: yield f"data: {json.dumps(chunk)}\n\n" yield "data: [DONE]\n\n" return StreamingResponse(generate(), media_type="text/event-stream") else: response = llm.create_chat_completion( messages=messages, max_tokens=1024, temperature=0.7, top_p=0.9 ) return JSONResponse(response) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)