diff --git a/generate_chart.py b/generate_chart.py new file mode 100644 index 0000000..1848c67 --- /dev/null +++ b/generate_chart.py @@ -0,0 +1,359 @@ +#!/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 + + + +
+

LINK/USD — HMM Regime Detection (12h)

+
+ Low Spread + Mid Spread + High Spread + ▽ transition +
+
+ +
+
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) diff --git a/index.html b/index.html index 8e4f4b8..7f99e46 100644 --- a/index.html +++ b/index.html @@ -1,6 +1,15 @@ + + + + LOB Latent Regimes — Early Detection of Market Micro-Instability