Upload folder using huggingface_hub

pipeline/fetcher.py

@@ -1,759 +1,838 @@
-"""
-pipeline/fetcher.py — Fetch 5-year financials.
-Priority: yfinance (free, Indian stocks) → FMP → screener.in
-"""
-from __future__ import annotations
-import sys, re, json, datetime
-from pathlib import Path
-from typing import Optional, List, Dict, Any
-
-import requests
-from bs4 import BeautifulSoup
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-from config import FMP_API_KEY, FMP_BASE, REQUEST_TIMEOUT, DEFAULT_YEARS
-from storage.cache import get_cached, set_cache
-from pipeline.resolver import CompanyMeta
-
-HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
-
-
-# ── Helpers ───────────────────────────────────────────────────────────────────
-
-def _s(v, default=None):
-    if v is None or (isinstance(v, float) and v != v): return default
-    return v
-
-
-def _row(df, *keys):
-    """Get a row from a yfinance DataFrame by trying multiple key names."""
-    for k in keys:
-        if k in df.index:
-            return df.loc[k].tolist()
-    return [None] * len(df.columns)
-
-
-# ── yfinance fetcher (primary for Indian + global) ───────────────────────────
-
-def _fetch_yfinance(ticker: str, years: int, log_fn=None) -> Optional[Dict]:
-    def log(m):
-        if log_fn: log_fn(m)
-    try:
-        import yfinance as yf
-        log(f"  → yfinance: fetching {ticker} …")
-        t = yf.Ticker(ticker)
-
-        fin = t.financials      # newest col first
-        bs  = t.balance_sheet
-        cf  = t.cashflow
-        info = t.info
-
-        if fin is None or fin.empty:
-            log("  ✗ yfinance returned empty financials")
-            return None
-
-        # Year labels — columns are Timestamps, newest first; we want oldest→newest
-        all_cols = list(fin.columns)          # newest first
-        use_cols = all_cols[:years]           # take up to `years` newest
-        use_cols_asc = list(reversed(use_cols))  # now oldest→newest
-        year_labels = []
-        for c in use_cols_asc:
-            try:
-                year_labels.append(f"{c.year}/{c.month:02d}")
-            except Exception:
-                year_labels.append(str(c)[:7])
-
-        n = len(year_labels)
-
-        def col(df, *keys):
-            """Extract a row from df, return list oldest→newest, length n."""
-            for k in keys:
-                if df is not None and not df.empty and k in df.index:
-                    # df columns are newest first; slice to use_cols then reverse
-                    vals = []
-                    for c in use_cols_asc:   # iterate oldest→newest
-                        if c in df.columns:
-                            v = df.at[k, c]
-        if currency == "INR":
-            scale = 1e7
-            units = "Crores"
-            units_label = "INR Crores (₹ Cr)"
-        else:
-            scale = 1e6
-            units = "Millions"
-            units_label = f"{currency} Millions"
-
-        def col(df, *keys):
-            for k in keys:
-                if k in df.columns:
-                    return df[k].tolist()
-            return [None] * len(df)
-
-        def sc(lst):
-            return [round(v / scale, 2) if v is not None and not pd.isna(v) else None for v in lst]
-
-        income = {
-            "revenue":            sc(col(inc, "TotalRevenue", "OperatingRevenue")),
-            "cost_of_revenue":    sc(col(inc, "CostOfRevenue", "ReconciledCostOfRevenue")),
-            "gross_profit":       sc(col(inc, "GrossProfit")),
-            "operating_expenses": sc(col(inc, "OperatingExpense")),
-            "operating_income":   sc(col(inc, "OperatingIncome", "EBIT")),
-            "ebitda":             sc(col(inc, "EBITDA", "NormalizedEBITDA")),
-            "depreciation_amort": sc(col(inc, "ReconciledDepreciation", "DepreciationAndAmortization")),
-            "interest_expense":   sc(col(inc, "InterestExpenseNonOperating", "InterestExpense")),
-            "pretax_income":      sc(col(inc, "PretaxIncome")),
-            "tax_expense":        sc(col(inc, "TaxProvision")),
-            "net_income":         sc(col(inc, "NetIncome", "NetIncomeCommonStockholders")),
-            "diluted_eps":        col(inc, "DilutedEPS"),
-            "diluted_shares":     col(inc, "DilutedAverageShares"),
-        }
-
-        balance = {
-            "cash_and_equivalents":       sc(col(bs, "CashAndCashEquivalents", "CashFinancial")),
-            "short_term_investments":     sc(col(bs, "OtherShortTermInvestments")),
-            "receivables":                sc(col(bs, "AccountsReceivable", "Receivables")),
-            "inventory":                  sc(col(bs, "Inventory")),
-            "other_current_assets":       sc(col(bs, "OtherCurrentAssets")),
-            "total_current_assets":       sc(col(bs, "CurrentAssets")),
-            "ppe_net":                    sc(col(bs, "NetPPE")),
-            "goodwill":                   sc(col(bs, "Goodwill")),
-            "intangibles":                sc(col(bs, "OtherIntangibleAssets")),
-            "other_long_term_assets":     sc(col(bs, "OtherNonCurrentAssets")),
-            "total_assets":               sc(col(bs, "TotalAssets")),
-            "accounts_payable":           sc(col(bs, "AccountsPayable", "Payables")),
-            "short_term_debt":            sc(col(bs, "CurrentDebt", "CurrentDebtAndCapitalLeaseObligation")),
-            "other_current_liabilities":  sc(col(bs, "OtherCurrentLiabilities")),
-            "total_current_liabilities":  sc(col(bs, "CurrentLiabilities")),
-            "long_term_debt":             sc(col(bs, "LongTermDebt", "LongTermDebtAndCapitalLeaseObligation")),
-            "other_long_term_liabilities":sc(col(bs, "OtherNonCurrentLiabilities")),
-            "total_liabilities":          sc(col(bs, "TotalLiabilitiesNetMinorityInterest", "TotalLiabilities")),
-            "total_equity":               sc(col(bs, "StockholdersEquity", "CommonStockEquity")),
-            "total_liab_and_equity":      sc(col(bs, "TotalAssets")),
-        }
-
-        cashflow = {
-            "net_income":              sc(col(cf, "NetIncomeFromContinuingOperations", "NetIncome")),
-            "depreciation_amort":      sc(col(cf, "DepreciationAndAmortization", "Depreciation")),
-            "stock_based_comp":        sc(col(cf, "StockBasedCompensation")),
-            "working_capital_changes": sc(col(cf, "ChangeInWorkingCapital")),
-            "other_operating":         [None] * n,
-            "cash_from_operations":    sc(col(cf, "OperatingCashFlow", "CashFlowFromContinuingOperatingActivities")),
-            "capex":                   sc(col(cf, "CapitalExpenditure", "PurchaseOfPPE")),
-            "acquisitions":            sc(col(cf, "PurchaseOfBusiness")),
-            "cash_from_investing":     sc(col(cf, "InvestingCashFlow", "CashFlowFromContinuingInvestingActivities")),
-            "debt_issuance_repayment": sc(col(cf, "NetLongTermDebtIssuance", "NetIssuancePaymentsOfDebt")),
-            "stock_buybacks":          sc(col(cf, "NetCommonStockIssuance", "RepurchaseOfCapitalStock")),
-            "dividends_paid":          sc(col(cf, "CashDividendsPaid")),
-            "cash_from_financing":     sc(col(cf, "FinancingCashFlow", "CashFlowFromContinuingFinancingActivities")),
-            "net_change_in_cash":      sc(col(cf, "ChangesInCash", "ChangeInCashSupplementalAsReported")),
-            "free_cash_flow":          sc(col(cf, "FreeCashFlow")),
-        }
-
-        # Compute FCF if missing
-        for i in range(n):
-            if cashflow["free_cash_flow"][i] is None:
-                cfo = cashflow["cash_from_operations"][i]
-                cap = cashflow["capex"][i]
-                if cfo is not None and cap is not None:
-                    cashflow["free_cash_flow"][i] = round(cfo + cap, 2)
-                    
-        # Extract valuation snapshot and analyst targets
-        val_snap = {}
-        fd = t.financial_data
-        if fd and ticker in fd and isinstance(fd[ticker], dict):
-            val_snap["target_mean_price"] = fd[ticker].get("targetMeanPrice")
-            val_snap["target_high_price"] = fd[ticker].get("targetHighPrice")
-            val_snap["target_low_price"] = fd[ticker].get("targetLowPrice")
-            val_snap["wacc_estimate"] = 0.10 # default fallback
-        
-        info = t.summary_detail
-        if info and ticker in info and isinstance(info[ticker], dict):
-            val_snap["pe_ttm"] = info[ticker].get("trailingPE")
-            val_snap["dividend_yield"] = info[ticker].get("dividendYield")
-
-        log(f"  ✓ yahooquery data ready ({n} years, {currency}, in {units})")
-
-        return {
-            "years": year_labels,
-            "income_statement": income,
-            "balance_sheet":    balance,
-            "cash_flow":        cashflow,
-            "currency":         currency,
-            "units":            units,
-            "units_label":      units_label,
-            "scale":            scale,
-            "source":           "Yahoo Finance (yahooquery)",
-            "valuation_snapshot": val_snap,
-        }
-    except Exception as e:
-        import traceback
-        if log_fn: log_fn(f"  ✗ yahooquery error: {e}\n{traceback.format_exc()}")
-        return None
-
-
-# ── FMP fetcher (secondary for global) ───────────────────────────────────────
-
-def _fmp_get(path, params=None):
-    if not FMP_API_KEY: return None
-    p = {"apikey": FMP_API_KEY}
-    if params: p.update(params)
-    try:
-        r = requests.get(f"{FMP_BASE}{path}", params=p, timeout=REQUEST_TIMEOUT, headers=HEADERS)
-        r.raise_for_status()
-        return r.json()
-    except Exception:
-        return None
-
-
-def _normalise_fmp(raw, years):
-    inc = list(reversed(raw.get("income", [])))[:years]
-    bal = list(reversed(raw.get("balance", [])))[:years]
-    cf  = list(reversed(raw.get("cashflow", [])))[:years]
-
-    def col(lst, key):
-        return [_s(d.get(key)) for d in lst]
-
-    year_labels = [d.get("calendarYear", str(d.get("date",""))[:4]) for d in inc]
-    income = {
-        "revenue":            col(inc, "revenue"),
-        "cost_of_revenue":    col(inc, "costOfRevenue"),
-        "gross_profit":       col(inc, "grossProfit"),
-        "operating_expenses": col(inc, "operatingExpenses"),
-        "operating_income":   col(inc, "operatingIncome"),
-        "ebitda":             col(inc, "ebitda"),
-        "depreciation_amort": col(inc, "depreciationAndAmortization"),
-        "interest_expense":   col(inc, "interestExpense"),
-        "pretax_income":      col(inc, "incomeBeforeTax"),
-        "tax_expense":        col(inc, "incomeTaxExpense"),
-        "net_income":         col(inc, "netIncome"),
-        "diluted_eps":        col(inc, "epsdiluted"),
-        "diluted_shares":     col(inc, "weightedAverageShsOutDil"),
-    }
-    balance = {
-        "cash_and_equivalents":       col(bal, "cashAndCashEquivalents"),
-        "short_term_investments":     col(bal, "shortTermInvestments"),
-        "receivables":                col(bal, "netReceivables"),
-        "inventory":                  col(bal, "inventory"),
-        "other_current_assets":       col(bal, "otherCurrentAssets"),
-        "total_current_assets":       col(bal, "totalCurrentAssets"),
-        "ppe_net":                    col(bal, "propertyPlantEquipmentNet"),
-        "goodwill":                   col(bal, "goodwill"),
-        "intangibles":                col(bal, "intangibleAssets"),
-        "other_long_term_assets":     col(bal, "otherNonCurrentAssets"),
-        "total_assets":               col(bal, "totalAssets"),
-        "accounts_payable":           col(bal, "accountPayables"),
-        "short_term_debt":            col(bal, "shortTermDebt"),
-        "other_current_liabilities":  col(bal, "otherCurrentLiabilities"),
-        "total_current_liabilities":  col(bal, "totalCurrentLiabilities"),
-        "long_term_debt":             col(bal, "longTermDebt"),
-        "other_long_term_liabilities":col(bal, "otherNonCurrentLiabilities"),
-        "total_liabilities":          col(bal, "totalLiabilities"),
-        "total_equity":               col(bal, "totalStockholdersEquity"),
-        "total_liab_and_equity":      col(bal, "totalLiabilitiesAndStockholdersEquity"),
-    }
-    cashflow = {
-        "net_income":              col(cf, "netIncome"),
-        "depreciation_amort":      col(cf, "depreciationAndAmortization"),
-        "stock_based_comp":        col(cf, "stockBasedCompensation"),
-        "working_capital_changes": col(cf, "changeInWorkingCapital"),
-        "other_operating":         col(cf, "otherWorkingCapital"),
-        "cash_from_operations":    col(cf, "netCashProvidedByOperatingActivities"),
-        "capex":                   col(cf, "capitalExpenditure"),
-        "acquisitions":            col(cf, "acquisitionsNet"),
-        "cash_from_investing":     col(cf, "netCashUsedForInvestingActivites"),
-        "debt_issuance_repayment": col(cf, "debtRepayment"),
-        "stock_buybacks":          col(cf, "commonStockRepurchased"),
-        "dividends_paid":          col(cf, "dividendsPaid"),
-        "cash_from_financing":     col(cf, "netCashUsedProvidedByFinancingActivities"),
-        "net_change_in_cash":      col(cf, "netChangeInCash"),
-        "free_cash_flow":          col(cf, "freeCashFlow"),
-    }
-    return {
-        "years": year_labels, "income_statement": income,
-        "balance_sheet": balance, "cash_flow": cashflow,
-        "currency": "USD", "units": "millions",
-        "source": "FMP API",
-        "valuation_snapshot": {},
-    }
-
-
-def _fetch_fmp(ticker, years, log_fn=None):
-    def log(m):
-        if log_fn: log_fn(m)
-    if not FMP_API_KEY:
-        return None
-    log("  → FMP: fetching statements …")
-    lim = years + 1
-    raw = {
-        "income":   _fmp_get(f"/v3/income-statement/{ticker}", {"limit": lim, "period": "annual"}) or [],
-        "balance":  _fmp_get(f"/v3/balance-sheet-statement/{ticker}", {"limit": lim, "period": "annual"}) or [],
-        "cashflow": _fmp_get(f"/v3/cash-flow-statement/{ticker}", {"limit": lim, "period": "annual"}) or [],
-    }
-    if not raw["income"]:
-        log("  ✗ FMP returned no data")
-        return None
-    log("  ✓ FMP data ready")
-    data = _normalise_fmp(raw, years)
-    # Add valuation
-    rtm = (_fmp_get(f"/v3/ratios-ttm/{ticker}") or [{}])[0]
-    km  = (_fmp_get(f"/v3/key-metrics-ttm/{ticker}") or [{}])[0]
-    data["valuation_snapshot"] = {
-        "pe_ttm":         _s(rtm.get("peRatioTTM")),
-        "pb":             _s(rtm.get("priceToBookRatioTTM")),
-        "ps":             _s(rtm.get("priceToSalesRatioTTM")),
-        "ev_ebitda":      _s(km.get("enterpriseValueOverEBITDATTM")),
-        "ev_revenue":     _s(km.get("evToSalesTTM")),
-        "fcf_yield":      _s(km.get("freeCashFlowYieldTTM")),
-        "dividend_yield": _s(rtm.get("dividendYieldTTM")),
-    }
-    return data
-
-
-# ── screener.in (Indian fallback) ────────────────────────────────────────────
-
-def _scrape_screener(ticker: str, log_fn=None) -> Optional[Dict]:
-    def log(m):
-        if log_fn: log_fn(m)
-    # Ticker usually like RELIANCE.NS, extract RELIANCE
-    base_ticker = ticker.split(".")[0].upper()
-    for url in [f"https://www.screener.in/company/{base_ticker}/consolidated/",
-                f"https://www.screener.in/company/{base_ticker}/"]:
-        try:
-            log(f"  → screener.in: {url}")
-            r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HEADERS)
-            if r.status_code == 200:
-                soup = BeautifulSoup(r.text, "lxml")
-                sections = {}
-                for sec_id, key in [("profit-loss","income"),("balance-sheet","balance"),
-                                     ("cash-flow","cashflow")]:
-                    sec = soup.find("section", {"id": sec_id})
-                    if sec:
-                        table = sec.find("table")
-                        if table:
-                            result = {}
-                            for row in table.find_all("tr"):
-                                cells = row.find_all(["td","th"])
-                                if len(cells) < 2: continue
-                                label = cells[0].get_text(strip=True)
-                                vals = []
-                                for c in cells[1:]:
-                                    t2 = c.get_text(strip=True).replace(",","").replace("+","")
-                                    try: vals.append(float(t2))
-                                    except: vals.append(None)
-                                if label: result[label] = vals
-                            sections[key] = result
-                            if "years" not in sections:
-                                hdr = table.find("tr")
-                                if hdr:
-                                    sections["years"] = [c.get_text(strip=True)
-                                                         for c in hdr.find_all(["th","td"])[1:]]
-                if sections:
-                    sections["source_url"] = url
-                    log("  ✓ screener.in scraped")
-                    return sections
-        except Exception as e:
-            log(f"  ✗ screener.in error: {e}")
-    return None
-
-
-def _normalise_screener(raw, years):
-    yr = raw.get("years", [])[:years]
-    n = len(yr)
-    inc = raw.get("income", {})
-    bal = raw.get("balance", {})
-    cf  = raw.get("cashflow", {})
-
-    def row(*keys):
-        for k in keys:
-            if k in inc: return inc[k][:n]
-            if k in bal: return bal[k][:n]
-            if k in cf:  return cf[k][:n]
-        return [None]*n
-
-    income = {
-        "revenue":            row("Revenue","Net Sales","Sales"),
-        "cost_of_revenue":    row("Expenses","Cost of Goods Sold"),
-        "gross_profit":       row("Gross Profit"),
-        "operating_expenses": [None]*n,
-        "operating_income":   row("Operating Profit","EBIT","PBDIT"),
-        "ebitda":             row("PBDIT","EBITDA"),
-        "depreciation_amort": row("Depreciation"),
-        "interest_expense":   row("Interest"),
-        "pretax_income":      row("Profit before tax","PBT"),
-        "tax_expense":        row("Tax"),
-        "net_income":         row("Net Profit","PAT"),
-        "diluted_eps":        row("EPS in Rs"),
-        "diluted_shares":     [None]*n,
-    }
-    balance = {
-        "cash_and_equivalents":       row("Cash Equivalents"),
-        "short_term_investments":     [None]*n,
-        "receivables":                [None]*n,
-        "inventory":                  [None]*n,
-        "other_current_assets":       [None]*n,
-        "total_current_assets":       row("Current Assets"),
-        "ppe_net":                    row("Fixed Assets"),
-        "goodwill":                   [None]*n,
-        "intangibles":                [None]*n,
-        "other_long_term_assets":     [None]*n,
-        "total_assets":               row("Total Assets","Balance Sheet Size"),
-        "accounts_payable":           [None]*n,
-        "short_term_debt":            row("Short Term Borrowing"),
-        "other_current_liabilities":  [None]*n,
-        "total_current_liabilities":  row("Current Liabilities"),
-        "long_term_debt":             row("Borrowings","Long Term Borrowing"),
-        "other_long_term_liabilities":[None]*n,
-        "total_liabilities":          row("Total Liabilities"),
-        "total_equity":               row("Equity Capital","Total Equity"),
-        "total_liab_and_equity":      row("Total Assets","Balance Sheet Size"),
-    }
-    cfo_vals = row("Cash from Operating Activity","Cash from Operations")
-    cap_vals = row("Fixed Assets Purchased","Capital Expenditure")
-    fcf_vals = [(c - abs(x)) if c is not None and x is not None else None
-                for c, x in zip(cfo_vals, cap_vals)]
-    cashflow = {
-        "net_income":              income["net_income"],
-        "depreciation_amort":      income["depreciation_amort"],
-        "stock_based_comp":        [None]*n,
-        "working_capital_changes": [None]*n,
-        "other_operating":         [None]*n,
-        "cash_from_operations":    cfo_vals,
-        "capex":                   cap_vals,
-        "acquisitions":            [None]*n,
-        "cash_from_investing":     row("Cash from Investing Activity"),
-        "debt_issuance_repayment": [None]*n,
-        "stock_buybacks":          [None]*n,
-        "dividends_paid":          row("Dividend Amount"),
-        "cash_from_financing":     row("Cash from Financing Activity"),
-        "net_change_in_cash":      row("Net Cash Flow"),
-        "free_cash_flow":          fcf_vals,
-    }
-    return {
-        "years": yr, "income_statement": income, "balance_sheet": balance, "cash_flow": cashflow,
-        "currency": "INR", "units": "crores",
-        "source": f"screener.in ({raw.get('source_url','')})",
-        "valuation_snapshot": {},
-    }
-
-# ── stockanalysis.com (Global fallback for >4 years) ──────────────────────────
-
-def _scrape_stockanalysis(ticker: str, years: int, log_fn=None) -> Optional[Dict]:
-    def log(m):
-        if log_fn: log_fn(m)
-    base_ticker = ticker.split(".")[0].lower()
-    
-    urls = {
-        "income": f"https://stockanalysis.com/stocks/{base_ticker}/financials/",
-        "balance": f"https://stockanalysis.com/stocks/{base_ticker}/financials/balance-sheet/",
-        "cashflow": f"https://stockanalysis.com/stocks/{base_ticker}/financials/cash-flow-statement/"
-    }
-    
-    raw = {}
-    try:
-        for key, url in urls.items():
-            log(f"  → stockanalysis.com: {url}")
-            r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HEADERS)
-            if r.status_code != 200:
-                log(f"  ✗ stockanalysis.com returned {r.status_code} for {key}")
-                return None
-            soup = BeautifulSoup(r.text, "lxml")
-            table = soup.find("table")
-            if not table: return None
-            
-            # Parse headers (Skip TTM, take `years` count)
-            thead = table.find("thead")
-            ths = thead.find("tr").find_all("th")
-            # Index 0 is "Fiscal Year", Index 1 is "TTM" (usually). Index 2+ are the years.
-            if len(ths) < 3: return None
-            
-            # Only need to extract year labels once
-            if "years" not in raw:
-                # Get the years newest to oldest, then reverse to oldest to newest
-                year_labels = [th.get_text(strip=True) for th in ths[2:2+years]]
-                raw["years"] = list(reversed(year_labels))
-                
-            data_rows = {}
-            tbody = table.find("tbody")
-            for tr in tbody.find_all("tr"):
-                tds = tr.find_all("td")
-                if len(tds) < 3: continue
-                label = tds[0].get_text(strip=True)
-                vals = []
-                for td in tds[2:2+years]:
-                    text = td.get_text(strip=True).replace(",", "")
-                    if text == "-" or text == "":
-                        vals.append(None)
-                    else:
-                        try: vals.append(float(text))
-                        except: vals.append(None)
-                data_rows[label] = list(reversed(vals))
-            raw[key] = data_rows
-            
-    except Exception as e:
-        log(f"  ✗ stockanalysis error: {e}")
-        return None
-        
-    n = len(raw.get("years", []))
-    if n == 0: return None
-    
-    def row(sec, *keys):
-        for k in keys:
-            if k in raw[sec]: return raw[sec][k]
-        return [None]*n
-
-    income = {
-        "revenue":            row("income", "Revenue"),
-        "cost_of_revenue":    row("income", "Cost of Revenue"),
-        "gross_profit":       row("income", "Gross Profit"),
-        "operating_expenses": row("income", "Operating Expenses"),
-        "operating_income":   row("income", "Operating Income"),
-        "ebitda":             row("income", "EBITDA"),
-        "depreciation_amort": row("income", "Depreciation & Amortization"),
-        "interest_expense":   row("income", "Interest Expense / Income"),
-        "pretax_income":      row("income", "Pretax Income"),
-        "tax_expense":        row("income", "Income Tax"),
-        "net_income":         row("income", "Net Income"),
-        "diluted_eps":        row("income", "EPS (Diluted)"),
-        "diluted_shares":     row("income", "Shares Outstanding (Diluted)"),
-    }
-    balance = {
-        "cash_and_equivalents":       row("balance", "Cash & Equivalents"),
-        "short_term_investments":     row("balance", "Short-Term Investments"),
-        "receivables":                row("balance", "Accounts Receivable"),
-        "inventory":                  row("balance", "Inventory"),
-        "other_current_assets":       row("balance", "Other Current Assets"),
-        "total_current_assets":       row("balance", "Total Current Assets"),
-        "ppe_net":                    row("balance", "Property, Plant & Equipment"),
-        "goodwill":                   row("balance", "Goodwill and Intangibles", "Goodwill"),
-        "intangibles":                [None]*n,
-        "other_long_term_assets":     row("balance", "Other Long-Term Assets"),
-        "total_assets":               row("balance", "Total Assets"),
-        "accounts_payable":           row("balance", "Accounts Payable"),
-        "short_term_debt":            row("balance", "Current Debt"),
-        "other_current_liabilities":  row("balance", "Other Current Liabilities"),
-        "total_current_liabilities":  row("balance", "Total Current Liabilities"),
-        "long_term_debt":             row("balance", "Long-Term Debt"),
-        "other_long_term_liabilities":row("balance", "Other Long-Term Liabilities"),
-        "total_liabilities":          row("balance", "Total Liabilities"),
-        "total_equity":               row("balance", "Shareholders' Equity"),
-        "total_liab_and_equity":      row("balance", "Total Liabilities and Equity"),
-    }
-    cashflow = {
-        "net_income":              row("cashflow", "Net Income"),
-        "depreciation_amort":      row("cashflow", "Depreciation & Amortization"),
-        "stock_based_comp":        row("cashflow", "Share-Based Compensation"),
-        "working_capital_changes": [None]*n,
-        "other_operating":         row("cashflow", "Other Operating Activities"),
-        "cash_from_operations":    row("cashflow", "Operating Cash Flow"),
-        "capex":                   row("cashflow", "Capital Expenditures"),
-        "acquisitions":            row("cashflow", "Acquisitions"),
-        "cash_from_investing":     row("cashflow", "Investing Cash Flow"),
-        "debt_issuance_repayment": row("cashflow", "Net Debt Issued/Paid"),
-        "stock_buybacks":          row("cashflow", "Common Stock Issued/Repurchased", "Stock Repurchases"),
-        "dividends_paid":          row("cashflow", "Dividends Paid"),
-        "cash_from_financing":     row("cashflow", "Financing Cash Flow"),
-        "net_change_in_cash":      row("cashflow", "Net Change in Cash"),
-        "free_cash_flow":          row("cashflow", "Free Cash Flow"),
-    }
-    return {
-        "years": raw["years"], "income_statement": income, "balance_sheet": balance, "cash_flow": cashflow,
-        "currency": "USD", "units": "millions",
-        "source": "stockanalysis.com",
-        "valuation_snapshot": {},
-    }
-
-# ── Peer snapshot ─────────────────────────────────────────────────────────────
-
-def fetch_peer_snapshot(ticker: str) -> Optional[Dict]:
-    try:
-        import yfinance as yf
-        t = yf.Ticker(ticker)
-        info = t.info
-        if not info.get("longName"): return None
-        fin = t.financials
-        rev = fin.loc["Total Revenue"].iloc[0] if fin is not None and "Total Revenue" in fin.index else None
-        ratios_ttm = (_fmp_get(f"/v3/ratios-ttm/{ticker}") or [{}])[0]
-        return {
-            "name": info.get("longName", ticker),
-            "ticker": ticker,
-            "rationale": f"Industry peer ({info.get('industry','')})",
-            "latest_year": "",
-            "currency": info.get("currency","INR"),
-            "units": "actuals",
-            "revenue": _s(rev),
-            "gross_margin": _s(info.get("grossMargins")),
-            "operating_margin": _s(info.get("operatingMargins")),
-            "net_margin": _s(info.get("profitMargins")),
-            "roe": _s(info.get("returnOnEquity")),
-            "roic": _s(ratios_ttm.get("returnOnCapitalEmployedTTM")),
-            "debt_to_equity": _s(info.get("debtToEquity")),
-            "current_ratio": _s(info.get("currentRatio")),
-            "fcf_margin": None,
-            "revenue_growth_3y_cagr": _s(info.get("revenueGrowth")),
-            "pe_ttm": _s(info.get("trailingPE")),
-            "ev_ebitda": _s(info.get("enterpriseToEbitda")),
-        }
-    except Exception:
-        return None
-
-def fetch_valuation(ticker: str) -> Dict:
-    try:
-        import yfinance as yf
-        info = yf.Ticker(ticker).info
-        return {
-            "pe_ttm":         _s(info.get("trailingPE")),
-            "pb":             _s(info.get("priceToBook")),
-            "ps":             _s(info.get("priceToSalesTrailing12Months")),
-            "ev_ebitda":      _s(info.get("enterpriseToEbitda")),
-            "ev_revenue":     _s(info.get("enterpriseToRevenue")),
-            "fcf_yield":      None,
-            "dividend_yield": _s(info.get("dividendYield")),
-        }
-    except Exception:
-        return {}
-
-def _scrape_edgar_10k(ticker: str, log_fn=None) -> str:
-    """Fetch the latest 10-K for a US company from SEC EDGAR and extract a chunk of text."""
-    headers = {"User-Agent": "FinancialAnalyzer/1.0 (contact@example.com)"}
-    try:
-        if log_fn: log_fn(f"  → Looking up SEC CIK for {ticker}...")
-        r = requests.get("https://www.sec.gov/files/company_tickers.json", headers=headers, timeout=10)
-        tickers = r.json()
-        cik = None
-        for k, v in tickers.items():
-            if v["ticker"].upper() == ticker.upper():
-                cik = str(v["cik_str"]).zfill(10)
-                break
-        
-        if not cik:
-            if log_fn: log_fn("  ✗ CIK not found in SEC database.")
-            return ""
-            
-        if log_fn: log_fn(f"  → Found CIK {cik}. Fetching recent filings...")
-        sub_r = requests.get(f"https://data.sec.gov/submissions/CIK{cik}.json", headers=headers, timeout=10)
-        data = sub_r.json()
-        filings = data.get("filings", {}).get("recent", {})
-        
-        form_types = filings.get("form", [])
-        accessions = filings.get("accessionNumber", [])
-        primary_docs = filings.get("primaryDocument", [])
-        
-        idx = -1
-        for i, form in enumerate(form_types):
-            if form == "10-K":
-                idx = i
-                break
-                
-        if idx == -1:
-            if log_fn: log_fn("  ✗ No 10-K found in recent filings.")
-            return ""
-            
-        acc_num = accessions[idx]
-        acc_num_no_dash = acc_num.replace("-", "")
-        doc = primary_docs[idx]
-        
-        url = f"https://www.sec.gov/Archives/edgar/data/{int(cik)}/{acc_num_no_dash}/{doc}"
-        if log_fn: log_fn(f"  → Downloading 10-K from {url}...")
-        
-        doc_r = requests.get(url, headers=headers, timeout=20)
-        from bs4 import BeautifulSoup
-        soup = BeautifulSoup(doc_r.content, "lxml")
-        
-        text = soup.get_text(separator="\n", strip=True)
-        import re
-        match = re.search(r'(?i)Item\s+7\.\s*Management[’\']s\s+Discussion', text)
-        if match:
-            start = match.start()
-            extract = text[start:start+40000] 
-        else:
-            extract = text[len(text)//4 : len(text)//4 + 40000]
-            
-        if log_fn: log_fn("  ✓ 10-K text extracted successfully.")
-        return extract
-    except Exception as e:
-        if log_fn: log_fn(f"  ✗ Failed to scrape EDGAR: {str(e)[:80]}")
-        return ""
-
-
-# ── Main Fetcher ──────────────────────────────────────────────────────────────
-
-def fetch_all(meta: CompanyMeta, years: int = DEFAULT_YEARS,
-              force_refresh: bool = False, log_fn=None) -> Dict:
-    def log(m):
-        if log_fn: log_fn(m)
-
-    if not meta.is_listed:
-        msg = (f"'{meta.name}' is unlisted/private. Public data not available. "
-               "Try the exact BSE/NSE ticker.")
-        log(f"  ⚠ {msg}")
-        return {"warnings": [msg], "years": [], "income_statement": {},
-                "balance_sheet": {}, "cash_flow": {}}
-
-    ticker = meta.ticker
-    cache_key = f"full_{years}yr"
-    if not force_refresh:
-        cached = get_cached(ticker, cache_key)
-        if cached:
-            log(f"  ✓ Loaded cached data for {ticker}")
-            return cached
-
-    warnings = []
-    result = None
-
-    # Re-prioritized Fetching Logic:
-    
-    # 1. Indian Stocks: Try screener.in first (up to 10+ years of data)
-    if meta.currency == "INR":
-        log("  → Indian Stock detected. Prioritizing screener.in ...")
-        scraped = _scrape_screener(ticker, log_fn)
-        if scraped:
-            result = _normalise_screener(scraped, years)
-            warnings.append("Data sourced from screener.in. Verify against company filings.")
-            
-        # Fallback to yfinance if screener fails or yields nothing
-        if not result or not result.get("years"):
-            log("  → screener.in failed. Falling back to yahooquery ...")
-            result = _fetch_yahooquery(ticker, years, log_fn)
-            if result and len(result.get("years", [])) < years:
-                warnings.append(f"yahooquery only provides {len(result['years'])} years of historical data.")
-
-    # 2. Global/US Stocks
-    else:
-        # If requesting > 4 years, try stockanalysis.com first because yahooquery limits to 4 years
-        if years > 4:
-            log("  → Horizon > 4 years requested. Prioritizing stockanalysis.com for global stock ...")
-            result = _scrape_stockanalysis(ticker, years, log_fn)
-            if result:
-                warnings.append("Data sourced from stockanalysis.com. Verify against company filings.")
-        
-        # Fallback to yahooquery
-        if not result or not result.get("years"):
-            log("  → Trying yahooquery ...")
-            result = _fetch_yahooquery(ticker, years, log_fn)
-            if result and len(result.get("years", [])) < years:
-                warnings.append(f"yahooquery only provides {len(result['years'])} years of historical data.")
-                
-        # Fallback to FMP
-        if (not result or not result.get("years")) and FMP_API_KEY:
-            result = _fetch_fmp(ticker, years, log_fn)
-
-    # 3. Final validation
-    if not result or not result.get("years"):
-        msg = (f"Could not fetch data for '{meta.name}' ({ticker}). "
-               "Check ticker, or the company may have limited public data coverage.")
-        log(f"  ⚠ {msg}")
-        return {"warnings": [msg], "years": [], "income_statement": {},
-                "balance_sheet": {}, "cash_flow": {}}
-
-    # 4. Optional Annual Report Scrape (for context)
-    if meta.currency != "INR":
-        ar_context = _scrape_edgar_10k(ticker, log_fn)
-        if ar_context:
-            result["annual_report_context"] = ar_context
-            warnings.append("AI analysis enhanced with raw MD&A excerpts from SEC EDGAR 10-K.")
-
-    result["warnings"] = warnings
-    result.setdefault("currency", meta.currency)
-    set_cache(ticker, cache_key, result)
-    log("  ✓ Data cached for 7 days")
-    return result
+"""
+pipeline/fetcher.py — Fetch 5-year financials.
+Priority: yfinance (free, Indian stocks) -> FMP -> screener.in
+"""
+from __future__ import annotations
+import sys, re, json, datetime
+from pathlib import Path
+from typing import Optional, List, Dict, Any
+
+import requests
+from bs4 import BeautifulSoup
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+from config import FMP_API_KEY, FMP_BASE, REQUEST_TIMEOUT, DEFAULT_YEARS
+from storage.cache import get_cached, set_cache
+from pipeline.resolver import CompanyMeta
+
+HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+def _s(v, default=None):
+    if v is None or (isinstance(v, float) and v != v): return default
+    return v
+
+
+def _row(df, *keys):
+    """Get a row from a yfinance DataFrame by trying multiple key names."""
+    for k in keys:
+        if k in df.index:
+            return df.loc[k].tolist()
+    return [None] * len(df.columns)
+
+
+# ── yahooquery fetcher (primary for Indian + global) ─────────────────────────
+
+def _fetch_yahooquery(ticker: str, years: int, log_fn=None) -> Optional[Dict]:
+    """Fetch financial statements via yahooquery. Returns normalized dict or None."""
+    def log(m):
+        if log_fn: log_fn(m)
+    try:
+        import yahooquery as yq
+        import pandas as pd
+        import math
+
+        log(f"  -> yahooquery: fetching {ticker} ...")
+        t = yq.Ticker(ticker)
+
+        inc_df = t.income_statement(frequency='annual')
+        bs_df  = t.balance_sheet(frequency='annual')
+        cf_df  = t.cash_flow(frequency='annual')
+
+        if inc_df is None or isinstance(inc_df, str) or inc_df.empty:
+            log("  [ERR] yahooquery returned empty income statement")
+            return None
+
+        # ── Filter to clean annual rows only (periodType == '12M') ──
+        def clean_df(df):
+            if df is None or isinstance(df, str) or df.empty:
+                return pd.DataFrame()
+            df = df.reset_index()
+            if 'asOfDate' not in df.columns:
+                return df
+            df = df.dropna(subset=['asOfDate'])
+            df['asOfDate'] = pd.to_datetime(df['asOfDate'])
+            # Keep only 12M (full annual) rows, not TTM or quarterly
+            if 'periodType' in df.columns:
+                annual = df[df['periodType'] == '12M']
+                if not annual.empty:
+                    df = annual
+            df = df.sort_values('asOfDate')
+            # Deduplicate: keep last row per year-month (safety net)
+            df['ym'] = df['asOfDate'].dt.to_period('M')
+            df = df.drop_duplicates(subset=['ym'], keep='last').drop(columns=['ym'])
+            return df.reset_index(drop=True)
+
+        inc = clean_df(inc_df)
+        bs  = clean_df(bs_df)
+        cf  = clean_df(cf_df)
+
+        if inc.empty:
+            log("  [ERR] yahooquery: no usable annual rows after filtering")
+            return None
+
+        # Take the most recent `years` rows (oldest→newest)
+        inc = inc.tail(years)
+        bs  = bs.tail(years) if not bs.empty else pd.DataFrame()
+        cf  = cf.tail(years) if not cf.empty else pd.DataFrame()
+
+        # Year labels
+        year_labels = [d.strftime('%Y/%m') for d in inc['asOfDate']]
+        n = len(year_labels)
+
+        # Determine currency and scaling
+        currency = "INR"
+        if 'currencyCode' in inc.columns:
+            cur_vals = inc['currencyCode'].dropna().unique()
+            if len(cur_vals) > 0:
+                currency = str(cur_vals[-1])
+
+        if currency == "INR":
+            scale = 1e7          # raw INR -> Crores
+            units = "Crores"
+            units_label = "INR Crores (₹ Cr)"
+        else:
+            scale = 1e6
+            units = "Millions"
+            units_label = f"{currency} Millions"
+
+        def gcol(df, *keys):
+            """Get column values (oldest→newest list of length n), scaled."""
+            for k in keys:
+                if k in df.columns:
+                    vals = df[k].tolist()
+                    result = []
+                    for v in vals:
+                        if v is None or (isinstance(v, float) and math.isnan(v)):
+                            result.append(None)
+                        else:
+                            result.append(round(float(v) / scale, 2))
+                    # Pad or trim to n
+                    while len(result) < n: result.append(None)
+                    return result[:n]
+            return [None] * n
+
+        def gcol_raw(df, *keys):
+            """Get column values without scaling."""
+            for k in keys:
+                if k in df.columns:
+                    vals = df[k].tolist()
+                    result = []
+                    for v in vals:
+                        if v is None or (isinstance(v, float) and math.isnan(v)):
+                            result.append(None)
+                        else:
+                            result.append(float(v))
+                    while len(result) < n: result.append(None)
+                    return result[:n]
+            return [None] * n
+
+        income = {
+            "revenue":            gcol(inc, "TotalRevenue", "OperatingRevenue"),
+            "cost_of_revenue":    gcol(inc, "CostOfRevenue", "ReconciledCostOfRevenue"),
+            "gross_profit":       gcol(inc, "GrossProfit"),
+            "operating_expenses": gcol(inc, "OperatingExpense", "TotalExpenses"),
+            "operating_income":   gcol(inc, "OperatingIncome", "EBIT"),
+            "ebitda":             gcol(inc, "EBITDA", "NormalizedEBITDA"),
+            "depreciation_amort": gcol(inc, "ReconciledDepreciation", "DepreciationAndAmortizationInIncomeStatement"),
+            "interest_expense":   gcol(inc, "InterestExpenseNonOperating", "InterestExpense"),
+            "pretax_income":      gcol(inc, "PretaxIncome"),
+            "tax_expense":        gcol(inc, "TaxProvision"),
+            "net_income":         gcol(inc, "NetIncome", "NetIncomeCommonStockholders"),
+            "diluted_eps":        gcol_raw(inc, "DilutedEPS", "BasicEPS"),
+            "diluted_shares":     gcol_raw(inc, "DilutedAverageShares", "BasicAverageShares"),
+        }
+
+        balance = {
+            "cash_and_equivalents":       gcol(bs, "CashAndCashEquivalents", "CashFinancial"),
+            "short_term_investments":     gcol(bs, "OtherShortTermInvestments", "InvestmentsAndAdvances"),
+            "receivables":                gcol(bs, "AccountsReceivable", "Receivables"),
+            "inventory":                  gcol(bs, "Inventory"),
+            "other_current_assets":       gcol(bs, "OtherCurrentAssets"),
+            "total_current_assets":       gcol(bs, "CurrentAssets"),
+            "ppe_net":                    gcol(bs, "NetPPE"),
+            "goodwill":                   gcol(bs, "Goodwill"),
+            "intangibles":                gcol(bs, "OtherIntangibleAssets"),
+            "other_long_term_assets":     gcol(bs, "OtherNonCurrentAssets"),
+            "total_assets":               gcol(bs, "TotalAssets"),
+            "accounts_payable":           gcol(bs, "AccountsPayable", "Payables"),
+            "short_term_debt":            gcol(bs, "CurrentDebt", "CurrentDebtAndCapitalLeaseObligation"),
+            "other_current_liabilities":  gcol(bs, "OtherCurrentLiabilities"),
+            "total_current_liabilities":  gcol(bs, "CurrentLiabilities"),
+            "long_term_debt":             gcol(bs, "LongTermDebt", "LongTermDebtAndCapitalLeaseObligation"),
+            "other_long_term_liabilities":gcol(bs, "OtherNonCurrentLiabilities"),
+            "total_liabilities":          gcol(bs, "TotalLiabilitiesNetMinorityInterest", "TotalLiabilities"),
+            "total_equity":               gcol(bs, "StockholdersEquity", "CommonStockEquity"),
+            "total_liab_and_equity":      gcol(bs, "TotalAssets"),
+        }
+
+        cfo = gcol(cf, "OperatingCashFlow", "CashFlowFromContinuingOperatingActivities")
+        cap = gcol(cf, "CapitalExpenditure")
+        fcf = gcol(cf, "FreeCashFlow")
+        # Compute FCF from CFO - CapEx if missing
+        for i in range(n):
+            if fcf[i] is None and cfo[i] is not None and cap[i] is not None:
+                fcf[i] = round(cfo[i] + cap[i], 2)  # CapEx is negative in yahooquery
+
+        cashflow = {
+            "net_income":              gcol(cf, "NetIncomeFromContinuingOperations", "NetIncome"),
+            "depreciation_amort":      gcol(cf, "DepreciationAndAmortization", "Depreciation"),
+            "stock_based_comp":        gcol(cf, "StockBasedCompensation"),
+            "working_capital_changes": gcol(cf, "ChangeInWorkingCapital"),
+            "other_operating":         [None] * n,
+            "cash_from_operations":    cfo,
+            "capex":                   cap,
+            "acquisitions":            gcol(cf, "PurchaseOfBusiness", "AcquisitionOfBusiness"),
+            "cash_from_investing":     gcol(cf, "InvestingCashFlow", "CashFlowFromContinuingInvestingActivities"),
+            "debt_issuance_repayment": gcol(cf, "NetLongTermDebtIssuance", "NetIssuancePaymentsOfDebt"),
+            "stock_buybacks":          gcol(cf, "NetCommonStockIssuance", "RepurchaseOfCapitalStock"),
+            "dividends_paid":          gcol(cf, "CashDividendsPaid", "DividendsPaid"),
+            "cash_from_financing":     gcol(cf, "FinancingCashFlow", "CashFlowFromContinuingFinancingActivities"),
+            "net_change_in_cash":      gcol(cf, "ChangesInCash", "ChangeInCashSupplementalAsReported"),
+            "free_cash_flow":          fcf,
+        }
+
+        # Valuation snapshot
+        val_snap = {}
+        try:
+            fd = t.financial_data.get(ticker, {})
+            if isinstance(fd, dict):
+                val_snap["target_mean_price"] = fd.get("targetMeanPrice")
+                val_snap["target_high_price"] = fd.get("targetHighPrice")
+                val_snap["target_low_price"]  = fd.get("targetLowPrice")
+                val_snap["wacc_estimate"]      = 0.10
+
+            sd = t.summary_detail.get(ticker, {})
+            if isinstance(sd, dict):
+                val_snap["pe_ttm"]        = sd.get("trailingPE")
+                val_snap["dividend_yield"] = sd.get("dividendYield")
+        except Exception:
+            pass
+
+        log(f"  [OK] yahooquery data ready ({n} years, {currency}, in {units})")
+
+        return {
+            "years":              year_labels,
+            "income_statement":   income,
+            "balance_sheet":      balance,
+            "cash_flow":          cashflow,
+            "currency":           currency,
+            "units":              units,
+            "units_label":        units_label,
+            "scale":              scale,
+            "source":             "Yahoo Finance (yahooquery)",
+            "valuation_snapshot": val_snap,
+        }
+
+    except Exception as e:
+        import traceback
+        log(f"  [ERR] yahooquery error: {e}\n{traceback.format_exc()}")
+        return None
+
+
+# Keep a stub for backward compatibility (if anything still calls _fetch_yfinance)
+def _fetch_yfinance(ticker: str, years: int, log_fn=None) -> Optional[Dict]:
+    return _fetch_yahooquery(ticker, years, log_fn)
+
+
+# ── FMP fetcher (secondary for global) ───────────────────────────────────────
+
+def _fmp_get(path, params=None):
+    if not FMP_API_KEY: return None
+    p = {"apikey": FMP_API_KEY}
+    if params: p.update(params)
+    try:
+        r = requests.get(f"{FMP_BASE}{path}", params=p, timeout=REQUEST_TIMEOUT, headers=HEADERS)
+        r.raise_for_status()
+        return r.json()
+    except Exception:
+        return None
+
+
+def _normalise_fmp(raw, years):
+    inc = list(reversed(raw.get("income", [])))[:years]
+    bal = list(reversed(raw.get("balance", [])))[:years]
+    cf  = list(reversed(raw.get("cashflow", [])))[:years]
+
+    def col(lst, key):
+        return [_s(d.get(key)) for d in lst]
+
+    year_labels = [d.get("calendarYear", str(d.get("date",""))[:4]) for d in inc]
+    income = {
+        "revenue":            col(inc, "revenue"),
+        "cost_of_revenue":    col(inc, "costOfRevenue"),
+        "gross_profit":       col(inc, "grossProfit"),
+        "operating_expenses": col(inc, "operatingExpenses"),
+        "operating_income":   col(inc, "operatingIncome"),
+        "ebitda":             col(inc, "ebitda"),
+        "depreciation_amort": col(inc, "depreciationAndAmortization"),
+        "interest_expense":   col(inc, "interestExpense"),
+        "pretax_income":      col(inc, "incomeBeforeTax"),
+        "tax_expense":        col(inc, "incomeTaxExpense"),
+        "net_income":         col(inc, "netIncome"),
+        "diluted_eps":        col(inc, "epsdiluted"),
+        "diluted_shares":     col(inc, "weightedAverageShsOutDil"),
+    }
+    balance = {
+        "cash_and_equivalents":       col(bal, "cashAndCashEquivalents"),
+        "short_term_investments":     col(bal, "shortTermInvestments"),
+        "receivables":                col(bal, "netReceivables"),
+        "inventory":                  col(bal, "inventory"),
+        "other_current_assets":       col(bal, "otherCurrentAssets"),
+        "total_current_assets":       col(bal, "totalCurrentAssets"),
+        "ppe_net":                    col(bal, "propertyPlantEquipmentNet"),
+        "goodwill":                   col(bal, "goodwill"),
+        "intangibles":                col(bal, "intangibleAssets"),
+        "other_long_term_assets":     col(bal, "otherNonCurrentAssets"),
+        "total_assets":               col(bal, "totalAssets"),
+        "accounts_payable":           col(bal, "accountPayables"),
+        "short_term_debt":            col(bal, "shortTermDebt"),
+        "other_current_liabilities":  col(bal, "otherCurrentLiabilities"),
+        "total_current_liabilities":  col(bal, "totalCurrentLiabilities"),
+        "long_term_debt":             col(bal, "longTermDebt"),
+        "other_long_term_liabilities":col(bal, "otherNonCurrentLiabilities"),
+        "total_liabilities":          col(bal, "totalLiabilities"),
+        "total_equity":               col(bal, "totalStockholdersEquity"),
+        "total_liab_and_equity":      col(bal, "totalLiabilitiesAndStockholdersEquity"),
+    }
+    cashflow = {
+        "net_income":              col(cf, "netIncome"),
+        "depreciation_amort":      col(cf, "depreciationAndAmortization"),
+        "stock_based_comp":        col(cf, "stockBasedCompensation"),
+        "working_capital_changes": col(cf, "changeInWorkingCapital"),
+        "other_operating":         col(cf, "otherWorkingCapital"),
+        "cash_from_operations":    col(cf, "netCashProvidedByOperatingActivities"),
+        "capex":                   col(cf, "capitalExpenditure"),
+        "acquisitions":            col(cf, "acquisitionsNet"),
+        "cash_from_investing":     col(cf, "netCashUsedForInvestingActivites"),
+        "debt_issuance_repayment": col(cf, "debtRepayment"),
+        "stock_buybacks":          col(cf, "commonStockRepurchased"),
+        "dividends_paid":          col(cf, "dividendsPaid"),
+        "cash_from_financing":     col(cf, "netCashUsedProvidedByFinancingActivities"),
+        "net_change_in_cash":      col(cf, "netChangeInCash"),
+        "free_cash_flow":          col(cf, "freeCashFlow"),
+    }
+    return {
+        "years": year_labels, "income_statement": income,
+        "balance_sheet": balance, "cash_flow": cashflow,
+        "currency": "USD", "units": "millions",
+        "source": "FMP API",
+        "valuation_snapshot": {},
+    }
+
+
+def _fetch_fmp(ticker, years, log_fn=None):
+    def log(m):
+        if log_fn: log_fn(m)
+    if not FMP_API_KEY:
+        return None
+    log("  -> FMP: fetching statements ...")
+    lim = years + 1
+    raw = {
+        "income":   _fmp_get(f"/v3/income-statement/{ticker}", {"limit": lim, "period": "annual"}) or [],
+        "balance":  _fmp_get(f"/v3/balance-sheet-statement/{ticker}", {"limit": lim, "period": "annual"}) or [],
+        "cashflow": _fmp_get(f"/v3/cash-flow-statement/{ticker}", {"limit": lim, "period": "annual"}) or [],
+    }
+    if not raw["income"]:
+        log("  [ERR] FMP returned no data")
+        return None
+    log("  [OK] FMP data ready")
+    data = _normalise_fmp(raw, years)
+    # Add valuation
+    rtm = (_fmp_get(f"/v3/ratios-ttm/{ticker}") or [{}])[0]
+    km  = (_fmp_get(f"/v3/key-metrics-ttm/{ticker}") or [{}])[0]
+    data["valuation_snapshot"] = {
+        "pe_ttm":         _s(rtm.get("peRatioTTM")),
+        "pb":             _s(rtm.get("priceToBookRatioTTM")),
+        "ps":             _s(rtm.get("priceToSalesRatioTTM")),
+        "ev_ebitda":      _s(km.get("enterpriseValueOverEBITDATTM")),
+        "ev_revenue":     _s(km.get("evToSalesTTM")),
+        "fcf_yield":      _s(km.get("freeCashFlowYieldTTM")),
+        "dividend_yield": _s(rtm.get("dividendYieldTTM")),
+    }
+    return data
+
+
+# ── screener.in (Indian fallback) ────────────────────────────────────────────
+
+def _scrape_screener(ticker: str, log_fn=None) -> Optional[Dict]:
+    def log(m):
+        if log_fn: log_fn(m)
+    # Ticker usually like RELIANCE.NS, extract RELIANCE
+    base_ticker = ticker.split(".")[0].upper()
+    for url in [f"https://www.screener.in/company/{base_ticker}/consolidated/",
+                f"https://www.screener.in/company/{base_ticker}/"]:
+        try:
+            log(f"  -> screener.in: {url}")
+            r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HEADERS)
+            if r.status_code == 200:
+                soup = BeautifulSoup(r.text, "lxml")
+                sections = {}
+                for sec_id, key in [("profit-loss","income"),("balance-sheet","balance"),
+                                     ("cash-flow","cashflow")]:
+                    sec = soup.find("section", {"id": sec_id})
+                    if sec:
+                        table = sec.find("table")
+                        if table:
+                            result = {}
+                            for row in table.find_all("tr"):
+                                cells = row.find_all(["td","th"])
+                                if len(cells) < 2: continue
+                                label = cells[0].get_text(strip=True)
+                                vals = []
+                                for c in cells[1:]:
+                                    t2 = c.get_text(strip=True).replace(",","").replace("+","")
+                                    try: vals.append(float(t2))
+                                    except: vals.append(None)
+                                if label: result[label] = vals
+                            sections[key] = result
+                            if "years" not in sections:
+                                hdr = table.find("tr")
+                                if hdr:
+                                    sections["years"] = [c.get_text(strip=True)
+                                                         for c in hdr.find_all(["th","td"])[1:]]
+                if sections:
+                    sections["source_url"] = url
+                    log("  [OK] screener.in scraped")
+                    return sections
+        except Exception as e:
+            log(f"  [ERR] screener.in error: {e}")
+    return None
+
+
+def _normalise_screener(raw, years):
+    yr = raw.get("years", [])[:years]
+    n = len(yr)
+    inc = raw.get("income", {})
+    bal = raw.get("balance", {})
+    cf  = raw.get("cashflow", {})
+
+    def row(*keys):
+        for k in keys:
+            if k in inc: return inc[k][:n]
+            if k in bal: return bal[k][:n]
+            if k in cf:  return cf[k][:n]
+        return [None]*n
+
+    income = {
+        "revenue":            row("Revenue","Net Sales","Sales"),
+        "cost_of_revenue":    row("Expenses","Cost of Goods Sold"),
+        "gross_profit":       row("Gross Profit"),
+        "operating_expenses": [None]*n,
+        "operating_income":   row("Operating Profit","EBIT","PBDIT"),
+        "ebitda":             row("PBDIT","EBITDA"),
+        "depreciation_amort": row("Depreciation"),
+        "interest_expense":   row("Interest"),
+        "pretax_income":      row("Profit before tax","PBT"),
+        "tax_expense":        row("Tax"),
+        "net_income":         row("Net Profit","PAT"),
+        "diluted_eps":        row("EPS in Rs"),
+        "diluted_shares":     [None]*n,
+    }
+    balance = {
+        "cash_and_equivalents":       row("Cash Equivalents"),
+        "short_term_investments":     [None]*n,
+        "receivables":                [None]*n,
+        "inventory":                  [None]*n,
+        "other_current_assets":       [None]*n,
+        "total_current_assets":       row("Current Assets"),
+        "ppe_net":                    row("Fixed Assets"),
+        "goodwill":                   [None]*n,
+        "intangibles":                [None]*n,
+        "other_long_term_assets":     [None]*n,
+        "total_assets":               row("Total Assets","Balance Sheet Size"),
+        "accounts_payable":           [None]*n,
+        "short_term_debt":            row("Short Term Borrowing"),
+        "other_current_liabilities":  [None]*n,
+        "total_current_liabilities":  row("Current Liabilities"),
+        "long_term_debt":             row("Borrowings","Long Term Borrowing"),
+        "other_long_term_liabilities":[None]*n,
+        "total_liabilities":          row("Total Liabilities"),
+        "total_equity":               row("Equity Capital","Total Equity"),
+        "total_liab_and_equity":      row("Total Assets","Balance Sheet Size"),
+    }
+    cfo_vals = row("Cash from Operating Activity","Cash from Operations")
+    cap_vals = row("Fixed Assets Purchased","Capital Expenditure")
+    fcf_vals = [(c - abs(x)) if c is not None and x is not None else None
+                for c, x in zip(cfo_vals, cap_vals)]
+    cashflow = {
+        "net_income":              income["net_income"],
+        "depreciation_amort":      income["depreciation_amort"],
+        "stock_based_comp":        [None]*n,
+        "working_capital_changes": [None]*n,
+        "other_operating":         [None]*n,
+        "cash_from_operations":    cfo_vals,
+        "capex":                   cap_vals,
+        "acquisitions":            [None]*n,
+        "cash_from_investing":     row("Cash from Investing Activity"),
+        "debt_issuance_repayment": [None]*n,
+        "stock_buybacks":          [None]*n,
+        "dividends_paid":          row("Dividend Amount"),
+        "cash_from_financing":     row("Cash from Financing Activity"),
+        "net_change_in_cash":      row("Net Cash Flow"),
+        "free_cash_flow":          fcf_vals,
+    }
+    return {
+        "years": yr, "income_statement": income, "balance_sheet": balance, "cash_flow": cashflow,
+        "currency": "INR", "units": "crores",
+        "source": f"screener.in ({raw.get('source_url','')})",
+        "valuation_snapshot": {},
+    }
+
+# ── stockanalysis.com (Global fallback for >4 years) ──────────────────────────
+
+def _scrape_stockanalysis(ticker: str, years: int, log_fn=None) -> Optional[Dict]:
+    def log(m):
+        if log_fn: log_fn(m)
+    base_ticker = ticker.split(".")[0].lower()
+    
+    urls = {
+        "income": f"https://stockanalysis.com/stocks/{base_ticker}/financials/",
+        "balance": f"https://stockanalysis.com/stocks/{base_ticker}/financials/balance-sheet/",
+        "cashflow": f"https://stockanalysis.com/stocks/{base_ticker}/financials/cash-flow-statement/"
+    }
+    
+    raw = {}
+    try:
+        for key, url in urls.items():
+            log(f"  -> stockanalysis.com: {url}")
+            r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HEADERS)
+            if r.status_code != 200:
+                log(f"  [ERR] stockanalysis.com returned {r.status_code} for {key}")
+                return None
+            soup = BeautifulSoup(r.text, "lxml")
+            table = soup.find("table")
+            if not table: return None
+            
+            # Parse headers (Skip TTM, take `years` count)
+            thead = table.find("thead")
+            ths = thead.find("tr").find_all("th")
+            # Index 0 is "Fiscal Year", Index 1 is "TTM" (usually). Index 2+ are the years.
+            if len(ths) < 3: return None
+            
+            # Only need to extract year labels once
+            if "years" not in raw:
+                # Get the years newest to oldest, then reverse to oldest to newest
+                year_labels = [th.get_text(strip=True) for th in ths[2:2+years]]
+                raw["years"] = list(reversed(year_labels))
+                
+            data_rows = {}
+            tbody = table.find("tbody")
+            for tr in tbody.find_all("tr"):
+                tds = tr.find_all("td")
+                if len(tds) < 3: continue
+                label = tds[0].get_text(strip=True)
+                vals = []
+                for td in tds[2:2+years]:
+                    text = td.get_text(strip=True).replace(",", "")
+                    if text == "-" or text == "":
+                        vals.append(None)
+                    else:
+                        try: vals.append(float(text))
+                        except: vals.append(None)
+                data_rows[label] = list(reversed(vals))
+            raw[key] = data_rows
+            
+    except Exception as e:
+        log(f"  [ERR] stockanalysis error: {e}")
+        return None
+        
+    n = len(raw.get("years", []))
+    if n == 0: return None
+    
+    def row(sec, *keys):
+        for k in keys:
+            if k in raw[sec]: return raw[sec][k]
+        return [None]*n
+
+    income = {
+        "revenue":            row("income", "Revenue"),
+        "cost_of_revenue":    row("income", "Cost of Revenue"),
+        "gross_profit":       row("income", "Gross Profit"),
+        "operating_expenses": row("income", "Operating Expenses"),
+        "operating_income":   row("income", "Operating Income"),
+        "ebitda":             row("income", "EBITDA"),
+        "depreciation_amort": row("income", "Depreciation & Amortization"),
+        "interest_expense":   row("income", "Interest Expense / Income"),
+        "pretax_income":      row("income", "Pretax Income"),
+        "tax_expense":        row("income", "Income Tax"),
+        "net_income":         row("income", "Net Income"),
+        "diluted_eps":        row("income", "EPS (Diluted)"),
+        "diluted_shares":     row("income", "Shares Outstanding (Diluted)"),
+    }
+    balance = {
+        "cash_and_equivalents":       row("balance", "Cash & Equivalents"),
+        "short_term_investments":     row("balance", "Short-Term Investments"),
+        "receivables":                row("balance", "Accounts Receivable"),
+        "inventory":                  row("balance", "Inventory"),
+        "other_current_assets":       row("balance", "Other Current Assets"),
+        "total_current_assets":       row("balance", "Total Current Assets"),
+        "ppe_net":                    row("balance", "Property, Plant & Equipment"),
+        "goodwill":                   row("balance", "Goodwill and Intangibles", "Goodwill"),
+        "intangibles":                [None]*n,
+        "other_long_term_assets":     row("balance", "Other Long-Term Assets"),
+        "total_assets":               row("balance", "Total Assets"),
+        "accounts_payable":           row("balance", "Accounts Payable"),
+        "short_term_debt":            row("balance", "Current Debt"),
+        "other_current_liabilities":  row("balance", "Other Current Liabilities"),
+        "total_current_liabilities":  row("balance", "Total Current Liabilities"),
+        "long_term_debt":             row("balance", "Long-Term Debt"),
+        "other_long_term_liabilities":row("balance", "Other Long-Term Liabilities"),
+        "total_liabilities":          row("balance", "Total Liabilities"),
+        "total_equity":               row("balance", "Shareholders' Equity"),
+        "total_liab_and_equity":      row("balance", "Total Liabilities and Equity"),
+    }
+    cashflow = {
+        "net_income":              row("cashflow", "Net Income"),
+        "depreciation_amort":      row("cashflow", "Depreciation & Amortization"),
+        "stock_based_comp":        row("cashflow", "Share-Based Compensation"),
+        "working_capital_changes": [None]*n,
+        "other_operating":         row("cashflow", "Other Operating Activities"),
+        "cash_from_operations":    row("cashflow", "Operating Cash Flow"),
+        "capex":                   row("cashflow", "Capital Expenditures"),
+        "acquisitions":            row("cashflow", "Acquisitions"),
+        "cash_from_investing":     row("cashflow", "Investing Cash Flow"),
+        "debt_issuance_repayment": row("cashflow", "Net Debt Issued/Paid"),
+        "stock_buybacks":          row("cashflow", "Common Stock Issued/Repurchased", "Stock Repurchases"),
+        "dividends_paid":          row("cashflow", "Dividends Paid"),
+        "cash_from_financing":     row("cashflow", "Financing Cash Flow"),
+        "net_change_in_cash":      row("cashflow", "Net Change in Cash"),
+        "free_cash_flow":          row("cashflow", "Free Cash Flow"),
+    }
+    return {
+        "years": raw["years"], "income_statement": income, "balance_sheet": balance, "cash_flow": cashflow,
+        "currency": "USD", "units": "millions",
+        "source": "stockanalysis.com",
+        "valuation_snapshot": {},
+    }
+
+# ── Peer snapshot ─────────────────────────────────────────────────────────────
+
+def fetch_peer_snapshot(ticker: str) -> Optional[Dict]:
+    try:
+        import yahooquery as yq
+        t = yq.Ticker(ticker)
+        price_data   = t.price.get(ticker, {})
+        profile_data = t.summary_profile.get(ticker, {})
+        key_stats    = t.key_stats.get(ticker, {})
+        fin_data     = t.financial_data.get(ticker, {})
+        summary_det  = t.summary_detail.get(ticker, {})
+
+        name = price_data.get("longName") or price_data.get("shortName", "")
+        if not name:
+            return None
+
+        # Try to get latest revenue from income statement
+        rev = None
+        try:
+            inc = t.income_statement(frequency='annual')
+            if inc is not None and not isinstance(inc, str) and not inc.empty:
+                annual_inc = inc[inc.get('periodType', '') == '12M'] if 'periodType' in inc.columns else inc
+                rev_col = annual_inc['TotalRevenue'].dropna() if 'TotalRevenue' in annual_inc.columns else None
+                if rev_col is not None and not rev_col.empty:
+                    rev = float(rev_col.iloc[-1])
+        except Exception:
+            pass
+
+        ratios_ttm = (_fmp_get(f"/v3/ratios-ttm/{ticker}") or [{}])[0]
+
+        return {
+            "name": name,
+            "ticker": ticker,
+            "rationale": f"Industry peer ({profile_data.get('industry', '')})",
+            "latest_year": "",
+            "currency": price_data.get("currency", "INR"),
+            "units": "actuals",
+            "revenue": _s(rev),
+            "gross_margin": _s(fin_data.get("grossMargins")),
+            "operating_margin": _s(fin_data.get("operatingMargins")),
+            "net_margin": _s(fin_data.get("profitMargins")),
+            "roe": _s(fin_data.get("returnOnEquity")),
+            "roic": _s(ratios_ttm.get("returnOnCapitalEmployedTTM")),
+            "debt_to_equity": _s(key_stats.get("debtToEquity")),
+            "current_ratio": _s(fin_data.get("currentRatio")),
+            "fcf_margin": None,
+            "revenue_growth_3y_cagr": _s(fin_data.get("revenueGrowth")),
+            "pe_ttm": _s(summary_det.get("trailingPE")),
+            "ev_ebitda": _s(key_stats.get("enterpriseToEbitda")),
+        }
+    except Exception:
+        return None
+
+
+def fetch_valuation(ticker: str) -> Dict:
+    try:
+        import yahooquery as yq
+        t = yq.Ticker(ticker)
+        key_stats   = t.key_stats.get(ticker, {})
+        summary_det = t.summary_detail.get(ticker, {})
+        fin_data    = t.financial_data.get(ticker, {})
+        return {
+            "pe_ttm":         _s(summary_det.get("trailingPE")),
+            "pb":             _s(key_stats.get("priceToBook")),
+            "ps":             _s(key_stats.get("priceToSalesTrailing12Months")),
+            "ev_ebitda":      _s(key_stats.get("enterpriseToEbitda")),
+            "ev_revenue":     _s(key_stats.get("enterpriseToRevenue")),
+            "fcf_yield":      None,
+            "dividend_yield": _s(summary_det.get("dividendYield")),
+        }
+    except Exception:
+        return {}
+
+def _scrape_edgar_10k(ticker: str, log_fn=None) -> str:
+    """Fetch the latest 10-K for a US company from SEC EDGAR and extract a chunk of text."""
+    headers = {"User-Agent": "FinancialAnalyzer/1.0 (contact@example.com)"}
+    try:
+        if log_fn: log_fn(f"  -> Looking up SEC CIK for {ticker}...")
+        r = requests.get("https://www.sec.gov/files/company_tickers.json", headers=headers, timeout=10)
+        tickers = r.json()
+        cik = None
+        for k, v in tickers.items():
+            if v["ticker"].upper() == ticker.upper():
+                cik = str(v["cik_str"]).zfill(10)
+                break
+        
+        if not cik:
+            if log_fn: log_fn("  [ERR] CIK not found in SEC database.")
+            return ""
+            
+        if log_fn: log_fn(f"  -> Found CIK {cik}. Fetching recent filings...")
+        sub_r = requests.get(f"https://data.sec.gov/submissions/CIK{cik}.json", headers=headers, timeout=10)
+        data = sub_r.json()
+        filings = data.get("filings", {}).get("recent", {})
+        
+        form_types = filings.get("form", [])
+        accessions = filings.get("accessionNumber", [])
+        primary_docs = filings.get("primaryDocument", [])
+        
+        idx = -1
+        for i, form in enumerate(form_types):
+            if form == "10-K":
+                idx = i
+                break
+                
+        if idx == -1:
+            if log_fn: log_fn("  [ERR] No 10-K found in recent filings.")
+            return ""
+            
+        acc_num = accessions[idx]
+        acc_num_no_dash = acc_num.replace("-", "")
+        doc = primary_docs[idx]
+        
+        url = f"https://www.sec.gov/Archives/edgar/data/{int(cik)}/{acc_num_no_dash}/{doc}"
+        if log_fn: log_fn(f"  -> Downloading 10-K from {url}...")
+        
+        doc_r = requests.get(url, headers=headers, timeout=20)
+        from bs4 import BeautifulSoup
+        soup = BeautifulSoup(doc_r.content, "lxml")
+        
+        text = soup.get_text(separator="\n", strip=True)
+        import re
+        match = re.search(r'(?i)Item\s+7\.\s*Management[’\']s\s+Discussion', text)
+        if match:
+            start = match.start()
+            extract = text[start:start+40000] 
+        else:
+            extract = text[len(text)//4 : len(text)//4 + 40000]
+            
+        if log_fn: log_fn("  [OK] 10-K text extracted successfully.")
+        return extract
+    except Exception as e:
+        if log_fn: log_fn(f"  [ERR] Failed to scrape EDGAR: {str(e)[:80]}")
+        return ""
+
+
+# ── Main Fetcher ──────────────────────────────────────────────────────────────
+
+def fetch_all(meta: CompanyMeta, years: int = DEFAULT_YEARS,
+              force_refresh: bool = False, log_fn=None) -> Dict:
+    def log(m):
+        if log_fn: log_fn(m)
+
+    if not meta.is_listed:
+        msg = (f"'{meta.name}' is unlisted/private. Public data not available. "
+               "Try the exact BSE/NSE ticker.")
+        log(f"  ⚠ {msg}")
+        return {"warnings": [msg], "years": [], "income_statement": {},
+                "balance_sheet": {}, "cash_flow": {}}
+
+    ticker = meta.ticker
+    cache_key = f"full_{years}yr"
+    if not force_refresh:
+        cached = get_cached(ticker, cache_key)
+        if cached:
+            log(f"  [OK] Loaded cached data for {ticker}")
+            return cached
+
+    warnings = []
+    result = None
+
+    # Re-prioritized Fetching Logic:
+    
+    # 1. Indian Stocks: Try screener.in first (up to 10+ years of data)
+    if meta.currency == "INR":
+        log("  -> Indian Stock detected. Prioritizing screener.in ...")
+        scraped = _scrape_screener(ticker, log_fn)
+        if scraped:
+            result = _normalise_screener(scraped, years)
+            warnings.append("Data sourced from screener.in. Verify against company filings.")
+            
+        # Fallback to yfinance if screener fails or yields nothing
+        if not result or not result.get("years"):
+            log("  -> screener.in failed. Falling back to yahooquery ...")
+            result = _fetch_yahooquery(ticker, years, log_fn)
+            if result and len(result.get("years", [])) < years:
+                warnings.append(f"yahooquery only provides {len(result['years'])} years of historical data.")
+
+    # 2. Global/US Stocks
+    else:
+        # If requesting > 4 years, try stockanalysis.com first because yahooquery limits to 4 years
+        if years > 4:
+            log("  -> Horizon > 4 years requested. Prioritizing stockanalysis.com for global stock ...")
+            result = _scrape_stockanalysis(ticker, years, log_fn)
+            if result:
+                warnings.append("Data sourced from stockanalysis.com. Verify against company filings.")
+        
+        # Fallback to yahooquery
+        if not result or not result.get("years"):
+            log("  -> Trying yahooquery ...")
+            result = _fetch_yahooquery(ticker, years, log_fn)
+            if result and len(result.get("years", [])) < years:
+                warnings.append(f"yahooquery only provides {len(result['years'])} years of historical data.")
+                
+        # Fallback to FMP
+        if (not result or not result.get("years")) and FMP_API_KEY:
+            result = _fetch_fmp(ticker, years, log_fn)
+
+    # 3. Final validation
+    if not result or not result.get("years"):
+        msg = (f"Could not fetch data for '{meta.name}' ({ticker}). "
+               "Check ticker, or the company may have limited public data coverage.")
+        log(f"  ⚠ {msg}")
+        return {"warnings": [msg], "years": [], "income_statement": {},
+                "balance_sheet": {}, "cash_flow": {}}
+
+    # 4. Optional Annual Report Scrape (for context)
+    if meta.currency != "INR":
+        ar_context = _scrape_edgar_10k(ticker, log_fn)
+        if ar_context:
+            result["annual_report_context"] = ar_context
+            warnings.append("AI analysis enhanced with raw MD&A excerpts from SEC EDGAR 10-K.")
+
+    result["warnings"] = warnings
+    result.setdefault("currency", meta.currency)
+    set_cache(ticker, cache_key, result)
+    log("  [OK] Data cached for 7 days")
+    return result