feat: 60/40 paper-forward harness (the ~0.7-Sharpe non-crypto baseline)
Local CLI paper-forward test of 60% SPY / 40% IEF (ETF analog of the ES/ZN 60/40 backtest, +0.72 Sharpe), mirroring the funding harness. Free Yahoo adjusted closes (dividends+coupons = true total return). Subcommands snapshot/run/status. Books each real trading day once with catch-up (handles weekends/missed runs); forward-start (no history backfill). Idempotent via last_date. Daytime cron 12/15/18 UTC. Always-invested baseline -> confirms forward Sharpe tracks the ~0.7 backtest, the clean no-phantom option available regardless of crypto regime. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
151
scripts/surfer/sixtyforty_paper.py
Normal file
151
scripts/surfer/sixtyforty_paper.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user