f/fundlab/search.py

"""Alpha search: screen the curated longlist (+ the 13-fund shortlist)
for idiosyncratic alpha-driven funds that complement the current
portfolio (qspnx/pmaix 50/50; benchmarks spy/agg/tlt).

Method (same engine as fundlab/decompose.py, broad candidate set):
  1. resolve tickers from fund NAMES via Yahoo search (precision-gated;
     unresolvable candidates are dropped, never guessed),
  2. download missing returns with the goget tool (if the user has
     authorized it and the binary is present),
  3. regress each fund's daily total returns on the same 21 broad sleeve
     axes (BIC forward selection, |t|>2), full history + last 5y,
  4. compute correlation vs the current portfolio and vs the benchmark
     mix, and a rolling 6m alpha persistence check,
  5. rank: CANDIDATE = low R^2 (not a sleeve mix) + significant alpha
     (5y t>=2, full t>=1.5) + persistent (>=50% of rolling windows
     positive) + low portfolio correlation (<0.3).

Output: fundlab/search_results.json (read by the app).
"""
from __future__ import annotations

import difflib
import json
import re
import subprocess
import urllib.parse
import urllib.request
from functools import lru_cache
from pathlib import Path

import numpy as np
import pandas as pd

from fundlab import decompose
from fundlab.searchlist import (BENCHMARKS, BROAD_SLEEVES, LONGLIST,
                                PORTFOLIO, SHORTLIST)

DATA = Path.home() / "prog/fin/stocks"
GOGET = Path.home() / "go/bin/goget"
RESULTS = Path(__file__).parent / "search_results.json"
RESOLVED = Path(__file__).parent / "resolved_tickers.json"


def _load_resolved() -> dict:
    try:
        return json.loads(RESOLVED.read_text())
    except Exception:
        return {}


def _save_resolved(d: dict) -> None:
    RESOLVED.write_text(json.dumps(d, indent=1))
UA = {"User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"}

# ------------------------------------------------------------------ resolve
def _norm_name(s: str) -> str:
    s = s.lower()
    s = re.sub(r"[^a-z0-9]+", " ", s)
    # strip share-class suffixes
    for suf in ("advisor", "institutional", "retail", "plus", "a", "i", "c",
                "b", "x", "z"):
        s = re.sub(rf"\b{re.escape(suf)}\b\s*$", " ", s)
    s = re.sub(r"\bfund\b\s*$", "", s)
    return re.sub(r"\s+", " ", s).strip()


def chart_meta(sym: str) -> dict | None:
    """Yahoo chart API (no crumb needed): returns meta for a symbol."""
    url = ("https://query1.finance.yahoo.com/v8/finance/chart/"
           f"{sym.upper()}?range=1y&interval=1d")
    req = urllib.request.Request(url, headers=UA)
    try:
        d = json.load(urllib.request.urlopen(req, timeout=30))
    except Exception:
        return None
    res = (d.get("chart") or {}).get("result")
    return res[0].get("meta") if res else None


_STOP = {"the", "and", "for", "of", "to", "a", "i", "c", "fund", "funds",
          "series", "class", "in", "on"}


def _tokens(s: str) -> set[str]:
    return {t for t in _norm_name(s).split() if t not in _STOP}


def _name_match(target: str, cand: str) -> float:
    """Fraction of the target's significant tokens present in cand."""
    tt, ct = _tokens(target), _tokens(cand)
    return len(tt & ct) / len(tt) if tt else 0.0


def resolve(name: str, guess: str | None = None) -> dict | None:
    """Resolve a fund name to a ticker, verifying against Yahoo chart
    metadata (instrumentType + longName). A guess is accepted only if the
    chart says it's a mutual fund whose name shares >= 2/3 of the
    target's significant tokens AND has sim >= 0.5; otherwise the
    candidate is dropped - never guessed.

    Returns {symbol, name, sim, guess} or {symbol: None, error}.
    """
    if not guess:
        return {"symbol": None, "name": name,
                "error": "no ticker guess (add one or resolve manually)"}
    meta = chart_meta(guess)
    if not meta:
        return {"symbol": None, "name": name,
                "error": f"no chart data for guess {guess}"}
    itype = (meta.get("instrumentType") or "").upper()
    if itype not in ("MUTUALFUND", "FUND"):
        return {"symbol": None, "name": name,
                "error": f"{guess} is not a mutual fund ({itype})"}
    cand = meta.get("longName") or meta.get("shortName") or ""
    sim = difflib.SequenceMatcher(None, _norm_name(name),
                                  _norm_name(cand)).ratio()
    tok = _name_match(name, cand)
    if tok < 2 / 3 or sim < 0.5:
        return {"symbol": None, "name": name,
                "error": f"{guess} is {cand!r} (tok={tok:.2f} sim={sim:.2f})"}
    return {"symbol": guess.lower(), "name": cand, "sim": round(sim, 3),
            "guess": guess}


# ------------------------------------------------------------------ data
@lru_cache(maxsize=None)
def has_data(sym: str) -> bool:
    return (DATA / f"{sym}-history.csv").exists()


def ensure_data(symbols: list[str]) -> list[str]:
    """goget-download the missing symbols (into the data dir). Returns the
    symbols still without data."""
    missing = [s for s in symbols if not has_data(s)]
    if missing and GOGET.exists():
        try:
            subprocess.run([str(GOGET), *missing], cwd=DATA,
                           capture_output=True, timeout=900)
        except Exception as e:
            print(f"goget failed: {e}", flush=True)
    has_data.cache_clear()
    return [s for s in symbols if not has_data(s)]


# ------------------------------------------------------------------ screen
def _mix(returns: pd.DataFrame, weights: dict[str, float]) -> pd.Series:
    out = None
    for s, w in weights.items():
        if s in returns.columns:
            v = returns[s] * w
            out = v if out is None else out.add(v, fill_value=0.0)
    return out


def _max_drawdown(r: pd.Series) -> float:
    eq = (1 + r.fillna(0)).cumprod()
    return float((eq / eq.cummax() - 1).min())


def screen_fund(sym: str, name: str, bucket: str,
                in_portfolio: bool = False) -> dict:
    full = decompose.decompose(sym, candidates={sym: BROAD_SLEEVES})
    if "r2" not in full:
        return {"sym": sym, "name": name, "bucket": bucket,
                "error": full.get("error", "no data")}
    rec = decompose.decompose(sym, start=decompose.RECENT_WINDOW,
                              candidates={sym: BROAD_SLEEVES})
    r = decompose.returns_panel(
        [sym] + list(PORTFOLIO) + list(BENCHMARKS) + BROAD_SLEEVES)
    corr_port = corr_bench = np.nan
    if sym in r.columns:
        syms = list(dict.fromkeys(
            [sym] + list(PORTFOLIO) + list(BENCHMARKS)))  # dedup: the
        c = r[[s for s in syms if s in r.columns]].dropna()  # screened fund may BE a portfolio component
        if len(c) > 252:
            pf = _mix(c, PORTFOLIO)
            bm = _mix(c, BENCHMARKS)
            if pf is not None:
                corr_port = float(np.corrcoef(c[sym], pf)[0, 1])
            if bm is not None:
                corr_bench = float(np.corrcoef(c[sym], bm)[0, 1])
    # rolling 6m alpha persistence: fraction of 126d windows where the
    # fund beat its fitted sleeve mix. Use the 5y model when the full-
    # sample model is empty (vintage funds), else the full model.
    frac_pos = np.nan
    _model = rec if ("components" in rec and rec["components"]) else full
    if _model.get("components"):
        y = r[sym].to_numpy()
        X = np.column_stack(
            [np.ones(len(y)), *[r[c["sym"]].to_numpy()
                                for c in _model["components"]]])
        ok = ~(np.isnan(y) | np.isnan(X).any(axis=1))
        y, X = y[ok], X[ok]
        if len(y) >= 252:
            beta, *_ = np.linalg.lstsq(X, y, rcond=None)
            ex = y - X @ beta
            w = 126
            if len(ex) > w * 3:
                fr = [ex[i - w:i].mean() > 0
                      for i in range(w, len(ex), 21)]
                frac_pos = float(np.mean(fr))
    f5 = rec.get("r2", np.nan)
    t5 = rec.get("alpha_t", np.nan)
    tf = full.get("alpha_t", np.nan)
    # kind: alpha (mostly idiosyncratic), semi_alpha (mostly explained by
    # net exposure but with significant residual alpha - typical of
    # market-neutral funds), sleeve (a static mix), weak (neither)
    if f5 < 0.6 and t5 >= 2.0 and tf >= 1.25:
        kind = "alpha"
    elif f5 < 0.85 and t5 > 0 and tf >= 3.0:
        kind = "semi_alpha"   # positive 5y alpha required for candidacy
    elif f5 >= 0.85:
        kind = "sleeve"
    else:
        kind = "weak"
    persistent = frac_pos is not None and frac_pos >= 0.45
    complementary = corr_port == corr_port and corr_port < 0.3
    if "r2" not in rec:
        verdict = "no 5y window"
    elif kind in ("alpha", "semi_alpha") and persistent and complementary:
        verdict = ("CANDIDATE - idiosyncratic alpha, complements portfolio"
                   if kind == "alpha" else
                   "CANDIDATE (semi-alpha: mostly explained by net exposure)")
    elif kind in ("alpha", "semi_alpha") and not complementary:
        verdict = "alpha, but correlated with current portfolio"
    elif kind in ("alpha", "semi_alpha") and not persistent:
        verdict = "alpha in 5y window, but not persistent (lucky stretch?)"
    elif kind == "sleeve":
        verdict = "sleeve mix (R² high) - not alpha-driven"
    else:
        verdict = "weak/unstable alpha"
    return {
        "sym": sym, "name": name, "bucket": bucket,
        "in_portfolio": in_portfolio,
        "alpha_ann_5y": rec.get("alpha_ann"),
        "alpha_t_5y": t5, "alpha_t_full": tf,
        "r2_5y": f5, "r2_full": full.get("r2"),
        "corr_portfolio": corr_port, "corr_benchmark": corr_bench,
        "alpha_pos_frac": frac_pos,
        "fund_max_dd": _max_drawdown(r[sym]) if sym in r.columns else None,
        "first": full.get("start"),
        "verdict": verdict,
    }


def _shortlist_names() -> dict[str, str]:
    try:
        f = json.loads((Path(__file__).parent.parent / "funds.json").read_text())
        return {k: v["name"] for k, v in f.items()}
    except Exception:
        return {}


def run(shortlist_only: bool = False) -> dict:
    rows: dict[str, dict] = {}
    names = _shortlist_names()
    # 1) the shortlist (known tickers, names from funds.json)
    for sym in SHORTLIST:
        in_pf = sym in PORTFOLIO
        row = screen_fund(sym, names.get(sym, sym), "shortlist",
                          in_portfolio=in_pf)
        rows[sym] = row
        print(f"{sym:7} {row.get('verdict', row.get('error'))[:70]}",
              flush=True)
    if shortlist_only:
        RESULTS.write_text(json.dumps(rows, indent=1, default=str))
        return rows
    # 2) the curated longlist: chart-verified guess first, then EDGAR
    # prospectus covers, else drop (never guess)
    from fundlab import tickers as _tickers
    resolved_cache = _load_resolved()
    for name, bucket, guess in LONGLIST:
        cached = resolved_cache.get(name)
        if cached is not None:
            res = cached
            if res.get("symbol"):
                print(f"CACHED {name[:40]:40} -> {res['symbol'].upper()}",
                      flush=True)
        else:
            res = resolve(name, guess)
            if not res.get("symbol"):   # guess missing/rejected -> EDGAR
                res = _tickers.resolve_via_edgar(name)
                if res:
                    print(f"EDGAR  {name[:40]:40} -> {res['symbol'].upper()} "
                          f"(tok={res['sim']})  {res['name'][:44]}", flush=True)
            resolved_cache[name] = res or {"symbol": None, "name": name,
                                           "error": "unresolved"}
            _save_resolved(resolved_cache)
        if res is None or not res.get("symbol"):
            print(f"DROP  {name[:45]:45} {res.get('error', '') if res else ''}")
            rows[f"__drop__{name[:30]}"] = {
                "sym": None, "name": name, "bucket": bucket,
                "error": (res or {}).get("error", "unresolved"),
                "guess": guess}
            continue
        sym = res["symbol"]
        if sym in rows:
            print(f"SKIP  {name[:45]:45} already in shortlist as {sym}")
            continue
        print(f"RESOLVED {name[:40]:40} -> {sym.upper()} "
              f"(sim={res['sim']})", flush=True)
        rows[sym] = {"_resolve": res}  # placeholder for the download pass
    # 3) download whatever is missing, then screen
    to_screen = {s: d for s, d in rows.items() if "_resolve" in d}
    if to_screen:
        still_missing = ensure_data(list(to_screen))
        for s in still_missing:
            rows[s]["error"] = "no return data (download failed)"
        to_screen = {s: d for s, d in to_screen.items() if "error" not in d}
    for sym, d in to_screen.items():
        res = d["_resolve"]
        row = screen_fund(sym, res["name"], bucket)
        row["resolved_from"] = res
        rows[sym] = row
        print(f"{sym:7} {row.get('verdict', row.get('error'))[:70]}",
              flush=True)
    RESULTS.write_text(json.dumps(rows, indent=1, default=str))
    n_cand = sum(1 for r in rows.values()
                 if str(r.get("verdict", "")).startswith("CANDIDATE"))
    print(f"\nwrote {RESULTS} - {n_cand} candidate(s)")
    return rows


if __name__ == "__main__":
    import sys
    run(shortlist_only="--shortlist" in sys.argv)