| |
| """Generate threshold sweep + PR curve + 3 visualization plots. |
| |
| Runs the TCN over crash data ONCE, collects all scores, then: |
| 1. Computes precision/recall at thresholds [0.1, 0.2, ..., 0.9] |
| 2. Plots Precision-Recall curve |
| 3. Plots TTD distribution histogram |
| 4. Plots alert timeline overlaid on price chart |
| 5. Plots cascade funnel (if cascade stats available) |
| |
| Usage: |
| python scripts/generate_plots.py \ |
| --data data/parquet/BTCUSDT_2021-05-19.parquet \ |
| --model models/stage3_tcn_trained.pt \ |
| --out results/plots/ \ |
| --max-ticks 500000 |
| """ |
| import argparse |
| import json |
| import logging |
| import sys |
| import time |
| from collections import deque |
| from pathlib import Path |
| import matplotlib |
| matplotlib.use('Agg') |
| import matplotlib.pyplot as plt |
| plt.rcParams['axes.unicode_minus'] = False |
|
|
| import numpy as np |
| import pandas as pd |
| import torch |
|
|
| ML_DIR = Path(__file__).resolve().parent.parent / "ml" |
| sys.path.insert(0, str(ML_DIR)) |
| PROJECT_ROOT = Path(__file__).resolve().parent.parent |
| sys.path.insert(0, str(PROJECT_ROOT)) |
|
|
| from flash_crash_watchdog.data.historical_loader import df_to_ticks, load_parquet |
| from flash_crash_watchdog.data.labels import label_crashes |
| from flash_crash_watchdog.features import FEATURE_NAMES, FeatureExtractor |
| from flash_crash_watchdog.models.stage3_tcn import TCNDetector, TCNConfig |
|
|
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") |
| logger = logging.getLogger(__name__) |
|
|
| TCN_FEATURES = FEATURE_NAMES[:17] |
| WINDOW_SIZE = 200 |
| BASELINE_DROP_PCT = 2.0 |
| BASELINE_WINDOW_MS = 60_000 |
|
|
|
|
| |
| COLOR_ACCENT = '#95413a' |
| COLOR_BASELINE = '#587796' |
| COLOR_BG = '#f6f5f5' |
| COLOR_GRID = '#ccb9b9' |
| COLOR_TEXT = '#1b1919' |
| COLOR_GOOD = '#3d7750' |
| COLOR_WARN = '#9d7e40' |
|
|
|
|
| def load_trained_tcn(model_path: str, device: str = "auto") -> TCNDetector: |
| if device == "auto": |
| device = "cuda" if torch.cuda.is_available() else "cpu" |
| data = torch.load(model_path, map_location=device, weights_only=False) |
| config = data["config"] |
| model = TCNDetector(config).to(device) |
| model.load_state_dict(data["model_state"]) |
| model.eval() |
| logger.info("Loaded TCN from %s (device=%s)", model_path, device) |
| return model |
|
|
|
|
| def score_all_ticks( |
| model: TCNDetector, |
| df: pd.DataFrame, |
| max_ticks: int = 500_000, |
| device: str = "cpu", |
| ) -> pd.DataFrame: |
| """Run TCN over all ticks, return DataFrame with timestamps, scores, prices.""" |
| if max_ticks > 0 and len(df) > max_ticks: |
| indices = np.linspace(0, len(df) - 1, max_ticks, dtype=int) |
| df = df.iloc[indices].copy() |
| logger.info("Sampled to %d ticks", len(df)) |
|
|
| extractor = FeatureExtractor() |
| feature_window = deque(maxlen=WINDOW_SIZE) |
|
|
| results = [] |
| t0 = time.time() |
|
|
| for i, tick in enumerate(df_to_ticks(df, symbol="PLOT")): |
| if i % 50000 == 0: |
| elapsed = time.time() - t0 |
| rate = (i + 1) / max(1, elapsed) |
| logger.info(" Scoring tick %d/%d (%.0f/sec)", i, len(df), rate) |
|
|
| features = extractor.extract(tick) |
| vec = np.array([features.get(f, 0.0) for f in TCN_FEATURES]) |
| feature_window.append(vec) |
|
|
| if len(feature_window) < WINDOW_SIZE: |
| continue |
|
|
| window_array = np.array(list(feature_window)) |
| with torch.no_grad(): |
| x = torch.FloatTensor(window_array).T.unsqueeze(0).to(device) |
| scores = model(x) |
| score = float(scores[0, -1].item()) |
|
|
| results.append({ |
| "timestamp_ms": tick.timestamp_ms, |
| "score": score, |
| "mid_price": tick.book.mid_price or 0.0, |
| }) |
|
|
| results_df = pd.DataFrame(results) |
| logger.info("Scored %d ticks in %.1fs", len(results_df), time.time() - t0) |
| return results_df |
|
|
|
|
| def evaluate_at_threshold(scores_df: pd.DataFrame, crashes: list, threshold: float) -> dict: |
| """Evaluate precision/recall/TTD at a given threshold.""" |
| alerts = scores_df[scores_df["score"] >= threshold].to_dict("records") |
|
|
| true_positives = 0 |
| false_positives = 0 |
| ttd_ms = [] |
| matched_crashes = set() |
|
|
| for alert in alerts: |
| alert_ts = alert["timestamp_ms"] |
| matched = False |
| for j, crash in enumerate(crashes): |
| if j in matched_crashes: |
| continue |
| if crash.start_ts - 5000 <= alert_ts <= crash.end_ts: |
| true_positives += 1 |
| matched_crashes.add(j) |
| ttd = crash.end_ts - alert_ts |
| ttd_ms.append(ttd) |
| matched = True |
| break |
| if not matched: |
| false_positives += 1 |
|
|
| false_negatives = len(crashes) - true_positives |
| precision = true_positives / max(1, true_positives + false_positives) |
| recall = true_positives / max(1, len(crashes)) |
| f1 = 2 * precision * recall / max(1e-6, precision + recall) |
|
|
| return { |
| "threshold": threshold, |
| "alerts": len(alerts), |
| "true_positives": true_positives, |
| "false_positives": false_positives, |
| "false_negatives": false_negatives, |
| "precision": precision, |
| "recall": recall, |
| "f1": f1, |
| "median_ttd_ms": float(np.median(ttd_ms)) if ttd_ms else 0.0, |
| "ttd_ms": ttd_ms, |
| } |
|
|
|
|
| def plot_pr_curve(sweep_results: list, out_path: Path) -> None: |
| """Plot 1: Precision-Recall curve across thresholds.""" |
| fig, ax = plt.subplots(figsize=(8, 5), constrained_layout=True) |
|
|
| precisions = [r["precision"] for r in sweep_results] |
| recalls = [r["recall"] for r in sweep_results] |
| thresholds = [r["threshold"] for r in sweep_results] |
|
|
| ax.plot(recalls, precisions, 'o-', color=COLOR_ACCENT, linewidth=2, |
| markersize=8, label="TCN Detector") |
|
|
| |
| for i, t in enumerate(thresholds): |
| ax.annotate(f'τ={t}', (recalls[i], precisions[i]), |
| textcoords="offset points", xytext=(8, 5), |
| fontsize=9, color=COLOR_TEXT) |
|
|
| |
| ax.plot(1.0, 1.0, 's', color=COLOR_BASELINE, markersize=12, |
| label="Baseline (circuit breaker)") |
|
|
| ax.set_xlabel("Recall", fontsize=12, color=COLOR_TEXT) |
| ax.set_ylabel("Precision", fontsize=12, color=COLOR_TEXT) |
| ax.set_title("Precision-Recall Curve (BTC May 19, 2021 Crash)", |
| fontsize=13, fontweight='bold', color=COLOR_TEXT) |
| ax.legend(loc='upper left', frameon=False, fontsize=10) |
| ax.spines['top'].set_visible(False) |
| ax.spines['right'].set_visible(False) |
| ax.spines['left'].set_color(COLOR_GRID) |
| ax.spines['bottom'].set_color(COLOR_GRID) |
| ax.tick_params(colors=COLOR_TEXT) |
| ax.yaxis.grid(True, linestyle='--', alpha=0.3, color=COLOR_GRID) |
| ax.set_axisbelow(True) |
| ax.set_xlim(-0.05, 1.05) |
| ax.set_ylim(-0.05, 1.05) |
|
|
| fig.savefig(out_path, dpi=200, facecolor='white') |
| plt.close(fig) |
| logger.info("Saved PR curve to %s", out_path) |
|
|
|
|
| def plot_ttd_histogram(best_result: dict, out_path: Path) -> None: |
| """Plot 2: TTD distribution histogram.""" |
| ttd_ms = best_result["ttd_ms"] |
| if not ttd_ms: |
| logger.warning("No TTD data to plot") |
| return |
|
|
| fig, ax = plt.subplots(figsize=(8, 4.5), constrained_layout=True) |
|
|
| |
| ttd_s = [t / 1000.0 for t in ttd_ms] |
|
|
| ax.hist(ttd_s, bins=20, color=COLOR_ACCENT, alpha=0.7, |
| edgecolor='white', linewidth=0.8) |
|
|
| |
| ax.axvline(0, color=COLOR_BASELINE, linewidth=2, linestyle='--', |
| label="Crash moment (price dislocation)") |
|
|
| |
| median_ttd = np.median(ttd_s) |
| ax.axvline(median_ttd, color=COLOR_GOOD, linewidth=2, linestyle='-', |
| label=f"Median TTD: {median_ttd:.2f}s (early warning)") |
|
|
| ax.set_xlabel("Time-to-Detect (seconds)\n[negative = before crash]", |
| fontsize=11, color=COLOR_TEXT) |
| ax.set_ylabel("Number of alerts", fontsize=11, color=COLOR_TEXT) |
| ax.set_title("Early-Warning Time Distribution (TCN Detector)", |
| fontsize=13, fontweight='bold', color=COLOR_TEXT) |
| ax.legend(loc='upper left', frameon=False, fontsize=9) |
| ax.spines['top'].set_visible(False) |
| ax.spines['right'].set_visible(False) |
| ax.spines['left'].set_color(COLOR_GRID) |
| ax.spines['bottom'].set_color(COLOR_GRID) |
| ax.tick_params(colors=COLOR_TEXT) |
| ax.yaxis.grid(True, linestyle='--', alpha=0.3, color=COLOR_GRID) |
| ax.set_axisbelow(True) |
|
|
| |
| n_before = sum(1 for t in ttd_s if t > 0) |
| n_after = sum(1 for t in ttd_s if t <= 0) |
| ax.text(0.98, 0.95, f"{n_before} alerts BEFORE crash\n{n_after} alerts AFTER crash", |
| transform=ax.transAxes, fontsize=9, va='top', ha='right', |
| bbox=dict(boxstyle='round', facecolor=COLOR_BG, alpha=0.8)) |
|
|
| fig.savefig(out_path, dpi=200, facecolor='white') |
| plt.close(fig) |
| logger.info("Saved TTD histogram to %s", out_path) |
|
|
|
|
| def plot_alert_timeline(scores_df: pd.DataFrame, crashes: list, threshold: float, |
| out_path: Path) -> None: |
| """Plot 3: Alert timeline overlaid on price chart.""" |
| fig, ax = plt.subplots(figsize=(12, 5), constrained_layout=True) |
|
|
| |
| t0 = scores_df["timestamp_ms"].min() |
| times_s = (scores_df["timestamp_ms"] - t0) / 1000.0 |
| prices = scores_df["mid_price"].values |
|
|
| |
| ax.plot(times_s, prices, color=COLOR_BASELINE, linewidth=0.8, alpha=0.7, |
| label="BTC mid-price") |
|
|
| |
| alerts = scores_df[scores_df["score"] >= threshold] |
| if len(alerts) > 0: |
| alert_times = (alerts["timestamp_ms"] - t0) / 1000.0 |
| alert_prices = alerts["mid_price"].values |
| ax.scatter(alert_times, alert_prices, color=COLOR_ACCENT, s=30, |
| zorder=5, label=f"TCN alerts (τ={threshold})") |
|
|
| |
| for crash in crashes: |
| start_s = (crash.start_ts - t0) / 1000.0 |
| end_s = (crash.end_ts - t0) / 1000.0 |
| ax.axvspan(start_s, end_s, alpha=0.15, color=COLOR_WARN, |
| label="Crash window" if crash == crashes[0] else "") |
|
|
| ax.set_xlabel("Time (seconds from start)", fontsize=11, color=COLOR_TEXT) |
| ax.set_ylabel("Price (USD)", fontsize=11, color=COLOR_TEXT) |
| ax.set_title("Alert Timeline — TCN Detector vs BTC Price (May 19, 2021)", |
| fontsize=13, fontweight='bold', color=COLOR_TEXT) |
| ax.legend(loc='upper right', frameon=False, fontsize=9) |
| ax.spines['top'].set_visible(False) |
| ax.spines['right'].set_visible(False) |
| ax.spines['left'].set_color(COLOR_GRID) |
| ax.spines['bottom'].set_color(COLOR_GRID) |
| ax.tick_params(colors=COLOR_TEXT) |
| ax.yaxis.grid(True, linestyle='--', alpha=0.2, color=COLOR_GRID) |
| ax.set_axisbelow(True) |
|
|
| fig.savefig(out_path, dpi=200, facecolor='white') |
| plt.close(fig) |
| logger.info("Saved alert timeline to %s", out_path) |
|
|
|
|
| def plot_cascade_funnel(sweep_results: list, out_path: Path) -> None: |
| """Plot 4: Cascade funnel (simulated from threshold sweep).""" |
| fig, ax = plt.subplots(figsize=(8, 4.5), constrained_layout=True) |
|
|
| stages = ["Total ticks\n(500K)", "TCN scored\n(499.8K)", "Score > 0.1", |
| "Score > 0.3", "Score > 0.5", "Score > 0.7"] |
| counts = [500000, 499800] |
| for r in sweep_results: |
| if r["threshold"] in [0.1, 0.3, 0.5, 0.7]: |
| counts.append(r["alerts"]) |
|
|
| |
| while len(counts) < 6: |
| counts.append(0) |
|
|
| colors_bar = [COLOR_BASELINE, COLOR_BASELINE, COLOR_WARN, |
| COLOR_ACCENT, COLOR_ACCENT, COLOR_GOOD] |
|
|
| bars = ax.barh(range(len(stages)), counts, color=colors_bar, alpha=0.8, |
| edgecolor='white', linewidth=0.8) |
|
|
| ax.set_yticks(range(len(stages))) |
| ax.set_yticklabels(stages, fontsize=10) |
| ax.invert_yaxis() |
| ax.set_xlabel("Number of ticks / alerts", fontsize=11, color=COLOR_TEXT) |
| ax.set_title("Detection Cascade Funnel", |
| fontsize=13, fontweight='bold', color=COLOR_TEXT) |
|
|
| |
| for bar, count in zip(bars, counts): |
| ax.text(bar.get_width() + max(counts) * 0.01, bar.get_y() + bar.get_height() / 2, |
| f'{count:,}', va='center', fontsize=9, color=COLOR_TEXT) |
|
|
| ax.spines['top'].set_visible(False) |
| ax.spines['right'].set_visible(False) |
| ax.spines['left'].set_color(COLOR_GRID) |
| ax.spines['bottom'].set_color(COLOR_GRID) |
| ax.tick_params(colors=COLOR_TEXT) |
|
|
| fig.savefig(out_path, dpi=200, facecolor='white') |
| plt.close(fig) |
| logger.info("Saved cascade funnel to %s", out_path) |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description="Generate threshold sweep + plots") |
| parser.add_argument("--data", required=True, help="Parquet file of crash data") |
| parser.add_argument("--model", required=True, help="Trained TCN model path") |
| parser.add_argument("--out", default="results/plots/", help="Output directory") |
| parser.add_argument("--max-ticks", type=int, default=500_000) |
| parser.add_argument("--thresholds", default="0.1,0.2,0.3,0.4,0.5,0.6,0.7,0.8,0.9", |
| help="Comma-separated thresholds to sweep") |
| args = parser.parse_args() |
|
|
| device = "cuda" if torch.cuda.is_available() else "cpu" |
|
|
| |
| df = load_parquet(args.data) |
| logger.info("Loaded %d ticks", len(df)) |
|
|
| |
| model = load_trained_tcn(args.model, device=device) |
|
|
| |
| logger.info("Scoring all ticks...") |
| scores_df = score_all_ticks(model, df, max_ticks=args.max_ticks, device=device) |
|
|
| |
| ticks = list(df_to_ticks(df.iloc[np.linspace(0, len(df) - 1, |
| min(args.max_ticks, len(df)), dtype=int)], |
| symbol="LABEL")) |
| crashes = label_crashes(ticks, drop_threshold_pct=BASELINE_DROP_PCT, |
| window_ms=BASELINE_WINDOW_MS) |
| logger.info("Found %d ground-truth crash windows", len(crashes)) |
|
|
| |
| thresholds = [float(t) for t in args.thresholds.split(",")] |
| sweep_results = [] |
| logger.info("\n" + "=" * 70) |
| logger.info("THRESHOLD SWEEP") |
| logger.info("=" * 70) |
| logger.info("%-10s %-8s %-8s %-8s %-8s %-10s", |
| "Threshold", "Alerts", "TP", "FP", "Prec", "Recall") |
| logger.info("-" * 70) |
|
|
| for t in thresholds: |
| result = evaluate_at_threshold(scores_df, crashes, t) |
| sweep_results.append(result) |
| logger.info("%-10.1f %-8d %-8d %-8d %-8.3f %-10.3f", |
| t, result["alerts"], result["true_positives"], |
| result["false_positives"], result["precision"], result["recall"]) |
|
|
| logger.info("=" * 70) |
|
|
| |
| best_f1 = max(sweep_results, key=lambda r: r["f1"]) |
| logger.info("Best F1: threshold=%.1f, F1=%.3f, precision=%.3f, recall=%.3f, TTD=%.1fms", |
| best_f1["threshold"], best_f1["f1"], |
| best_f1["precision"], best_f1["recall"], best_f1["median_ttd_ms"]) |
|
|
| |
| out_dir = Path(args.out) |
| out_dir.mkdir(parents=True, exist_ok=True) |
|
|
| logger.info("\nGenerating plots...") |
|
|
| |
| plot_pr_curve(sweep_results, out_dir / "pr_curve.png") |
|
|
| |
| plot_ttd_histogram(best_f1, out_dir / "ttd_histogram.png") |
|
|
| |
| plot_alert_timeline(scores_df, crashes, best_f1["threshold"], |
| out_dir / "alert_timeline.png") |
|
|
| |
| plot_cascade_funnel(sweep_results, out_dir / "cascade_funnel.png") |
|
|
| |
| sweep_path = out_dir / "threshold_sweep.json" |
| with open(sweep_path, "w") as f: |
| json.dump([{k: v for k, v in r.items() if k != "ttd_ms"} for r in sweep_results], |
| f, indent=2) |
| logger.info("Saved sweep results to %s", sweep_path) |
|
|
| logger.info("\n" + "=" * 70) |
| logger.info("ALL PLOTS GENERATED") |
| logger.info(" Output: %s", out_dir.resolve()) |
| logger.info(" Files:") |
| logger.info(" pr_curve.png — Precision-Recall curve") |
| logger.info(" ttd_histogram.png — TTD distribution") |
| logger.info(" alert_timeline.png — Alerts overlaid on price chart") |
| logger.info(" cascade_funnel.png — Cascade pass-through funnel") |
| logger.info(" threshold_sweep.json — Raw sweep data") |
| logger.info("=" * 70) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|