diff --git a/scripts/surfer/sixtyforty_paper.py b/scripts/surfer/sixtyforty_paper.py new file mode 100644 index 000000000..ea101d180 --- /dev/null +++ b/scripts/surfer/sixtyforty_paper.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +"""sixtyforty — local CLI paper-forward test of a simple 60/40 ETF portfolio (the ~0.7-Sharpe +non-crypto baseline). No agents, no cloud, no capital. Free Yahoo daily adjusted closes. + +60% SPY (S&P 500) + 40% IEF (7-10yr Treasuries) — the deployable ETF analog of the ES/ZN 60/40 +backtest (+0.72 Sharpe). Always invested (no regime filter); the test just confirms the realized +forward Sharpe/return tracks the backtest. Books each real trading day exactly once (catch-up on +weekends / missed runs), using ADJUSTED closes (dividends + coupons = true total return). + + python3 sixtyforty_paper.py snapshot latest prices + recent daily portfolio returns + python3 sixtyforty_paper.py run book new trading days, persist (cron-compatible) + python3 sixtyforty_paper.py status cumulative return + annualized Sharpe/vol/maxDD + (alias: 'paper' == 'run') +""" +import datetime +import json +import math +import os +import sys +import urllib.request + +_REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +STATE = os.path.join(_REPO, "data/surfer/sixtyforty_state.json") +INSTR = [("SPY", 0.6), ("IEF", 0.4)] + + +def get(url, tries=4): + import time + for a in range(tries): + try: + req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) + return json.loads(urllib.request.urlopen(req, timeout=30).read()) + except Exception: + if a == tries - 1: + raise + time.sleep(2 * (a + 1)) + + +def daily_adjclose(sym): + """{ 'YYYY-MM-DD': adjclose } for ~3 months of trading days (Yahoo, free).""" + u = f"https://query1.finance.yahoo.com/v8/finance/chart/{sym}?interval=1d&range=3mo" + res = get(u)["chart"]["result"][0] + ts = res["timestamp"] + ind = res["indicators"] + adj = ind.get("adjclose", [{}])[0].get("adjclose") or ind["quote"][0]["close"] + out = {} + for t, c in zip(ts, adj): + if c is not None: + d = datetime.datetime.utcfromtimestamp(t).strftime("%Y-%m-%d") + out[d] = float(c) + return out + + +def load_state(): + if os.path.exists(STATE): + return json.load(open(STATE)) + return {"last_date": "", "days": 0, "sum_r": 0.0, "sumsq_r": 0.0, + "equity": 1.0, "peak": 1.0, "max_dd": 0.0} + + +def metrics(st): + n = st["days"] + if n < 2: + return None + mean = st["sum_r"] / n + var = max(st["sumsq_r"] / n - mean * mean, 0.0) + std = math.sqrt(var) + apr = mean * 252 + vol = std * math.sqrt(252) + return dict(apr=apr, vol=vol, sharpe=(apr / vol if vol > 0 else float("nan")), + total=st["equity"] - 1.0, maxdd=st["max_dd"]) + + +def series(): + data = {s: daily_adjclose(s) for s, _ in INSTR} + dates = sorted(set.intersection(*[set(d) for d in data.values()])) + return data, dates + + +def cmd_run(): + st = load_state() + data, dates = series() + if len(dates) < 2: + print("not enough data"); return + os.makedirs(os.path.dirname(STATE), exist_ok=True) + if not st["last_date"]: # fresh: start forward from latest close, no backfill + st["last_date"] = dates[-1] + json.dump(st, open(STATE, "w")) + print(f"initialized forward tracking from {dates[-1]}; first booked return on the next trading day.") + return + booked = 0 + for i in range(1, len(dates)): + d, dprev = dates[i], dates[i - 1] + if d <= st["last_date"]: + continue + r = sum(w * (data[s][d] / data[s][dprev] - 1.0) for s, w in INSTR) # daily-rebalanced port return + st["days"] += 1 + st["sum_r"] += r + st["sumsq_r"] += r * r + st["equity"] *= (1 + r) + st["peak"] = max(st["peak"], st["equity"]) + st["max_dd"] = min(st["max_dd"], st["equity"] / st["peak"] - 1.0) + st["last_date"] = d + booked += 1 + os.makedirs(os.path.dirname(STATE), exist_ok=True) + json.dump(st, open(STATE, "w")) + m = metrics(st) + tail = f" | Sharpe {m['sharpe']:+.2f} APR {100*m['apr']:+.1f}% maxDD {100*m['maxdd']:+.1f}%" if m else "" + print(f"{datetime.date.today()} booked {booked} new trading day(s) -> day {st['days']} " + f"(through {st['last_date']}); total {100*(st['equity']-1):+.2f}%{tail}") + + +def cmd_status(): + st = load_state() + m = metrics(st) + print(f"60/40 (SPY 60% / IEF 40%) paper — day {st['days']} (through {st['last_date'] or 'n/a'})") + print(f" total return: {100*(st['equity']-1):+.2f}%") + if m: + print(f" annualized: Sharpe {m['sharpe']:+.2f} return {100*m['apr']:+.1f}% vol {100*m['vol']:.1f}% maxDD {100*m['maxdd']:+.1f}%") + print(f" backtest reference: ~0.72 Sharpe (ES/ZN 60/40, 2010-2026)") + else: + print(" (need >=2 booked trading days for annualized stats)") + + +def cmd_snapshot(): + data, dates = series() + last = dates[-1] + print(f"60/40 snapshot {datetime.date.today()} (latest close {last})") + for s, w in INSTR: + print(f" {s} ({int(w*100)}%): {data[s][last]:.2f}") + print(" recent daily portfolio returns:") + for i in range(max(1, len(dates) - 5), len(dates)): + d, dprev = dates[i], dates[i - 1] + r = sum(w * (data[s][d] / data[s][dprev] - 1.0) for s, w in INSTR) + print(f" {d}: {100*r:+.2f}%") + + +def main(): + cmd = sys.argv[1] if len(sys.argv) > 1 else "status" + if cmd in ("run", "paper"): + cmd_run() + elif cmd == "status": + cmd_status() + elif cmd == "snapshot": + cmd_snapshot() + else: + print(__doc__) + + +if __name__ == "__main__": + main()