#!/usr/bin/env python3 # MIT License # # Copyright (c) 2026 Options Lab (ashares options learning site) # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. """How often did A-share indices fall 20 % in a month, and does the model see it? Reproduce the numbers. For learning only. Past results do not guarantee future ones; not investment advice. What it does ------------ 1. Downloads public data with AKShare: daily closes of six A-share indices (Sina) and the QVIX implied-vol index of 50ETF and 300ETF options. 2. For every index and every trading day from 2005-01-04, takes the window of the next h trading days (h = 10, 21, 42) and asks: did the index close at least X lower at the end of the window (X = 10 / 15 / 20 / 30 / 40 %)? Also the "path" version: was any close inside the window X lower? 3. Overlapping windows are not independent (one crash makes many hit days), so hit days closer than h trading days to the previous one form one episode. The 95 % interval counts episodes (an exact Poisson interval on the episode count, scaled to the rate); a second interval resamples whole calendar years (2000 draws, seed 7). Pooled over the six indices, the episodes are merged across indices (one market crash = one episode). 4. The model side: the probability of the same fall under the site's model at a typical vol, both a single lognormal and the vol range the site's Kelly sizing averages over (the forecast vol is uncertain: log sd 0.253, the high-vol half widened by 1.143). The typical vol is FORECAST_RATIO x the median at-the-money implied vol (ATM IV) the site stored for each option underlying, 2024-2026 (TYPICAL_ATM_IV below). 5. Prints: the rate of each fall at each horizon with both intervals, the model's rate, the 2016+ cut, the pooled rate without each of the three largest episodes, the horizon ratios, terminal vs path, the worst fall in named periods, the day-by-day check with QVIX, and the rate split by the realised vol of the 21 days before the start. Not included: the Kelly replay on real option chains (the recorded chains of 2026-09-24 come from a commercial data vendor and cannot be redistributed). Run --- pip install -r requirements.txt python crash_floor.py # download, then print the numbers (~1 minute) python crash_floor.py --cache-dir data # keep the download as CSV and reuse it next time python crash_floor.py --end none # use every close available, not just the published run Python 3.10+. Returns and vols are decimals inside the code (0.20 = 20 %). """ from __future__ import annotations import argparse import math from functools import lru_cache from pathlib import Path import numpy as np import pandas as pd from scipy.stats import chi2, norm # --------------------------------------------------------------------------- # Inputs # --------------------------------------------------------------------------- # code: (name, Sina symbol, option underlying it stands for, QVIX series, AKShare QVIX function) INDICES = { "000016": ("SSE 50", "sh000016", "510050", "50etf", "index_option_50etf_qvix"), "000300": ("CSI 300", "sh000300", "510300", "300etf", "index_option_300etf_qvix"), "000905": ("CSI 500", "sh000905", "510500", None, None), "000852": ("CSI 1000", "sh000852", "IM", None, None), "399006": ("ChiNext", "sz399006", "159915", None, None), "000688": ("STAR 50", "sh000688", "588000", None, None), } # The typical ATM IV of each option underlying: the median of the ATM IV the # site stored from its own option chains, 2024-01 to 2026-10 (about 646 # trading days each). That one input is not public, so it is fixed here # (decimal; 0.1874 = 18.74 %). IM = CSI 1000 index options. TYPICAL_ATM_IV = { "510050": 0.18741892857142858, "510300": 0.20317, "510500": 0.29247192857142856, "IM": 0.29168908571428576, "159915": 0.3234338428571429, "588000": 0.39030000000000004, } FORECAST_RATIO = 0.95 # the site's vol forecast = 0.95 x ATM IV FORECAST_FLOOR = 0.01 START = "2005-01-04" # first start day (or the index's first close, if later) PUBLISHED_END = "2026-09-30" # the last close the published numbers used HORIZONS = (10, 21, 42) # trading days FALLS = (0.10, 0.15, 0.20, 0.30, 0.40) PRIMARY_H = 21 PRIMARY_X = 0.20 CUT_2016 = pd.Timestamp("2016-01-04") ZERO_EPISODE_UPPER = 3.689 # 95 % upper bound on a Poisson count of 0 N_BOOT = 2000 BOOT_SEED = 7 TRADING_YEAR = 252 CALIBRATION_FALLS = (0.10, 0.15, 0.20) RV_WINDOW = 21 RV_BUCKETS = ((0.0, 0.20, "< 20 %"), (0.20, 0.30, "20-30 %"), (0.30, np.inf, ">= 30 %")) PERIODS = { "2008": ("2008-01-01", "2008-12-31"), "2015": ("2015-06-01", "2015-09-30"), "2016-01": ("2016-01-01", "2016-02-29"), "2018": ("2018-01-01", "2018-12-31"), "2020-02": ("2020-01-15", "2020-03-31"), "2022": ("2022-01-01", "2022-10-31"), "2024-02": ("2024-01-01", "2024-02-29"), } # The vol range (same as kelly_range.py): log RV ~ N(log c - s^2/2, s^2) RANGE_ALPHA = 0.064 LOG_SD = math.sqrt(RANGE_ALPHA) # 0.253, the same at every horizon TAIL_ABOVE_Q95 = 0.075 _Z95 = float(norm.ppf(0.95)) TAIL_K = _Z95 / float(norm.ppf(1.0 - TAIL_ABOVE_Q95)) # 1.143: upper-half widening QUADRATURE_NODES = 31 # 16 Gauss points on each half VOL_FLOOR = 0.01 MAX_RATIO = 5.0 # nodes above 5x the centre are clipped to it # --------------------------------------------------------------------------- # Data # --------------------------------------------------------------------------- def fetch(cache_dir: str | Path | None = None) -> dict[str, pd.DataFrame]: """Download (or reload) the index closes and QVIX. Returns long frames: ``idx`` (code, date, close) and ``qvix`` (series, date, close); QVIX is in index points (18.5 = 18.5 %). """ names = {"idx": "crash_floor_idx.csv", "qvix": "crash_floor_qvix.csv"} if cache_dir is not None: paths = {n: Path(cache_dir) / f for n, f in names.items()} if all(p.exists() for p in paths.values()): return {n: pd.read_csv(p, parse_dates=["date"], dtype={"code": str, "series": str}) for n, p in paths.items()} import akshare as ak # imported here so compute() works without it idx, qv = [], [] for code, (_, sina, _, series, qfunc) in INDICES.items(): i = ak.stock_zh_index_daily(symbol=sina)[["date", "close"]] idx.append(i.assign(code=code)) if qfunc: q = getattr(ak, qfunc)().rename(columns=str.lower)[["date", "close"]] qv.append(q.assign(series=series)) out = {"idx": pd.concat(idx, ignore_index=True), "qvix": pd.concat(qv, ignore_index=True)} for df in out.values(): df["date"] = pd.to_datetime(df["date"]) if cache_dir is not None: Path(cache_dir).mkdir(parents=True, exist_ok=True) for n, df in out.items(): df.to_csv(Path(cache_dir) / names[n], index=False) return out def closes(frames: dict[str, pd.DataFrame], end: str | None = PUBLISHED_END) -> dict[str, pd.Series]: """Close by date for each index, from START to ``end`` (None = all).""" idx = frames["idx"] out = {} for code in INDICES: s = idx[idx["code"] == code].set_index("date")["close"].astype(float).sort_index() s.index = pd.to_datetime(s.index) s = s[~s.index.duplicated(keep="last")] s = s[s.index >= pd.Timestamp(START)] out[code] = s if end is None else s[s.index <= pd.Timestamp(end)] return out def qvix_series(frames: dict[str, pd.DataFrame], series: str) -> pd.Series: """QVIX close (decimal) by date; zero / missing rows dropped.""" q = frames["qvix"] s = q[q["series"] == series].set_index("date")["close"].astype(float) s.index = pd.to_datetime(s.index) return (s[s > 0] / 100.0).sort_index() # --------------------------------------------------------------------------- # Windows, episodes and intervals # --------------------------------------------------------------------------- def windows(s: pd.Series, h: int) -> pd.DataFrame: """Terminal and path return of every closed h-day window of one series.""" v = s.to_numpy(dtype=float) n = len(v) - h if n <= 0: return pd.DataFrame(columns=["date", "term", "path"]) term = v[h:] / v[:n] - 1.0 ahead = np.lib.stride_tricks.sliding_window_view(v[1:], h)[:n] # the next h closes path = ahead.min(axis=1) / v[:n] - 1.0 return pd.DataFrame({"date": s.index[:n], "term": term, "path": path}) def panel(cl: dict[str, pd.Series], h: int) -> pd.DataFrame: parts = [] for code, s in cl.items(): w = windows(s, h) w.insert(0, "code", code) parts.append(w) p = pd.concat(parts, ignore_index=True) p["date"] = pd.to_datetime(p["date"]) return p def episode_spans(positions: np.ndarray, h: int) -> list[tuple[int, int, int]]: """(first, last, hits) per episode of hit positions on one calendar; a gap of more than h trading days starts a new episode.""" pos = np.unique(np.asarray(positions, dtype=int)) if pos.size == 0: return [] breaks = np.nonzero(np.diff(pos) > h)[0] starts = np.concatenate([[0], breaks + 1]) ends = np.concatenate([breaks, [pos.size - 1]]) return [(int(pos[a]), int(pos[b]), int(b - a + 1)) for a, b in zip(starts, ends)] def garwood(k: int) -> tuple[float, float]: """Exact 95 % Poisson interval on a count k.""" lo = 0.0 if k == 0 else chi2.ppf(0.025, 2 * k) / 2.0 hi = chi2.ppf(0.975, 2 * k + 2) / 2.0 return float(lo), float(hi) def episode_ci(hits: int, n: int, episodes: int, h: int) -> tuple[float, float, float]: """Rate and its 95 % interval: the episode count is Poisson, the hits per episode are taken as fixed.""" if n == 0: return float("nan"), float("nan"), float("nan") p = hits / n if episodes == 0: return 0.0, 0.0, min(1.0, ZERO_EPISODE_UPPER * h / n) lo, hi = garwood(episodes) return p, p * lo / episodes, min(1.0, p * hi / episodes) def year_bootstrap(years: np.ndarray, hit: np.ndarray, n_boot: int = N_BOOT, seed: int = BOOT_SEED) -> tuple[float, float]: """Percentile 95 % interval of the rate, resampling calendar years.""" cats, idx = np.unique(years, return_inverse=True) cnt = np.bincount(idx, minlength=len(cats)).astype(float) hits = np.bincount(idx, weights=hit.astype(float), minlength=len(cats)) rng = np.random.default_rng(seed) w = rng.multinomial(len(cats), np.full(len(cats), 1.0 / len(cats)), size=n_boot).astype(float) with np.errstate(invalid="ignore", divide="ignore"): rates = (w @ hits) / (w @ cnt) lo, hi = np.nanpercentile(rates, [2.5, 97.5]) return float(lo), float(hi) def calendar(p: pd.DataFrame) -> pd.DatetimeIndex: return pd.DatetimeIndex(np.sort(p["date"].unique())) def cell(p: pd.DataFrame, x: float, h: int, kind: str = "term", cal: pd.DatetimeIndex | None = None, boot: bool = True) -> dict: """Rate of a fall >= x on panel ``p`` (one index, or pooled). Episodes are counted on ``cal`` (default: the panel's own union calendar).""" cal = calendar(p) if cal is None else cal hit = (p[kind] <= -x).to_numpy() hit_dates = p.loc[hit, "date"] spans = episode_spans(cal.get_indexer(pd.DatetimeIndex(hit_dates.unique())), h) n_hits, n = int(hit.sum()), len(p) rate, lo, hi = episode_ci(n_hits, n, len(spans), h) blo, bhi = (year_bootstrap(p["date"].dt.year.to_numpy(), hit) if boot and n else (float("nan"), float("nan"))) hd = hit_dates.to_numpy() dated = [] for a, b, _ in spans: first, last = cal[a], cal[b] dated.append((first, last, int(((hd >= first.to_datetime64()) & (hd <= last.to_datetime64())).sum()))) mean_fall = float(-p.loc[hit, kind].mean()) if n_hits else float("nan") return {"hits": n_hits, "n": n, "episodes": len(spans), "p": rate, "lo": lo, "hi": hi, "boot_lo": blo, "boot_hi": bhi, "spans": dated, "mean_fall": mean_fall} def leave_one_out(p: pd.DataFrame, x: float, h: int, k: int = 3) -> list[dict]: """The pooled rate with each of the k largest episodes left out (every index-day dated inside the episode's span leaves both counts).""" full = cell(p, x, h, boot=False) out = [] for first, last, n_hits in sorted(full["spans"], key=lambda s: -s[2])[:k]: q = p[(p["date"] < first) | (p["date"] > last)] out.append({"first": first, "last": last, "hits_dropped": n_hits, **cell(q, x, h, boot=False)}) return out # --------------------------------------------------------------------------- # The model # --------------------------------------------------------------------------- @lru_cache(maxsize=16) def half_range_rule(n: int) -> tuple[np.ndarray, np.ndarray]: """n-point Gauss rule for the half-normal (weight e^(-z^2/2) on [0, inf)), weights normalised to sum 1. Built by the discretised Stieltjes procedure and the Golub-Welsch eigenvalue method.""" x, wx = np.polynomial.legendre.leggauss(400) z = 6.0 * (x + 1.0) dz = 6.0 * wx * np.exp(-0.5 * z * z) a, b = np.zeros(n), np.zeros(n) p_prev, p = np.zeros_like(z), np.ones_like(z) norm_prev = 1.0 for j in range(n): norm_j = float(dz @ (p * p)) a[j] = float(dz @ (z * p * p)) / norm_j b[j] = norm_j if j == 0 else norm_j / norm_prev p_prev, p = p, (z - a[j]) * p - (b[j] if j > 0 else 0.0) * p_prev norm_prev = norm_j off = np.sqrt(b[1:]) nodes, vecs = np.linalg.eigh(np.diag(a) + np.diag(off, 1) + np.diag(off, -1)) weights = vecs[0] ** 2 return nodes, weights / weights.sum() def forecast_range(centre: float, sd: float = LOG_SD) -> tuple[np.ndarray, np.ndarray]: """Quadrature nodes (sigma_k, w_k) of the vol range around ``centre``: log RV ~ N(log c - s^2/2, s^2), the upper half's log sd times TAIL_K, each half with its own Gauss rule.""" median = centre * math.exp(-0.5 * sd * sd) u, wu = half_range_rule((QUADRATURE_NODES + 1) // 2) z = np.concatenate([-u[::-1], u * TAIL_K]) w = np.concatenate([wu[::-1], wu]) / 2.0 sigmas = np.clip(median * np.exp(sd * z), VOL_FLOOR, MAX_RATIO * centre) return sigmas, w def forecast(atm_iv: float) -> float: return max(atm_iv * FORECAST_RATIO, FORECAST_FLOOR) def p_lognormal(sigma, h: int, x: float): """P(S_T <= F (1 - x)) for a lognormal with mean F (no drift), T = h / 252.""" sigma = np.asarray(sigma, dtype=float) sd = sigma * math.sqrt(h / TRADING_YEAR) out = norm.cdf((math.log(1.0 - x) + 0.5 * sd * sd) / sd) return float(out) if out.ndim == 0 else out def p_mixture(centre: float, h: int, x: float) -> float: """The same probability averaged over the vol range around ``centre``.""" sigmas, w = forecast_range(centre) return float(w @ p_lognormal(sigmas, h, x)) # --------------------------------------------------------------------------- # The study # --------------------------------------------------------------------------- def _cut(p: pd.DataFrame, cut: str) -> pd.DataFrame: return p if cut == "full" else p[p["date"] >= CUT_2016] def model_rates(panels: dict[int, pd.DataFrame], sigma: dict[str, float]) -> dict: """{(scope, cut, h, x): (point, vol range)}; pooled = the mean of the six indices weighted by their number of start days in that cut.""" out = {} for h, p in panels.items(): for x in FALLS: per = {c: (p_lognormal(sigma[c], h, x), p_mixture(sigma[c], h, x)) for c in sigma} for c, v in per.items(): out[(c, "full", h, x)] = v for cut in ("full", "2016+"): n = _cut(p, cut)["code"].value_counts() w = np.array([n.get(c, 0) for c in per], dtype=float) w = w / w.sum() out[("pooled", cut, h, x)] = (float(w @ [v[0] for v in per.values()]), float(w @ [v[1] for v in per.values()])) return out def calibration(frames: dict[str, pd.DataFrame], p21: pd.DataFrame) -> list[dict]: """Day by day: the model at FORECAST_RATIO x that day's QVIX vs the hit rate (h = 21).""" rows = [] for code, (_, _, _, series, _) in INDICES.items(): if series is None: continue q = qvix_series(frames, series).rename("qvix") g = p21[p21["code"] == code].set_index("date").join(q, how="inner") sig = np.maximum(FORECAST_RATIO * g["qvix"].to_numpy(), FORECAST_FLOOR) mix: dict[float, list[float]] = {} for x in CALIBRATION_FALLS: hit = (g["term"] <= -x).to_numpy() pm = [] for s in sig: key = round(float(s), 4) if key not in mix: mix[key] = [p_mixture(key, PRIMARY_H, xx) for xx in CALIBRATION_FALLS] pm.append(mix[key][CALIBRATION_FALLS.index(x)]) cal = pd.DatetimeIndex(g.index) spans = episode_spans(cal.get_indexer(cal[hit]), PRIMARY_H) rows.append({"code": code, "first": g.index.min(), "last": g.index.max(), "n": len(g), "x": x, "hits": int(hit.sum()), "episodes": len(spans), "rate": float(hit.mean()), "p_point": float(p_lognormal(sig, PRIMARY_H, x).mean()), "p_mix": float(np.mean(pm))}) return rows def by_realised_vol(cl: dict[str, pd.Series], p21: pd.DataFrame) -> dict: """The h = 21 rate split by the realised vol of the 21 trading days before the start (annualised, sqrt(252)); episodes on each scope's full calendar.""" parts = [] for code, s in cl.items(): rv = np.log(s).diff().rolling(RV_WINDOW).std() * np.sqrt(TRADING_YEAR) g = p21[p21["code"] == code].merge(rv.rename("rv"), left_on="date", right_index=True) parts.append(g.dropna(subset=["rv"])) p = pd.concat(parts, ignore_index=True) out = {} for lo, hi, label in RV_BUCKETS: b = p[(p["rv"] >= lo) & (p["rv"] < hi)] for scope, g in [("pooled", b)] + [(c, b[b["code"] == c]) for c in sorted(b["code"].unique())]: cal = calendar(p if scope == "pooled" else p[p["code"] == scope]) for x in CALIBRATION_FALLS: out[(label, scope, x)] = cell(g, x, PRIMARY_H, cal=cal, boot=False) return out def worst_in_periods(p21: pd.DataFrame, kind: str = "term") -> dict: out = {} for name, (a, b) in PERIODS.items(): q = p21[(p21["date"] >= a) & (p21["date"] <= b)] for code, g in q.groupby("code"): out[(name, code)] = float(g[kind].min()) return out def compute(frames: dict[str, pd.DataFrame], end: str | None = PUBLISHED_END) -> dict: cl = closes(frames, end) sigma = {c: forecast(TYPICAL_ATM_IV[u]) for c, (_, _, u, _, _) in INDICES.items()} panels = {h: panel(cl, h) for h in HORIZONS} cells = {} for h, p in panels.items(): for cut in ("full", "2016+"): q = _cut(p, cut) for scope, g in [("pooled", q)] + [(c, q[q["code"] == c]) for c in INDICES]: for kind in ("term", "path"): for x in FALLS: cells[(scope, cut, kind, h, x)] = cell(g, x, h, kind) p21 = panels[PRIMARY_H] return { "closes": cl, "sigma": sigma, "panels": panels, "cells": cells, "models": model_rates(panels, sigma), "loo": leave_one_out(p21, PRIMARY_X, PRIMARY_H), "worst": worst_in_periods(p21), "calibration": calibration(frames, p21), "by_rv": by_realised_vol(cl, p21), } # --------------------------------------------------------------------------- # Printout # --------------------------------------------------------------------------- def _pct(v: float, d: int = 2) -> str: return "n/a" if v != v else f"{100 * v:.{d}f} %" def _name(scope: str) -> str: return "pooled (6)" if scope == "pooled" else f"{INDICES[scope][0]} ({scope})" def _rate_table(r: dict, cut: str, h: int, x: float, kind: str = "term") -> None: print(f" {'index':<20}{'starts':>7}{'hits':>6}{'ep.':>5} {'rate [95 % episode CI]':<27}" f"{'year-block CI':<21}{'point':>9}{'vol range':>11}{'ratio':>8}") for scope in ("pooled", *INDICES): c = r["cells"][(scope, cut, kind, h, x)] pt, mx = r["models"][(scope, cut if scope == "pooled" else "full", h, x)] ratio = c["p"] / mx if mx > 0 else float("nan") print(f" {_name(scope):<20}{c['n']:>7}{c['hits']:>6}{c['episodes']:>5} " f"{_pct(c['p']):>8} [{_pct(c['lo'])}, {_pct(c['hi'])}] " f"[{_pct(c['boot_lo'])}, {_pct(c['boot_hi'])}] {pt:>9.2e} {_pct(mx, 3):>9} {ratio:>5.2g}x") def main(argv=None) -> int: ap = argparse.ArgumentParser(description="Crash floor study (public data)") ap.add_argument("--cache-dir", type=Path, help="save / reuse the downloaded CSVs here") ap.add_argument("--end", default=PUBLISHED_END, help="ignore closes after this date ('none' = use all)") args = ap.parse_args(argv) r = compute(fetch(args.cache_dir), None if args.end == "none" else args.end) print("Typical forecast vol (FORECAST_RATIO x TYPICAL_ATM_IV): " + ", ".join(f"{INDICES[c][0]} {_pct(s, 1)}" for c, s in r["sigma"].items())) print("ratio = history / the vol-range model\n") for x in (0.20, 0.15): print(f"1. A fall of {x:.0%} or more at the end of 21 trading days, starts from 2005") _rate_table(r, "full", PRIMARY_H, x) print() print("2. The same, starts from 2016-01-04 (fall >= 20 %)") _rate_table(r, "2016+", PRIMARY_H, PRIMARY_X) print("\n3. Pooled, fall >= 20 % in 21 days, without each of the three largest episodes") for d in r["loo"]: print(f" without {d['first']:%Y-%m-%d} -> {d['last']:%Y-%m-%d} ({d['hits_dropped']} hits): " f"{d['hits']} / {d['n']}, {d['episodes']} episodes, {_pct(d['p'])} [{_pct(d['lo'])}, {_pct(d['hi'])}]") print("\n4. Pooled, by horizon (terminal), and terminal vs path at 21 days") for x in FALLS: rates = [r["cells"][("pooled", "full", "term", h, x)]["p"] for h in HORIZONS] path = r["cells"][("pooled", "full", "path", PRIMARY_H, x)] ratios = f"10/21 {rates[0] / rates[1]:.2f}, 42/21 {rates[2] / rates[1]:.2f}" if rates[1] > 0 else "-" print(f" fall >= {x:.0%}: h 10 {_pct(rates[0])}, h 21 {_pct(rates[1])}, h 42 {_pct(rates[2])} " f"({ratios}); path at 21 days {_pct(path['p'])} ({path['episodes']} episodes)") print("\n5. Worst 21-day change at expiry, start dates inside each period") print(" " + f"{'period':<9}" + "".join(f"{INDICES[c][0]:>10}" for c in INDICES)) for name in PERIODS: vals = [r["worst"].get((name, c)) for c in INDICES] print(" " + f"{name:<9}" + "".join(f"{100 * v:>9.1f}%" if v is not None else f"{'-':>10}" for v in vals)) print("\n6. Day by day: the model at FORECAST_RATIO x QVIX vs history (21 days)") for d in r["calibration"]: print(f" {INDICES[d['code']][0]:<8} {d['first']:%Y-%m-%d} -> {d['last']:%Y-%m-%d} ({d['n']}), " f"fall >= {d['x']:.0%}: {d['hits']} hits ({d['episodes']} ep.), history {_pct(d['rate'])}, " f"point {_pct(d['p_point'])}, vol range {_pct(d['p_mix'])}") print("\n7. By the realised vol of the 21 days before the start (fall >= 20 %, 21 days)") for scope in ("pooled", *INDICES): row = [] for _, _, label in RV_BUCKETS: c = r["by_rv"].get((label, scope, PRIMARY_X)) row.append(f"RV {label}: " + ("-" if c is None else f"{_pct(c['p'], 3)} ({c['episodes']} ep., n {c['n']})")) print(f" {_name(scope):<20}" + "; ".join(row)) return 0 if __name__ == "__main__": raise SystemExit(main())