| """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" |
|
|
|
|
| |
|
|
| 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 |
|
|
|
|
| |
|
|
| def _sector_map() -> dict: |
| try: |
| return json.loads((DATA_DIR / "sector_map.json").read_text()) |
| except Exception: |
| return {} |
|
|
|
|
| SECTOR_MAP = _sector_map() |
|
|
|
|
| |
|
|
| 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 |
|
|
|
|
| |
|
|
| _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) |
| |
| 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"), |
| |
| "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_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"): |
| 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] |
|
|
|
|
| |
|
|
| _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 |
|
|
|
|
| |
|
|
| _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", |
| } |
|
|
| |
| |
| |
| |
| |
| 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: |
| 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)) |
| else: |
| ov = _cache_get(f"avov_{t}") |
| 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 = {} |
| |
| |
| 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"))) |
| for attempt in range(2 if need_info else 0): |
| 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: |
| 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 |
| 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) |
| |
| 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 |
| 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"): |
| 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 |
| |
| |
| 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: |
| |
| 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) |
| except Exception: |
| pass |
| d["fetched_at"] = time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime()) |
| if str(d.get("currency")) == "GBp": |
| |
| |
| |
| 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: |
| 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() |
| 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(): |
| 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 |
|
|
|
|
| |
|
|
| 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") |
|
|
|
|
| |
| |
| _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 |
|
|
|
|
| |
|
|
| _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 |
|
|
|
|
| |
|
|
| _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") |
| |
| eu_w = sum(v for k, v in by_mkt.items() if k in EU_SUFFIXES) |
| if eu_w >= 0.6: |
| return "^STOXX" |
| 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: |
| 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() |
|
|
|
|
| |
|
|
| 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"): |
| 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 |
|
|
|
|
| |
|
|
| 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}%") |
| 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"): |
| 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"): |
| 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) |
|
|
|
|
| |
|
|
| 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 [] |
|
|
|
|
| |
| |
| |
| 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() |
| |
| AV_SUFFIX = {"BO": "BSE", "NS": "BSE", |
| "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 |
| 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 |
| 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")): |
| 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: |
| 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]): |
| 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: |
| 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()): |
| 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 |
| 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_put(key, out or [{"_empty": True}]) |
| return [o for o in out if "av" in o] |
|
|
|
|
| |
| |
| |
| |
| |
| |
|
|
| _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) |
| |
| |
| |
| 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 |
| |
| |
| if strong and (ok_loc(meta) or len(toks) < 2 or toks[1] in mname): |
| return {"adr": base, "location": meta.get("location")} |
| |
| 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"] |
|
|
|
|
| |
|
|
| 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 |
|
|
