""" Per-ticker auto-configuration + trust report. Instead of guessing lookback/T, this searches a small grid by *walk-forward evidence* on the ticker's own recent history, rejects configs whose fed window violates Kronos' normalisation assumption (wide price range), and picks the one that best beats a random-walk baseline. It writes a profile JSON the dashboard consumes so every served forecast carries a validated track record, a calibrated band, and an honest verdict. Example ------- python us/autoconfig.py --ticker COHR --pred_len 20 --n_windows 6 --n_paths 15 python us/autoconfig.py --ticker AAPL # defaults Note: this is an *offline* step (grid x windows x paths predictions ~ minutes on CPU). The result is cached; serving then loads it instantly. """ import argparse import json import sys import time from datetime import datetime, timezone from pathlib import Path import numpy as np from predict_us import fetch_ohlcv from engine import get_predictor, _RANGE_THRESH import validation def profile_path(outdir: Path, ticker: str, interval: str) -> Path: return outdir / f"{ticker.upper()}_{interval}.json" def _num(x, default): """None/NaN -> a sentinel so it sorts as the worst option.""" return x if isinstance(x, (int, float)) and x == x else default def select_best(results: list, objective: str = "blend") -> dict: """ Pick a config (all objectives reject out-of-distribution windows first): * blend : (default) among in-distribution configs that beat a random walk (skill < 1.1), take the lowest MAPE — MAPE-driven but guarded against persistence-mimics / normalisation-broken windows. * mape : lowest MAPE (naive: a 'tomorrow = today' mimic also scores low). * skill : lowest skill_vs_rw (best at genuinely beating persistence). * ic : highest Information Coefficient (predicted-vs-realised return correlation) — how the Kronos paper judges skill; direction/ranking, not price level. * dir : highest directional accuracy. """ def key(r): ind = 0 if r["_in_distribution"] else 1 m = r["metrics"] mape = _num(m.get("MAPE%"), 1e9) skill = _num(m.get("skill_vs_rw"), 1e9) ic = _num(m.get("IC"), -1e9) da = _num(m.get("DirAcc%"), -1e9) if objective == "mape": return (ind, mape) if objective == "skill": return (ind, skill, mape) if objective == "ic": return (ind, -ic) if objective == "dir": return (ind, -da) competitive = 0 if skill < 1.1 else 1 return (ind, competitive, mape) return sorted(results, key=key)[0] def _grade(sel: dict) -> dict: """Turn the selected config's evidence into an honest verdict.""" m = sel["metrics"] in_dist = sel["_in_distribution"] skill = _num(m.get("skill_vs_rw"), 1e9) da = _num(m.get("DirAcc%"), 0.0) ic = _num(m.get("IC"), float("nan")) reasons = [] if not in_dist: reasons.append(f"視窗價格範圍 {sel['validity']['range_ratio_mean']}× 超出 Kronos " f"正規化可靠區(≥{_RANGE_THRESH}×),預測可能失真") return {"grade": "D", "trust": "unreliable-ood", "reasons": reasons} reasons.append(f"vs 隨機漫步 skill={skill:.2f}（<1 才算贏過『明天=今天』）") reasons.append(f"方向命中 {da:.0f}%（50% 為亂猜）" + (f" · IC={ic:.2f}" if ic == ic else "")) if skill < 0.97 and da >= 55: grade, trust = "A", "usable" elif skill < 1.0: grade, trust = "B", "marginal" elif skill <= 1.05: grade, trust = "C", "no-edge" reasons.append("點預測未能勝過隨機漫步 → 別當方向訊號用,只看校準後的區間") else: grade, trust = "D", "worse-than-naive" reasons.append("比隨機漫步還差 → 不建議使用此預測") return {"grade": grade, "trust": trust, "reasons": reasons} def run_ticker(predictor, df, ticker, interval, period, pred_len, lookbacks, temps, top_p, n_paths, n_windows, n_outer, target_coverage, objective, seed, verbose=False) -> dict: """ Nested walk-forward for one ticker → JSON trust profile. Rolling origins are split in time: the earlier *inner* windows drive config selection and band-scale calibration; the most recent *outer* windows — which the selection never saw — are used ONLY to report the honest track record. This removes the selection-bias optimism you'd get from grading a config on the same windows you picked it with. """ max_lb = max(lookbacks) windows = validation.make_windows(len(df), max_lb, pred_len, n_windows) if len(windows) < n_outer + 2: raise ValueError(f"need >= {n_outer + 2} windows to nest; got {len(windows)} " f"(raise --n_windows or --period)") n_inner = len(windows) - n_outer configs = [] for lb in lookbacks: if len(df) < max_lb + pred_len * 2: continue for T in temps: recs = validation.run_windows(predictor, df, lb, pred_len, T=T, top_p=top_p, n_paths=n_paths, windows=windows, seed=seed) if len(recs) < len(windows): continue inner, outer = recs[:n_inner], recs[n_inner:] r_in = validation.aggregate(inner, target_coverage) # fits band_scale on inner item = {"config": {"lookback": lb, "pred_len": pred_len, "T": T, "top_p": top_p, "top_k": 0, "n_paths": n_paths}, "metrics": r_in["metrics"], "calibration": r_in["calibration"], "validity": r_in["validity"], "n_windows": r_in["n_windows"], "_in_distribution": r_in["validity"]["range_ratio_mean"] < _RANGE_THRESH, "_outer": outer} configs.append(item) if verbose: m = r_in["metrics"] print(f" [inner] lb={lb:>3} T={T} MAPE={m['MAPE%']}% skill={m['skill_vs_rw']} " f"IC={m['IC']} Dir={m['DirAcc%']}% range={item['validity']['range_ratio_mean']}x" f"{'' if item['_in_distribution'] else ' ⚠OOD'}") if not configs: raise ValueError("no runnable configs (need more history / smaller pred_len)") sel = select_best(configs, objective) # honest report: apply the inner-fitted band scale to the untouched outer windows r_out = validation.aggregate(sel["_outer"], target_coverage, band_scale=sel["calibration"]["band_scale"]) verdict = _grade({"metrics": r_out["metrics"], "_in_distribution": sel["_in_distribution"], "validity": r_out["validity"]}) return { "ticker": ticker.upper(), "interval": interval, "period": period, "generated": datetime.now(timezone.utc).isoformat(timespec="seconds"), "objective": objective, "validation": {"scheme": "nested", "n_windows": len(windows), "n_inner": n_inner, "n_outer": n_outer}, "recommended": {"lookback": sel["config"]["lookback"], "pred_len": pred_len, "T": sel["config"]["T"], "top_p": sel["config"]["top_p"], "top_k": 0, "n_paths": n_paths}, "calibration": {"target_coverage": target_coverage, "band_scale": sel["calibration"]["band_scale"], "coverage_raw": r_out["calibration"]["coverage_raw"], "coverage_calibrated": r_out["calibration"]["coverage_calibrated"]}, "track_record": {**r_out["metrics"], "n_windows": n_outer, "note": "honest: outer windows the selection never saw"}, "selection_metrics": {**sel["metrics"], "n_windows": n_inner}, "validity": {**sel["validity"], "in_distribution": sel["_in_distribution"], "range_thresh": _RANGE_THRESH}, "verdict": verdict, "all_configs": [ {"lookback": c["config"]["lookback"], "T": c["config"]["T"], **c["metrics"], **c["calibration"], "range_ratio_mean": c["validity"]["range_ratio_mean"], "in_distribution": c["_in_distribution"]} for c in configs], } def main(): try: # Windows consoles default to cp950/utf-16; make Chinese output readable sys.stdout.reconfigure(encoding="utf-8") except Exception: pass ap = argparse.ArgumentParser() ap.add_argument("--ticker", default="AAPL") ap.add_argument("--interval", default="1d") ap.add_argument("--period", default="5y") ap.add_argument("--pred_len", type=int, default=20) ap.add_argument("--n_windows", type=int, default=8, help="total rolling origins (inner+outer)") ap.add_argument("--n_outer", type=int, default=3, help="most-recent windows held out for the honest report") ap.add_argument("--n_paths", type=int, default=15, help="ensemble size (also used for serving, keeps calibration valid)") ap.add_argument("--lookbacks", default="60,120,240,400") ap.add_argument("--temps", default="0.7,1.0") ap.add_argument("--top_p", type=float, default=0.9) ap.add_argument("--target_coverage", type=float, default=0.8) ap.add_argument("--objective", choices=["blend", "mape", "skill", "ic", "dir"], default="blend", help="blend=lowest MAPE among in-dist configs that beat RW; ic=max Information Coefficient; dir=max direction") ap.add_argument("--seed", type=int, default=123) ap.add_argument("--outdir", default=str(Path(__file__).resolve().parent / "out" / "profiles")) args = ap.parse_args() lookbacks = [int(x) for x in args.lookbacks.split(",") if x.strip()] temps = [float(x) for x in args.temps.split(",") if x.strip()] df = fetch_ohlcv(args.ticker, args.interval, args.period) print(f"[data] {args.ticker} {args.interval}: {len(df)} bars -> {df['timestamps'].iloc[-1].date()}") predictor = get_predictor() grid = len(lookbacks) * len(temps) print(f"[search] {grid} configs x {args.n_windows} windows x {args.n_paths} paths " f"(nested: {args.n_windows-args.n_outer} inner / {args.n_outer} outer)\n") t0 = time.time() profile = run_ticker(predictor, df, args.ticker, args.interval, args.period, args.pred_len, lookbacks, temps, args.top_p, args.n_paths, args.n_windows, args.n_outer, args.target_coverage, args.objective, args.seed, verbose=True) outdir = Path(args.outdir); outdir.mkdir(parents=True, exist_ok=True) pth = profile_path(outdir, args.ticker, args.interval) pth.write_text(json.dumps(profile, indent=2, ensure_ascii=False), encoding="utf-8") rec = profile["recommended"]; v = profile["verdict"] tr = profile["track_record"]; sm = profile["selection_metrics"]; c = profile["calibration"] print(f"\n=== TRUST REPORT · {profile['ticker']} (objective={args.objective}) ===") print(f"grade {v['grade']} ({v['trust']})") for r in v["reasons"]: print(" -", r) print(f"recommended: lookback={rec['lookback']} T={rec['T']} top_p={rec['top_p']} " f"pred_len={rec['pred_len']} n_paths={rec['n_paths']}") print(f"HONEST (outer {tr['n_windows']} win, unseen): MAPE={tr['MAPE%']}% " f"skill={tr['skill_vs_rw']} IC={tr['IC']} Dir={tr['DirAcc%']}% " f"cover={c['coverage_calibrated']*100:.0f}%") print(f"selection (inner {sm['n_windows']} win, optimistic): MAPE={sm['MAPE%']}% " f"skill={sm['skill_vs_rw']} IC={sm['IC']}") print(f"[done] {time.time()-t0:.0f}s profile -> {pth}") if __name__ == "__main__": main()