254 lines
9.9 KiB
Python
254 lines
9.9 KiB
Python
"""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]
|
|
try:
|
|
subprocess.run([str(GOGET), *chunk], cwd=DATA,
|
|
capture_output=True, timeout=7200)
|
|
except Exception as e: # never kill the whole run for one batch
|
|
log(f"download: batch {i // batch + 1} FAILED ({e}) - continuing")
|
|
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]}")
|
|
|
|
|
|
PIDFILE = HERE / "overnight.pid"
|
|
|
|
|
|
def run(stages: list[str] | None = None) -> None:
|
|
import os
|
|
stages = stages or ["verify", "select", "download", "screen", "finalize"]
|
|
PIDFILE.write_text(str(os.getpid())) # liveness marker for watchdog
|
|
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)
|