#!/usr/bin/env python3
"""ROTATION MAP v2 (28 Sep 2026). Where the money should sit: cash, BTC or alts,
and which alt basket is topping. Runs every 4 hours on closed UTC daily candles.
THE BOOK (three layers, nothing else drives a trade)
1 RISK BTC above its 100 day average (1% band) = risk on, else CASH.
2 ALT CLOCK share of 64 alts beating BTC. Alts are bought only after a
CONFIRMED BOTTOM: 90d share 5% or less inside 60 days, 30d share
back to 40%+, risk on. They are sold when risk turns off.
3 BASKET TOP a basket whose own members mostly beat BTC (70%+ over 90d) and
have now stopped (under 50% over 30d) is dropped; its weight
moves to BTC.
HOW IT IS GRADED. major_rotations() labels, with hindsight, every stretch since
June 2020 where cash, BTC or alts beat the next best place by 20%+ (best place
over the next 45 days, stretches under 21 days merged). Hindsight only grades;
no rule reads it. v1 caught 4 of 7, v2 catches 6 of 7 (median 2 days late).
WHAT CHANGED IN v2, and why
* risk average 50 -> 100 days. About 35 variants tried. 100 days with a
0, 1 or 3% band, and 150 days, beat 50 on return AND max drop in 2020-24
and 2024-26. 125 days had smaller drops but made less since 2024; 75 days
was worse in 2020-24. Worst calendar year -34% vs -52%. Survives 1% fees, 4 day late fills, and removing the best
2020-21 coins from every basket. Chosen after seeing both periods, so
2020-24 is no longer fully unseen: the dashboard says so.
* baskets start with their first live coin (was: three), so MEMES covers
the 2021 DOGE rotation instead of starting in 2023.
* candles are UTC days (plain OKX 1D is Hong Kong aligned).
* funding crowding removed as a warning colour: crowded baskets did BETTER
over the next 30 days, not worse.
REJECTED, shown as information or not at all: alts/BTC, ETH/BTC and per basket
relative strength trends (18 versions, all lost since 2024); breadth thrust
entries; the market wide alt top as an exit; stablecoin supply growth; a basket
bouncing off its own low. The trench window predicts launchpad VOLUME (11 of
14), not meme prices.
KNOWN LIMITS. Alt baskets are today's surviving coins (flatters 2020-21).
Seven major rotations is a small sample. The 2023-24 alt run was caught late
(68 days) because the book waits for a washout.
"""
import json, math, os, sys, time, statistics as st, urllib.request, datetime as dt
HERE = os.path.dirname(os.path.abspath(__file__))
STORE = os.path.join(HERE, "store")
PUB_JSON = os.environ.get("PUB_JSON", "/opt/launchwatch/pub/rotmap.json")
TRENCH_DB = os.path.join(HERE, "rot.db")
CFG = os.path.join(HERE, "config.json")
UA = {"User-Agent": "Mozilla/5.0 (rotation-map)"}
BKK = 7 * 3600
FEE = 0.003
BASKETS = {
"BTC": ["BTC"], "ETH": ["ETH"], "SOL": ["SOL"],
"MAJORS": ["BNB", "XRP", "ADA", "AVAX", "DOT", "LINK", "TRX", "LTC", "BCH", "ATOM", "NEAR", "APT",
"SUI", "ICP", "FIL", "HBAR", "XLM", "ETC", "XTZ", "ALGO"],
"MEMES": ["DOGE", "SHIB", "PEPE", "WIF", "BONK", "FLOKI", "MEW", "TRUMP", "BOME", "TURBO", "NEIRO",
"PNUT", "GOAT", "ACT", "PENGU"],
"AI": ["FET", "RENDER", "WLD", "GRT", "ARKM", "VIRTUAL", "AIXBT", "NMR", "PHA", "ATH"],
"DEFI": ["UNI", "AAVE", "CRV", "LDO", "PENDLE", "SNX", "COMP", "SUSHI", "1INCH", "DYDX", "JUP", "RAY",
"ENA", "ETHFI", "EIGEN", "YFI", "UMA"],
}
PRICE_BASKETS = ["BTC", "ETH", "SOL", "MAJORS", "MEMES", "AI", "DEFI"]
ALT_BASKETS = ["ETH", "SOL", "MAJORS", "MEMES", "AI", "DEFI"]
FUND = {"BTC": ["BTC"], "ETH": ["ETH"], "SOL": ["SOL"], "MAJORS": ["XRP", "ADA", "AVAX", "LINK", "SUI"],
"MEMES": ["DOGE", "kPEPE", "WIF", "kBONK", "kSHIB"], "AI": ["FET", "RENDER", "WLD", "TAO"],
"DEFI": ["UNI", "AAVE", "ENA", "PENDLE", "LDO"]}
LABEL = {"BTC": "Bitcoin", "ETH": "Ether", "SOL": "Solana", "MAJORS": "Large caps",
"MEMES": "Memes", "AI": "AI", "DEFI": "DeFi"}
CLOCK_TOP, CLOCK_WASH = 0.70, 0.05
RISK_BAND = 0.01
RISK_MA = 100 # v2: was 50. 100d (any band 0-3%) and 150d beat 50d in both periods; worst year -34% vs -52%
INDEX_NEED = 1 # v2: was 3, which started MEMES only in 2023 and hid the 2021 DOGE rotation
# ------------------------------------------------------------------ fetch --
def get(u, body=None, tries=5):
for i in range(tries):
try:
if body is None:
r = urllib.request.Request(u, headers=UA)
else:
r = urllib.request.Request(u, json.dumps(body).encode(),
dict(UA, **{"content-type": "application/json"}))
return json.load(urllib.request.urlopen(r, timeout=45))
except Exception:
if i == tries - 1:
return None
time.sleep(2 + 3 * i)
def okx_closed(inst, want):
"""Closed UTC daily candles only (plain 1D is Hong Kong aligned and would
mislabel every day). OKX marks the live candle confirm=0; a signal
computed on a half-finished day flips back and forth all day."""
out, after = {}, ""
while len(out) < want:
u = ("https://www.okx.com/api/v5/market/history-candles?instId=%s&bar=1Dutc&limit=100"
% inst) + (("&after=" + after) if after else "")
rows = ((get(u) or {}).get("data")) or []
if not rows:
break
for r in rows:
if r[-1] == "1":
out[dt.datetime.utcfromtimestamp(int(r[0]) / 1000).strftime("%Y-%m-%d")] = float(r[4])
after = rows[-1][0]
time.sleep(0.12)
return out
def hl_funding(coin, start_ms):
t, daily = start_ms, {}
while True:
rows = get("https://api.hyperliquid.xyz/info", {"type": "fundingHistory", "coin": coin, "startTime": t})
if not rows:
break
for r in rows:
d = dt.datetime.utcfromtimestamp(r["time"] / 1000).strftime("%Y-%m-%d")
daily.setdefault(d, []).append(float(r["fundingRate"]))
t = rows[-1]["time"] + 1
if len(rows) < 500:
break
time.sleep(0.6)
return {d: 100 * 24 * 365 * sum(v) / len(v) for d, v in daily.items()}
def refresh(full=False):
os.makedirs(STORE, exist_ok=True)
pxp, fdp = os.path.join(STORE, "px.json"), os.path.join(STORE, "fund.json")
px = json.load(open(pxp)) if os.path.exists(pxp) else {}
fd = json.load(open(fdp)) if os.path.exists(fdp) else {}
dead = []
for b, syms in BASKETS.items():
for s in syms:
new = okx_closed(s + "-USDT", 2500 if (full or s not in px) else 60)
if new:
px.setdefault(s, {}).update(new)
elif s in ("BTC", "ETH", "SOL"):
dead.append("price " + s)
json.dump(px, open(pxp + ".tmp", "w"))
os.replace(pxp + ".tmp", pxp)
for coins in FUND.values():
for c in coins:
if full or c not in fd:
start = int(dt.datetime(2024, 1, 1).timestamp()) * 1000
else:
start = int(time.time() - 12 * 86400) * 1000
new = hl_funding(c, start)
if new:
fd.setdefault(c, {}).update(new)
elif c == "BTC":
dead.append("funding BTC")
json.dump(fd, open(fdp + ".tmp", "w"))
os.replace(fdp + ".tmp", fdp)
return px, fd, dead
# ------------------------------------------------------------------ model --
def _add(day, n):
return (dt.date.fromisoformat(day) + dt.timedelta(days=n)).isoformat()
class Model:
def __init__(self, px, fd):
self.px, self.fd = px, fd
self.mem = {b: [s for s in syms if s in px and len(px[s]) >= 120] for b, syms in BASKETS.items()}
days = sorted(set().union(*[set(v) for v in px.values()]))
last = max(px["BTC"]) # never run ahead of BTC's last closed day
self.days = [d for d in days if "2019-12-01" <= d <= last]
self.N = len(self.days)
self.pos = {d: i for i, d in enumerate(self.days)}
self.first = {s: min(v) for s, v in px.items()}
self.idx = {b: self._index(self.mem[b]) for b in PRICE_BASKETS}
self.alts = sorted({s for b in ALT_BASKETS for s in self.mem[b]})
self.c90 = [self._share(i, 90, self.alts) for i in range(self.N)]
self.c30 = [self._share(i, 30, self.alts) for i in range(self.N)]
self.b90 = {b: [self._share(i, 90, self.mem[b]) for i in range(self.N)] for b in ALT_BASKETS if len(self.mem[b]) > 1}
self.b30 = {b: [self._share(i, 30, self.mem[b]) for i in range(self.N)] for b in ALT_BASKETS if len(self.mem[b]) > 1}
# THE RISK SWITCH HAS A 1% BAND. Without it BTC crossed its average
# average six times in Jul-Aug 2026 and would have sent six buy/sell
# alerts in a month. Four bands were tested (0, 1, 2, 3%); 1% was best
# in BOTH test periods and halved the switching.
self._risk, s = [], False
for i in range(self.N):
m, x = self.ma("BTC", i, RISK_MA), self.idx["BTC"][i]
if m and x:
if x > m * (1 + RISK_BAND):
s = True
elif x < m * (1 - RISK_BAND):
s = False
self._risk.append(s)
def p(self, s, i):
return self.px[s].get(self.days[i]) if 0 <= i < self.N else None
def _index(self, members):
out, lvl = [None] * self.N, None
need = 1 if len(members) == 1 else INDEX_NEED
for i in range(1, self.N):
d, q = self.days[i], self.days[i - 1]
r = []
for s in members:
a, b = self.px[s].get(q), self.px[s].get(d)
if a and b and d >= _add(self.first[s], 30):
r.append(max(-0.5, min(0.5, math.log(b / a))))
if len(r) >= need:
lvl = (lvl or 100.0) * math.exp(sum(r) / len(r))
out[i] = lvl
return out
def _share(self, i, n, members):
if i < n:
return None
b0, b1 = self.p("BTC", i - n), self.p("BTC", i)
if not b0 or not b1:
return None
rb, v = b1 / b0 - 1, []
for s in members:
a0, a1 = self.p(s, i - n), self.p(s, i)
if a0 and a1 and self.days[i - n] >= _add(self.first[s], 30):
v.append(a1 / a0 - 1 > rb)
return sum(v) / len(v) if len(v) >= (10 if len(members) > 15 else 3) else None
def ret(self, b, i, n):
x = self.idx[b]
return None if i - n < 0 or i >= self.N or not x[i] or not x[i - n] else x[i] / x[i - n] - 1
def fwd(self, b, i, n):
return self.ret(b, i + n, n) if i + n < self.N else None
def ma(self, b, i, n=50):
x = self.idx[b]
w = [v for v in x[max(0, i - n + 1):i + 1] if v]
return sum(w) / len(w) if len(w) >= int(0.8 * n) else None
def risk_on(self, i):
return self._risk[i]
# --- the validated signals ---------------------------------------------
def washed_recent(self, i, look=60):
return any(c is not None and c <= CLOCK_WASH for c in self.c90[max(0, i - look):i + 1])
def alt_bottom(self, i):
c = self.c30[i]
return self.washed_recent(i) and c is not None and c >= 0.40 and self.risk_on(i)
def alt_top_warning(self, i):
c = self.c30[i]
return any(x is not None and x >= CLOCK_TOP for x in self.c90[max(0, i - 30):i + 1]) and c is not None and c < 0.50
def basket_extended(self, b, i):
return b in self.b90 and any(x is not None and x >= CLOCK_TOP for x in self.b90[b][max(0, i - 30):i + 1])
def basket_top(self, b, i):
return self.basket_extended(b, i) and self.b30[b][i] is not None and self.b30[b][i] < 0.50
def funding(self, b, i, win=7):
v = []
for c in FUND[b]:
w = [self.fd.get(c, {}).get(self.days[k]) for k in range(max(0, i - win + 1), i + 1)]
w = [x for x in w if x is not None]
if w:
v.append(sum(w) / len(w))
return sum(v) / len(v) if v else None
def funding_pct(self, b, i):
"""Where today's crowding sits in this basket's own history since 2024."""
now = self.funding(b, i)
if now is None:
return None, None
hist = [self.funding(b, k) for k in range(max(0, i - 540), i, 3)]
hist = [h for h in hist if h is not None]
if len(hist) < 30:
return now, None
return now, sum(h < now for h in hist) / len(hist)
# --- the book --------------------------------------------------------------
def book_rule(self):
s = {"on": False, "dropped": set()}
def rule(i):
up = self.risk_on(i)
if s["on"] and not up:
s["on"] = False
elif not s["on"] and self.alt_bottom(i):
s["on"], s["dropped"] = True, set()
if not s["on"]:
return {"BTC": 1.0} if up else {}
for b in ALT_BASKETS:
if self.basket_top(b, i):
s["dropped"].add(b)
have = [b for b in ALT_BASKETS if self.idx[b][i]]
live = [b for b in have if b not in s["dropped"]]
w = {b: round(1 / len(have), 6) for b in live}
if len(live) < len(have):
w["BTC"] = round(1 - sum(w.values()), 6)
return w
return rule, s
def backtest(self, rule, lo, hi):
eq, w, pend, curve, log = 1.0, {}, [], [], []
for i in range(lo, hi + 1):
if w:
g = sum(wt * ((self.idx[b][i] / self.idx[b][i - 1]) if self.idx[b][i] and self.idx[b][i - 1] else 1.0)
for b, wt in w.items())
eq *= g + 1.0 - sum(w.values())
while pend and pend[0][0] <= i:
_, new = pend.pop(0)
moved = sum(abs(new.get(b, 0) - w.get(b, 0)) for b in set(new) | set(w))
if moved > 1e-9:
eq *= 1 - FEE * moved
log.append((self.days[i], new))
w = new
new = rule(i)
if new != (pend[-1][1] if pend else w):
pend.append((i + 1, new))
curve.append((self.days[i], eq, dict(w)))
return curve, log
def stats(curve):
e = [c[1] for c in curve]
if len(e) < 30:
return {}
yrs = len(e) / 365
peak, mdd = e[0], 0.0
for v in e:
peak = max(peak, v)
mdd = min(mdd, v / peak - 1)
return {"total": round(100 * (e[-1] / e[0] - 1), 1),
"cagr": round(100 * ((e[-1] / e[0]) ** (1 / yrs) - 1), 1),
"maxdd": round(100 * mdd, 1)}
# ------------------------------------------------------------ scoreboard --
def alts_index(M):
"""Equal weight of the six alt baskets, daily rebalanced (two or more live)."""
out, lvl = [None] * M.N, None
for i in range(1, M.N):
rs = [math.log(M.idx[b][i] / M.idx[b][i - 1]) for b in ALT_BASKETS if M.idx[b][i] and M.idx[b][i - 1]]
if len(rs) >= 2:
lvl = (lvl or 100.0) * math.exp(sum(rs) / len(rs))
out[i] = lvl
return out
def major_rotations(M, lo, H=45, min_run=21, major=0.20):
"""HINDSIGHT LABELS, used only to grade the system, never inside a rule.
Each day gets the best of CASH / BTC / ALTS over the next H days; runs
shorter than min_run are merged into the longer neighbour. A run is a
MAJOR rotation when its bucket beat the next best by 20%+ over the run."""
A = alts_index(M)
idx = {"BTC": M.idx["BTC"], "ALTS": A}
def lr(k, a, b):
return 0.0 if k == "CASH" else (math.log(idx[k][b] / idx[k][a]) if idx[k][a] and idx[k][b] else None)
lab = [None] * M.N
for i in range(lo, M.N - H):
c = {k: lr(k, i, i + H) for k in ("CASH", "BTC", "ALTS")}
c = {k: v for k, v in c.items() if v is not None}
lab[i] = max(c, key=c.get)
runs, s0 = [], lo
for i in range(lo + 1, M.N - H + 1):
if i == M.N - H or lab[i] != lab[s0]:
runs.append([s0, i - 1, lab[s0]])
s0 = i
changed = True
while changed and len(runs) > 1:
changed = False
for k, r in enumerate(runs):
if r[1] - r[0] + 1 < min_run:
if 0 < k < len(runs) - 1:
j = k - 1 if runs[k - 1][1] - runs[k - 1][0] >= runs[k + 1][1] - runs[k + 1][0] else k + 1
else:
j = k - 1 if k else k + 1
a, b = sorted((k, j))
runs[a] = [runs[a][0], runs[b][1], runs[j][2]]
del runs[b]
changed = True
break
k = 0
while k + 1 < len(runs):
if runs[k][2] == runs[k + 1][2]:
runs[k][1] = runs[k + 1][1]
del runs[k + 1]
changed = True
else:
k += 1
out = []
for a, b, l in runs:
z = min(b + 1, M.N - 1)
r = {k: lr(k, a, z) for k in ("CASH", "BTC", "ALTS")}
if None in r.values():
continue
edge = r[l] - max(v for k, v in r.items() if k != l)
out.append({"a": a, "b": b, "bucket": l, "edge": math.exp(edge) - 1, "r": r})
return [x for x in out if x["edge"] >= major]
# ------------------------------------------------------------- the reading --
def holding_word(w):
if not w:
return "CASH"
alts = [b for b in w if b != "BTC"]
if not alts:
return "BTC"
return "ALTS"
def read(M):
i = M.N - 1
today = M.days[i]
btc, m50 = M.idx["BTC"][i], M.ma("BTC", i, RISK_MA)
up = M.risk_on(i)
cross = next((k for k in range(i, 0, -1) if M.risk_on(k) != M.risk_on(k - 1)), 0)
rule, st_ = M.book_rule()
lo = M.pos.get("2020-06-01", 200)
curve, log = M.backtest(rule, lo, i)
held = curve[-1][2]
want = log[-1][1] if log else {}
# what the rule wants as of today's close (it is traded at tomorrow's close)
rule2, _ = M.book_rule()
for k in range(lo, i + 1):
want = rule2(k)
# ---- baskets
cards = []
for b in PRICE_BASKETS:
c = {"key": b, "name": LABEL[b], "members": M.mem[b]}
r30, rb30 = M.ret(b, i, 30), M.ret("BTC", i, 30)
c["vs_btc_30d"] = round(100 * (r30 - rb30), 1) if (r30 is not None and rb30 is not None and b != "BTC") else None
c["ret_30d"] = round(100 * r30, 1) if r30 is not None else None
m = M.ma(b, i)
c["trend"] = "up" if (m and M.idx[b][i] > m) else "down"
fnow, fpct = M.funding_pct(b, i)
c["funding"] = round(fnow, 1) if fnow is not None else None
c["funding_pct"] = round(100 * fpct) if fpct is not None else None
c["clock90"] = round(100 * M.b90[b][i]) if b in M.b90 and M.b90[b][i] is not None else None
c["clock30"] = round(100 * M.b30[b][i]) if b in M.b30 and M.b30[b][i] is not None else None
c["weight"] = round(100 * want.get(b, 0))
checks = []
if b == "BTC":
dist = 100 * (btc / m50 - 1) if m50 else None
c["shade"] = "green" if up else "red"
c["word"] = "RISK ON" if up else "RISK OFF"
c["why"] = ("BTC is %.1f%% %s its 100 day average. Risk switched %s on %s. It flips on at +1%% "
"and off at -1%%, so a wobble on the line cannot flip it." %
(abs(dist or 0), "above" if (dist or 0) >= 0 else "below", "on" if up else "off", M.days[cross]))
checks = [["BTC over its 100 day average, with the 1% band", up]]
elif b in M.b90:
ext, top = M.basket_extended(b, i), M.basket_top(b, i)
if top:
c["shade"], c["word"] = "red", "TOP CONFIRMED"
c["why"] = ("Most of the %s have beaten BTC for three months and have now stopped. "
"This lagged BTC over the next 60 days {EV_TOP}." % LABEL[b].lower())
elif ext:
c["shade"], c["word"] = "lightred", "EXTENDED"
c["why"] = "Most members have beaten BTC for three months. Not rolled over yet, so no exit, just no new money."
elif M.alt_bottom(i):
c["shade"], c["word"] = "green", "BOTTOM CONFIRMED"
c["why"] = "The whole alt market washed out and has turned, and BTC risk is on. This basket is part of the next rotation."
elif M.washed_recent(i):
c["shade"], c["word"] = "lightgreen", "WASHED OUT"
c["why"] = "The alt market washed out. Waiting for alts to start beating BTC again before this turns green."
else:
c["shade"], c["word"] = "grey", ("LEADING BTC" if (c["vs_btc_30d"] or 0) > 0 else "LAGGING BTC")
c["why"] = "Nothing has crossed. Shown for context, not a signal."
checks = [["extended: 70%+ of members beat BTC over 90 days", ext],
["rolled over: under 50% beat BTC over the last 30 days", bool(c["clock30"] is not None and c["clock30"] < 50)]]
else:
# ETH and SOL are single coins, so there is no basket clock to read.
if M.alt_bottom(i):
c["shade"], c["word"] = "green", "BOTTOM CONFIRMED"
c["why"] = "Part of the alt rotation the market clock just confirmed."
elif M.washed_recent(i):
c["shade"], c["word"] = "lightgreen", "WASHED OUT"
c["why"] = "The alt market washed out. Waiting for the turn."
else:
c["shade"], c["word"] = "grey", ("LEADING BTC" if (c["vs_btc_30d"] or 0) > 0 else "LAGGING BTC")
c["why"] = "Single coin, no basket clock. Shown for context."
checks = [["the alt market washed out in the last 60 days", M.washed_recent(i)],
["alts are beating BTC again (30 day share 40%+)", bool(M.c30[i] is not None and M.c30[i] >= 0.40)]]
c["checks"] = checks
x = M.idx[b]
c["spark"] = [round(x[k] / x[max(0, i - 180)], 4) if x[k] and x[max(0, i - 180)] else None
for k in range(max(0, i - 180), i + 1, 3)]
cards.append(c)
# ---- the one call
now_word, want_word = holding_word(held), holding_word(want)
dropped = sorted(st_["dropped"]) if st_["on"] else []
if not up:
call = ("GO TO CASH" if now_word != "CASH" else "STAY IN CASH",
"BTC is below its 100 day average. Risk is off, so nothing is held. {EV_RISK}")
elif want_word == "ALTS" and now_word != "ALTS":
call = ("ROTATE INTO ALTS",
"The alt market washed out and has turned, with BTC risk on. Split across the alt baskets that are not red.")
elif want_word == "ALTS":
red = [c["name"] for c in cards if c["shade"] == "red" and c["key"] != "BTC"]
call = ("HOLD ALTS" + (", TRIM " + " AND ".join(n.upper() for n in red) if red else ""),
"Inside an alt rotation." + (" Red baskets move to BTC." if red else " No basket has topped yet."))
else:
call = ("HOLD BTC" if now_word == "BTC" else "BUY BTC",
"BTC risk is on. Alts have not washed out and turned, so they are not the trade yet.")
# ---- trench card (from Rotation Watch's collector)
trench = None
try:
import sqlite3
c = sqlite3.connect(TRENCH_DB)
rows = list(c.execute("select date, btc, dex_all, pad from d where btc is not null order by date"))
if len(rows) > 40:
dd = [r[0] for r in rows]
b = [r[1] for r in rows]
sh = [(100 * r[3] / r[2]) if r[2] and r[3] else None for r in rows]
n = len(rows) - 1
def sm(x, k, w):
v = [y for y in x[max(0, k - w + 1):k + 1] if y is not None]
return sum(v) / len(v) if v else None
r7 = [(b[k] / b[k - 7] - 1) if k >= 7 and b[k] and b[k - 7] else None for k in range(len(b))]
imp = [k for k in range(8, len(b)) if r7[k] is not None and r7[k - 1] is not None and r7[k] >= 0.10 > r7[k - 1]]
last_cross = dd[imp[-1]] if imp else None
open_ = bool(imp and n - imp[-1] <= 14)
trench = {"share_7d": round(sm(sh, n - 1, 7) or 0, 2), "share_90d_avg": round(sm(sh, n - 1, 90) or 0, 2),
"btc_7d": round(100 * (r7[n] or 0), 1), "window_open": open_, "last_impulse": last_cross,
"window_closes": _add(last_cross, 14) if open_ else None}
except Exception as e:
trench = {"error": str(e)[:80]}
# ---- evidence, recomputed every run
def events(pred, lo_, hi_, gap=45):
out = []
for k in range(lo_ + 1, hi_):
if pred(k) and not pred(k - 1) and (not out or k - out[-1] > gap):
out.append(k)
return out
ev = {}
for per, a, z in (("holdout", "2020-06-01", "2024-06-30"), ("recent", "2024-07-01", today)):
pa, pz = M.pos.get(a, lo), M.pos.get(z, i)
n_ok = n_all = 0
for b in M.b90:
for k in events(lambda k: M.basket_top(b, k), pa, min(pz, i - 60)):
f, g = M.fwd(b, k, 60), M.fwd("BTC", k, 60)
if f is not None and g is not None:
n_all += 1
n_ok += f < g
ev["basket_top_" + per] = [n_ok, n_all]
books = {}
for per, a, z in (("holdout", "2020-06-01", "2024-06-30"), ("recent", "2024-07-01", today), ("full", "2020-06-01", today)):
pa, pz = M.pos.get(a, lo), M.pos.get(z, i)
r1, _ = M.book_rule()
books[per] = {"book": stats(M.backtest(r1, pa, pz)[0]),
"btc_trend": stats(M.backtest(lambda k: {"BTC": 1.0} if M.risk_on(k) else {}, pa, pz)[0]),
"hold_btc": stats(M.backtest(lambda k: {"BTC": 1.0}, pa, pz)[0])}
# ---- the rotation timeline, every leg the book took
legs, prev = [], None
eqd = {d: e for d, e, _ in curve}
for d, w in log:
wd = holding_word(w)
if prev and prev[1] != wd:
legs.append({"from": prev[0], "to": d, "held": prev[1],
"ret": round(100 * (eqd[d] / eqd[prev[0]] - 1), 1)})
prev = (d, wd)
elif not prev:
prev = (d, wd)
if prev:
legs.append({"from": prev[0], "to": None, "held": prev[1], "ret": round(100 * (curve[-1][1] / eqd[prev[0]] - 1), 1)})
# ---- scoreboard: every major rotation since 2020 and whether the book was in it
def bucket(w): # half or more of the book in alts counts as holding ALTS
return "CASH" if not w else ("ALTS" if sum(v for k, v in w.items() if k != "BTC") >= 0.5 else "BTC")
heldw = {M.pos[d]: bucket(w) for d, _, w in curve}
rots = []
for r in major_rotations(M, lo):
span = range(r["a"], r["b"] + 1)
inb = [k for k in span if heldw.get(k) == r["bucket"]]
lag = (inb[0] - r["a"]) if inb else None
share = len(inb) / len(span)
rots.append({"from": M.days[r["a"]], "to": M.days[r["b"]], "bucket": r["bucket"],
"edge": round(100 * r["edge"]), "btc": round(100 * (math.exp(r["r"]["BTC"]) - 1)),
"alts": round(100 * (math.exp(r["r"]["ALTS"]) - 1)),
"lag": lag, "held": round(100 * share), "caught": share >= 0.5})
lags = sorted(x["lag"] for x in rots if x["lag"] is not None)
score = {"major": len(rots), "caught": sum(x["caught"] for x in rots),
"median_lag": lags[len(lags) // 2] if lags else None}
# ---- fill the evidence into the words
t = ev.get("basket_top_holdout", [0, 0])
u = ev.get("basket_top_recent", [0, 0])
ev_top = "%d of %d times in 2020 to 2024 and %d of %d since" % (t[0], t[1], u[0], u[1])
hb, bt = books["holdout"]["hold_btc"], books["holdout"]["btc_trend"]
ev_risk = "On 2020 to 2024 this rule alone cut the worst drop of holding BTC from %d%% to %d%%." % (hb["maxdd"], bt["maxdd"])
for c in cards:
c["why"] = c["why"].replace("{EV_TOP}", ev_top)
call = (call[0], call[1].replace("{EV_RISK}", ev_risk))
# ---- weekly series for the charts
wk = list(range(lo, i + 1, 7)) + ([i] if (i - lo) % 7 else [])
btc_hold = M.idx["BTC"]
series = {"d": [M.days[k] for k in wk],
"clock90": [round(100 * M.c90[k]) if M.c90[k] is not None else None for k in wk],
"clock30": [round(100 * M.c30[k]) if M.c30[k] is not None else None for k in wk],
"risk": [1 if M.risk_on(k) else 0 for k in wk],
"book": [round(curve[k - lo][1], 4) for k in wk],
"btc": [round(btc_hold[k] / btc_hold[lo], 4) for k in wk],
"held": [holding_word(curve[k - lo][2]) for k in wk]}
return {"as_of": today, "call": {"word": call[0], "line": call[1]},
"held_now": {b: round(100 * v) for b, v in held.items()},
"want": {b: round(100 * v) for b, v in want.items()},
"dropped": dropped,
"risk": {"on": up, "btc": round(btc), "ma": round(m50) if m50 else None, "ma_days": RISK_MA,
"dist": round(100 * (btc / m50 - 1), 1) if m50 else None, "since": M.days[cross]},
"clock": {"c90": round(100 * M.c90[i]) if M.c90[i] is not None else None,
"c30": round(100 * M.c30[i]) if M.c30[i] is not None else None,
"washed_recent": M.washed_recent(i), "bottom": M.alt_bottom(i),
"top_warning": M.alt_top_warning(i), "n_alts": len(M.alts)},
"baskets": cards, "trench": trench, "evidence": ev, "books": books,
"legs": legs[-40:], "series": series, "rotations": rots, "score": score,
"settings": {"risk_ma": RISK_MA, "risk_band": RISK_BAND, "clock_top": CLOCK_TOP, "clock_wash": CLOCK_WASH}}
# ----------------------------------------------------------------- alerts --
def tg(msg):
try:
cfg = json.load(open(CFG))["telegram"]
body = json.dumps({"chat_id": cfg["chat_id"], "text": msg, "parse_mode": "HTML",
"disable_web_page_preview": True}).encode()
urllib.request.urlopen(urllib.request.Request(
"https://api.telegram.org/bot%s/sendMessage" % cfg["bot_token"], body,
{"content-type": "application/json"}), timeout=25).read()
return True
except Exception as e:
print("telegram failed:", str(e)[:80])
return False
def alerts(R, dry=False):
"""One message per real change: the call word, a basket's colour, the risk
switch, the trench window. State lives in a file so a restart cannot
replay old alerts."""
sp = os.path.join(HERE, "rotmap_state.json")
old = json.load(open(sp)) if os.path.exists(sp) else None
new = {"call": R["call"]["word"], "risk": R["risk"]["on"],
"shade": {c["key"]: c["shade"] for c in R["baskets"]},
"trench": bool((R.get("trench") or {}).get("window_open"))}
msgs = []
if old is not None:
if new["call"] != old.get("call"):
icon = {"ROTATE INTO ALTS": "\U0001F7E2", "GO TO CASH": "\U0001F534"}.get(new["call"], "\U0001F504")
msgs.append("%s %s\n%s" % (icon, new["call"], R["call"]["line"]))
for c in R["baskets"]:
a, b = old.get("shade", {}).get(c["key"]), c["shade"]
if a and a != b and b in ("red", "green"):
msgs.append("%s %s: %s\n%s" % ("\U0001F7E5" if b == "red" else "\U0001F7E9",
c["name"], c["word"], c["why"]))
if new["trench"] and not old.get("trench"):
t = R["trench"]
msgs.append("\U0001F525 TRENCH WINDOW OPEN\nBTC just did +10%% in a week. The last 14 times, launchpad volume "
"was higher two weeks later in 11. Window closes %s." % t.get("window_closes"))
if msgs:
body = "\n\n".join(msgs) + "\n\n" + R["as_of"] + " close ยท http://159.223.88.168/rotation.html"
if dry:
print("DRY ALERT:\n" + body)
elif tg(body):
print("alert sent")
if not dry:
json.dump(new, open(sp, "w"))
# ------------------------------------------------------------------- main --
def main():
full = "--full" in sys.argv
dry = "--dry" in sys.argv
if "--offline" in sys.argv:
px = json.load(open(os.path.join(STORE, "px.json")))
fd = json.load(open(os.path.join(STORE, "fund.json")))
dead = []
else:
px, fd, dead = refresh(full)
M = Model(px, fd)
R = read(M)
R["dead"] = dead
R["generated"] = int(time.time())
R["generated_bkk"] = dt.datetime.utcfromtimestamp(R["generated"] + BKK).strftime("%Y-%m-%d %H:%M")
tmp = PUB_JSON + ".tmp"
json.dump(R, open(tmp, "w"), separators=(",", ":"))
os.replace(tmp, PUB_JSON)
print("%s %s risk %s clock90 %s clock30 %s %s" % (
R["generated_bkk"], R["call"]["word"], "on" if R["risk"]["on"] else "off",
R["clock"]["c90"], R["clock"]["c30"],
" ".join("%s:%s" % (c["key"], c["shade"]) for c in R["baskets"])))
if dead:
print("DEAD SOURCES:", dead)
alerts(R, dry)
if __name__ == "__main__":
main()