# btc options signal analyser # # reads live IV data from deribit, reconstructs the risk-neutral pdf using # breeden-litzenberger, and outputs directional signals based on skew/tail/peak/band. # # mode 1: live fetch for a given expiry, plots pdf + signals # mode 2: synthetic backtest over randomly generated IV surfaces import math import re import threading from datetime import datetime import matplotlib matplotlib.use("TkAgg") import matplotlib.pyplot as plt from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg import numpy as np import requests import tkinter as tk from tkinter import ttk, scrolledtext from scipy.integrate import simpson from scipy.interpolate import UnivariateSpline from scipy.ndimage import gaussian_filter1d from scipy.stats import norm # --- config --- DENSE_POINTS = 100 SMOOTHING_IV = 1.0 MIN_POINTS = 4 TAIL_THRESHOLD = 0.02 # 2% cutoff for tail mass BAND_WIDTH = 0.03 # +-3% band for concentration signal SIGNAL_DELTA = 0.03 # min abs difference to fire a signal BAND_DELTA_MULT = 8 BAND_SIGNAL_DELTA = min(0.49, SIGNAL_DELTA * BAND_DELTA_MULT) RISK_FREE_R = 0.0 # synthetic backtest params SYNTH_SEED = 56 DEFAULT_N_CASES = 500 MIN_MONEYN_LOW = 0.60 MAX_MONEYN_HIGH = 1.50 MIN_STRIKES = 22 MAX_STRIKES = 48 ATM_VOL_RANGE = (0.30, 0.85) SKEW_RANGE = (-0.25, 0.20) CURV_RANGE = (0.05, 0.60) MAX_ADJ_DELTA = 0.015 CORR_NOISE_AMP = 0.015 CORR_NOISE_SIGMA = 3.0 VOLFLOOR = 0.05 VOLCAP = 2.00 SPOT_LOG_MEAN = math.log(40_000) SPOT_LOG_SD = 0.45 HORIZON_DAYS = 1 LABEL_EPS = 0.002 # catppuccin mocha C_BG = "#1e1e2e" C_SURFACE = "#181825" C_OVERLAY = "#313244" C_MUTED = "#585b70" C_TEXT = "#cdd6f4" C_SUBTEXT = "#a6adc8" C_BLUE = "#89b4fa" C_GREEN = "#a6e3a1" C_RED = "#f38ba8" C_YELLOW = "#f9e2af" C_MAUVE = "#cba6f7" # --- black-scholes + breeden-litzenberger --- def bs_call_price(S: float, K: float, T: float, r: float, sigma: float) -> float: if sigma <= 0 or T <= 0: return max(S - K, 0.0) d1 = (np.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * np.sqrt(T)) d2 = d1 - sigma * np.sqrt(T) return float(S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)) def build_pdf( spot: float, strike_arr: np.ndarray, iv_arr: np.ndarray, T: float, r: float = RISK_FREE_R, ) -> tuple[np.ndarray, np.ndarray, np.ndarray]: # breeden-litzenberger: spline the IV surface, reprice calls, take second derivative # returns (strike grid, normalised pdf, gaussian-smoothed pdf) sort_idx = np.argsort(strike_arr) strike_arr, iv_arr = strike_arr[sort_idx], iv_arr[sort_idx] iv_spline = UnivariateSpline(strike_arr, iv_arr, s=SMOOTHING_IV) K_dense = np.linspace(strike_arr.min(), strike_arr.max(), DENSE_POINTS) iv_dense = np.clip(iv_spline(K_dense), 1e-4, None) prices = np.array([bs_call_price(spot, K, T, r, iv) for K, iv in zip(K_dense, iv_dense)]) price_spline = UnivariateSpline(K_dense, prices, s=0) pdf_raw = price_spline.derivative(2)(K_dense) * np.exp(r * T) area = simpson(pdf_raw, x=K_dense) pdf_norm = pdf_raw / (area if area > 0 else 1.0) pdf_smooth = gaussian_filter1d(pdf_norm, sigma=2) return K_dense, pdf_norm, pdf_smooth # --- signal engine --- def compute_signals( spot: float, K_dense: np.ndarray, pdf_norm: np.ndarray, pdf_smooth: np.ndarray, tail_threshold: float = TAIL_THRESHOLD, band_width: float = BAND_WIDTH, signal_delta: float = SIGNAL_DELTA, ) -> dict: # four signals from the pdf: # skew - which side carries more mass? # tails - are far-otm tails lopsided? # peak - where is the mode relative to spot? # band - is most mass outside +-band_width (big move priced in)? # verdict fires only if all active signals point the same way band_signal_delta = min(0.49, signal_delta * BAND_DELTA_MULT) pos = np.clip(pdf_norm, 0, None) # skew up, dn = K_dense > spot, K_dense < spot p_up = simpson(pos[up], x=K_dense[up]) if up.any() else 0.0 p_dn = simpson(pos[dn], x=K_dense[dn]) if dn.any() else 0.0 skew_diff = abs(p_up - p_dn) skew_sig = ("LONG" if p_up > p_dn else "SHORT") if skew_diff >= signal_delta else "NO TRADE" # tails cut_up, cut_dn = spot * (1 + tail_threshold), spot * (1 - tail_threshold) m_tu, m_td = K_dense > cut_up, K_dense < cut_dn p_tail_up = simpson(pos[m_tu], x=K_dense[m_tu]) if m_tu.any() else 0.0 p_tail_dn = simpson(pos[m_td], x=K_dense[m_td]) if m_td.any() else 0.0 tail_diff = abs(p_tail_up - p_tail_dn) tails_sig = ("LONG" if p_tail_up > p_tail_dn else "SHORT") if tail_diff >= signal_delta else "NO TRADE" # peak shift peak_strike = K_dense[np.argmax(pdf_smooth)] if peak_strike > spot * 1.005: peak_sig = "LONG" elif peak_strike < spot * 0.995: peak_sig = "SHORT" else: peak_sig = "NO TRADE" # band concentration band_lo, band_hi = spot * (1 - band_width), spot * (1 + band_width) m_band = (K_dense >= band_lo) & (K_dense <= band_hi) p_band = simpson(pos[m_band], x=K_dense[m_band]) if m_band.any() else 0.0 p_outside = 1.0 - p_band if (p_outside > 0.5) and (abs(p_outside - p_band) >= band_signal_delta): band_sig = skew_sig else: band_sig = "NO TRADE" # unanimous → fire, otherwise sit out active = [s for s in (skew_sig, tails_sig, peak_sig, band_sig) if s != "NO TRADE"] if not active: final = "NO TRADE" elif all(s == active[0] for s in active): final = active[0] else: final = "NO TRADE" return { "skew": {"signal": skew_sig, "p_up": p_up, "p_dn": p_dn, "diff": skew_diff}, "tails": {"signal": tails_sig, "p_up": p_tail_up, "p_dn": p_tail_dn, "diff": tail_diff}, "peak": {"signal": peak_sig, "peak_strike": peak_strike}, "band": {"signal": band_sig, "p_band": p_band, "p_outside": p_outside}, "final": final, } # --- deribit data fetcher --- def fetch_deribit(expiry: str) -> tuple[float, np.ndarray, np.ndarray, float]: # pulls all BTC options for the given expiry (e.g. "22AUG25") from deribit # returns (spot, strikes, ivs as decimals, T in years) expiry = expiry.upper().strip() try: expiry_date = datetime.strptime(expiry, "%d%b%y").date() except ValueError: raise ValueError(f"Cannot parse expiry '{expiry}'. Expected format: 22AUG25") resp = requests.get( "https://www.deribit.com/api/v2/public/get_instruments", params={"currency": "BTC", "kind": "option", "expired": "false"}, timeout=15, ) resp.raise_for_status() pattern = re.compile(rf"-{expiry}-", re.IGNORECASE) instruments = resp.json()["result"] strikes, ivs, spot = [], [], None for opt in instruments: if not pattern.search(opt["instrument_name"]): continue ob = requests.get( "https://www.deribit.com/api/v2/public/get_order_book_by_instrument_id", params={"instrument_id": opt["instrument_id"]}, timeout=10, ).json()["result"] if spot is None: spot = ob.get("index_price") iv = ob.get("mark_iv") if iv and iv > 0: strikes.append(float(opt["strike"])) ivs.append(float(iv) / 100.0) if not strikes: raise ValueError(f"No option data found for expiry '{expiry}'. " "Check that the expiry is active on Deribit.") if len(strikes) < MIN_POINTS: raise ValueError(f"Only {len(strikes)} strikes found for '{expiry}' " f"(minimum {MIN_POINTS} required).") T = max((expiry_date - datetime.utcnow().date()).days, 0) / 365.0 return float(spot), np.array(strikes), np.array(ivs), T # --- synthetic case generator (for backtesting) --- def _gaussian_corr_noise(n: int, amp: float, sigma_pts: float, rng) -> np.ndarray: if n <= 1 or amp <= 0: return np.zeros(n) white = rng.normal(0.0, 1.0, size=n) win = max(3, int(round(6 * sigma_pts))) if win % 2 == 0: win += 1 x = np.linspace(-(win // 2), win // 2, win) kernel = np.exp(-0.5 * (x / sigma_pts) ** 2) kernel /= kernel.sum() pad = np.pad(white, (win // 2, win // 2), mode="edge") return amp * np.convolve(pad, kernel, mode="valid") def _adj_delta_cap(iv: np.ndarray, max_delta: float) -> np.ndarray: # caps jump between adjacent IV points so the surface stays smooth-ish out = iv.copy() for i in range(1, len(out)): diff = out[i] - out[i - 1] if diff > max_delta: out[i] = out[i - 1] + max_delta elif diff < -max_delta: out[i] = out[i - 1] - max_delta return out def generate_one_case(rng) -> dict: # builds a fake but plausible BTC option snapshot # label is derived from a 1-day GBM draw off the ATM vol spot = float(np.exp(rng.normal(SPOT_LOG_MEAN, SPOT_LOG_SD))) n = int(rng.integers(MIN_STRIKES, MAX_STRIKES + 1)) lo, hi = spot * MIN_MONEYN_LOW, spot * MAX_MONEYN_HIGH strikes = np.linspace(lo, hi, n) strikes += rng.normal(0.0, 0.0005 * spot, size=n) strikes = np.sort(np.clip(strikes, lo * 0.995, hi * 1.005)) atm = float(rng.uniform(*ATM_VOL_RANGE)) skew = float(rng.uniform(*SKEW_RANGE)) curv = float(rng.uniform(*CURV_RANGE)) x = np.log(strikes / spot) base = atm * (1.0 + skew * x + curv * x ** 2) noise = _gaussian_corr_noise(len(strikes), CORR_NOISE_AMP, CORR_NOISE_SIGMA, rng) ivs = np.clip(_adj_delta_cap(base + noise, MAX_ADJ_DELTA), VOLFLOOR, VOLCAP) atm_vol = float(np.interp(spot, strikes, ivs)) dt = HORIZON_DAYS / 365.0 z = float(rng.normal()) spot_future = float(spot * np.exp(-0.5 * atm_vol ** 2 * dt + atm_vol * math.sqrt(dt) * z)) ret = (spot_future / spot) - 1.0 label = "LONG" if ret > LABEL_EPS else ("SHORT" if ret < -LABEL_EPS else "NO TRADE") return { "spot": spot, "strikes": strikes, "ivs": ivs, "spot_future": spot_future, "label": label, "T": 30 / 365.0, } # --- backtest runner --- def run_backtest(n_cases: int, seed: int = SYNTH_SEED, params: dict | None = None, progress_cb=None) -> dict: # runs n_cases synthetic snapshots through the signal engine and tallies accuracy if params is None: params = {} rng = np.random.default_rng(seed) correct = wrong = no_trade = skipped = 0 by_dir = {"LONG": [0, 0], "SHORT": [0, 0]} # [correct, wrong] for i in range(n_cases): case = generate_one_case(rng) try: K_d, pdf_n, pdf_s = build_pdf(case["spot"], case["strikes"], case["ivs"], case["T"]) sigs = compute_signals(case["spot"], K_d, pdf_n, pdf_s, **params) except Exception: skipped += 1 continue pred = sigs["final"] truth = case["label"] if pred == "NO TRADE": no_trade += 1 elif pred == truth: correct += 1 by_dir[pred][0] += 1 else: wrong += 1 if pred in by_dir: by_dir[pred][1] += 1 if progress_cb and (i + 1) % max(1, n_cases // 100) == 0: progress_cb((i + 1) / n_cases) traded = correct + wrong accuracy = correct / traded if traded else 0.0 return { "n_total": n_cases, "n_correct": correct, "n_wrong": wrong, "n_no_trade": no_trade, "n_skipped": skipped, "n_traded": traded, "accuracy": accuracy, "by_dir": by_dir, } # --- gui helpers --- def _sig_color(sig: str) -> str: return {"LONG": C_GREEN, "SHORT": C_RED}.get(sig, C_YELLOW) def _style_axes(ax): ax.set_facecolor(C_SURFACE) ax.tick_params(colors=C_MUTED, labelsize=9) for spine in ax.spines.values(): spine.set_edgecolor(C_OVERLAY) ax.grid(True, color=C_OVERLAY, linewidth=0.5, linestyle="--") # --- live analysis tab --- class LiveAnalysisTab(tk.Frame): def __init__(self, parent): super().__init__(parent, bg=C_BG) self._build() def _build(self): ctrl = tk.Frame(self, bg=C_BG) ctrl.pack(fill="x", padx=16, pady=(14, 6)) tk.Label(ctrl, text="Expiry (e.g. 22AUG25):", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 10)).pack(side="left") self._expiry_var = tk.StringVar(value="22AUG25") tk.Entry(ctrl, textvariable=self._expiry_var, width=14, bg=C_OVERLAY, fg=C_TEXT, insertbackground=C_TEXT, relief="flat", font=("Segoe UI", 10)).pack(side="left", padx=8) self._btn = tk.Button(ctrl, text="Fetch & Analyse", command=self._on_fetch, bg=C_OVERLAY, fg=C_TEXT, activebackground=C_MUTED, relief="flat", font=("Segoe UI", 10, "bold"), padx=12, pady=4, cursor="hand2") self._btn.pack(side="left") self._status = tk.Label(ctrl, text="", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 9)) self._status.pack(side="left", padx=16) params = tk.Frame(self, bg=C_BG) params.pack(fill="x", padx=16, pady=(0, 6)) _PCT_OPTS = ["0.5%", "1%", "2%", "3%", "5%", "7%", "10%"] def _pct_menu(parent, var): m = tk.OptionMenu(parent, var, *_PCT_OPTS) m.config(bg=C_OVERLAY, fg=C_TEXT, activebackground=C_MUTED, activeforeground=C_TEXT, relief="flat", font=("Segoe UI", 9), highlightthickness=0, width=5) m["menu"].config(bg=C_OVERLAY, fg=C_TEXT, activebackground=C_MUTED, font=("Segoe UI", 9)) return m tk.Label(params, text="Tail cutoff:", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 9)).pack(side="left") self._tail_var = tk.StringVar(value="2%") _pct_menu(params, self._tail_var).pack(side="left", padx=(4, 12)) tk.Label(params, text="Band width +-:", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 9)).pack(side="left") self._band_var = tk.StringVar(value="3%") _pct_menu(params, self._band_var).pack(side="left", padx=(4, 12)) tk.Label(params, text="Signal gate:", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 9)).pack(side="left") self._gate_var = tk.StringVar(value="3%") _pct_menu(params, self._gate_var).pack(side="left", padx=(4, 0)) tk.Label(params, text=" Tail: min distance from spot for tail mass · " "Band: +-% range around spot · Gate: min prob diff to fire a signal", bg=C_BG, fg=C_MUTED, font=("Segoe UI", 8)).pack(side="left", padx=12) paned = tk.PanedWindow(self, orient="horizontal", bg=C_BG, sashwidth=5, relief="flat", sashpad=2) paned.pack(fill="both", expand=True, padx=8, pady=4) chart_f = tk.Frame(paned, bg=C_BG) paned.add(chart_f, minsize=520) self._fig, self._ax = plt.subplots(facecolor=C_BG) self._canvas = FigureCanvasTkAgg(self._fig, master=chart_f) self._canvas.get_tk_widget().pack(fill="both", expand=True) self._draw_empty_chart() sig_f = tk.Frame(paned, bg=C_SURFACE) paned.add(sig_f, minsize=300) self._sig_text = scrolledtext.ScrolledText( sig_f, font=("Consolas", 10), bg=C_SURFACE, fg=C_TEXT, insertbackground=C_TEXT, relief="flat", state="disabled", wrap="word", ) self._sig_text.pack(fill="both", expand=True, padx=4, pady=4) def _on_fetch(self): expiry = self._expiry_var.get().strip() if not expiry: return self._btn.config(state="disabled") self._status.config(text="Fetching…", fg=C_YELLOW) params = self._get_params() threading.Thread(target=self._worker, args=(expiry, params), daemon=True).start() def _get_params(self) -> dict: def pct(var): return float(var.get().rstrip("%")) / 100.0 return { "tail_threshold": pct(self._tail_var), "band_width": pct(self._band_var), "signal_delta": pct(self._gate_var), } def _worker(self, expiry: str, params: dict): try: spot, strikes, ivs, T = fetch_deribit(expiry) K_d, pdf_n, pdf_s = build_pdf(spot, strikes, ivs, T) sigs = compute_signals(spot, K_d, pdf_n, pdf_s, **params) self.after(0, self._on_result, expiry, spot, strikes, K_d, pdf_s, sigs, params) except Exception as exc: self.after(0, self._on_error, str(exc)) def _on_result(self, expiry, spot, strikes, K_d, pdf_s, sigs, params): self._update_chart(expiry, spot, K_d, pdf_s, sigs) self._update_signals(expiry, spot, strikes, sigs, params) self._status.config( text=f"{len(strikes)} strikes loaded · T = {round((datetime.strptime(expiry,'%d%b%y').date() - datetime.utcnow().date()).days)} days", fg=C_GREEN, ) self._btn.config(state="normal") def _on_error(self, msg: str): self._status.config(text=f"Error: {msg}", fg=C_RED) self._btn.config(state="normal") def _draw_empty_chart(self): self._ax.clear() self._ax.set_facecolor(C_SURFACE) self._ax.text(0.5, 0.5, "Enter an expiry and click Fetch & Analyse", transform=self._ax.transAxes, ha="center", va="center", color=C_MUTED, fontsize=11) for spine in self._ax.spines.values(): spine.set_edgecolor(C_OVERLAY) self._ax.tick_params(colors=C_MUTED) self._canvas.draw() def _update_chart(self, expiry, spot, K_d, pdf_s, sigs): self._ax.clear() _style_axes(self._ax) self._ax.plot(K_d, pdf_s, color=C_BLUE, linewidth=2, label="Risk-Neutral PDF") self._ax.fill_between(K_d, pdf_s, alpha=0.12, color=C_BLUE) self._ax.axvline(spot, color=C_RED, linestyle="--", linewidth=1.4, label=f"Spot ${spot:,.0f}") peak = sigs["peak"]["peak_strike"] self._ax.axvline(peak, color=C_MAUVE, linestyle=":", linewidth=1.2, label=f"PDF mode ${peak:,.0f}") final = sigs["final"] color = _sig_color(final) self._ax.set_title(f"BTC Risk-Neutral PDF · {expiry} → {final}", color=color, fontsize=12, fontweight="bold") self._ax.set_xlabel("Strike", color=C_SUBTEXT, fontsize=9) self._ax.set_ylabel("Probability Density", color=C_SUBTEXT, fontsize=9) legend = self._ax.legend(facecolor=C_OVERLAY, labelcolor=C_TEXT, framealpha=0.9, fontsize=9) self._fig.tight_layout() self._canvas.draw() def _update_signals(self, expiry, spot, strikes, sigs, params: dict | None = None): if params is None: params = {"tail_threshold": TAIL_THRESHOLD, "band_width": BAND_WIDTH, "signal_delta": SIGNAL_DELTA} tail = params["tail_threshold"] gate = params["signal_delta"] final = sigs["final"] fc = _sig_color(final) W = 38 lines = [ "─" * W, f" Expiry : {expiry}", f" Spot : ${spot:>12,.2f}", f" Strikes: {len(strikes)} loaded", "─" * W, "", " SIGNALS", "", ] details = [ ("SKEW", sigs["skew"]["signal"], f" P(above spot) = {sigs['skew']['p_up']:.4f}\n" f" P(below spot) = {sigs['skew']['p_dn']:.4f}\n" f" diff = {sigs['skew']['diff']:.4f} (gate {gate:.3f})"), ("TAILS", sigs["tails"]["signal"], f" P(far up >+{tail:.0%}) = {sigs['tails']['p_up']:.4f}\n" f" P(far dn -{tail:.0%}) = {sigs['tails']['p_dn']:.4f}\n" f" diff = {sigs['tails']['diff']:.4f} (gate {gate:.3f})"), ("PEAK", sigs["peak"]["signal"], f" PDF mode = ${sigs['peak']['peak_strike']:>10,.2f}\n" f" Spot = ${spot:>10,.2f}"), ("BAND", sigs["band"]["signal"], f" Mass inside +-{params['band_width']:.0%} = {sigs['band']['p_band']:.4f}\n" f" Mass outside = {sigs['band']['p_outside']:.4f}\n" f" (gate {min(0.49, gate * BAND_DELTA_MULT):.3f})"), ] arrow = {"LONG": "▲ LONG", "SHORT": "▼ SHORT", "NO TRADE": "— NO TRADE"} self._sig_text.config(state="normal") self._sig_text.delete("1.0", "end") for line in lines: self._sig_text.insert("end", line + "\n") for name, sig, detail in details: tag = f"sig_{name}" label = f" {name:<6} {arrow.get(sig, sig)}\n" self._sig_text.insert("end", label, tag) self._sig_text.tag_config(tag, foreground=_sig_color(sig), font=("Consolas", 10, "bold")) self._sig_text.insert("end", detail + "\n\n") sep = "─" * W + "\n" vtag = "verdict" vline = f" VERDICT: {final}\n" self._sig_text.insert("end", sep) self._sig_text.insert("end", vline, vtag) self._sig_text.tag_config(vtag, foreground=fc, font=("Consolas", 12, "bold")) self._sig_text.insert("end", sep) self._sig_text.insert("end", "\n What does this mean?\n\n") explanations = { "LONG": " The market is pricing in more upside\n" " than downside. Probability mass, tail\n" " weight, and the PDF mode all lean above\n" " the current spot price.", "SHORT": " The market is pricing in more downside\n" " than upside. Skew and tails favour a\n" " move lower from current spot.", "NO TRADE": " Signals disagree or are below the noise\n" " gate. No clear directional bias is priced\n" " into the options market right now.", } self._sig_text.insert("end", explanations.get(final, ""), "explain") self._sig_text.tag_config("explain", foreground=C_SUBTEXT, font=("Consolas", 9)) self._sig_text.config(state="disabled") # --- backtest tab --- class BacktestTab(tk.Frame): def __init__(self, parent): super().__init__(parent, bg=C_BG) self._build() def _build(self): ctrl = tk.Frame(self, bg=C_BG) ctrl.pack(fill="x", padx=16, pady=(14, 4)) tk.Label(ctrl, text="Synthetic cases:", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 10)).pack(side="left") self._n_var = tk.StringVar(value=str(DEFAULT_N_CASES)) tk.Entry(ctrl, textvariable=self._n_var, width=10, bg=C_OVERLAY, fg=C_TEXT, insertbackground=C_TEXT, relief="flat", font=("Segoe UI", 10)).pack(side="left", padx=8) self._btn = tk.Button(ctrl, text="Run Backtest", command=self._on_run, bg=C_OVERLAY, fg=C_TEXT, activebackground=C_MUTED, relief="flat", font=("Segoe UI", 10, "bold"), padx=12, pady=4, cursor="hand2") self._btn.pack(side="left") self._status = tk.Label(ctrl, text="", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 9)) self._status.pack(side="left", padx=16) params = tk.Frame(self, bg=C_BG) params.pack(fill="x", padx=16, pady=(0, 4)) _PCT_OPTS = ["0.5%", "1%", "2%", "3%", "5%", "7%", "10%"] def _pct_menu(parent, var): m = tk.OptionMenu(parent, var, *_PCT_OPTS) m.config(bg=C_OVERLAY, fg=C_TEXT, activebackground=C_MUTED, activeforeground=C_TEXT, relief="flat", font=("Segoe UI", 9), highlightthickness=0, width=5) m["menu"].config(bg=C_OVERLAY, fg=C_TEXT, activebackground=C_MUTED, font=("Segoe UI", 9)) return m tk.Label(params, text="Tail cutoff:", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 9)).pack(side="left") self._tail_var = tk.StringVar(value="2%") _pct_menu(params, self._tail_var).pack(side="left", padx=(4, 12)) tk.Label(params, text="Band width +-:", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 9)).pack(side="left") self._band_var = tk.StringVar(value="3%") _pct_menu(params, self._band_var).pack(side="left", padx=(4, 12)) tk.Label(params, text="Signal gate:", bg=C_BG, fg=C_SUBTEXT, font=("Segoe UI", 9)).pack(side="left") self._gate_var = tk.StringVar(value="3%") _pct_menu(params, self._gate_var).pack(side="left", padx=(4, 0)) style = ttk.Style() style.theme_use("clam") style.configure("BT.Horizontal.TProgressbar", troughcolor=C_OVERLAY, background=C_BLUE, thickness=6) self._progress = ttk.Progressbar(self, style="BT.Horizontal.TProgressbar", mode="determinate") self._progress.pack(fill="x", padx=16, pady=(0, 6)) paned = tk.PanedWindow(self, orient="horizontal", bg=C_BG, sashwidth=5, relief="flat", sashpad=2) paned.pack(fill="both", expand=True, padx=8, pady=4) stats_f = tk.Frame(paned, bg=C_SURFACE) paned.add(stats_f, minsize=280) self._stats_text = scrolledtext.ScrolledText( stats_f, font=("Consolas", 10), bg=C_SURFACE, fg=C_TEXT, insertbackground=C_TEXT, relief="flat", state="disabled", ) self._stats_text.pack(fill="both", expand=True, padx=4, pady=4) charts_f = tk.Frame(paned, bg=C_BG) paned.add(charts_f, minsize=480) self._fig, self._axes = plt.subplots(1, 2, facecolor=C_BG, figsize=(7, 4)) self._canvas = FigureCanvasTkAgg(self._fig, master=charts_f) self._canvas.get_tk_widget().pack(fill="both", expand=True) self._draw_empty_charts() def _on_run(self): try: n = int(self._n_var.get()) if not (1 <= n <= 100_000): raise ValueError except ValueError: self._status.config(text="Enter a whole number between 1 and 100 000.", fg=C_RED) return self._btn.config(state="disabled") self._progress["value"] = 0 self._status.config(text="Running…", fg=C_YELLOW) params = self._get_params() threading.Thread(target=self._worker, args=(n, params), daemon=True).start() def _get_params(self) -> dict: def pct(var): return float(var.get().rstrip("%")) / 100.0 return { "tail_threshold": pct(self._tail_var), "band_width": pct(self._band_var), "signal_delta": pct(self._gate_var), } def _worker(self, n: int, params: dict): def cb(p): self.after(0, lambda: self._progress.configure(value=p * 100)) result = run_backtest(n, params=params, progress_cb=cb) self.after(0, self._on_result, result) def _on_result(self, r: dict): self._update_stats(r) self._update_charts(r) acc = r["accuracy"] self._status.config( text=f"Done · {r['n_traded']:,} traded · accuracy {acc:.1%}", fg=C_GREEN, ) self._progress["value"] = 100 self._btn.config(state="normal") def _update_stats(self, r: dict): W = 40 acc = r["accuracy"] if acc > 0.55: verdict = " EDGE DETECTED above random baseline." vc = C_GREEN elif acc > 0.48: verdict = " INCONCLUSIVE — near random (50%)." vc = C_YELLOW else: verdict = " UNDERPERFORMS random baseline." vc = C_RED lines = [ ("─" * W + "\n", None), (" BACKTEST RESULTS\n", None), ("─" * W + "\n", None), (f" Total cases : {r['n_total']:>8,}\n", None), (f" Skipped (bad PDF): {r['n_skipped']:>7,}\n", None), (f" Traded (signal) : {r['n_traded']:>7,}\n", None), (f" No-trade filter : {r['n_no_trade']:>7,}\n", None), ("─" * W + "\n", None), (f" Correct : {r['n_correct']:>7,}\n", "correct"), (f" Wrong : {r['n_wrong']:>7,}\n", "wrong"), (f" Accuracy : {acc:>8.2%}\n", "accuracy"), ("─" * W + "\n", None), (" By direction\n\n", None), (f" LONG correct / wrong : " f"{r['by_dir']['LONG'][0]:>4} / {r['by_dir']['LONG'][1]}\n", "long_row"), (f" SHORT correct / wrong : " f"{r['by_dir']['SHORT'][0]:>4} / {r['by_dir']['SHORT'][1]}\n", "short_row"), ("─" * W + "\n", None), ("\n", None), (verdict + "\n", "verdict"), ("\n", None), (" Note: labels are GBM-simulated futures, not\n" " live prices. Treat this as a sanity-check on\n" " signal coherence, not a live P&L forecast.\n", "note"), ] self._stats_text.config(state="normal") self._stats_text.delete("1.0", "end") for text, tag in lines: if tag: self._stats_text.insert("end", text, tag) else: self._stats_text.insert("end", text) self._stats_text.tag_config("correct", foreground=C_GREEN) self._stats_text.tag_config("wrong", foreground=C_RED) self._stats_text.tag_config("accuracy", foreground=C_BLUE, font=("Consolas", 11, "bold")) self._stats_text.tag_config("long_row", foreground=C_GREEN) self._stats_text.tag_config("short_row", foreground=C_RED) self._stats_text.tag_config("verdict", foreground=vc, font=("Consolas", 10, "bold")) self._stats_text.tag_config("note", foreground=C_MUTED, font=("Consolas", 9)) self._stats_text.config(state="disabled") def _draw_empty_charts(self): for ax in self._axes: ax.clear() ax.set_facecolor(C_SURFACE) ax.text(0.5, 0.5, "No data yet", transform=ax.transAxes, ha="center", va="center", color=C_MUTED, fontsize=10) for spine in ax.spines.values(): spine.set_edgecolor(C_OVERLAY) ax.tick_params(colors=C_MUTED) self._fig.tight_layout() self._canvas.draw() def _update_charts(self, r: dict): ax_pie, ax_bar = self._axes ax_pie.clear() ax_bar.clear() # pie: overall outcome split labels = ["Correct", "Wrong", "No-Trade"] values = [r["n_correct"], r["n_wrong"], r["n_no_trade"]] colors = [C_GREEN, C_RED, C_YELLOW] explode = (0.04, 0.04, 0.0) non_zero = [(l, v, c, e) for l, v, c, e in zip(labels, values, colors, explode) if v > 0] if non_zero: ls, vs, cs, es = zip(*non_zero) ax_pie.pie(vs, labels=ls, colors=cs, explode=es, autopct="%1.1f%%", startangle=120, textprops={"color": C_TEXT, "fontsize": 9}, wedgeprops={"linewidth": 0.6, "edgecolor": C_BG}) ax_pie.set_facecolor(C_SURFACE) ax_pie.set_title("Outcome Breakdown", color=C_TEXT, fontsize=10, pad=8) # bar: correct vs wrong broken down by direction _style_axes(ax_bar) directions = ["LONG", "SHORT"] x = np.arange(len(directions)) w = 0.35 correct_v = [r["by_dir"][d][0] for d in directions] wrong_v = [r["by_dir"][d][1] for d in directions] bars_c = ax_bar.bar(x - w / 2, correct_v, w, label="Correct", color=C_GREEN, edgecolor=C_BG, linewidth=0.5) bars_w = ax_bar.bar(x + w / 2, wrong_v, w, label="Wrong", color=C_RED, edgecolor=C_BG, linewidth=0.5) for bars in (bars_c, bars_w): for bar in bars: h = bar.get_height() if h > 0: ax_bar.text(bar.get_x() + bar.get_width() / 2, h + 0.5, str(int(h)), ha="center", va="bottom", color=C_TEXT, fontsize=8) ax_bar.set_xticks(x) ax_bar.set_xticklabels(directions, color=C_TEXT) ax_bar.set_ylabel("Cases", color=C_SUBTEXT, fontsize=9) ax_bar.set_title("Correct vs Wrong by Direction", color=C_TEXT, fontsize=10) ax_bar.legend(facecolor=C_OVERLAY, labelcolor=C_TEXT, fontsize=9, framealpha=0.9) self._fig.tight_layout() self._canvas.draw() # --- app shell --- class App(tk.Tk): def __init__(self): super().__init__() self.title("BTC Options Signal Analyser") self.minsize(900, 620) self.configure(bg=C_BG) self._build() def _build(self): style = ttk.Style(self) style.theme_use("clam") style.configure("TNotebook", background=C_BG, borderwidth=0) style.configure("TNotebook.Tab", background=C_OVERLAY, foreground=C_SUBTEXT, padding=[18, 7], font=("Segoe UI", 10)) style.map("TNotebook.Tab", background=[("selected", C_SURFACE)], foreground=[("selected", C_MAUVE)]) nb = ttk.Notebook(self) nb.pack(fill="both", expand=True, padx=8, pady=8) live_tab = LiveAnalysisTab(nb) nb.add(live_tab, text=" Mode 1 — Live Analysis ") bt_tab = BacktestTab(nb) nb.add(bt_tab, text=" Mode 2 — Backtest ") if __name__ == "__main__": App().mainloop()