"""Market data layer — yfinance primary (global coverage), Alpha Vantage + Sharadar/MarketFM
panel as US-only enrichment. Everything cached with a TTL; every source fails open.

Patterns reused from: marketguru_space/app.py (fact_sheet / news_block), crucible
predict_monthly.api_call (TTL file cache that refuses to cache error payloads),
marketfm_space (ThreadPool fan-out), deepquant (HF dataset snapshot read).
"""
from __future__ import annotations

import concurrent.futures as cf
import json
import os
import time
import urllib.parse
import urllib.request
from pathlib import Path

import numpy as np
import pandas as pd

from .llm import CTX

AV_KEY = os.environ.get("ALPHAVANTAGE_API_KEY", "")
HF_TOKEN = os.environ.get("HF_TOKEN", "")
PANEL_REPO = os.environ.get("MARKETFM_REPO", "s-ttp/marketfm-panel")
CACHE_DIR = Path(os.environ.get("CACHE_DIR", Path(__file__).resolve().parent.parent / "data" / "cache"))
CACHE_TTL_H = float(os.environ.get("CACHE_TTL_HOURS", "6"))
DATA_DIR = Path(__file__).resolve().parent.parent / "data"


# ---------------- TTL file cache (never caches error/empty payloads) ----------------

def _cache_get(key: str):
    p = CACHE_DIR / f"{key}.json"
    try:
        if p.exists() and time.time() - p.stat().st_mtime < CACHE_TTL_H * 3600:
            return json.loads(p.read_text())
    except Exception:
        pass
    return None


def _cache_put(key: str, val) -> None:
    if not val:
        return
    try:
        CACHE_DIR.mkdir(parents=True, exist_ok=True)
        (CACHE_DIR / f"{key}.json").write_text(json.dumps(val))
    except Exception:
        pass


# ---------------- static sector map (slipstream, 974 US tickers, zero API calls) ----------------

def _sector_map() -> dict:
    try:
        return json.loads((DATA_DIR / "sector_map.json").read_text())
    except Exception:
        return {}


SECTOR_MAP = _sector_map()


# ---------------- Alpha Vantage (US enrichment) ----------------

def _av(params: dict) -> dict:
    params = {**params, "apikey": AV_KEY}
    url = "https://www.alphavantage.co/query?" + urllib.parse.urlencode(params)
    with urllib.request.urlopen(url, timeout=25, context=CTX) as r:
        return json.loads(r.read())


def _f(x):
    try:
        if x in (None, "None", "-", "", "NaN"):
            return None
        v = float(str(x).replace("%", "").replace(",", ""))
        return v if v == v else None
    except Exception:
        return None


# ---------------- MarketFM panel snapshot (optional, US top-1000) ----------------

_panel_cache: dict | None = None


def load_panel() -> dict:
    """{ticker: {sector, marketcap, pred_excess_12m, xs_rank_pct, decile, as_of}} or {}."""
    global _panel_cache
    if _panel_cache is not None:
        return _panel_cache
    _panel_cache = {}
    if not HF_TOKEN:
        return _panel_cache
    try:
        from huggingface_hub import hf_hub_download
        p = hf_hub_download(PANEL_REPO, "marketfm_snapshot.parquet",
                            repo_type="dataset", token=HF_TOKEN)
        df = pd.read_parquet(p)
        # live snapshot columns: rank, ticker, name, sector, price, marketcap, pred, pctile, ...
        fac = [c for c in df.columns if c.startswith("n_")] + ["sector_pctile"]
        _panel_cache = {str(r["ticker"]).upper(): {
            "name": r.get("name"), "sector": r.get("sector"), "marketcap": r.get("marketcap"),
            "pred_excess_12m": r.get("pred"), "xs_rank_pct": r.get("pctile"),
            "rank": r.get("rank"),
            # normalized cross-sectional factor scores (−1…+1) → peer percentiles & tilts
            "factors": {c: float(r[c]) for c in fac if isinstance(r.get(c), (int, float))
                        and r[c] == r[c]},
        } for r in df.to_dict("records")}
        print(f"  marketfm panel loaded: {len(_panel_cache)} tickers", flush=True)
    except Exception as e:
        print(f"  marketfm panel unavailable ({type(e).__name__}) — continuing without", flush=True)
    return _panel_cache


# ---------------- FX to USD (mixed-currency portfolios must weigh in one unit) ----------------

_fx_cache: dict[str, float] = {}


def fx_to_usd(currency) -> float:
    """1 unit of `currency` in USD. 1.0 for USD; degrades to 1.0 (with a warning) if the
    FX quote is unavailable — weights then mix currencies, better than crashing."""
    cur = str(currency or "USD")
    if cur in _fx_cache:
        return _fx_cache[cur]
    base, scale = cur.upper(), 1.0
    if cur in ("GBp", "GBX"):        # LSE quotes in pence
        base, scale = "GBP", 0.01
    rate = 1.0
    if base != "USD":
        c = _cache_get(f"fx_{base}")
        if c:
            rate = float(c["rate"])
        else:
            try:
                import yfinance as yf
                rate = float(yf.Ticker(f"{base}USD=X").fast_info["lastPrice"])
                _cache_put(f"fx_{base}", {"rate": rate})
            except Exception:
                print(f"  FX {base}USD unavailable — weights may mix currencies", flush=True)
    _fx_cache[cur] = rate * scale
    return _fx_cache[cur]


# ---------------- Slipstream smart-money candidates (13F + insider, quality-gated) ----------------

_slip_cache: dict | None = None


def load_slipstream() -> dict:
    """{ticker: {sector, score, size}} — slipstream's quality-gated smart-money candidate
    pool (13F superinvestor + insider accumulation, ROE/D-E gated). Local checkout first
    (dev), then the s-ttp/slipstream Space repo. {} on any failure."""
    global _slip_cache
    if _slip_cache is not None:
        return _slip_cache
    _slip_cache = {}
    try:
        local = "/Users/sttp/Desktop/AI Projects/mitpoc/slipstream/data/candidates.parquet"
        p = local if os.path.exists(local) else None
        if not p:
            from huggingface_hub import hf_hub_download
            p = hf_hub_download("s-ttp/slipstream", "data/candidates.parquet",
                                repo_type="space", token=HF_TOKEN or None)
        df = pd.read_parquet(p)
        _slip_cache = {str(r["ticker"]).upper(): {"sector": r.get("sector"),
                                                  "score": float(r.get("score", 0)),
                                                  "size": r.get("size")}
                       for r in df.to_dict("records")}
        print(f"  slipstream candidates loaded: {len(_slip_cache)}", flush=True)
    except Exception as e:
        print(f"  slipstream candidates unavailable ({type(e).__name__})", flush=True)
    return _slip_cache


# ---------------- per-holding fundamentals ----------------

_YF_KEYS = {
    "name": "shortName", "sector": "sector", "industry": "industry",
    "mcap": "marketCap", "pe": "trailingPE", "fwd_pe": "forwardPE", "peg": "pegRatio",
    "pb": "priceToBook", "ps": "priceToSalesTrailing12Months", "ev_ebitda": "enterpriseToEbitda",
    "div_yield": "dividendYield", "roe": "returnOnEquity", "roa": "returnOnAssets",
    "op_margin": "operatingMargins", "profit_margin": "profitMargins", "beta": "beta",
    "hi52": "fiftyTwoWeekHigh", "lo52": "fiftyTwoWeekLow", "sma50": "fiftyDayAverage",
    "sma200": "twoHundredDayAverage", "rev_growth": "revenueGrowth",
    "eps_growth": "earningsGrowth", "eps_ttm": "trailingEps", "eps_fwd": "forwardEps",
    "target": "targetMeanPrice", "de": "debtToEquity", "payout": "payoutRatio",
    "currency": "currency", "bvps": "bookValue", "avg_volume": "averageVolume",
    "shares_out": "sharesOutstanding",
}

# ---------------- field sanity gates (institutional data-integrity rule #1) ----------------
# Every fundamentals field has a plausible range; a value outside it is a unit or feed
# error (D/E in percent, EPS in pounds vs pence prices, yield as a fraction — all seen
# in production). Out-of-range values are NULLED and flagged, never fed to the DCF or
# the analysts. Ranges are deliberately generous; they catch units, not opinions.
SANITY: dict[str, tuple[float, float]] = {
    "pe": (0.0, 500.0), "fwd_pe": (0.0, 500.0), "peg": (-50.0, 50.0), "pb": (0.0, 200.0),
    "ps": (0.0, 200.0), "ev_ebitda": (-500.0, 500.0), "div_yield": (0.0, 25.0),
    "roe": (-5.0, 5.0), "roa": (-5.0, 5.0), "op_margin": (-5.0, 1.0),
    "profit_margin": (-5.0, 1.0), "beta": (-3.0, 5.0), "rev_growth": (-1.0, 20.0),
    "eps_growth": (-1.0, 50.0), "de": (0.0, 20.0), "payout": (-1.0, 5.0),
    "avg_volume": (0.0, 1e11), "shares_out": (0.0, 1e12),
}


def sanitize(d: dict) -> dict:
    """Null out-of-range fields, record what was dropped in d['data_flags']. Price-relative
    fields (52w range, SMAs, targets) must sit within 0.2×–5× the price."""
    flags = list(d.get("data_flags") or [])
    for k, (lo, hi) in SANITY.items():
        v = d.get(k)
        if isinstance(v, (int, float)) and (v != v or v < lo or v > hi):
            flags.append(f"{k}={v:g} outside [{lo:g},{hi:g}] — dropped")
            d[k] = None
    price = d.get("price")
    if isinstance(price, (int, float)) and price > 0:
        for k in ("hi52", "lo52", "sma50", "sma200", "target"):
            v = d.get(k)
            if isinstance(v, (int, float)) and not (price * 0.2 <= v <= price * 5):
                flags.append(f"{k}={v:g} implausible vs price {price:g} — dropped")
                d[k] = None
        eps = d.get("eps_ttm")
        if isinstance(eps, (int, float)) and eps > 0 and price / eps > 2000:
            flags.append(f"eps_ttm={eps:g} implies P/E {price / eps:.0f} — unit error, dropped")
            d["eps_ttm"] = None
    if flags:
        d["data_flags"] = flags
    return d


_FAST_KEYS = [("price", "lastPrice"), ("currency", "currency"), ("mcap", "marketCap"),
              ("sma50", "fiftyDayAverage"), ("sma200", "twoHundredDayAverage"),
              ("hi52", "yearHigh"), ("lo52", "yearLow")]


class DataUnavailable(ValueError):
    """No price AND the market-data layer itself is not answering (a canary quote fails):
    an outage, not a verdict on the ticker."""


_alive_memo = {"t": 0.0, "ok": True}


def _data_alive() -> bool:
    """Canary: can we quote SPY right now? Memoised for 60s so a burst of misses costs one call."""
    now = time.time()
    if now - _alive_memo["t"] < 60:
        return _alive_memo["ok"]
    ok = False
    try:
        if NDL_KEY:
            from . import sharadar_prices as SP
            ok = bool((SP.snapshot("SPY") or {}).get("price"))
    except Exception:
        ok = False
    if not ok:
        try:
            import yfinance as yf
            ok = bool(_f(yf.Ticker("SPY").fast_info["lastPrice"]))
        except Exception:
            ok = False
    _alive_memo.update(t=now, ok=ok)
    return ok


def fetch_holding(ticker: str) -> dict:
    """Fundamentals + price for one ticker. Layered for resilience:
    1) yfinance fast_info (chart API — rarely rate-limited): price, mcap, SMAs, 52w range
    2) yfinance .info (quoteSummary — often 429s): full fundamentals, best-effort w/ one retry
    3) AV OVERVIEW fill-in for plain US symbols; 4) marketfm panel rank when covered.
    Cached; raises only if no price can be found at all."""
    t = ticker.upper()
    cached = _cache_get(f"hold_{t}")
    if cached:
        return cached
    d: dict = {"ticker": t}
    if NDL_KEY and "." not in t:    # US: Sharadar price/meta snapshot — no rate-limited call needed
        try:
            from . import sharadar_prices as SP
            d.update(SP.snapshot(t))
        except Exception as e:
            print(f"  sharadar snapshot {t} failed ({type(e).__name__})", flush=True)
    if not d.get("price"):
        d.update(_av_primary(t))    # Alpha Vantage premium (mapped exchanges, and US when Sharadar lacks it)
    else:
        ov = _cache_get(f"avov_{t}")        # analyst targets / growth from the overview, if pre-warmed
        if isinstance(ov, dict):
            for k, v in ov.items():
                if v is not None and d.get(k) is None:
                    d[k] = v
    try:
        import yfinance as yf
        tk = yf.Ticker(t)
        try:
            fi = tk.fast_info
            for k, fk in _FAST_KEYS:
                if d.get(k) is not None:
                    continue
                try:
                    v = fi[fk]
                    d[k] = v if k == "currency" else _f(v)
                except Exception:
                    pass
            try:
                d["quote_type"] = str(fi["quoteType"] or "").upper() or None
            except Exception:
                pass
        except Exception:
            pass
        info = {}
        # Sharadar already supplied name/type/price for US names (and SF1 supplies EPS below):
        # the slow quoteSummary call is only for names nothing else covers
        need_info = ((d.get("price_source") != "sharadar"
                      and any(d.get(k) is None for k in ("name", "sector", "industry", "eps_ttm", "avg_volume", "quote_type")))
                     or (str(d.get("quote_type") or "EQUITY").upper() == "EQUITY"
                         and str(d.get("sector") or "").title() in ("", "Unknown")))   # Sharadar lacks a sector: fill it
        for attempt in range(2 if need_info else 0):  # quoteSummary 429s freely; one spaced retry, then shrug
            try:
                info = tk.info or {}
                break
            except Exception as e:
                if attempt == 0:
                    time.sleep(2.5)
                else:
                    print(f"  yfinance info {t} unavailable ({type(e).__name__})", flush=True)
        for k, yk in _YF_KEYS.items():
            v = info.get(yk)
            if v is None or d.get(k) is not None:
                continue
            d[k] = v if k in ("name", "sector", "industry", "currency") else _f(v)
        if info.get("quoteType") and not d.get("quote_type"):
            d["quote_type"] = str(info["quoteType"]).upper()
        mt = info.get("regularMarketTime")
        if isinstance(mt, (int, float)) and mt > 0:   # as-of stamp for the price
            d["price_asof"] = time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime(mt))
        if isinstance(d.get("de"), (int, float)):
            d["de"] = d["de"] / 100.0   # Yahoo reports debtToEquity in percent, always
        if not d.get("price"):
            d["price"] = _f(info.get("currentPrice") or info.get("regularMarketPrice")
                            or info.get("previousClose"))
    except Exception as e:
        print(f"  yfinance {t} failed ({type(e).__name__})", flush=True)
    # AV fill-in: only for plain US-style symbols, only fields yfinance left empty
    if AV_KEY and "." not in t:
        try:
            ov = _av({"function": "OVERVIEW", "symbol": t})
            if ov.get("Symbol"):
                for k, avk in [("name", "Name"), ("sector", "Sector"), ("pe", "PERatio"),
                               ("fwd_pe", "ForwardPE"), ("peg", "PEGRatio"), ("pb", "PriceToBookRatio"),
                               ("ev_ebitda", "EVToEBITDA"), ("roe", "ReturnOnEquityTTM"),
                               ("profit_margin", "ProfitMargin"), ("op_margin", "OperatingMarginTTM"),
                               ("beta", "Beta"), ("eps_ttm", "EPS"), ("target", "AnalystTargetPrice"),
                               ("rev_growth", "QuarterlyRevenueGrowthYOY"),
                               ("eps_growth", "QuarterlyEarningsGrowthYOY"),
                               ("div_yield", "DividendYield"), ("mcap", "MarketCapitalization")]:
                    if d.get(k) in (None, "") :
                        d[k] = ov.get(avk) if k in ("name", "sector") else _f(ov.get(avk))
                        if k == "div_yield" and isinstance(d[k], (int, float)):
                            d[k] = d[k] * 100   # AV gives a fraction; yfinance (and we) use percent
                if not d.get("price"):
                    q = _av({"function": "GLOBAL_QUOTE", "symbol": t})
                    d["price"] = _f(q.get("Global Quote", {}).get("05. price"))
        except Exception:
            pass
    if not d.get("price"):  # last resort: the chart API via a tiny download
        try:
            import yfinance as yf
            h = yf.download(t, period="5d", interval="1d", progress=False, auto_adjust=True)
            if h is not None and len(h):
                close = h["Close"]
                d["price"] = _f(float(close.iloc[-1].iloc[0] if hasattr(close.iloc[-1], "iloc")
                                      else close.iloc[-1]))
        except Exception:
            pass
    # Sharadar SF1 (licensed, point-in-time, unit-consistent) is the PRIMARY fundamentals
    # source for plain US equities; the scraped feeds above become the fallback
    if NDL_KEY and "." not in t and str(d.get("quote_type") or "EQUITY").upper() == "EQUITY":
        sh = fetch_sharadar(t)
        if sh:
            for k, v in sh.items():
                if v is not None:
                    d[k] = v
            d["fund_source"] = "sharadar"
            try:   # US-listed ADRs: Sharadar per-share fields sit on the ordinary-share / local-
                   # currency basis (TSM EPS looked 4× too big) — derive them from the ADR price
                meta = _sharadar_ticker_meta(t)
                if "ADR" in str(meta.get("category") or "").upper():
                    d["adr"] = True
                    price = d.get("price")
                    if isinstance(price, (int, float)) and price > 0:
                        if isinstance(d.get("pe"), (int, float)) and d["pe"] > 0:
                            d["eps_ttm"] = round(price / d["pe"], 4)
                        else:
                            d.pop("eps_ttm", None)
                        if isinstance(d.get("pb"), (int, float)) and d["pb"] > 0:
                            d["bvps"] = round(price / d["pb"], 4)
                        else:
                            d.pop("bvps", None)
                        d.pop("fcfps", None)
                    d["fund_source"] = "sharadar (ADR — per-share values derived from ratios)"
            except Exception:
                pass
    d.setdefault("fund_source", "alphavantage" if d.get("fund_source_av") else "yfinance")
    if "." in t and NDL_KEY and str(d.get("quote_type") or "EQUITY").upper() == "EQUITY":
        try:
            bridge_adr_fundamentals(d)     # licensed ratios for the big EM/international names
        except Exception:
            pass
    d["fetched_at"] = time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime())
    if str(d.get("currency")) == "GBp":
        # Yahoo quotes LSE prices in pence but often reports EPS in pounds; the reported
        # P/E reveals which — rescale so price and EPS share a unit (BP.L: 538p, EPS 0.26
        # "GBP", P/E 20.7 → EPS is really 26p; without this the DCF is 100× too low)
        pe, eps, price = d.get("pe"), d.get("eps_ttm"), d.get("price")
        if pe and eps and price and eps > 0 and price / pe / eps > 30:
            for k in ("eps_ttm", "eps_fwd"):
                if isinstance(d.get(k), (int, float)):
                    d[k] = d[k] * 100
    if not d.get("name") and AV_KEY:        # name is what the analysts and the UI show — never blank
        try:
            want = av_symbol(t)
            for m in av_symbol_search(t.rsplit(".", 1)[0]):
                if m["av"] == want or m["yahoo"] == t:
                    d["name"] = m["name"]
                    break
        except Exception:
            pass
    if not d.get("sector"):
        d["sector"] = SECTOR_MAP.get(t, "Unknown")
    d["sector"] = str(d["sector"]).title()   # AV yells "TECHNOLOGY"; normalize across sources
    panel = load_panel().get(t)
    if panel:
        d["panel"] = {k: (float(v) if isinstance(v, (int, float, np.floating)) else v)
                      for k, v in panel.items() if v is not None}
    if not d.get("price"):
        if not _data_alive():      # SQ (renamed XYZ) is a miss; GOOG with every source down is an outage
            raise DataUnavailable(f"No live price for '{t}' and the market-data sources are not responding")
        raise ValueError(f"No live price found for '{t}' — check the ticker.")
    sanitize(d)
    _cache_put(f"hold_{t}", d)
    return d


# ---------------- Sharadar SF1 — primary US fundamentals (licensed, point-in-time) ----------------

NDL_KEY = os.environ.get("NASDAQ_DATA_LINK_API_KEY", "")
_SF1_COLS = ("ticker,datekey,calendardate,pe,pb,ps,evebitda,divyield,roe,roa,netmargin,"
             "grossmargin,de,eps,epsdil,bvps,fcfps,depamor,netinc,sharesbas,marketcap,revenue")


# Sharadar files a company under its primary share class; the other classes share the same
# statements and ratios (Alphabet C → GOOGL, Under Armour C → UAA, Liberty Formula One K → FWONA)
_SHARE_CLASS_ALIAS = {"GOOG": "GOOGL", "UA": "UAA", "FWONK": "FWONA", "LBRDK": "LBRDA", "LSXMK": "LSXMA",
                      "BATRK": "BATRA", "NWS": "NWSA", "FOX": "FOXA", "RUSHB": "RUSHA", "ZG": "Z"}


def fetch_sharadar(ticker: str) -> dict:
    """SF1 fundamentals for a ticker; a secondary share class falls back to the filing class
    (identical statements, ratios and per-share economics), tagged with a fund_note."""
    t = ticker.upper()
    out = _fetch_sharadar_raw(t)
    if not out and t in _SHARE_CLASS_ALIAS:
        out = dict(_fetch_sharadar_raw(_SHARE_CLASS_ALIAS[t]))
        if out:
            out["fund_note"] = f"statements filed under {_SHARE_CLASS_ALIAS[t]}"
    return out


def _fetch_sharadar_raw(ticker: str) -> dict:
    """Latest as-reported trailing-twelve-month (ART) fundamentals from SHARADAR/SF1, mapped
    onto our field names. {} on any failure; cached. Adds REIT-grade inputs (net income,
    D&A, shares) the scraped feeds don't expose."""
    t = ticker.upper()
    cached = _cache_get(f"sf1_{t}")
    if cached is not None:
        return cached
    out: dict = {}
    try:
        url = ("https://data.nasdaq.com/api/v3/datatables/SHARADAR/SF1.json?"
               + urllib.parse.urlencode({"ticker": t, "dimension": "ART",
                                         "qopts.columns": _SF1_COLS, "api_key": NDL_KEY}))
        with urllib.request.urlopen(url, timeout=30, context=CTX) as r:
            dt = json.loads(r.read())["datatable"]
        cols = [c["name"] for c in dt["columns"]]
        rows = [dict(zip(cols, row)) for row in dt["data"]]
        if rows:
            r0 = max(rows, key=lambda x: str(x.get("datekey") or ""))
            f = lambda k: _f(r0.get(k))
            dy = f("divyield")
            out = {"pe": f("pe"), "pb": f("pb"), "ps": f("ps"), "ev_ebitda": f("evebitda"),
                   "div_yield": dy * 100 if dy is not None else None,
                   "roe": f("roe"), "roa": f("roa"), "profit_margin": f("netmargin"),
                   "gross_margin": f("grossmargin"), "de": f("de"),
                   "eps_ttm": f("epsdil") or f("eps"), "bvps": f("bvps"), "fcfps": f("fcfps"),
                   "sh_depamor": f("depamor"), "sh_netinc": f("netinc"),
                   "shares_out": f("sharesbas"), "mcap": f("marketcap"),
                   "fund_asof": str(r0.get("datekey") or "")}
            out = {k: v for k, v in out.items() if v is not None and v != ""}
        _cache_put(f"sf1_{t}", out or {"_empty": True})
    except Exception as e:
        print(f"  sharadar {t} unavailable ({type(e).__name__})", flush=True)
    return {} if out.get("_empty") else out


def ffo_per_share(d: dict) -> float | None:
    """REIT valuation base: (net income + D&A) / shares — Sharadar fields first, then the
    yfinance cash-flow statement. None when unavailable."""
    ni, da, sh = d.get("sh_netinc"), d.get("sh_depamor"), d.get("shares_out")
    if not (isinstance(ni, (int, float)) and isinstance(da, (int, float))):
        try:
            import yfinance as yf
            cf = yf.Ticker(d["ticker"]).cashflow
            row = lambda pat: next((float(cf.loc[i].iloc[0]) for i in cf.index if pat in str(i)), None)
            ni, da = row("Net Income"), row("Depreciation And Amortization") or row("Depreciation")
        except Exception:
            return None
    if not (isinstance(ni, (int, float)) and isinstance(da, (int, float)) and isinstance(sh, (int, float)) and sh > 0):
        return None
    ffo = (ni + da) / sh
    return ffo if ffo > 0 else None


# ---------------- fund look-through (real constituents, not inference) ----------------

_SECTOR_LABEL = {"realestate": "Real Estate", "consumer_cyclical": "Consumer Cyclical",
                 "basic_materials": "Basic Materials", "consumer_defensive": "Consumer Defensive",
                 "technology": "Technology", "communication_services": "Communication Services",
                 "financial_services": "Financial Services", "utilities": "Utilities",
                 "industrials": "Industrials", "energy": "Energy", "healthcare": "Healthcare"}


def fetch_fund_lookthrough(ticker: str) -> dict:
    """{'top': [[symbol, weight], ...], 'sectors': {Sector: weight}} for an ETF/fund from
    the provider's disclosed holdings. {} when unavailable; cached."""
    t = ticker.upper()
    cached = _cache_get(f"lt_{t}")
    if cached is not None:
        return {} if cached.get("_empty") else cached
    out: dict = {}
    try:
        import yfinance as yf
        fd = yf.Ticker(t).funds_data
        top = fd.top_holdings
        if top is not None and len(top):
            out["top"] = [[str(i).upper(), round(float(v), 4)]
                          for i, v in top["Holding Percent"].items() if float(v) > 0][:15]
        sw = fd.sector_weightings or {}
        if sw:
            out["sectors"] = {_SECTOR_LABEL.get(k, k.title()): round(float(v), 4)
                              for k, v in sw.items() if float(v) > 0}
        _cache_put(f"lt_{t}", out or {"_empty": True})
    except Exception:
        pass
    return out


# ---------------- benchmark by domicile ----------------

_BENCH_BY_SUFFIX = {"NS": "^NSEI", "BO": "^BSESN", "L": "^FTSE", "T": "^N225", "DE": "^GDAXI", "F": "^GDAXI",
                    "SS": "000001.SS", "SZ": "399001.SZ", "V": "^GSPTSE", "LS": "PSI20.LS", "BR": "^BFX",
                    "SW": "^SSMI", "KS": "^KS11", "AX": "^AXJO", "HK": "^HSI", "PA": "^FCHI",
                    "TO": "^GSPTSE", "AS": "^AEX", "MI": "FTSEMIB.MI", "SA": "^BVSP"}
EU_SUFFIXES = {"L", "DE", "PA", "AS", "SW", "MI", "CO", "MC", "ST", "HE", "BR", "VI", "LS", "IR", "OL"}
BENCH_NAMES = {"SPY": "S&P 500", "^STOXX": "STOXX Europe 600", "000001.SS": "SSE Composite",
               "399001.SZ": "SZSE Component", "PSI20.LS": "PSI 20", "^BFX": "BEL 20", "^NSEI": "NIFTY 50", "^BSESN": "SENSEX", "^FTSE": "FTSE 100",
               "^N225": "Nikkei 225", "^GDAXI": "DAX", "^SSMI": "SMI", "^KS11": "KOSPI",
               "^AXJO": "ASX 200", "^HSI": "Hang Seng", "^FCHI": "CAC 40", "^GSPTSE": "TSX",
               "^AEX": "AEX", "FTSEMIB.MI": "FTSE MIB", "^BVSP": "Bovespa", "ACWI": "MSCI ACWI"}


def benchmark_for(holdings: list[dict]) -> str:
    """Benchmark symbol for the book: the dominant market by weight (>=60%), a global
    index (ACWI) for a genuinely mixed book, S&P 500 for US."""
    by_mkt: dict[str, float] = {}
    for h in holdings:
        sfx = h["symbol"].rsplit(".", 1)[1] if "." in h["symbol"] else "US"
        by_mkt[sfx] = by_mkt.get(sfx, 0.0) + float(h.get("weight") or 0.0)
    if not by_mkt:
        return "SPY"
    mkt, w = max(by_mkt.items(), key=lambda kv: kv[1])
    if w >= 0.6:
        return "SPY" if mkt == "US" else _BENCH_BY_SUFFIX.get(mkt, "ACWI")
    # no single market dominates — a European book still deserves a European yardstick
    eu_w = sum(v for k, v in by_mkt.items() if k in EU_SUFFIXES)
    if eu_w >= 0.6:
        return "^STOXX"          # STOXX Europe 600 (includes the UK)
    return "ACWI"


def fetch_news(ticker: str) -> list[dict]:
    """[{headline, sentiment}] — AV NEWS_SENTIMENT for US tickers, yfinance titles otherwise. Fail-open []."""
    t = ticker.upper()
    cached = _cache_get(f"news_{t}")
    if cached is not None and cached != []:
        return cached
    out: list[dict] = []
    if AV_KEY and "." not in t:
        try:
            news = _av({"function": "NEWS_SENTIMENT", "tickers": t, "limit": 10, "sort": "LATEST"})
            for a in (news or {}).get("feed", [])[:8]:
                ts = next((s for s in a.get("ticker_sentiment", []) if s.get("ticker") == t), {})
                out.append({"headline": str(a.get("title", ""))[:140],
                            "sentiment": ts.get("ticker_sentiment_label",
                                                a.get("overall_sentiment_label", ""))})
        except Exception:
            pass
    if not out:
        try:
            import yfinance as yf
            for a in (yf.Ticker(t).news or [])[:8]:
                title = (a.get("content") or {}).get("title") or a.get("title") or ""
                if title:
                    out.append({"headline": str(title)[:140], "sentiment": ""})
        except Exception:
            pass
    _cache_put(f"news_{t}", out)
    return out


def fetch_news_batch(tickers: list[str], av_limit: int = 25) -> dict[str, list[dict]]:
    """Headlines for many names: Alpha Vantage per ticker for the first `av_limit` (callers
    pass the book ordered by weight), yfinance titles for the rest. NEWS_SENTIMENT's multi-
    ticker form returns only articles mentioning EVERY ticker (market round-ups), so it is
    not used. Parallel, cached per ticker."""
    out: dict[str, list[dict]] = {}
    order = [t.upper() for t in tickers]

    def one(pair):
        t, use_av = pair
        try:
            if use_av:
                return t, fetch_news(t)
            c = _cache_get(f"news_{t}")
            if c:
                return t, c
            titles = []
            try:
                import yfinance as yf
                for a in (yf.Ticker(t).news or [])[:8]:
                    title = (a.get("content") or {}).get("title") or a.get("title") or ""
                    if title:
                        titles.append({"headline": str(title)[:140], "sentiment": ""})
            except Exception:
                pass
            if titles:
                _cache_put(f"news_{t}", titles)
            return t, titles
        except Exception:
            return t, []
    pairs = [(t, i < av_limit or "." in t) for i, t in enumerate(order)]
    with cf.ThreadPoolExecutor(max_workers=4) as ex:
        for t, items in ex.map(one, pairs):
            out[t] = items
    return out


def fetch_all(tickers: list[str], progress=None) -> dict[str, dict]:
    """Parallel fan-out over holdings (marketfm enrich pattern). Returns {ticker: data|{'error':...}}."""
    out: dict[str, dict] = {}
    done = 0
    with cf.ThreadPoolExecutor(max_workers=4) as ex:
        futs = {ex.submit(fetch_holding, t): t for t in tickers}
        for fut in cf.as_completed(futs):
            t = futs[fut]
            try:
                out[t] = fut.result()
            except Exception as e:
                out[t] = {"ticker": t, "error": str(e)}
            done += 1
            if progress:
                progress(done, len(tickers), t)
    return out


def fetch_history(tickers: list[str], period: str = "1y", bench: str | None = None) -> pd.DataFrame:
    """Daily close prices, columns = tickers + SPY (+ a domicile benchmark when given).
    Empty DataFrame on failure."""
    syms = sorted(set([t.upper() for t in tickers] + ["SPY"] + ([bench] if bench else [])))
    key = "hist_" + "_".join(syms) + "_" + period
    cached = _cache_get(key)
    if cached:
        df = pd.DataFrame(cached)
        try:    # to_json stores the index as epoch-milliseconds strings
            df.index = pd.to_datetime([int(x) for x in df.index], unit="ms")
        except (ValueError, TypeError):
            df.index = pd.to_datetime(df.index)
        return df.sort_index()
    try:
        import yfinance as yf
        raw = yf.download(syms, period=period, interval="1d", progress=False, auto_adjust=True)
        if raw is None or len(raw) == 0:
            return pd.DataFrame()
        px = raw["Close"] if isinstance(raw.columns, pd.MultiIndex) else raw[["Close"]]
        if isinstance(px, pd.Series):
            px = px.to_frame(syms[0])
        px = px.dropna(how="all")
        _cache_put(key, json.loads(px.to_json() or "{}"))
        return px
    except Exception as e:
        print(f"  history fetch failed ({type(e).__name__})", flush=True)
        return pd.DataFrame()


# ---------------- entry-price sanity data (split history + long price range) ----------------

def fetch_splits(ticker: str) -> list[dict]:
    """Stock splits, most recent first: [{date, ratio}] (a 3-for-1 split has ratio 3.0).
    Yahoo reports Indian bonus issues as splits too. [] on any failure; cached."""
    t = ticker.upper()
    cached = _cache_get(f"splits_{t}")
    if cached is not None:
        return cached
    out: list[dict] = []
    try:
        import yfinance as yf
        s = yf.Ticker(t).splits
        if s is not None and len(s):
            out = [{"date": str(idx)[:10], "ratio": float(v)}
                   for idx, v in s.items() if float(v) > 0][-6:][::-1]
        _cache_put(f"splits_{t}", out)
    except Exception:
        pass
    return out


def fetch_long_range(ticker: str) -> tuple[float, float] | None:
    """(min, max) of ~10y of SPLIT-ADJUSTED monthly closes — the ground truth for
    "could this entry price ever have been paid (in today's per-share terms)?".
    None on failure; cached."""
    t = ticker.upper()
    cached = _cache_get(f"lrange_{t}")
    if cached:
        return float(cached[0]), float(cached[1])
    try:
        import yfinance as yf
        h = yf.download(t, period="10y", interval="1mo", progress=False, auto_adjust=True)
        if h is None or len(h) < 6:
            return None
        close = h["Close"]
        if hasattr(close, "columns"):       # MultiIndex frame for single ticker
            close = close.iloc[:, 0]
        lo, hi = float(close.min()), float(close.max())
        if hi > 0:
            _cache_put(f"lrange_{t}", [lo, hi])
            return lo, hi
    except Exception:
        pass
    return None


# ---------------- fact sheet — the compact grounded brief fed to personas ----------------

def _cap(x):
    x = _f(x)
    if x is None:
        return "n/a"
    for u, dvd in [("T", 1e12), ("B", 1e9), ("M", 1e6)]:
        if x >= dvd:
            return f"${x / dvd:.1f}{u}"
    return f"${x:.0f}"


def _pct(x, signed=True):
    if not isinstance(x, (int, float)) or x != x:
        return "n/a"
    return f"{x * 100:+.1f}%" if signed else f"{x * 100:.1f}%"


def fact_sheet(d: dict, holding: dict, news: list[dict], dcf: dict | None) -> str:
    """One holding → the grounded text block every persona sees (marketguru schema, extended
    with the user's cost basis, DCF intrinsic value, and panel rank)."""
    g = d.get
    price = g("price")
    L = [f"{d['ticker']} — {g('name') or '?'} | Sector: {g('sector') or '?'} | "
         f"Industry: {g('industry') or '?'} | Mkt cap: {_cap(g('mcap'))} | Currency: {g('currency') or 'USD'}"]
    pos = [f"Price {price:.2f}", f"user entry {holding['entry']:.2f}",
           f"P/L since entry {_pct(holding.get('pl_pct'))}"]
    if holding.get("weight") is not None:
        pos.append(f"portfolio weight {_pct(holding['weight'], signed=False)}")
    L.append("Position: " + " | ".join(pos))
    val = [f"{lab} {v:.1f}" for lab, v in [("P/E", g("pe")), ("Fwd P/E", g("fwd_pe")),
           ("PEG", g("peg")), ("P/B", g("pb")), ("P/S", g("ps")), ("EV/EBITDA", g("ev_ebitda"))]
           if isinstance(v, (int, float)) and v == v]
    dy = g("div_yield")
    if isinstance(dy, (int, float)):
        val.append(f"Div yield {dy:.2f}%")   # always percent (0.86 = 0.86%, not a fraction)
    if val:
        L.append("Valuation: " + ", ".join(val))
    if dcf and dcf.get("intrinsic"):
        model = dcf.get("model") or "Damodaran 2-stage DCF"
        rng = ""
        if isinstance(dcf.get("intrinsic_low"), (int, float)) and isinstance(dcf.get("intrinsic_high"), (int, float)):
            rng = f" | bear–bull range {dcf['intrinsic_low']:.2f}–{dcf['intrinsic_high']:.2f}"
        imp = ""
        if isinstance(dcf.get("implied_growth"), (int, float)):
            imp = (f" | REVERSE DCF: today's price implies {dcf['implied_growth'] * 100:+.0f}%/yr "
                   f"earnings growth over the high-growth phase")
        L.append(f"Intrinsic value ({model}, lifecycle {dcf.get('lifecycle', '?')}): "
                 f"{dcf['intrinsic']:.2f}/share = {_pct(dcf.get('upside'))} vs price "
                 f"({_pct(dcf.get('upside_vs_entry'))} vs user entry){rng}{imp}"
                 + (f" | guards: {','.join(dcf['flags'])}" if dcf.get("flags") else ""))
    elif dcf and dcf.get("note"):
        L.append(f"Valuation note: {dcf['note']} — judge on yield, quality and price trend instead of a DCF gap.")
    fac = (d.get("panel") or {}).get("factors") or {}
    if fac:
        pk = lambda k: f"{(fac[k] + 1) / 2 * 100:.0f}" if isinstance(fac.get(k), (int, float)) else None
        peers = [f"{lab} {pk(k)}th" for lab, k in [("earnings yield", "n_earnings_yield"),
                 ("EV/EBITDA cheapness", "n_evebitda"), ("ROE", "n_roe"), ("net margin", "n_netmargin"),
                 ("12-1m momentum", "n_mom_12_1")] if pk(k)]
        if peers:
            uni = (d.get("panel") or {}).get("universe") or "US"
            L.append(f"Peer percentiles vs {uni} universe (higher = better): " + ", ".join(peers))
    src = d.get("fund_source")
    if d.get("data_flags"):     # provenance lives in the report; the brief only carries what changes a judgment
        L.append("Data dropped as implausible: " + "; ".join(d["data_flags"]))
    qual = [f"{lab} {_pct(v, signed=False)}" for lab, v in [("ROE", g("roe")), ("ROA", g("roa")),
            ("Op margin", g("op_margin")), ("Net margin", g("profit_margin"))]
            if isinstance(v, (int, float)) and v == v]
    de = g("de")
    if isinstance(de, (int, float)):
        qual.append(f"D/E {de:.2f}")
    if isinstance(g("beta"), (int, float)):
        qual.append(f"Beta {g('beta'):.2f}")
    if qual:
        L.append("Quality: " + ", ".join(qual))
    if d.get("us"):                       # deep US fundamentals (Sharadar history + estimates)
        from .fundamentals import us_lines
        L.extend(us_lines(d["us"]))
    gro = [f"{lab} {_pct(v)}" for lab, v in [("Revenue growth", g("rev_growth")),
           ("EPS growth", g("eps_growth"))] if isinstance(v, (int, float)) and v == v]
    if isinstance(g("target"), (int, float)) and price:
        gro.append(f"Analyst target {g('target'):.0f} ({_pct(g('target') / price - 1)})")
    if gro:
        L.append("Growth: " + ", ".join(gro))
    tech = []
    for lab, ref in [("vs 52w high", g("hi52")), ("vs 52w low", g("lo52")),
                     ("vs 50d SMA", g("sma50")), ("vs 200d SMA", g("sma200"))]:
        if price and isinstance(ref, (int, float)) and ref:
            tech.append(f"{lab} {_pct(price / ref - 1)}")
    if tech:
        L.append("Technicals: " + ", ".join(tech))
    p = d.get("panel")
    if p and isinstance(p.get("xs_rank_pct"), (int, float)):
        L.append(f"Quant panel (MarketFM factor model): cross-sectional percentile "
                 f"{p['xs_rank_pct']:.0f}/100 (rank #{p.get('rank', '?')} of ~3600 US names), "
                 f"ML-predicted 12m excess return {_pct(p.get('pred_excess_12m'))}")
    if news:
        L.append("Recent headlines: " + " | ".join(
            f'"{n["headline"]}"' + (f" ({n['sentiment']})" if n.get("sentiment") else "")
            for n in news[:5]))
    return "\n".join(L)


# ---------------- event calendar + instrument universe ----------------

def fetch_events(ticker: str) -> dict:
    """{'earnings': 'YYYY-MM-DD'|None, 'ex_div': 'YYYY-MM-DD'|None} — the dates a desk checks
    before recommending a switch (never buy into earnings blind; sell after the ex-date)."""
    t = ticker.upper()
    cached = _cache_get(f"ev_{t}")
    if cached is not None:
        return cached
    out = {"earnings": None, "ex_div": None}
    try:
        import yfinance as yf
        cal = yf.Ticker(t).calendar
        if isinstance(cal, dict):
            ed = cal.get("Earnings Date")
            if isinstance(ed, (list, tuple)) and ed:
                out["earnings"] = str(ed[0])[:10]
            xd = cal.get("Ex-Dividend Date")
            if xd:
                out["ex_div"] = str(xd)[:10]
        _cache_put(f"ev_{t}", out if any(out.values()) else {"earnings": None, "ex_div": None, "_": 1})
    except Exception:
        pass
    out.pop("_", None)
    return out


def days_until(date_str: str | None) -> int | None:
    if not date_str:
        return None
    try:
        from datetime import date
        y, m, d = (int(x) for x in str(date_str)[:10].split("-"))
        return (date(y, m, d) - date.today()).days
    except Exception:
        return None


def load_instruments() -> list[dict]:
    try:
        return json.loads((DATA_DIR / "instruments.json").read_text())
    except Exception:
        return []


# ═══════════════ Alpha Vantage premium — PRIMARY market-data layer ═══════════════
# Licensed, rate-limited, unit-consistent. yfinance (scraped) is demoted to fallback and to
# exchanges Alpha Vantage does not map. Sharadar stays primary for point-in-time fundamentals.
import collections as _collections
import threading as _threading

AV_CALLS_PER_MIN = int(os.environ.get("AV_CALLS_PER_MIN", "75"))
_av_lock = _threading.Lock()
_av_times: "_collections.deque[float]" = _collections.deque()
# Yahoo suffix → Alpha Vantage suffix (documented/verified mappings only; others fall back)
AV_SUFFIX = {"BO": "BSE", "NS": "BSE",            # NSE names are dual-listed on BSE at near-identical prices
             "L": "LON", "DE": "DEX", "F": "FRK", "TO": "TRT", "V": "TRV", "PA": "PAR", "AS": "AMS",
             "LS": "LIS", "BR": "BRU", "SS": "SHH", "SZ": "SHZ"}
AV_TO_YAHOO = {"BSE": ".BO", "LON": ".L", "DEX": ".DE", "FRK": ".F", "TRT": ".TO", "TRV": ".V", "PAR": ".PA",
               "AMS": ".AS", "LIS": ".LS", "BRU": ".BR", "SHH": ".SS", "SHZ": ".SZ", "SAO": ".SA"}
_AV_SECTOR = {"INFORMATION TECHNOLOGY": "Technology", "HEALTHCARE": "Healthcare", "HEALTH CARE": "Healthcare",
              "FINANCIALS": "Financial Services", "FINANCE": "Financial Services",
              "CONSUMER DISCRETIONARY": "Consumer Cyclical", "CONSUMER STAPLES": "Consumer Defensive",
              "COMMUNICATION SERVICES": "Communication Services", "INDUSTRIALS": "Industrials",
              "ENERGY": "Energy", "MATERIALS": "Basic Materials", "BASIC MATERIALS": "Basic Materials",
              "UTILITIES": "Utilities", "REAL ESTATE": "Real Estate", "TECHNOLOGY": "Technology"}


def av_symbol(ticker: str) -> str | None:
    """Alpha Vantage symbol for a Yahoo-style ticker, or None if AV can't serve it."""
    if not AV_KEY:
        return None
    t = ticker.upper()
    if t.startswith("^"):
        return None
    if "." not in t:
        return t.replace("-", ".") if t.endswith("-B") or t.endswith("-A") else t   # BRK-B → BRK.B
    base, sfx = t.rsplit(".", 1)
    return f"{base}.{AV_SUFFIX[sfx]}" if sfx in AV_SUFFIX else None


def _av_call(params: dict, raw: bool = False):
    """Rate-limited (AV_CALLS_PER_MIN) premium call; one retry; returns dict/str or None."""
    with _av_lock:
        now = time.time()
        while _av_times and now - _av_times[0] > 60:
            _av_times.popleft()
        if len(_av_times) >= AV_CALLS_PER_MIN:
            time.sleep(max(0.0, 60 - (now - _av_times[0])) + 0.05)
        _av_times.append(time.time())
    url = "https://www.alphavantage.co/query?" + urllib.parse.urlencode({**params, "apikey": AV_KEY})
    for _ in range(2):
        try:
            with urllib.request.urlopen(url, timeout=30, context=CTX) as r:
                body = r.read()
            if raw:
                return body.decode("utf-8", "ignore")
            out = json.loads(body)
            if isinstance(out, dict) and ("Note" in out or "Information" in out) and len(out) == 1:
                time.sleep(2)
                continue
            return out
        except Exception:
            time.sleep(1)
    return None


def _av_primary(ticker: str) -> dict:
    """OVERVIEW + GLOBAL_QUOTE (+ ETF_PROFILE when OVERVIEW is empty) mapped onto our fields."""
    sym = av_symbol(ticker)
    if not sym:
        return {}
    d: dict = {}
    ov = _av_call({"function": "OVERVIEW", "symbol": sym}) or {}
    if isinstance(ov, dict) and ov.get("Symbol"):
        for k, avk in [("name", "Name"), ("sector", "Sector"), ("industry", "Industry"), ("currency", "Currency"),
                       ("mcap", "MarketCapitalization"), ("pe", "TrailingPE"), ("fwd_pe", "ForwardPE"),
                       ("peg", "PEGRatio"), ("pb", "PriceToBookRatio"), ("ps", "PriceToSalesRatioTTM"),
                       ("ev_ebitda", "EVToEBITDA"), ("div_yield", "DividendYield"), ("roe", "ReturnOnEquityTTM"),
                       ("roa", "ReturnOnAssetsTTM"), ("op_margin", "OperatingMarginTTM"),
                       ("profit_margin", "ProfitMargin"), ("beta", "Beta"), ("hi52", "52WeekHigh"),
                       ("lo52", "52WeekLow"), ("sma50", "50DayMovingAverage"), ("sma200", "200DayMovingAverage"),
                       ("rev_growth", "QuarterlyRevenueGrowthYOY"), ("eps_growth", "QuarterlyEarningsGrowthYOY"),
                       ("eps_ttm", "EPS"), ("target", "AnalystTargetPrice"), ("bvps", "BookValue"),
                       ("shares_out", "SharesOutstanding")]:
            v = ov.get(avk)
            d[k] = v if k in ("name", "sector", "industry", "currency") else _f(v)
        if d.get("div_yield") is not None:
            d["div_yield"] = d["div_yield"] * 100          # AV gives a fraction; we use percent
        at = str(ov.get("AssetType") or "").upper()
        d["quote_type"] = "ETF" if "ETF" in at else "EQUITY"
        d = {k: v for k, v in d.items() if v is not None and v != ""}
    else:
        prof = _av_call({"function": "ETF_PROFILE", "symbol": sym}) or {}
        if isinstance(prof, dict) and prof.get("holdings"):
            d["quote_type"] = "ETF"
            dy = _f(prof.get("dividend_yield"))
            if dy is not None:
                d["div_yield"] = dy * 100
    q = _av_call({"function": "GLOBAL_QUOTE", "symbol": sym}) or {}
    gq = q.get("Global Quote") if isinstance(q, dict) else None
    if gq and _f(gq.get("05. price")):
        d["price"] = _f(gq.get("05. price"))
        d["price_asof"] = str(gq.get("07. latest trading day") or "")
        d["price_source"] = "alphavantage" + (" (BSE line)" if ticker.upper().endswith(".NS") else "")
    if any(k in d for k in ("pe", "roe", "eps_ttm", "mcap")):   # fundamentals really came from AV
        d["fund_source_av"] = True
    return d


_yf_fetch_history = fetch_history


def _av_daily(ticker: str) -> pd.Series | None:
    """Full adjusted daily close history (20y) from TIME_SERIES_DAILY_ADJUSTED; cached."""
    sym = av_symbol(ticker)
    if not sym:
        return None
    cached = _cache_get(f"avh_{ticker.upper()}")
    if cached:
        s = pd.Series(cached, dtype=float)
        s.index = pd.to_datetime(s.index)
        return s.sort_index()
    out = _av_call({"function": "TIME_SERIES_DAILY_ADJUSTED", "symbol": sym, "outputsize": "full"})
    ts = next((v for k, v in out.items() if "Time Series" in k), None) if isinstance(out, dict) else None
    if not ts:
        return None
    ser = {dt: _f(row.get("5. adjusted close")) for dt, row in ts.items()}
    ser = {k: v for k, v in ser.items() if v is not None}
    if not ser:
        return None
    _cache_put(f"avh_{ticker.upper()}", ser)
    s = pd.Series(ser, dtype=float)
    s.index = pd.to_datetime(s.index)
    return s.sort_index()


def fetch_history(tickers: list[str], period: str = "1y", bench: str | None = None) -> pd.DataFrame:
    """Daily adjusted closes: Alpha Vantage per ticker (parallel, rate-limited) for every
    symbol it maps; yfinance for the rest (indices, unmapped exchanges) and as fallback."""
    syms = sorted(set([t.upper() for t in tickers] + ["SPY"] + ([bench] if bench else [])))
    if not AV_KEY:
        return _yf_fetch_history(tickers, period, bench)
    years = int(period[:-1]) if period.endswith("y") and period[:-1].isdigit() else 1
    cutoff = pd.Timestamp.today() - pd.DateOffset(years=years)
    series: dict[str, pd.Series] = {}
    if NDL_KEY:                     # US listings in batched Sharadar calls — no per-ticker AV budget
        try:
            from . import sharadar_prices as SP
            for t, sr in SP.closes([x for x in syms if SP.is_us_ticker(x)], years=max(years, 2)).items():
                if len(sr) > 20:
                    series[t] = sr[sr.index >= cutoff]
        except Exception as e:
            print(f"  sharadar history failed ({type(e).__name__})", flush=True)
    av_syms = [t for t in syms if av_symbol(t) and t not in series]
    with cf.ThreadPoolExecutor(max_workers=4) as ex:
        for t, s in zip(av_syms, ex.map(_av_daily, av_syms)):
            if s is not None and len(s) > 20:
                series[t] = s[s.index >= cutoff]
    rest = [t for t in syms if t not in series]
    if rest:
        try:
            yf_df = _yf_fetch_history([t for t in rest if t != "SPY" and t != bench], period,
                                      bench if bench in rest else None)
            for t in rest:
                if t in yf_df.columns:
                    series[t] = yf_df[t]
        except Exception:
            pass
    if not series:
        return pd.DataFrame()
    df = pd.DataFrame(series).sort_index()
    df.index = pd.to_datetime(df.index)
    return df.dropna(how="all")


_yf_fetch_splits = fetch_splits


def fetch_splits(ticker: str) -> list[dict]:
    if NDL_KEY and "." not in ticker:
        try:
            from . import sharadar_prices as SP
            sp = SP.splits(ticker)
            if sp or SP.history([ticker]):          # known to Sharadar → its (possibly empty) answer stands
                return sp
        except Exception:
            pass
    sym = av_symbol(ticker)
    if sym:
        cached = _cache_get(f"avsp_{ticker.upper()}")
        if cached is not None:
            return cached
        out = _av_call({"function": "SPLITS", "symbol": sym})
        if isinstance(out, dict) and isinstance(out.get("data"), list):
            rows = [{"date": str(r.get("effective_date"))[:10], "ratio": float(_f(r.get("split_factor")) or 0)}
                    for r in out["data"] if _f(r.get("split_factor"))]
            rows = sorted([r for r in rows if r["ratio"] > 0], key=lambda r: r["date"], reverse=True)[:6]
            if not rows and "." in ticker:        # AV split history is thin outside the US
                rows = _yf_fetch_splits(ticker)
            _cache_put(f"avsp_{ticker.upper()}", rows or [{"_": 1}])
            return [r for r in rows if "date" in r]
    return _yf_fetch_splits(ticker)


def fetch_long_range(ticker: str) -> tuple[float, float] | None:
    """10y split-adjusted monthly range: Sharadar for US, AV history otherwise (yfinance fallback)."""
    if NDL_KEY and "." not in ticker:
        try:
            from . import sharadar_prices as SP
            rng = SP.long_range(ticker)
            if rng:
                return rng
        except Exception:
            pass
    s = _av_daily(ticker)
    if s is not None and len(s) > 250:
        s = s[s.index >= pd.Timestamp.today() - pd.DateOffset(years=10)]
        m = s.resample("ME").last().dropna()
        if len(m) >= 6 and float(m.max()) > 0:
            return float(m.min()), float(m.max())
    try:
        import yfinance as yf
        h = yf.download(ticker, period="10y", interval="1mo", progress=False, auto_adjust=True)
        if h is None or len(h) < 6:
            return None
        close = h["Close"]
        if hasattr(close, "columns"):
            close = close.iloc[:, 0]
        return float(close.min()), float(close.max())
    except Exception:
        return None


_yf_fetch_fund_lookthrough = fetch_fund_lookthrough


def fetch_fund_lookthrough(ticker: str) -> dict:
    """ETF_PROFILE gives the FULL constituent list and sector weights (yfinance: top 10)."""
    sym = av_symbol(ticker)
    if sym:
        cached = _cache_get(f"avlt_{ticker.upper()}")
        if cached is not None:
            return {} if cached.get("_empty") else cached
        prof = _av_call({"function": "ETF_PROFILE", "symbol": sym})
        if isinstance(prof, dict) and prof.get("holdings"):
            top = [[str(h.get("symbol") or "").upper(), float(_f(h.get("weight")) or 0)]
                   for h in prof["holdings"] if h.get("symbol") and _f(h.get("weight"))]
            top = [x for x in top if x[0] and x[0] != "N/A" and x[1] > 0][:40]
            sectors = {}
            for sc in prof.get("sectors") or []:
                lab = _AV_SECTOR.get(str(sc.get("sector") or "").upper(), str(sc.get("sector") or "").title())
                w = _f(sc.get("weight"))
                if w:
                    sectors[lab] = round(sectors.get(lab, 0.0) + w, 4)
            out = {"top": top, "sectors": sectors, "source": "alphavantage"}
            _cache_put(f"avlt_{ticker.upper()}", out)
            return out
        _cache_put(f"avlt_{ticker.upper()}", {"_empty": True})
    return _yf_fetch_fund_lookthrough(ticker)


_yf_fetch_events = fetch_events


def _earnings_calendar_all() -> dict[str, str]:
    """The whole US market's next report dates in ONE Alpha Vantage call, cached for a day —
    instead of one call per holding and per candidate."""
    c = _cache_get("avcal_all")
    if isinstance(c, dict) and c.get("map"):
        return c["map"]
    out: dict[str, str] = {}
    csv_text = _av_call({"function": "EARNINGS_CALENDAR", "horizon": "3month"}, raw=True)
    if isinstance(csv_text, str):
        for line in csv_text.splitlines()[1:]:
            parts = line.split(",")
            if len(parts) >= 3 and parts[2][:4].isdigit():
                out.setdefault(parts[0].strip().upper(), parts[2][:10])
    if out:
        _cache_put("avcal_all", {"map": out})
    return out


def fetch_events(ticker: str) -> dict:
    """Earnings date + next/last ex-dividend date. US: the one-call market calendar plus
    Sharadar ACTIONS; other markets: AV per symbol, then yfinance."""
    t = ticker.upper()
    if NDL_KEY and "." not in t:
        cached = _cache_get(f"shev_{t}")
        if cached is not None:
            return {k: cached.get(k) for k in ("earnings", "ex_div")}
        out = {"earnings": None, "ex_div": None}
        try:
            out["earnings"] = _earnings_calendar_all().get(t)
        except Exception:
            pass
        try:
            from . import sharadar_prices as SP
            out["ex_div"] = SP.ex_dividend(t)
        except Exception:
            pass
        _cache_put(f"shev_{t}", {**out, "_": 1})
        return out
    sym = av_symbol(ticker)
    if sym:
        cached = _cache_get(f"avev_{ticker.upper()}")
        if cached is not None:
            return {k: cached.get(k) for k in ("earnings", "ex_div")}
        out = {"earnings": None, "ex_div": None}
        csv_text = _av_call({"function": "EARNINGS_CALENDAR", "symbol": sym, "horizon": "3month"}, raw=True)
        if isinstance(csv_text, str):
            for line in csv_text.splitlines()[1:]:
                parts = line.split(",")
                if len(parts) >= 3 and parts[2][:4].isdigit():
                    out["earnings"] = parts[2][:10]
                    break
        dv = _av_call({"function": "DIVIDENDS", "symbol": sym})
        if isinstance(dv, dict) and isinstance(dv.get("data"), list) and dv["data"]:
            today = time.strftime("%Y-%m-%d")
            future = sorted((str(r.get("ex_dividend_date"))[:10] for r in dv["data"]
                             if str(r.get("ex_dividend_date"))[:10] >= today))
            out["ex_div"] = future[0] if future else str(dv["data"][0].get("ex_dividend_date"))[:10]
        if not any(out.values()):          # AV calendars are US-centric — fall back
            out = _yf_fetch_events(ticker)
        _cache_put(f"avev_{ticker.upper()}", {**out, "_": 1})
        return out
    return _yf_fetch_events(ticker)


_yf_fx_to_usd = fx_to_usd


def fx_to_usd(currency) -> float:
    cur = str(currency or "USD")
    if cur in _fx_cache:
        return _fx_cache[cur]
    base, scale = cur.upper(), 1.0
    if cur in ("GBp", "GBX"):
        base, scale = "GBP", 0.01
    if base == "USD" or not AV_KEY:
        return _yf_fx_to_usd(currency)
    c = _cache_get(f"avfx_{base}")
    rate = float(c["rate"]) if c else None
    if rate is None:
        out = _av_call({"function": "CURRENCY_EXCHANGE_RATE", "from_currency": base, "to_currency": "USD"})
        r = (out or {}).get("Realtime Currency Exchange Rate", {}) if isinstance(out, dict) else {}
        rate = _f(r.get("5. Exchange Rate"))
        if rate:
            _cache_put(f"avfx_{base}", {"rate": rate})
    if not rate:
        return _yf_fx_to_usd(currency)
    _fx_cache[cur] = rate * scale
    return _fx_cache[cur]


_SS_TYPES = {"Equity", "ETF"}


def av_symbol_search(keywords: str) -> list[dict]:
    """SYMBOL_SEARCH → [{av, yahoo, name, region, currency, type}] (cached). The Yahoo symbol
    is derived from the AV suffix; Indian results map to the NSE line when it exists (the
    primary market) else BSE."""
    import hashlib as _hl   # exact-string key: "TATAMOTORS" and "Tata Motors" are different searches
    key = "avss_" + _hl.md5(keywords.strip().lower().encode()).hexdigest()[:16]
    cached = _cache_get(key)
    if cached is not None:
        return [o for o in cached if "av" in o]
    out: list[dict] = []
    res = _av_call({"function": "SYMBOL_SEARCH", "keywords": keywords}) if AV_KEY else None
    for m in (res or {}).get("bestMatches", []) if isinstance(res, dict) else []:
        sym, typ = str(m.get("1. symbol") or ""), str(m.get("3. type") or "")
        if not sym or typ not in _SS_TYPES:
            continue
        if "." in sym:
            base, sfx = sym.rsplit(".", 1)
            ysfx = AV_TO_YAHOO.get(sfx)
            if not ysfx:
                continue
            yahoo = base + ysfx
        else:
            yahoo = sym.replace(".", "-")
        out.append({"av": sym, "yahoo": yahoo, "name": str(m.get("2. name") or ""),
                    "region": str(m.get("4. region") or ""), "currency": str(m.get("8. currency") or ""),
                    "type": typ})
    if isinstance(res, dict) and "bestMatches" in res:   # cache only a REAL answer — never a
        _cache_put(key, out or [{"_empty": True}])          # throttled/failed call (6h of misses)
    return [o for o in out if "av" in o]


# ═══════════ ADR bridge — licensed Sharadar fundamentals for non-US holdings ═══════════
# Sharadar SF1 covers US-listed ADRs (HDB, INFY, TSM, BABA, VALE, SHEL, SAP…) with current,
# audited, point-in-time data for the UNDERLYING company. Ratio fields are scale-free and
# therefore valid for the home-market line; per-share fields mix currencies/ADR ratios and
# are never copied — instead EPS/BVPS for the local share are DERIVED from local price and
# the Sharadar P/E and P/B, which keeps them currency-consistent.

_SUFFIX_COUNTRY = {"NS": "India", "BO": "India", "SS": "China", "SZ": "China", "HK": "Hong Kong",
                   "SA": "Brazil", "KS": "South Korea", "TW": "Taiwan", "L": "United Kingdom",
                   "DE": "Germany", "F": "Germany", "PA": "France", "AS": "Netherlands", "SW": "Switzerland",
                   "MI": "Italy", "T": "Japan", "AX": "Australia", "TO": "Canada", "MX": "Mexico",
                   "JK": "Indonesia", "JO": "South Africa", "ST": "Sweden", "CO": "Denmark", "MC": "Spain"}
_RATIO_FIELDS = ("pe", "pb", "ps", "ev_ebitda", "div_yield", "roe", "roa", "profit_margin", "gross_margin", "de")


def _sharadar_ticker_meta(ticker: str) -> dict:
    """SHARADAR/TICKERS row (category, location) for a US symbol; {} if absent. Cached."""
    cached = _cache_get(f"sftk_{ticker}")
    if cached is not None:
        return {} if cached.get("_empty") else cached
    out: dict = {}
    try:
        url = ("https://data.nasdaq.com/api/v3/datatables/SHARADAR/TICKERS.json?"
               + urllib.parse.urlencode({"ticker": ticker, "table": "SF1",
                                         "qopts.columns": "ticker,name,category,location,isdelisted",
                                         "api_key": NDL_KEY}))
        with urllib.request.urlopen(url, timeout=30, context=CTX) as r:
            dt = json.loads(r.read())["datatable"]
        cols = [c["name"] for c in dt["columns"]]
        for row in dt["data"]:
            rec = dict(zip(cols, row))
            if rec.get("isdelisted") == "N":
                out = rec
                break
        _cache_put(f"sftk_{ticker}", out or {"_empty": True})
    except Exception:
        pass
    return out


def adr_for(ticker: str, name: str | None) -> dict | None:
    """Find a live NYSE/NASDAQ ADR of the same company (never an OTC line): AV Symbol Search
    on the company name → US primary match whose name shares the tokens → confirmed as an
    ADR from the right country in Sharadar's ticker table. Returns {adr, location} or None."""
    if not NDL_KEY or not name or "." not in ticker:
        return None
    sfx = ticker.rsplit(".", 1)[1]
    country = _SUFFIX_COUNTRY.get(sfx)
    stop = {"ltd", "limited", "plc", "inc", "corp", "the", "ord", "shs", "usd", "eur", "gbp", "co",
            "sa", "ag", "nv", "se", "spa", "asa", "abb", "group", "holdings", "holding", "company"}
    toks = [t for t in "".join(ch if ch.isalnum() else " " for ch in name.lower()).split()
            if len(t) > 2 and t.isalpha() and t not in stop][:2] \
        or [t for t in "".join(ch if ch.isalnum() else " " for ch in name.lower()).split() if t.isalpha()][:2]
    if not toks:
        return None
    ok_loc = lambda meta: (not country) or str(meta.get("location")) in (country, "Hong Kong" if country == "China" else country)
    # 1) many ADRs reuse the home-market base symbol (SHEL.L→SHEL, SAP.DE→SAP, INFY.NS→INFY):
    #    check it directly — guarded by ADR category, country AND a name-token match, so a
    #    coincidental US ticker (TCS = The Container Store) can never be mistaken for one
    base = ticker.rsplit(".", 1)[0].replace("-", ".")
    meta = _sharadar_ticker_meta(base) if base.isalnum() and len(base) <= 5 else {}
    if meta and "ADR" in str(meta.get("category")):
        mname = str(meta.get("name") or "").lower()
        strong = len(toks[0]) >= 4 and toks[0] in mname      # a distinctive name token matched
        # the country guard yields to a strong name match: Shell's ADR is still filed under
        # its pre-2022 Dutch domicile while the shares trade in London
        if strong and (ok_loc(meta) or len(toks) < 2 or toks[1] in mname):
            return {"adr": base, "location": meta.get("location")}
    # 2) otherwise search by name
    for m in av_symbol_search(" ".join(toks)):
        if m["region"] != "United States" or (len(m["av"]) == 5 and m["av"][-1] in "YF"):
            continue
        lname = m["name"].lower()
        if not all(t in lname for t in toks[:2]):
            continue
        meta = _sharadar_ticker_meta(m["av"].replace("-", "."))
        if meta and "ADR" in str(meta.get("category")) and ok_loc(meta):
            return {"adr": m["av"].replace("-", "."), "location": meta.get("location")}
    return None


def bridge_adr_fundamentals(d: dict) -> None:
    """Overlay Sharadar ratio fields from the company's ADR onto a non-US holding, and derive
    currency-consistent EPS/BVPS for the local share from local price × Sharadar P/E, P/B."""
    link = adr_for(d["ticker"], d.get("name"))
    if not link:
        return
    sh = fetch_sharadar(link["adr"])
    if not sh:
        return
    for k in _RATIO_FIELDS:
        if sh.get(k) is not None:
            d[k] = sh[k]
    price = d.get("price")
    if isinstance(price, (int, float)) and price > 0:
        if isinstance(sh.get("pe"), (int, float)) and sh["pe"] > 0:
            d["eps_ttm"] = round(price / sh["pe"], 4)
        if isinstance(sh.get("pb"), (int, float)) and sh["pb"] > 0:
            d["bvps"] = round(price / sh["pb"], 4)
    d["fund_source"] = f"sharadar (via {link['adr']} ADR)"
    d["fund_asof"] = sh.get("fund_asof")
    d["adr"] = link["adr"]


# ---------------- street view: analyst targets & ratings (pre-warmed, cached 24h) ----------------

def analyst_view(ticker: str) -> dict:
    """Consensus target with range, analyst count, rating split and the overview's growth
    fields. One Alpha Vantage OVERVIEW call (kept for US names because the target is a real
    signal) plus yfinance's target range when reachable. Cached; {} when unavailable."""
    t = ticker.upper()
    c = _cache_get(f"avov_{t}")
    if isinstance(c, dict):
        return {} if c.get("_empty") else c
    out: dict = {}
    sym = av_symbol(t)
    if AV_KEY and sym and "." not in t:
        ov = _av_call({"function": "OVERVIEW", "symbol": sym}) or {}
        if isinstance(ov, dict) and ov.get("Symbol"):
            f = lambda k: _f(ov.get(k))
            out.update({k: v for k, v in {"target": f("AnalystTargetPrice"), "rev_growth": f("QuarterlyRevenueGrowthYOY"),
                                          "eps_growth": f("QuarterlyEarningsGrowthYOY"), "fwd_pe": f("ForwardPE"),
                                          "peg": f("PEGRatio")}.items() if v is not None})
            ratings = {k: int(f(a) or 0) for k, a in [("strong_buy", "AnalystRatingStrongBuy"), ("buy", "AnalystRatingBuy"),
                                                       ("hold", "AnalystRatingHold"), ("sell", "AnalystRatingSell"),
                                                       ("strong_sell", "AnalystRatingStrongSell")]}
            if sum(ratings.values()):
                out["ratings"] = ratings
                out["n_analysts"] = sum(ratings.values())
    try:
        import yfinance as yf
        pt = yf.Ticker(t).analyst_price_targets or {}
        for k in ("low", "high", "mean", "median"):
            if isinstance(pt.get(k), (int, float)):
                out[f"target_{k}"] = float(pt[k])
        if out.get("target") is None and out.get("target_mean"):
            out["target"] = out["target_mean"]
    except Exception:
        pass
    _cache_put(f"avov_{t}", out or {"_empty": True})
    return out