LINK HMM Regime — Canvas Chart

#!/usr/bin/env python3 """ Generate HMM regime chart as single-file HTML with pure Canvas rendering. No external dependencies — works everywhere. """ import json, sys, os, glob import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler from hmmlearn.hmm import GaussianHMM import warnings warnings.filterwarnings("ignore") SEED = 42 N_REGIMES = 3 STEP = 5 def load_snapshots(filepath): snaps = [] with open(filepath) as f: for line in f: line = line.strip() if line: snaps.append(json.loads(line)) return snaps def extract_features(snapshots): records = [] for s in snapshots: ts = s["_collect_ts"] bids = sorted(s["bids"], key=lambda x: x[0], reverse=True) asks = sorted(s["asks"], key=lambda x: x[0]) bb = bids[0][0] if bids else 0 ba = asks[0][0] if asks else 0 mid = (bb + ba) / 2 if bb and ba else 0 spread = ba - bb if bb and ba else 0 bd = sum(b[1] for b in bids[:10]) ad = sum(a[1] for a in asks[:10]) depth = bd + ad bv = sum(b[0] * b[1] for b in bids[:10]) av = sum(a[0] * a[1] for a in asks[:10]) imbalance = (bv - av) / (bv + av + 1e-12) records.append({"ts": ts, "mid": mid, "spread": spread, "depth": depth, "imbalance": imbalance}) return pd.DataFrame(records) def engineer_features(df, win=20): s, d, imb = df["spread"].values, df["depth"].values, df["imbalance"].values rv = pd.Series(df["mid"]).pct_change().rolling(win, min_periods=1).std().fillna(0).values ofi = pd.Series(df["mid"]).diff().rolling(win, min_periods=1).mean().fillna(0).values return np.column_stack([s, d, imb, rv, ofi]) def fit_hmm(X): best_score, best_model = -np.inf, None for k in range(12): m = GaussianHMM(n_components=N_REGIMES, covariance_type="full", n_iter=400, tol=1e-7, random_state=SEED + k, init_params="stmc", params="stmc") try: m.fit(X) sc = m.score(X) if sc > best_score: best_score, best_model = sc, m except Exception: continue if best_model is None: raise RuntimeError("HMM fitting failed.") return best_model HTML = """ LINK HMM Regime — Canvas Chart

Loading data...

""" def main(datafile): print(f"[1/3] Loading {datafile}...") snaps = load_snapshots(datafile) df = extract_features(snaps) print(f" {len(snaps)} snapshots — price range: {df.mid.min():.4f} ~ {df.mid.max():.4f}") print(f"[2/3] Fitting HMM...") X_raw = engineer_features(df) X_scaled = StandardScaler().fit_transform(X_raw) model = fit_hmm(X_scaled[::STEP]) Z = model.predict(X_scaled) for k in range(N_REGIMES): mask = Z == k print(f" Regime {k}: {mask.mean()*100:.1f}% spread={X_raw[mask,0].mean():.6f}") pts = [{"t": int(df.iloc[i]["ts"]), "v": round(df.iloc[i]["mid"], 6), "r": int(Z[i])} for i in range(len(df))] base = os.path.splitext(datafile)[0] json_file = base + "_data.json" html_file = base + "_canvas_chart.html" with open(json_file, "w") as f: json.dump(pts, f, separators=(',', ':')) html = HTML.replace("DATA_URL_PLACEHOLDER", os.path.basename(json_file)) with open(html_file, "w") as f: f.write(html) print(f"[3/3] Done: {html_file} ({os.path.getsize(html_file)/1024:.0f} KB)") print(f" {json_file} ({os.path.getsize(json_file)/1024:.0f} KB)") if __name__ == "__main__": if len(sys.argv) < 2: files = sorted(glob.glob("data/l2_LINK_*.jsonl")) if not files: print("No data files found in data/") sys.exit(1) datafile = files[-1] else: datafile = sys.argv[1] main(datafile)

LINK/USD — HMM Regime Detection (12h)