""" Stock Fetcher Module — V2.0 Fetches stock prices using yfinance and saves them to an Excel workbook. Replaces the old Google Finance scraping approach with a reliable public API. """ import json import logging import sys import time from pathlib import Path from typing import Generator import openpyxl import yfinance as yf # --------------------------------------------------------------------------- # Path constants — works both in dev and as a PyInstaller one-file exe. # # BASE_DIR : directory of the .exe (or script in dev). Used for writable # outputs: log file, Excel file. # BUNDLE_DIR : directory where --add-data files are extracted at runtime. # When frozen this is sys._MEIPASS (temp folder); in dev it # is the same as BASE_DIR. # --------------------------------------------------------------------------- if getattr(sys, "frozen", False): BASE_DIR = Path(sys.executable).resolve().parent # folder holding the .exe BUNDLE_DIR = Path(sys._MEIPASS) # temp extraction folder else: BASE_DIR = Path(__file__).resolve().parent BUNDLE_DIR = BASE_DIR logger = logging.getLogger("StockFetcher") logger.setLevel(logging.DEBUG) # File handler — log goes next to the exe (BASE_DIR), not the temp bundle dir _fh = logging.FileHandler(BASE_DIR / "stock_generator.log", encoding="utf-8") _fh.setLevel(logging.DEBUG) _fh.setFormatter(logging.Formatter("%(asctime)s | %(levelname)-8s | %(message)s")) logger.addHandler(_fh) # Console handler — info and above _ch = logging.StreamHandler() _ch.setLevel(logging.INFO) _ch.setFormatter(logging.Formatter("%(levelname)-8s | %(message)s")) logger.addHandler(_ch) # --------------------------------------------------------------------------- # Configuration loader # --------------------------------------------------------------------------- def load_config() -> dict: """Load configuration from config.json (bundled with the exe or next to script).""" config_path = BUNDLE_DIR / "config.json" # bundled copy inside exe if not config_path.exists(): config_path = BASE_DIR / "config.json" # fallback: next to exe/script desktop = Path.home() / "Desktop" / "Share Analysis.xlsx" defaults = { "stock_urls_file": "StockUrls.csv", "excel_file": str(desktop), "excel_sheet": "Future Sight", "stock_code_column": 2, "stock_price_column": 4, "extra_save_paths": [], "request_timeout": 15, "max_retries": 3, "retry_delay": 2, "log_file": "stock_generator.log", } if config_path.exists(): try: with open(config_path, "r", encoding="utf-8") as f: user_cfg = json.load(f) defaults.update(user_cfg) logger.debug("Loaded config from %s", config_path) except Exception as exc: logger.warning("Failed to read config.json, using defaults: %s", exc) return defaults # --------------------------------------------------------------------------- # NSE index definitions — fetched live at runtime from NSE servers # --------------------------------------------------------------------------- INDEX_URLS: dict[str, str] = { "Nifty 50": "https://nsearchives.nseindia.com/content/indices/ind_nifty50list.csv", "Nifty 200": "https://nsearchives.nseindia.com/content/indices/ind_nifty200list.csv", "Nifty 500": "https://nsearchives.nseindia.com/content/indices/ind_nifty500list.csv", } INDEX_NAMES = list(INDEX_URLS.keys()) # ["Nifty 50", "Nifty 200", "Nifty 500"] DEFAULT_INDEX = "Nifty 200" def fetch_index_stocks(index_name: str, timeout: int = 20) -> list[tuple[str, str, str]]: """ Download the NSE constituent CSV for *index_name* and return a list of ("SYMBOL.NS", "Company Name", "Industry") tuples. Raises ValueError for unknown index names. Raises requests.RequestException on network failure. """ import csv import requests if index_name not in INDEX_URLS: raise ValueError( f"Unknown index '{index_name}'. Valid choices: {INDEX_NAMES}" ) url = INDEX_URLS[index_name] logger.info("Downloading %s constituent list from NSE...", index_name) resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=timeout) resp.raise_for_status() stocks: list[tuple[str, str, str]] = [] for row in csv.DictReader(resp.text.splitlines()): symbol = row.get("Symbol", "").strip() name = row.get("Company Name", "").strip().rstrip(".") sector = row.get("Industry", "").strip() if symbol and name: stocks.append((symbol + ".NS", name, sector)) logger.info("Loaded %d stocks from %s", len(stocks), index_name) return stocks # --------------------------------------------------------------------------- # StockFetcher class # --------------------------------------------------------------------------- class StockFetcher: """Fetches live stock prices via yfinance and writes them to Excel.""" def __init__(self, config: dict | None = None, index_name: str = DEFAULT_INDEX): self.config = config or load_config() # Which NSE index to fetch (live download — no local CSV needed) if index_name not in INDEX_URLS: logger.warning("Unknown index '%s', falling back to %s", index_name, DEFAULT_INDEX) index_name = DEFAULT_INDEX self.index_name = index_name self.excel_file = self._resolve_path(self.config["excel_file"]) self.sheet_name = self.config["excel_sheet"] self.code_col = int(self.config["stock_code_column"]) self.price_col = int(self.config["stock_price_column"]) self.extra_save_paths = [ Path(p) for p in self.config.get("extra_save_paths", []) ] self.max_retries = int(self.config.get("max_retries", 3)) self.retry_delay = float(self.config.get("retry_delay", 2)) self.timeout = int(self.config.get("request_timeout", 20)) # Results populated after fetch self.stock_prices: dict[str, float | str] = {} self.failed_stocks: dict[str, str] = {} self.all_success: bool = True # Cancellation flag (set from GUI thread) self._cancel_requested = False # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ @staticmethod def _resolve_path(raw: str) -> Path: """Resolve a path from config. - Absolute paths are used as-is (e.g. Desktop Excel path). - Relative paths are resolved against BUNDLE_DIR first (for bundled read-only files like StockUrls.csv), then BASE_DIR as fallback. """ p = Path(raw) if p.is_absolute(): return p # Check BUNDLE_DIR (temp _MEIPASS when frozen, same as BASE_DIR in dev) candidate = BUNDLE_DIR / p if candidate.exists(): return candidate # Fallback to BASE_DIR (next to exe / script) return BASE_DIR / p def request_cancel(self): """Thread-safe cancellation signal.""" self._cancel_requested = True @staticmethod def _bare_code(ticker: str) -> str: """Strip exchange suffix (.NS, .BO, etc.) for matching bare Excel codes.""" return ticker.split(".")[0] # ------------------------------------------------------------------ # Load stock list (live download from NSE) # ------------------------------------------------------------------ def _load_stocks(self) -> list[tuple[str, str, str]]: """ Download the NSE constituent list for self.index_name and return list of (ticker, display_name, sector). No local file required. """ return fetch_index_stocks(self.index_name, timeout=self.timeout) # ------------------------------------------------------------------ # Fetch prices # ------------------------------------------------------------------ def _fetch_price(self, ticker: str) -> float | None: """ Fetch the current/latest price for a single ticker using yfinance. Retries up to max_retries times on failure. """ for attempt in range(1, self.max_retries + 1): try: stock = yf.Ticker(ticker) info = stock.fast_info price = getattr(info, "last_price", None) if price is None: # Fallback: try to get from recent history hist = stock.history(period="1d") if not hist.empty: price = float(hist["Close"].iloc[-1]) if price is not None: return round(float(price), 2) logger.warning( "Attempt %d/%d — No price data for %s", attempt, self.max_retries, ticker, ) except Exception as exc: logger.warning( "Attempt %d/%d — Error fetching %s: %s", attempt, self.max_retries, ticker, exc, ) if attempt < self.max_retries: time.sleep(self.retry_delay) return None def fetch_all(self) -> Generator[int, None, None]: """ Generator that fetches all stock prices and yields progress (0-100). Results are stored in self.stock_prices and self.failed_stocks. """ self._cancel_requested = False self.stock_prices.clear() self.failed_stocks.clear() self.all_success = True stocks = self._load_stocks() total = len(stocks) if total == 0: logger.warning("No stocks to fetch.") yield 100 return for idx, (ticker, name, _sector) in enumerate(stocks, start=1): if self._cancel_requested: logger.info("Fetch cancelled by user at %d/%d.", idx, total) self.all_success = False self.failed_stocks["_cancelled"] = "Operation cancelled by user" break price = self._fetch_price(ticker) if price is not None: self.stock_prices[ticker] = price logger.info("%-30s ₹ %.2f", name, price) else: self.all_success = False self.failed_stocks[name] = "No data" logger.error("FAILED: %s (%s)", name, ticker) progress = int((idx / total) * 100) yield progress # ------------------------------------------------------------------ # Save to Excel # ------------------------------------------------------------------ def save_to_excel(self) -> dict[str, str]: """ Write fetched prices into the Excel workbook. Returns a status dict: {stock_name: value, ..., ResponseStatus: "True"/"False"}. The template is (re)created whenever: • the Excel file does not exist yet, OR • the sheet's row count doesn't match the number of fetched stocks (e.g. user switched from Nifty 200 to Nifty 500). """ expected_rows = len(self.stock_prices) + 1 # +1 for header row needs_template = not self.excel_file.exists() if not needs_template: # Check whether the existing sheet has the right number of rows wb_check = openpyxl.load_workbook(str(self.excel_file), read_only=True) if self.sheet_name in wb_check.sheetnames: existing_rows = wb_check[self.sheet_name].max_row or 0 if existing_rows != expected_rows: logger.info( "Row count mismatch (file has %d rows, index has %d stocks) " "— recreating template for %s.", existing_rows - 1, len(self.stock_prices), self.index_name, ) needs_template = True else: needs_template = True wb_check.close() if needs_template: logger.info( "Creating fresh template at %s for %s (%d stocks).", self.excel_file, self.index_name, len(self.stock_prices), ) self.create_template_excel(self.excel_file) wb = openpyxl.load_workbook(str(self.excel_file)) if self.sheet_name not in wb.sheetnames: raise ValueError( f"Sheet '{self.sheet_name}' not found in {self.excel_file.name}. " f"Available sheets: {wb.sheetnames}" ) ws = wb[self.sheet_name] rows = ws.max_row updated_count = 0 for ticker, price in self.stock_prices.items(): bare = self._bare_code(ticker) for row_num in range(1, rows + 1): cell_val = ws.cell(row=row_num, column=self.code_col).value if cell_val is None: continue cell_str = str(cell_val).strip() # Match against full ticker (KOTAKBANK.NS) or bare code (KOTAKBANK) if cell_str == ticker or cell_str == bare: ws.cell(row=row_num, column=self.price_col).value = price updated_count += 1 break # Save to primary location wb.save(str(self.excel_file)) logger.info("Saved workbook to %s (%d stocks updated)", self.excel_file, updated_count) # Save to additional locations for extra_path in self.extra_save_paths: try: extra_path.parent.mkdir(parents=True, exist_ok=True) wb.save(str(extra_path)) logger.info("Saved extra copy to %s", extra_path) except Exception as exc: logger.error("Failed to save extra copy to %s: %s", extra_path, exc) wb.close() # Build response dict result: dict[str, str] = {} for name, reason in self.failed_stocks.items(): result[name] = reason if self.all_success: result["ResponseStatus"] = "True" logger.info("All stock details retrieved successfully \u2713") else: result["ResponseStatus"] = "False" logger.warning("Stock details retrieved partially \u2717") return result # ------------------------------------------------------------------ # Create template Excel # ------------------------------------------------------------------ def create_template_excel(self, save_path: str | Path) -> Path: """ Create a brand-new Excel workbook pre-populated with all stocks. New column layout (9 columns): Col A: NSE Stocks — display name Col B: CODE — bare ticker (KOTAKBANK) Col C: Sector — pre-filled from built-in sector map Col D: Current Value — price written here by fetch (default 0) Col E: Share Quantity — default 100 Col F: Buying Amount — =D*E (Current Value * Quantity) Col G: Expected Price — blank (user fills) Col H: Selling Amount — =G*E Col I: Profit/Loss — =H-F Col J: Profit/Loss % — =IF(F=0,"",((H-F)/F)*100) """ from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter # Sector comes directly from the NSE CSV — no local lookup dict needed stocks = self._load_stocks() save_path = Path(save_path) wb = openpyxl.Workbook() ws = wb.active ws.title = self.sheet_name # --- Styles --- header_font = Font(name="Calibri", bold=True, color="FFFFFF", size=11) header_fill = PatternFill("solid", fgColor="1F4E79") # dark blue header_align = Alignment(horizontal="center", vertical="center", wrap_text=True) thin = Side(border_style="thin", color="AAAAAA") border = Border(left=thin, right=thin, top=thin, bottom=thin) data_font = Font(name="Calibri", size=10) data_align = Alignment(horizontal="center", vertical="center") price_fill = PatternFill("solid", fgColor="E8F5E9") # light green — current value qty_fill = PatternFill("solid", fgColor="FFF9C4") # light yellow — qty formula_fill = PatternFill("solid", fgColor="F3F3F3") # light grey — formula cols pnl_pos_fill = PatternFill("solid", fgColor="C8E6C9") # green — profit blank_fill = PatternFill("solid", fgColor="FFF3E0") # orange-tint — user input # --- Headers (9 columns) --- headers = [ "NSE Stocks", # A "CODE", # B "Sector", # C "Current Value", # D ← price written here "Share Quantity", # E ← default 100 "Buying Amount", # F =D*E "Expected Price", # G blank (user fills) "Selling Amount", # H =G*E "Profit / Loss", # I =H-F "Profit / Loss %", # J =IF(F=0,"",((H-F)/F)*100) ] col_widths = [28, 16, 24, 14, 14, 14, 16, 14, 12, 14] for col_idx, header in enumerate(headers, start=1): cell = ws.cell(row=1, column=col_idx, value=header) cell.font = header_font cell.fill = header_fill cell.alignment = header_align cell.border = border ws.row_dimensions[1].height = 40 for i, w in enumerate(col_widths, start=1): ws.column_dimensions[get_column_letter(i)].width = w # --- Data rows --- for row_idx, (ticker, name, sector) in enumerate(stocks, start=2): bare = self._bare_code(ticker) r = row_idx values = { 1: name, # A: Stock Name 2: bare, # B: CODE 3: sector, # C: Sector 4: 0, # D: Current Value 5: 100, # E: Share Quantity (default 100) 6: f"=D{r}*E{r}", # F: Buying Amount 7: None, # G: Expected Price (blank) 8: f"=IF(G{r}=\"\",\"\",G{r}*E{r})", # H: Selling Amount 9: f"=IF(F{r}=0,\"\",H{r}-F{r})", # I: Profit/Loss 10: f"=IF(F{r}=0,\"\",((H{r}-F{r})/F{r})*100)", # J: Profit/Loss % } fill_map = { 4: price_fill, 5: qty_fill, 6: formula_fill, 7: blank_fill, 8: formula_fill, 9: formula_fill, 10: pnl_pos_fill, } for col_idx, val in values.items(): cell = ws.cell(row=r, column=col_idx, value=val) cell.font = data_font cell.alignment = data_align cell.border = border if col_idx in fill_map: cell.fill = fill_map[col_idx] # Freeze header row, auto-filter ws.freeze_panes = "A2" ws.auto_filter.ref = f"A1:J{len(stocks) + 1}" save_path.parent.mkdir(parents=True, exist_ok=True) wb.save(str(save_path)) wb.close() logger.info("Template Excel created at %s (%d stocks)", save_path, len(stocks)) return save_path