""" data_utils.py - Hybrid data utilities (Colab/Streamlit friendly) Option B implementation: - Historical data: uses yfinance.history(...) (good for building ML features) - Live/current quote: uses NSE or BSE public APIs (when available) for up-to-date quotes - Fallbacks: if exchange API fails, returns best-effort data or empty dict - Also includes a simple fetch_news() placeholder and create_pdf_report() Usage: from data_utils import fetch_historical_data, fetch_live_quote, fetch_news, create_pdf_report Notes: - For NSE calls we use a requests.Session with common headers (NSE sometimes blocks non-browser agents). - For BSE calls we use the public BSE JSON endpoint (uses numeric security code or symbol depending on CSV). - This file assumes tickers in Yahoo format: e.g., "RELIANCE.NS" (NSE) or "500325.BO" (BSE). """ import pandas as pd import yfinance as yf import requests import time from datetime import datetime import matplotlib.pyplot as plt from fpdf import FPDF import os import re from typing import Optional, Dict, Any # ----------------------- # Helper: parse ticker # ----------------------- def _parse_ticker(ticker: str) -> (str, Optional[str]): """ Normalize ticker string and return (base, suffix) Examples: "RELIANCE.NS" -> ("RELIANCE", "NS") "500325.BO" -> ("500325", "BO") "RELIANCE" -> ("RELIANCE", None) """ if not isinstance(ticker, str): return ("", None) t = ticker.strip() if "." in t: parts = t.split(".") base = ".".join(parts[:-1]).upper() suffix = parts[-1].upper() return (base, suffix) return (t.upper(), None) # ----------------------- # HISTORICAL: yfinance # ----------------------- def fetch_historical_data(ticker: str, start_date: Optional[str], end_date: Optional[str]): """ Fetch OHLCV historical data using yfinance. - ticker: e.g., 'RELIANCE.NS' or '^NSEI' - start_date, end_date: 'YYYY-MM-DD' strings or None Returns pandas DataFrame (Open, High, Low, Close, Volume) or empty DataFrame. """ try: t = yf.Ticker(ticker) # if end_date is None, yfinance uses today's date automatically df = t.history(start=start_date, end=end_date if end_date else None, interval="1d", auto_adjust=False) if df is None or df.empty: return pd.DataFrame() df = df[['Open', 'High', 'Low', 'Close', 'Volume']] df.index = pd.to_datetime(df.index) return df except Exception as e: print("fetch_historical_data error:", e) return pd.DataFrame() # ----------------------- # LIVE QUOTE: NSE public API # ----------------------- def _get_nse_quote(symbol: str, session: Optional[requests.Session] = None) -> Dict[str, Any]: """ Get live quote for NSE symbol using NSE public JSON endpoint. symbol: plain symbol like 'RELIANCE' (no .NS) Returns a dict with keys: LTP, DayHigh, DayLow, PrevClose, Volume, timestamp (if found) """ if not symbol: return {} url_index = "https://www.nseindia.com" api_url = f"https://www.nseindia.com/api/quote-equity?symbol={symbol}" s = session or requests.Session() headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Accept-Language": "en-US,en;q=0.9", "Accept": "application/json, text/plain, */*", "Referer": "https://www.nseindia.com/", } s.headers.update(headers) try: # initial GET to obtain cookies and avoid 403 s.get(url_index, timeout=5) time.sleep(0.1) r = s.get(api_url, timeout=5) r.raise_for_status() j = r.json() # navigate JSON safely info = j.get("priceInfo", {}) or {} secinfo = j.get("securityInfo", {}) or {} intra = info.get("intraDayHighLow", {}) or {} result = { "symbol": j.get("metadata", {}).get("symbol", symbol), "LTP": info.get("lastPrice"), "DayHigh": intra.get("max"), "DayLow": intra.get("min"), "PrevClose": info.get("close"), "Volume": secinfo.get("totalTradedVolume") or j.get("tradeInfo", {}).get("totalTradedVolume"), "timestamp": datetime.now().isoformat() } return result except Exception as e: # NSE endpoint sometimes blocks — return empty and let caller fallback # print("NSE quote fetch failed:", e) return {} # ----------------------- # LIVE QUOTE: BSE public API # ----------------------- def _get_bse_quote(security_code: str) -> Dict[str, Any]: """ Get live quote for BSE security code using BSE public API. security_code: numeric code like '500325' OR alphanumeric symbol sometimes works Returns a dict with keys: LTP, DayHigh, DayLow, PrevClose, Volume, timestamp (if found) """ if not security_code: return {} url = ("https://api.bseindia.com/BseIndiaAPI/api/GetStkQuote/w" f"?flag=EQ&securitycode={security_code}") try: r = requests.get(url, timeout=5) r.raise_for_status() j = r.json() # BSE JSON contains 'CurrRate' dict cr = j.get("CurrRate", {}) or {} result = { "symbol": cr.get("Scripname") or security_code, "LTP": cr.get("LTP") or cr.get("LastPrice"), "Open": cr.get("Open"), "DayHigh": cr.get("High"), "DayLow": cr.get("Low"), "PrevClose": cr.get("PreviousClose"), "Volume": cr.get("TotalTradedQuantity") or cr.get("TradedQty"), "timestamp": datetime.now().isoformat() } return result except Exception as e: # print("BSE quote fetch failed:", e) return {} # ----------------------- # Public: fetch_live_quote # ----------------------- def fetch_live_quote(ticker: str) -> Dict[str, Any]: """ Public function to get a live/current quote for a ticker. - ticker: yahoo-style e.g., 'RELIANCE.NS' or '500325.BO' or 'RELIANCE' (best-effort) Returns dictionary with live fields or empty dict if none. Strategy: - parse ticker suffix .NS/.BO - if NSE -> use NSE API (_get_nse_quote) - if BSE -> use BSE API (_get_bse_quote) - fallback: query yfinance fast info (may be delayed) """ base, suffix = _parse_ticker(ticker) # Prefer NSE API if suffix == "NS" or (suffix is None and ticker.upper().endswith("NS")): # NSE expects plain symbol like 'RELIANCE' (without .NS) s = requests.Session() res = _get_nse_quote(base, session=s) if res: return res # BSE path if suffix == "BO" or (suffix is None and ticker.upper().endswith("BO")): # BSE often requires numeric security code (like 500325). If base is numeric, pass it. res = _get_bse_quote(base) if res: return res # If suffix not provided, try both: # Try NSE first res = _get_nse_quote(base, session=requests.Session()) if res: return res # Try BSE res = _get_bse_quote(base) if res: return res # Final fallback: use yfinance quick info (may be delayed) try: t = yf.Ticker(ticker) info = t.info if hasattr(t, "info") else {} # Some yfinance versions/queries may have 'regularMarketPrice' or 'previousClose' ltp = info.get("regularMarketPrice") or info.get("previousClose") or info.get("currentPrice") return { "symbol": base, "LTP": ltp, "DayHigh": info.get("dayHigh"), "DayLow": info.get("dayLow"), "PrevClose": info.get("previousClose"), "Volume": info.get("volume"), "timestamp": datetime.now().isoformat() } except Exception: return {} # ----------------------- # Simple news placeholder # ----------------------- def fetch_news(query: str, ticker: Optional[str] = None): """ Placeholder for news. Replace with NewsAPI or other provider for real news. Returns a small list of demo news dictionaries. """ return [ {"title": f"Latest update about {query}", "source": "DemoNews", "summary": "Replace with a real news API (NewsAPI/TwelveData news etc)."}, {"title": f"{query} quarterly results announced", "source": "DemoNews", "summary": "Demo placeholder."} ] # ----------------------- # PDF Report: include latest live info if available # ----------------------- def create_pdf_report(path: str, ticker: str, df: pd.DataFrame, live_info: Optional[Dict[str, Any]] = None): """ Create a simple PDF report including: - Close price time series plot - Histogram of daily returns - Optionally include small live quote box path: output pdf file path (e.g., 'report_RELIANCE_NS.pdf') ticker: ticker string used for headings df: DataFrame from fetch_historical_data live_info: optional dict returned by fetch_live_quote """ if df is None or df.empty: raise ValueError("DataFrame is empty; cannot create report.") out_dir = os.path.dirname(path) if out_dir and not os.path.exists(out_dir): os.makedirs(out_dir, exist_ok=True) # plot price plt.figure(figsize=(8, 3)) df['Close'].plot(title=f"{ticker} Close Price") plt.tight_layout() price_png = "temp_price.png" plt.savefig(price_png) plt.close() # histogram returns = df['Close'].pct_change().dropna() plt.figure(figsize=(8, 3)) returns.hist(bins=40) plt.title("Histogram of daily returns") plt.tight_layout() hist_png = "temp_hist.png" plt.savefig(hist_png) plt.close() # Build PDF pdf = FPDF() pdf.add_page() pdf.set_font("Arial", size=14) pdf.cell(0, 10, f"Report - {ticker}", ln=True) pdf.set_font("Arial", size=10) pdf.ln(4) pdf.cell(0, 8, f"Data rows: {len(df)}", ln=True) pdf.ln(4) # Live info block if live_info: pdf.set_font("Arial", size=10) pdf.cell(0, 7, f"Live snapshot ({live_info.get('timestamp','')})", ln=True) pdf.set_font("Arial", size=9) pdf.cell(0, 6, f"LTP: {live_info.get('LTP')} DayHigh: {live_info.get('DayHigh')} DayLow: {live_info.get('DayLow')}", ln=True) pdf.cell(0, 6, f"PrevClose: {live_info.get('PrevClose')} Volume: {live_info.get('Volume')}", ln=True) pdf.ln(6) pdf.image(price_png, w=180) pdf.ln(6) pdf.image(hist_png, w=180) pdf.ln(6) pdf.output(path) # cleanup try: os.remove(price_png) os.remove(hist_png) except: pass # ----------------------- # Quick test block (not run on import unless called) # ----------------------- if __name__ == "__main__": # quick manual test print("Historical sample (RELIANCE.NS): rows ->", len(fetch_historical_data("RELIANCE.NS", "2022-01-01", None))) print("Live NSE sample for RELIANCE:", fetch_live_quote("RELIANCE.NS")) print("Live BSE sample for 500325:", fetch_live_quote("500325.BO"))