f/fundlab/overnight.py

"""Overnight comprehensive screen (v2): EVERY OTC open-end fund in the
497 universe, no name pre-filter.

Everything is cached and resumable - safe to kill and restart:
  universe_cache/covers.json     CIK -> series+tickers (from edgar_universe)
  universe_cache/yahoo_meta.json ticker -> {name, type, exch, days}
  universe_cache/selected.json   final (sym, name, days) work list
  search_all.json                screen results, streamed + resumed

Stages:
  1. verify  - Yahoo chart call per non-local ticker (threaded, 429
               backoff); local tickers measured from CSV row counts
  2. select  - one class per series (longest history), >=5y,
               instrumentType MUTUALFUND; name-tag (not a filter)
  3. download- goget the missing symbols in resumable batches
  4. screen  - screen_fund per fund, skip already-screened, save often
  5. finalize- summary stats into RESEARCH.md

Usage: python -m fundlab.overnight [stage]   (default: all stages)
"""
from __future__ import annotations

import json
import re
import subprocess
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from fundlab import search
from fundlab.dbmine import PATTERN as ALPHA_KW

HERE = Path(__file__).parent
CACHE = HERE / "universe_cache"
COVERS = CACHE / "covers.json"
META = CACHE / "yahoo_meta.json"
SELECTED = CACHE / "selected.json"
RESULTS = HERE / "search_all.json"
RESEARCH = HERE / "RESEARCH.md"
DATA = Path.home() / "prog/fin/stocks"
GOGET = Path.home() / "go/bin/goget"
MIN_DAYS = 1250          # ~5y of daily bars
UA = {"User-Agent": "research test@example.com"}


def log(msg: str) -> None:
    print(time.strftime("%H:%M:%S"), msg, flush=True)
    with RESEARCH.open("a") as f:
        f.write(f"- {time.strftime('%Y-%m-%d %H:%M')} {msg}\n")


def all_tickers() -> dict[str, str]:
    """ticker -> series name (first series seen wins)."""
    d = json.loads(COVERS.read_text())
    out: dict[str, str] = {}
    for v in d.values():
        for s in v.get("series", []):
            name = s.get("name", "")
            for t in s.get("tickers", []):
                out.setdefault(t, name)
    return out


# ---------------------------------------------------------------- stage 1
def _yahoo_one(t: str) -> tuple[str, dict]:
    url = (f"https://query1.finance.yahoo.com/v8/finance/chart/{t}"
           f"?range=20y&interval=1d")
    for attempt in range(4):
        try:
            req = urllib.request.Request(url, headers=search.UA)
            d = json.load(urllib.request.urlopen(req, timeout=30))
            res = (d.get("chart") or {}).get("result")
            if not res:
                err = ((d.get("chart") or {}).get("error") or {})
                return t, {"error": str(err.get("code", "no result"))}
            meta = res[0].get("meta", {})
            return t, {"name": meta.get("longName") or
                       meta.get("shortName") or "",
                       "type": (meta.get("instrumentType") or "").upper(),
                       "exch": meta.get("fullExchangeName") or "",
                       "days": len(res[0].get("timestamp", []))}
        except urllib.error.HTTPError as e:
            if e.code == 429 and attempt < 3:
                time.sleep(5 * (attempt + 1))
                continue
            return t, {"error": f"HTTP {e.code}"}
        except Exception as e:
            return t, {"error": str(e)}
    return t, {"error": "retries exhausted"}


def stage_verify(workers: int = 4) -> dict:
    tickers = all_tickers()
    local = {p.name[:-5].lower() for p in DATA.glob("*.json")}
    meta: dict = (json.loads(META.read_text()) if META.exists() else {})
    # local tickers: measure from CSV row counts (no Yahoo call)
    for t in tickers:
        if t.lower() in local and t not in meta:
            csv = DATA / f"{t.lower()}-history.csv"
            if not csv.exists():
                meta[t] = {"local": True, "days": 0,
                           "type": "MUTUALFUND",
                           "name": tickers[t], "exch": "local"}
                continue
            with csv.open() as f:
                days = max(0, sum(1 for _ in f) - 1)
            meta[t] = {"local": True, "days": days, "type": "MUTUALFUND",
                       "name": tickers[t], "exch": "local"}
    todo = [t for t in tickers if t not in meta
            and t.lower() not in local]
    log(f"verify: {len(tickers)} tickers, {len(todo)} to hit Yahoo")
    done = 0
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futs = {ex.submit(_yahoo_one, t): t for t in todo}
        for fut in as_completed(futs):
            t, res = fut.result()
            meta[t] = res
            done += 1
            if done % 500 == 0:
                META.write_text(json.dumps(meta))
                log(f"verify: {done}/{len(todo)}")
    META.write_text(json.dumps(meta))
    ok = sum(1 for v in meta.values()
             if isinstance(v.get("days"), int))
    log(f"verify done: {ok}/{len(meta)} with data")
    return meta


# ---------------------------------------------------------------- stage 2
def select_rows(tickers: dict[str, str], meta: dict, min_days: int = MIN_DAYS) -> list[dict]:
    """Pure: (ticker->series-name, ticker->meta) -> one class per fund.

    Keeps MUTUALFUND classes with >= min_days bars; collapses share
    classes (same series name) to the longest-history one; tags each
    fund with whether its NAME matches the alpha pattern (a label, not
    a filter in v2).
    """
    out = []
    for t, name in tickers.items():
        m = meta.get(t) or {}
        if not isinstance(m.get("days"), int):
            continue
        if m.get("type", "MUTUALFUND") != "MUTUALFUND":
            continue
        if m["days"] < min_days:
            continue
        out.append({"sym": t.lower(), "ticker": t, "name": name or
                    m.get("name", ""), "days": m["days"],
                    "local": bool(m.get("local")),
                    "alpha_name": bool(ALPHA_KW.search(name or ""))})
    by_name: dict[str, dict] = {}
    for row in out:
        key = row["name"] or row["sym"]
        cur = by_name.get(key)
        if cur is None or row["days"] > cur["days"]:
            by_name[key] = row
    return sorted(by_name.values(), key=lambda r: r["sym"])


def stage_select() -> list[dict]:
    sel = select_rows(all_tickers(), json.loads(META.read_text()))
    SELECTED.write_text(json.dumps(sel, indent=1))
    log(f"select: {len(sel)} funds "
        f"({sum(1 for r in sel if r['local'])} local, "
        f"{sum(1 for r in sel if not r['local'])} external)")
    return sel


# ---------------------------------------------------------------- stage 3
def stage_download(batch: int = 200) -> None:
    sel = json.loads(SELECTED.read_text())
    missing = [r["sym"] for r in sel
               if not (DATA / f"{r['sym']}-history.csv").exists()]
    log(f"download: {len(missing)} missing symbols")
    for i in range(0, len(missing), batch):
        chunk = missing[i:i + batch]
        subprocess.run([str(GOGET), *chunk], cwd=DATA, capture_output=True,
                       timeout=7200)
        got = sum(1 for s in chunk if (DATA / f"{s}-history.csv").exists())
        log(f"download: batch {i//batch + 1} -> {got}/{len(chunk)}")
    still = sum(1 for s in missing
                if not (DATA / f"{s}-history.csv").exists())
    log(f"download done: {still} still missing (no Yahoo data?)")


# ---------------------------------------------------------------- stage 4
def stage_screen() -> None:
    sel = json.loads(SELECTED.read_text())
    results = (json.loads(RESULTS.read_text()) if RESULTS.exists() else {})
    todo = [r for r in sel if r["sym"] not in results]
    log(f"screen: {len(sel)} funds, {len(todo)} to do")
    for i, r in enumerate(todo):
        try:
            row = search.screen_fund(r["sym"], r["name"], "all")
        except Exception as e:
            row = {"sym": r["sym"], "name": r["name"], "source": "all",
                   "error": str(e)}
        row["alpha_name"] = r["alpha_name"]
        results[r["sym"]] = row
        if (i + 1) % 100 == 0:
            RESULTS.write_text(json.dumps(results, default=str))
            log(f"screen: {i + 1}/{len(todo)}")
    RESULTS.write_text(json.dumps(results, default=str))
    log(f"screen done: {len(results)} funds")


# ---------------------------------------------------------------- stage 5
def stage_finalize() -> None:
    results = json.loads(RESULTS.read_text())
    rows = [v for v in results.values() if isinstance(v, dict)]
    from collections import Counter
    verdicts = Counter(str(v.get("verdict", v.get("error", "?"))).split(" -")[0]
                       .split(" (")[0] for v in rows)
    log(f"finalize: {len(rows)} funds screened")
    for k, n in verdicts.most_common():
        log(f"  {n:5d}  {k}")
    # alpha funds whose names did NOT match the pattern (v1 blind spot)
    missed = [v for v in rows if v.get("alpha_name") is False
              and str(v.get("verdict", "")).startswith("CANDIDATE")]
    log(f"  candidates v1 name-filter would have missed: "
        f"{[v['sym'] for v in missed]}")


def run(stages: list[str] | None = None) -> None:
    stages = stages or ["verify", "select", "download", "screen", "finalize"]
    t0 = time.time()
    for s in stages:
        log(f"=== stage {s} ===")
        if s == "verify":
            stage_verify()
        elif s == "select":
            stage_select()
        elif s == "download":
            stage_download()
        elif s == "screen":
            stage_screen()
        elif s == "finalize":
            stage_finalize()
    log(f"=== overnight run finished in {(time.time() - t0) / 3600:.1f}h ===")


if __name__ == "__main__":
    import sys
    run(sys.argv[1:] or None)