In this tutorial, we’ll rebuild that experience using historical tick data. We’ll take a full AAPL trading session from EODHD, normalize more than one million trades into a deterministic event tape, and replay them according to their original timing through a controllable market clock.

Along the way, we’ll add adjustable playback speeds, pause and resume controls, seeking, and a FastAPI service that exposes the controls through REST while streaming trades over WebSockets.

We’ll also build a separate consumer that calculates rolling VWAP and market state only from the events it receives. By the end, we’ll have a complete local replay system that can feed an already-finished trading day back to event-driven software as a timed stream, while correctly rebuilding downstream state after seeks and validating the result with automated tests.

Table of Contents

• Prerequisites

• What We’re Building

• Set Up the Python Project

• Download a Full Trading Session from EODHD

  • Create replay/config.py

  • Create replay/loader.py

• Normalize Tick Data into a Replay Tape

  • Create replay/events.py

  • Create a Smaller Tape for Benchmarks and Tests

• Build the Historical Replay Clock

  • Create replay/clock.py

  • Benchmark the Replay Clock

• Add Playback Controls with a Replay Session

  • Create replay/session.py

• Expose the Replay with FastAPI and WebSockets

  • Create api/server.py

  • Create api/run.py

• Build a Stateful WebSocket Consumer

  • Create consumer/consumer.py

• Make Seeking State-Safe

  • Reset and Warm Up the Consumer

  • Check the Rebuilt State

• Replay the Full AAPL Trading Day

  • Run the Full Replay

• Test the Replay Engine

  • Create tests/testreplay.py

  • Verify State Reconstruction Independently

  • Configure pytest

• Conclusion

Prerequisites

Before starting, make sure you have:

• Python 3.10 or later installed.

• An EODHD API key with access to the historical tick-data endpoint. You can create a developer account from the EODHD pricing page.

• A terminal and code editor.

• Basic Python knowledge, including functions, classes, dictionaries, and working with packages.

• Basic familiarity with HTTP and WebSockets. You don't need prior FastAPI experience.

• Enough local disk space to store the downloaded raw tick data and processed replay tapes. The full AAPL session used in this tutorial contains more than one million trade records.

The shell commands in this tutorial use Unix-style syntax, so they work directly on macOS and Linux. On Windows, you can run them through WSL, Git Bash, or use the equivalent PowerShell commands.

What We’re Building

Before touching the code, it helps to see the full system once. The replay engine will take historical trades from EODHD, convert them into a consistent internal format, restore their timing, and stream them to a separate consumer as if the trading day were unfolding again.

The complete flow looks like this:

Each layer has one job. The loader retrieves and preserves the raw historical session. The normalizer validates those records and turns them into a deterministic replay tape. The clock maps historical timestamps onto wall-clock time, while the replay session adds controls such as start, pause, resume, speed changes, seek, and stop.

FastAPI sits around that replay engine. REST endpoints form the control plane, while a WebSocket carries the actual trade and replay-control events. On the other side, the consumer maintains its own rolling state only from what reaches it through that stream.

We’ll keep those responsibilities separated in the project structure:

market-time-machine/
├── data/
│   ├── raw/
│   └── processed/
├── replay/
│   ├── __init__.py
│   ├── config.py
│   ├── loader.py
│   ├── events.py
│   ├── clock.py
│   └── session.py
├── api/
│   ├── __init__.py
│   ├── server.py
│   └── run.py
├── consumer/
│   ├── __init__.py
│   └── consumer.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_replay.py
├── .env
├── .gitignore
└── pytest.ini

The important rule for the whole build is simple: the consumer should know only what has already arrived through the replay stream. It should never read ahead from the historical tape. That constraint is what makes timing, pause/resume behavior, and state reconstruction after a seek worth implementing correctly.

Set Up the Python Project

Start by creating the project directories and installing the packages we’ll use for data retrieval, replay timing, the API layer, WebSocket communication, and testing.

mkdir -p market-time-machine/data/raw
mkdir -p market-time-machine/data/processed
mkdir -p market-time-machine/replay
mkdir -p market-time-machine/api
mkdir -p market-time-machine/consumer
mkdir -p market-time-machine/tests

cd market-time-machine

pip install requests fastapi "uvicorn[standard]" websockets httpx python-dotenv numpy pytest pytest-asyncio

Create empty __init__.py files inside replay, api, consumer, and tests so Python treats each directory as a package:

replay/__init__.py
api/__init__.py
consumer/__init__.py
tests/__init__.py

We’ll fetch the historical trades from EODHD, so create a .env file in the project root and store your API key there:

The downloaded session will also be fairly large, so neither the credentials nor the local market-data files should be committed. Create .gitignore:

.env
data/
__pycache__/
*.pyc
.ipynb_checkpoints/

Note: If you don’t have an EODHD API key, you can easily get it by opening an EODHD developer account.

After this setup, the project should look like this:

market-time-machine/
├── data/
│   ├── raw/
│   └── processed/
├── replay/
│   └── __init__.py
├── api/
│   └── __init__.py
├── consumer/
│   └── __init__.py
├── tests/
│   └── __init__.py
├── .env
└── .gitignore

The raw/ directory will preserve the responses received from EODHD, while processed/ will hold the normalized replay tapes we build from them.

Download a Full Trading Session from EODHD

The replay engine needs a complete trading session before it can restore any sense of time. We’ll use EODHD’s historical tick API to retrieve AAPL trades for July 15, 2026, but keep the retrieval layer separate from everything related to replay.

Two files handle this part of the project:

market-time-machine/
└── replay/
    ├── __init__.py
    ├── config.py
    └── loader.py

config.py keeps the shared API, path, and market-session settings in one place. loader.py uses those settings to retrieve the session and preserve the raw responses under data/raw/.

Create replay/config.py

Add the following:

import os
from pathlib import Path
from dotenv import load_dotenv

ROOT = Path(__file__).resolve().parent.parent
load_dotenv(ROOT / ".env")

TOKEN = os.environ.get("EODHD_API_TOKEN")
TICKS_URL = "https://eodhd.com/api/ticks/"

RAW = ROOT / "data" / "raw"
PROCESSED = ROOT / "data" / "processed"

MARKET_TZ = "America/New_York"
OPEN = "09:30:00"
CLOSE = "16:00:00"

MAX_LIMIT = 10_000
MIN_WINDOW_S = 1
CLOSE_GRACE_S = 5

FIELDS = ("mkt", "price", "seq", "shares", "sl", "sub_mkt", "ts")
NON_LAST_SALE = frozenset("IWVT47")

def token():
    if not TOKEN:
        raise RuntimeError("EODHD_API_TOKEN not set")
    return TOKEN

def redact(text):
    return str(text).replace(TOKEN, "<TOKEN>") if TOKEN else str(text)

The regular US equity session is defined in America/New_York rather than with fixed UTC timestamps. That matters because the UTC equivalent of 09:30 changes with daylight saving time.

We also extend the request window five seconds beyond 16:00 with CLOSE_GRACE_S. The session used in this tutorial contains closing activity immediately after 16:00:00, so the grace window keeps those records inside the download.

Create replay/loader.py

A single large request is not a safe way to retrieve a dense tick-data session. Activity changes substantially throughout the day, and any request that reaches the configured 10,000-record limit could represent a truncated interval.

Instead, the loader will adjust its request window based on the density of the previous response.

Create replay/loader.py:

import json, time
from datetime import datetime
from zoneinfo import ZoneInfo

import requests

from . import config

def fetch(symbol, frm, to, limit=None):
    limit = limit or config.MAX_LIMIT

    r = requests.get(config.TICKS_URL, timeout=180, params={
        "s": symbol,
        "from": frm,
        "to": to,
        "limit": limit,
        "api_token": config.token(),
        "fmt": "json"
    })

    if r.status_code != 200:
        raise RuntimeError(
            f"HTTP {r.status_code} {config.redact(r.text[:200])}"
        )

    return r.json()

def bounds(date_str, grace=None):
    grace = config.CLOSE_GRACE_S if grace is None else grace
    tz = ZoneInfo(config.MARKET_TZ)
    d = datetime.strptime(date_str, "%Y-%m-%d").date()

    def at(hms):
        h, m, s = map(int, hms.split(":"))
        return datetime(
            d.year, d.month, d.day, h, m, s, tzinfo=tz
        ).timestamp()

    return int(at(config.OPEN)), int(at(config.CLOSE)) + grace

def fetch_session(symbol, date_str, tag="session", window=None,
                  force=False, verbose=True):

    raw = config.RAW / f"{symbol}_{date_str}_{tag}.jsonl"
    man = config.RAW / f"{symbol}_{date_str}_{tag}.manifest.json"

    if raw.exists() and man.exists() and not force:
        m = json.loads(man.read_text())
        print(f"cached {raw.name}: {m['ticks']:,} ticks")
        return m, raw

    start, end = window or bounds(date_str)
    cursor, win = start, 30

    total = pages = retries = 0
    first_ts = last_ts = None
    seen_fields = set()
    t0 = time.perf_counter()

    with open(raw, "w") as fh:
        while cursor < end:
            b = min(cursor + win, end)
            span = b - cursor

            payload = fetch(symbol, cursor, b)
            n = len(payload.get("ts", []))

            if n >= config.MAX_LIMIT:
                if span <= config.MIN_WINDOW_S:
                    raise RuntimeError(
                        f"second {cursor} has >= {config.MAX_LIMIT} ticks "
                        "and cannot be paginated"
                    )

                win = max(1, span // 2)
                retries += 1
                continue

            if n:
                seen_fields.update(payload.keys())

                if first_ts is None:
                    first_ts = payload["ts"][0]

                last_ts = payload["ts"][-1]

                fh.write(json.dumps({
                    "from": cursor,
                    "to": b,
                    "payload": payload
                }) + "\n")

            total += n
            pages += 1
            cursor = b

            density = n / span if span else 0
            win = int(min(
                1800,
                max(1, config.MAX_LIMIT * 0.75 / max(density, 0.01))
            ))

            if verbose and pages % 20 == 0:
                pct = 100 * (cursor - start) / (end - start)
                print(f"{pct:5.1f}% {total:,} ticks")

    m = {
        "symbol": symbol,
        "date": date_str,
        "tag": tag,
        "ticks": total,
        "pages": pages,
        "retries": retries,
        "window_from_utc": start,
        "window_to_utc": end,
        "first_timestamp_ms": first_ts,
        "last_timestamp_ms": last_ts,
        "fields": sorted(seen_fields),
        "elapsed_s": round(time.perf_counter() - t0, 1),
        "api_calls": pages * 10
    }

    man.write_text(json.dumps(m, indent=2))
    return m, raw

def read_pages(path):
    with open(path) as fh:
        for line in fh:
            if line.strip():
                yield json.loads(line)

The loader starts with a 30-second window. If that interval reaches the record ceiling, it retries with a smaller one instead of accepting a potentially incomplete response. For quieter periods, the next window can expand up to 30 minutes.

Each accepted response is written directly to JSONL before any normalization takes place. A manifest is stored alongside it with the session bounds, tick count, timestamps, observed fields, and retrieval statistics.

Now fetch the full AAPL session:

from replay.loader import fetch_session

SYMBOL = "AAPL"
DATE = "2026-07-15"

print("=== fullday ===")

manifest, raw_path = fetch_session(
    SYMBOL,
    DATE,
    tag="fullday"
)

print(
    f" window {manifest['window_from_utc']}..{manifest['window_to_utc']} | "
    f"{manifest['ticks']:,} ticks, {manifest['pages']} pages, "
    f"{manifest['retries']} retries | "
    f"{manifest['elapsed_s']}s, {manifest['api_calls']} metered api calls"
)

print(
    f" first_ts {manifest['first_timestamp_ms']} "
    f"last_ts {manifest['last_timestamp_ms']}"
)

print(" fields:", manifest["fields"])

The final clean run reused the already downloaded session and produced:

We now have 1,032,411 raw trade records covering the full regular session and closing grace window. The 145 accepted pages and 16 retries also show why a fixed request window would have been a weak assumption for tick data this dense.

These records are still stored exactly as they came from EODHD, though. Before the replay engine can use them, they need to become a deterministic internal event sequence.

Normalize Tick Data into a Replay Tape

The loader gives us the complete session, but the replay engine shouldn't work directly with EODHD’s raw response format. The tick endpoint returns fields such as timestamps, prices, sizes, sequence numbers, and market codes as parallel arrays.

Before replaying them, we need to verify those arrays line up, establish a deterministic event order, remove duplicates, and convert the result into one internal format.

That logic belongs in replay/events.py:

market-time-machine/
└── replay/
    ├── config.py
    ├── loader.py
    └── events.py

We’ll use two objects here. TradeEvent represents a single trade in the format that will eventually travel over the WebSocket. TradeTape stores the full session efficiently in columnar NumPy arrays and materializes individual TradeEvent objects only when they are needed.

Create replay/events.py

Create replay/events.py with:

from dataclasses import dataclass
import numpy as np

from . import config
from .loader import read_pages

@dataclass(frozen=True)
class TradeEvent:
    symbol: str
    timestamp_ms: int
    price: float
    size: int
    sequence: int
    market: str
    sub_market: str
    sale_condition: str
    source: str = "replay"

    def to_wire(self):
        sl = self.sale_condition

        return {
            "type": "trade",
            "symbol": self.symbol,
            "timestamp_ms": self.timestamp_ms,
            "price": self.price,
            "size": self.size,
            "sequence": self.sequence,
            "source": self.source,
            "metadata": {
                "market": self.market,
                "sub_market": self.sub_market or None,
                "sale_condition": sl,
                "odd_lot": "I" in sl,
                "zero_size": self.size == 0,
                "last_sale_eligible": not (
                    set(sl) & config.NON_LAST_SALE
                )
            }
        }


class TradeTape:
    def __init__(self, symbol, ts, price, size, seq, mkt, sub, sl):
        self.symbol = symbol
        self.ts = ts
        self.price = price
        self.size = size
        self.seq = seq
        self.mkt = mkt
        self.sub = sub
        self.sl = sl

    def __len__(self):
        return len(self.ts)

    def __getitem__(self, i):
        return TradeEvent(
            self.symbol,
            int(self.ts[i]),
            float(self.price[i]),
            int(self.size[i]),
            int(self.seq[i]),
            str(self.mkt[i]),
            str(self.sub[i]),
            str(self.sl[i])
        )

    def index_at(self, ts_ms):
        return int(np.searchsorted(self.ts, ts_ms, side="left"))

    def span(self):
        if not len(self):
            return None, None

        return int(self.ts[0]), int(self.ts[-1])

    def save(self, path):
        np.savez_compressed(
            path,
            ts=self.ts,
            price=self.price,
            size=self.size,
            seq=self.seq,
            mkt=self.mkt,
            sub=self.sub,
            sl=self.sl,
            symbol=np.array([self.symbol])
        )

    @classmethod
    def load(cls, path):
        z = np.load(path, allow_pickle=False)

        return cls(
            str(z["symbol"][0]),
            z["ts"],
            z["price"],
            z["size"],
            z["seq"],
            z["mkt"],
            z["sub"],
            z["sl"]
        )


def normalize(raw_path, symbol, verbose=True):
    cols = {k: [] for k in config.FIELDS}
    pages = 0

    for page in read_pages(raw_path):
        pages += 1
        p = page["payload"]

        lens = {k: len(p.get(k, [])) for k in config.FIELDS}

        if len(set(lens.values())) != 1:
            raise ValueError(
                f"ragged page {page['from']}: {lens}"
            )

        for k in config.FIELDS:
            cols[k].extend(p[k])

    ts = np.asarray(cols["ts"], dtype=np.int64)
    price = np.asarray(cols["price"], dtype=np.float64)
    size = np.asarray(cols["shares"], dtype=np.int64)
    seq = np.asarray(cols["seq"], dtype=np.int64)
    mkt = np.asarray(cols["mkt"], dtype=str)
    sub = np.asarray(cols["sub_mkt"], dtype=str)
    sl = np.asarray(cols["sl"], dtype=str)

    raw_n = len(ts)

    def arrays(mask):
        return tuple(
            a[mask]
            for a in (ts, price, size, seq, mkt, sub, sl)
        )

    keep = (
        (ts > 0)
        & np.isfinite(price)
        & (price > 0)
        & (size >= 0)
    )

    ts, price, size, seq, mkt, sub, sl = arrays(keep)

    order = np.lexsort((seq, ts))
    ts, price, size, seq, mkt, sub, sl = arrays(order)

    dup = np.zeros(len(ts), dtype=bool)

    if len(ts) > 1:
        dup[1:] = (
            (ts[1:] == ts[:-1])
            & (seq[1:] == seq[:-1])
        )

    ts, price, size, seq, mkt, sub, sl = arrays(~dup)

    tape = TradeTape(
        symbol,
        ts,
        price,
        size,
        seq,
        mkt,
        sub,
        sl
    )

    odd = sum("I" in str(s) for s in sl)
    elig = sum(
        not (set(str(s)) & config.NON_LAST_SALE)
        for s in sl
    )

    rep = {
        "pages": pages,
        "raw": raw_n,
        "kept": len(ts),
        "dropped": raw_n - len(ts) - int(dup.sum()),
        "dupes": int(dup.sum()),
        "zero_size": int((size == 0).sum()),
        "odd_lot": int(odd),
        "last_sale_eligible": int(elig),
        "seq_strict": bool(
            np.all(seq[1:] > seq[:-1])
        ) if len(seq) > 1 else True,
        "span": tape.span()
    }

    if verbose:
        n = max(1, len(ts))

        print(
            f"{rep['raw']:,} raw -> {rep['kept']:,} kept "
            f"({rep['dupes']} dupes, {rep['dropped']} invalid)"
        )

        print(
            f"zero-size {100*rep['zero_size']/n:.1f}% | "
            f"odd-lot {100*odd/n:.1f}% | "
            f"last-sale-eligible {100*elig/n:.1f}%"
        )

        print(
            f"seq strictly increasing: {rep['seq_strict']}"
        )

    return tape, rep

The first validation happens before we construct any trades. Since the source fields arrive as parallel arrays, every field on a page must contain the same number of observations. Otherwise, combining them could silently attach one trade’s price to another trade’s timestamp.

After that, the arrays are converted to NumPy, basic invalid records are removed, and the trades are sorted by (timestamp, sequence). The timestamp gives us chronological order, while the sequence number provides deterministic ordering when several trades share the same millisecond.

Exact duplicates with the same timestamp and sequence are then removed. TradeTape keeps the resulting columns as arrays rather than allocating more than a million permanent Python objects, which keeps the full-day session considerably lighter in memory.

Now normalize the raw session and save it under data/processed/:

import json

from replay import config
from replay.events import normalize

tape, report = normalize(raw_path, SYMBOL)

tape.save(
    config.PROCESSED / f"{SYMBOL}_{DATE}_fullday.npz"
)

lo, hi = tape.span()

print(
    f"span {lo}..{hi} "
    f"({(hi-lo)/3_600_000:.2f} market hours)"
)

print("first 3 normalized events:")

for i in range(3):
    print(json.dumps(tape[i].to_wire()))

The actual normalization run produced:

Only two duplicate records disappear from more than one million raw observations, and none fail the basic timestamp, price, or size checks. More importantly for replay, the normalized sequence is strictly increasing.

Create a Smaller Tape for Benchmarks and Tests

The full-day tape will power the final replay. For the timing benchmark and automated tests, though, we don't need to run through all 6.5 hours every time.

We’ll derive a 15-minute slice from 12:00 to 12:15 ET directly from the normalized full-day tape:

from datetime import datetime
from zoneinfo import ZoneInfo

from replay.events import TradeTape

tz = ZoneInfo(config.MARKET_TZ)

quiet_start = int(
    datetime(
        2026, 7, 15, 12, 0,
        tzinfo=tz
    ).timestamp() * 1000
)

quiet_end = quiet_start + 15 * 60_000

i = tape.index_at(quiet_start)
j = tape.index_at(quiet_end)

quiet_tape = TradeTape(
    tape.symbol,
    tape.ts[i:j],
    tape.price[i:j],
    tape.size[i:j],
    tape.seq[i:j],
    tape.mkt[i:j],
    tape.sub[i:j],
    tape.sl[i:j]
)

quiet_tape.save(config.PROCESSED / f"{SYMBOL}_{DATE}_quiet15m.npz")

We now have two processed tapes: the full session for the end-to-end replay and a smaller real market interval for repeatable timing and control tests.

Build the Historical Replay Clock

We now have a deterministic sequence of trades, but there's still nothing making those trades behave like a market stream. If we simply iterate through the tape, Python will process the session as quickly as the machine allows.

The replay clock solves that by mapping historical market time onto real wall-clock time. It also lets us change the playback speed without changing the original timestamps.

A naïve version might sleep for the historical gap between every pair of trades:

gap = (next_ts - current_ts) / 1000
await asyncio.sleep(gap / speed)

At 10x, a 500 ms historical gap becomes 50 ms. At 100x, it becomes 5 ms.

The problem is that asyncio.sleep() only guarantees that execution will resume after the requested delay. If each sleep wakes slightly late and the next delay is measured from that late wake-up, those errors can accumulate across a long replay.

Instead, we’ll anchor the whole replay to time.monotonic():

historical elapsed time
        ÷
replay speed
        +
wall-clock start
        =
target wall-clock time

Every event is therefore scheduled relative to the same anchor rather than relative to when the previous event happened to finish.

Create replay/clock.py

Add the clock to the replay package:

market-time-machine/
└── replay/
    ├── config.py
    ├── loader.py
    ├── events.py
    └── clock.py

Create replay/clock.py:

import asyncio, time
import numpy as np

MIN_SLEEP_S = 0.0005

class ReplayClock:
    def __init__(self, start_ms, speed=1.0):
        self.speed = float(speed)
        self._anchor_ms = float(start_ms)
        self._anchor_wall = None
        self.running = False
        self.epoch = 0

    def start(self):
        self._anchor_wall = time.monotonic()
        self.running = True
        return self

    def now_ms(self, now=None):
        if not self.running or self._anchor_wall is None:
            return self._anchor_ms

        now = now if now is not None else time.monotonic()

        return (
            self._anchor_ms
            + (now - self._anchor_wall) * 1000 * self.speed
        )

    def wall_for(self, ms):
        return (
            self._anchor_wall
            + (ms - self._anchor_ms) / 1000 / self.speed
        )

    def _reanchor(self, ms):
        self._anchor_ms = float(ms)
        self._anchor_wall = time.monotonic()
        self.epoch += 1

    def set_speed(self, speed):
        self._reanchor(self.now_ms())
        self.speed = float(speed)

    def pause(self):
        if self.running:
            self._anchor_ms = self.now_ms()
            self.running = False

    def resume(self):
        if not self.running:
            self._anchor_wall = time.monotonic()
            self.running = True
            self.epoch += 1

    def seek(self, ms):
        self._reanchor(ms)


def new_stats(speed):
    return {
        "emitted": 0,
        "batches": 0,
        "lateness": [],
        "dropped": 0,
        "speed": speed,
        "wall0": None,
        "market0": None,
        "market1": None
    }


def summarize(st):
    if not st["lateness"]:
        return {
            "emitted": st["emitted"],
            "batches": st["batches"]
        }

    a = np.asarray(st["lateness"])

    wall = (
        time.monotonic() - st["wall0"]
        if st["wall0"] else 0.0
    )

    mkt = (
        (st["market1"] - st["market0"]) / 1000
        if st["market0"] is not None else 0.0
    )

    ok = st["dropped"] == 0 and wall > 0
    realized = round(mkt / wall, 2) if ok else None

    return {
        "emitted": st["emitted"],
        "batches": st["batches"],
        "mean_batch": round(
            st["emitted"] / max(1, st["batches"]), 1
        ),
        "market_s": round(mkt, 3),
        "wall_s": round(wall, 3),
        "requested_speed": st["speed"],
        "realized_speed": realized,
        "speed_error_pct": (
            round(
                100 * (realized - st["speed"]) / st["speed"],
                2
            )
            if ok else None
        ),
        "lateness_p50_ms": round(
            float(np.percentile(a, 50)), 2
        ),
        "lateness_p95_ms": round(
            float(np.percentile(a, 95)), 2
        ),
        "lateness_max_ms": round(
            float(a.max()), 2
        ),
        "reanchor_batches_dropped": st["dropped"]
    }


async def replay_batches(
    tape,
    clock,
    start,
    stats,
    max_batch=4096
):
    i, n = start, len(tape)
    last_epoch = clock.epoch

    if stats["wall0"] is None:
        stats["wall0"] = time.monotonic()
        stats["market0"] = int(tape.ts[start])

    while i < n:
        if not clock.running:
            await asyncio.sleep(0.005)
            continue

        now = time.monotonic()

        j = min(
            int(
                np.searchsorted(
                    tape.ts,
                    clock.now_ms(now),
                    side="right"
                )
            ),
            n,
            i + max_batch
        )

        if j > i and not clock.running:
            continue

        if j > i:
            if clock.epoch == last_epoch:
                targets = clock.wall_for(
                    tape.ts[i:j].astype(np.float64)
                )

                stats["lateness"].extend(
                    ((now - targets) * 1000).tolist()
                )
            else:
                stats["dropped"] += 1
                last_epoch = clock.epoch

            stats["emitted"] += j - i
            stats["batches"] += 1
            stats["market1"] = int(tape.ts[j - 1])

            yield i, j
            i = j
            continue

        wait = clock.wall_for(float(tape.ts[i])) - now

        await asyncio.sleep(
            wait if wait > MIN_SLEEP_S else 0
        )

now_ms() tells us where the replay currently is in historical market time. wall_for() performs the opposite conversion and tells us when a historical timestamp should become due on the machine’s monotonic clock.

Pause, resume, speed changes, and seeking can then re-anchor that mapping without modifying the underlying tape.

The other important part is batching. At high replay speeds, scheduling one sleep for every trade would create substantial overhead of its own. replay_batches() instead asks how far market time has advanced and releases all trades that are already due, up to the configured batch size.

If the event loop falls slightly behind, the next batch gets larger rather than introducing another artificial delay.

Benchmark the Replay Clock

Now load the midday tape we created in the previous section and test the first 30 seconds of market time:

import numpy as np

from replay import config
from replay.events import TradeTape
from replay.clock import (
    ReplayClock,
    replay_batches,
    new_stats,
    summarize
)

tape = TradeTape.load(
    config.PROCESSED / "AAPL_2026-07-15_quiet15m.npz"
)

end = int(
    np.searchsorted(
        tape.ts,
        tape.ts[0] + 30_000,
        side="right"
    )
)

print(
    f"{end:,} events in the first "
    "30 market seconds of AAPL quiet15m\n"
)

async def measure():
    print(
        f"{'speed':>6} {'market_s':>9} "
        f"{'wall_s':>8} {'realized':>9} "
        f"{'err_%':>7} {'p50_ms':>7} "
        f"{'p95_ms':>7} {'max_ms':>7}"
    )

    for speed in [1, 10, 50, 100]:
        clock = ReplayClock(
            tape.ts[0],
            speed
        ).start()

        st = new_stats(speed)

        async for i, j in replay_batches(
            tape,
            clock,
            0,
            st
        ):
            if j >= end:
                break

        r = summarize(st)

        print(
            f"{r['requested_speed']:>6} "
            f"{r['market_s']:>9} "
            f"{r['wall_s']:>8} "
            f"{r['realized_speed']:>9} "
            f"{r['speed_error_pct']:>7} "
            f"{r['lateness_p50_ms']:>7} "
            f"{r['lateness_p95_ms']:>7} "
            f"{r['lateness_max_ms']:>7}"
        )

await measure()

The actual run produced:

Thirty seconds of historical market time took 30.001 seconds at 1x, 3.001 seconds at 10x, 0.6 seconds at 50x, and 0.3 seconds at 100x. The realized speeds therefore stayed very close to what we requested.

The lateness values tell us how far the scheduler missed individual event deadlines. At 10x, for example, the median lateness was 0.36 ms, the 95th percentile was 1.12 ms, and the worst observation in this run was 11.75 ms.

These numbers measure the replay clock itself. They're not end-to-end WebSocket latency measurements, and this is still best-effort scheduling on Python’s event loop rather than exchange-grade timing.

Add Playback Controls with a Replay Session

The replay clock knows when trades are due, but it doesn't know where the replay currently is or whether playback should be running at all. We need another layer to own the tape, track the current cursor, manage the event queue, and coordinate controls such as start, pause, resume, speed changes, seek, and stop.

That logic belongs in replay/session.py:

market-time-machine/
└── replay/
    ├── config.py
    ├── loader.py
    ├── events.py
    ├── clock.py
    └── session.py

The distinction is useful to keep clear: the clock owns time, while the session owns state.

A replay session moves through a small set of states:

Create replay/session.py

Create replay/session.py:

import asyncio, collections, contextlib, uuid
from enum import Enum

from .clock import ReplayClock, replay_batches, new_stats, summarize

class State(str, Enum):
    CREATED, RUNNING, PAUSED, COMPLETED, STOPPED = (
        "created", "running", "paused", "completed", "stopped"
    )

class ReplaySession:
    PRIORITY = {
        "paused", "resumed", "speed_changed",
        "replay_reset", "session_stopped"
    }

    def __init__(self, tape, speed=1.0, warmup_ms=120_000, maxsize=256):
        self.id = uuid.uuid4().hex[:12]
        self.tape = tape
        self.warmup_ms = warmup_ms
        self.maxsize = maxsize

        self.state = State.CREATED
        self.cursor = 0
        self.clock = ReplayClock(tape.ts[0], speed)
        self.stats = new_stats(speed)

        self._q = collections.deque()
        self._wake = asyncio.Event()
        self._task = None
        self._epoch = 0
        self._lock = asyncio.Lock()

    def info(self):
        lo, hi = self.tape.span()

        return {
            "session_id": self.id,
            "symbol": self.tape.symbol,
            "state": self.state.value,
            "speed": self.clock.speed,
            "cursor": self.cursor,
            "total_events": len(self.tape),
            "market_ts_ms": int(
                self.tape.ts[min(self.cursor, len(self.tape)-1)]
            ),
            "session_start_ms": lo,
            "session_end_ms": hi,
            "queued": len(self._q)
        }

    def _ctrl(self, kind, **kw):
        msg = {
            "type": kind,
            "session_id": self.id,
            "source": "replay",
            **kw
        }

        if kind in self.PRIORITY:
            self._q.appendleft(msg)
        else:
            self._q.append(msg)

        self._wake.set()

    async def _put(self, msg):
        while len(self._q) >= self.maxsize:
            self._wake.set()
            await asyncio.sleep(0)

        self._q.append(msg)
        self._wake.set()

    async def _kill(self):
        t, self._task = self._task, None

        if t and not t.done():
            t.cancel()

            with contextlib.suppress(
                asyncio.CancelledError,
                Exception
            ):
                await t

    async def start(self):
        self.clock.start()
        self.state = State.RUNNING
        self._task = asyncio.create_task(self._run())

        self._ctrl(
            "session_started",
            info=self.info()
        )

        return self.info()

    async def pause(self):
        if self.state is State.RUNNING:
            async with self._lock:
                self.clock.pause()
                self.stats["dropped"] += 1
                self.state = State.PAUSED

                self._ctrl(
                    "paused",
                    market_ts_ms=self.info()["market_ts_ms"]
                )

        return self.info()

    async def resume(self):
        if self.state is State.PAUSED:
            async with self._lock:
                self.clock.resume()
                self.state = State.RUNNING

                if self._task is None or self._task.done():
                    self._task = asyncio.create_task(self._run())

                self._ctrl(
                    "resumed",
                    market_ts_ms=self.info()["market_ts_ms"]
                )

        return self.info()

    async def set_speed(self, speed):
        async with self._lock:
            old = self.clock.speed
            self.clock.set_speed(speed)
            self.stats["speed"] = speed

            self._ctrl(
                "speed_changed",
                old_speed=old,
                new_speed=speed
            )

        return self.info()

    async def seek(self, target_ms):
        was = self.state
        await self._kill()

        async with self._lock:
            idx = max(
                0,
                min(
                    self.tape.index_at(target_ms),
                    len(self.tape)-1
                )
            )

            self._epoch += 1
            self.cursor = idx
            self.state = State.PAUSED
            self.clock.pause()

            warm = max(
                0,
                self.tape.index_at(
                    int(self.tape.ts[idx]) - self.warmup_ms
                )
            )

            purged = sum(
                1 for m in self._q
                if m.get("type") == "trade"
            )

            self._q = collections.deque(
                m for m in self._q
                if m.get("type") != "trade"
            )

            self._ctrl(
                "replay_reset",
                reason="seek",
                target_timestamp_ms=int(self.tape.ts[idx]),
                warmup_from_ms=int(self.tape.ts[warm]),
                warmup_events=idx-warm,
                purged_stale_events=purged,
                epoch=self._epoch
            )

        for k in range(warm, idx):
            await self._put({
                **self.tape[k].to_wire(),
                "warmup": True
            })

        self._ctrl(
            "warmup_complete",
            market_ts_ms=int(self.tape.ts[idx])
        )

        async with self._lock:
            self.clock.seek(float(self.tape.ts[idx]))

            if was is State.RUNNING:
                self.clock.start()
                self.state = State.RUNNING
                self._task = asyncio.create_task(self._run())

        return self.info()

    async def stop(self):
        self.state = State.STOPPED
        await self._kill()

        self._ctrl(
            "session_stopped",
            info=self.info(),
            timing=summarize(self.stats)
        )

        return self.info()

    async def _run(self):
        epoch = self._epoch

        async for i, j in replay_batches(
            self.tape,
            self.clock,
            self.cursor,
            self.stats
        ):
            if self._epoch != epoch or self.state is State.STOPPED:
                return

            for k in range(i, j):
                await self._put(self.tape[k].to_wire())
                self.cursor = k+1

        if self._epoch == epoch and self.cursor >= len(self.tape):
            self.state = State.COMPLETED

            self._ctrl(
                "session_completed",
                info=self.info(),
                timing=summarize(self.stats)
            )

    async def events(self):
        while True:
            if not self._q:
                self._wake.clear()
                await self._wake.wait()
                continue

            m = self._q.popleft()
            yield m

            if m.get("type") in (
                "session_completed",
                "session_stopped"
            ):
                return

The main piece of session state is cursor, which points to the next position in the TradeTape. The producer uses replay_batches() from the clock layer, converts each due tape position into a wire-ready trade event, and places it onto the session queue.

Pausing freezes the clock without changing the cursor. Resuming gives the clock a new wall-time anchor and continues from the same historical position. A speed change works similarly: the clock first anchors itself at the current replay timestamp, then applies the new speed from that point forward.

The queue contains more than trades. Controls such as paused, resumed, speed_changed, and replay_reset also become events, which means the downstream consumer can react to changes in replay state instead of trying to infer them from the trade timestamps.

seek() is the most involved control. It stops the current producer, finds the requested position with TradeTape.index_at(), removes stale queued trades, and prepares a warmup window before playback continues. We’ll look at why that warmup is necessary once the stateful consumer is in place.

There's no separate terminal run for ReplaySession at this point. We’ll exercise these controls through the actual API and WebSocket stream once the remaining pieces of the system are connected.

Expose the Replay with FastAPI and WebSockets

The replay session now has everything needed to control historical playback, but it still exists only as a Python object. To let another program create a session, control it, and receive the resulting trade stream, we’ll put a small API layer around it.

That layer lives in a separate api/ package:

market-time-machine/
├── replay/
│   └── ...
└── api/
    ├── __init__.py
    ├── server.py
    └── run.py

We’ll use two communication paths. REST endpoints form the control plane, while one persistent WebSocket carries the event stream.

Control plane

POST /sessions
POST /sessions/{id}/start
POST /sessions/{id}/pause
POST /sessions/{id}/resume
POST /sessions/{id}/speed
POST /sessions/{id}/seek
POST /sessions/{id}/stop


Event stream

WS /sessions/{id}/stream

A command such as pause or seek therefore arrives over HTTP, while trades and replay-control events continue flowing to the consumer through the WebSocket.

Create api/server.py

Create api/server.py:

from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
from pydantic import BaseModel, Field

from replay import config
from replay.events import TradeTape
from replay.session import ReplaySession
from replay.clock import summarize

app = FastAPI(title="Market Time Machine")

SESSIONS = {}
ATTACHED = set()

class Create(BaseModel):
    symbol: str = "AAPL"
    date: str
    tag: str = "fullday"
    speed: float = Field(1.0, gt=0)
    warmup_ms: int = 120_000

class Speed(BaseModel):
    speed: float = Field(..., gt=0)

class Seek(BaseModel):
    target_timestamp_ms: int

def get(sid):
    if sid not in SESSIONS:
        raise HTTPException(404, f"no session {sid}")
    return SESSIONS[sid]

@app.post("/sessions")
async def create(b: Create):
    path = config.PROCESSED / f"{b.symbol}_{b.date}_{b.tag}.npz"

    if not path.exists():
        raise HTTPException(404, f"no tape {path.name}")

    s = ReplaySession(
        TradeTape.load(path),
        b.speed,
        b.warmup_ms
    )

    SESSIONS[s.id] = s
    return s.info()

@app.get("/sessions/{sid}")
async def info(sid: str):
    return get(sid).info()

@app.get("/sessions/{sid}/timing")
async def timing(sid: str):
    return summarize(get(sid).stats)

@app.post("/sessions/{sid}/start")
async def start(sid: str):
    return await get(sid).start()

@app.post("/sessions/{sid}/pause")
async def pause(sid: str):
    return await get(sid).pause()

@app.post("/sessions/{sid}/resume")
async def resume(sid: str):
    return await get(sid).resume()

@app.post("/sessions/{sid}/stop")
async def stop(sid: str):
    return await get(sid).stop()

@app.post("/sessions/{sid}/speed")
async def speed(sid: str, b: Speed):
    return await get(sid).set_speed(b.speed)

@app.post("/sessions/{sid}/seek")
async def seek(sid: str, b: Seek):
    return await get(sid).seek(b.target_timestamp_ms)

@app.websocket("/sessions/{sid}/stream")
async def stream(ws: WebSocket, sid: str):
    await ws.accept()

    if sid not in SESSIONS:
        return await ws.close(4004, "unknown session")

    if sid in ATTACHED:
        return await ws.close(4009, "consumer already attached")

    ATTACHED.add(sid)

    try:
        await ws.send_json({
            "type": "attached",
            "session_id": sid
        })

        async for msg in SESSIONS[sid].events():
            await ws.send_json(msg)

    except (WebSocketDisconnect, Exception):
        pass

    finally:
        ATTACHED.discard(sid)

Creating a session loads the processed .npz tape and wraps it in a ReplaySession. At this point, the API never needs to call EODHD or read the raw JSONL responses again. The replay works entirely from the normalized tape.

The REST handlers stay intentionally thin. /pause, for example, doesn't contain any pause logic of its own:

@app.post("/sessions/{sid}/pause")
async def pause(sid: str):
    return await get(sid).pause()

It simply passes the command to ReplaySession. The same pattern applies to resume, speed changes, seek, and stop. This keeps the replay behavior inside replay/ instead of coupling it to FastAPI.

The WebSocket endpoint handles the other direction. Once a consumer connects, the server forwards everything produced by session.events():

async for msg in SESSIONS[sid].events():
    await ws.send_json(msg)

That can be a normal trade:

{
  "type": "trade",
  "symbol": "AAPL",
  "timestamp_ms": 1784122200009,
  "price": 317.46,
  "size": 3,
  "sequence": 61530328,
  "source": "replay"
}

or a replay-control message:

{
  "type": "paused",
  "market_ts_ms": 1784122200009
}

Seeking will later introduce another important control event:

{
  "type": "replay_reset",
  "reason": "seek",
  "target_timestamp_ms": 1784136600030
}

The server allows one WebSocket consumer per replay session. The current queue is a FIFO handoff, not a broadcast system, so attaching multiple consumers to the same session would cause them to divide the events rather than each receiving a complete stream.

Create api/run.py

The second API file only needs to launch the FastAPI application.

Create api/run.py:

import argparse
import uvicorn

from api.server import app

if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--port", type=int, default=8765)
    a = p.parse_args()

    uvicorn.run(
        app,
        host="127.0.0.1",
        port=a.port,
        log_level="warning"
    )

Start the service from the project root:

python -m api.run --port 8765

The replay engine now has an external control interface and a WebSocket event stream. The next piece is the program on the other end of that stream: a consumer that builds market state only from the events it receives.

Build a Stateful WebSocket Consumer

The replay service can now stream historical trades, but we still need something on the other side of the WebSocket that behaves like a real downstream application.

That consumer shouldn't load the historical tape or call EODHD directly. Its entire view of the market should come from the messages arriving through the replay stream.

We’ll keep it in a separate package:

market-time-machine/
├── replay/
│   └── ...
├── api/
│   └── ...
└── consumer/
    ├── __init__.py
    └── consumer.py

For this tutorial, the consumer will maintain:

• the latest trade

• the latest last-sale-eligible trade

• cumulative volume

• a 30-second VWAP

• a 2-minute VWAP

• odd-lot and zero-size percentages

• a simple SHORT_ABOVE / SHORT_BELOW state

That final state isn't meant to be a trading strategy. We just need something genuinely stateful so we can later verify that replay controls, especially seeking, don't leave the consumer with stale market history.

Create consumer/consumer.py

Create consumer/consumer.py:

import argparse, asyncio, collections, json
import websockets

class VWAP:
    def __init__(self, window_ms):
        self.w = window_ms
        self.buf = collections.deque()
        self.pv = 0.0
        self.vol = 0.0

    def add(self, ts, px, sz):
        self.buf.append((ts, px, sz))
        self.pv += px * sz
        self.vol += sz

        cut = ts - self.w

        while self.buf and self.buf[0][0] < cut:
            _, p, s = self.buf.popleft()
            self.pv -= p * s
            self.vol -= s

        if self.vol <= 0:
            self.pv = self.vol = 0.0

    @property
    def value(self):
        return self.pv / self.vol if self.vol > 0 else None


class State:
    def __init__(self, short_ms=30_000, long_ms=120_000):
        self.short = VWAP(short_ms)
        self.long = VWAP(long_ms)

        self.last_trade = None
        self.last_sale = None
        self.signal = None

        self.n = 0
        self.vol = 0
        self.odd = 0
        self.zero = 0
        self.warming = False

    def apply(self, m):
        ts = m["timestamp_ms"]
        px = m["price"]
        sz = m["size"]
        meta = m["metadata"]

        self.short.add(ts, px, sz)
        self.long.add(ts, px, sz)

        self.last_trade = px

        if meta["last_sale_eligible"]:
            self.last_sale = px

        self.n += 1
        self.vol += sz
        self.odd += meta["odd_lot"]
        self.zero += meta["zero_size"]

        s = self.short.value
        l = self.long.value

        if s is not None and l is not None:
            self.signal = (
                "SHORT_ABOVE"
                if s > l
                else "SHORT_BELOW"
            )

    def line(self):
        f = lambda v: "--" if v is None else f"{v:.4f}"

        return (
            f"n={self.n:>7,} "
            f"vol={self.vol:>9,} "
            f"trade={f(self.last_trade):>9} "
            f"sale={f(self.last_sale):>9} "
            f"vwap30s={f(self.short.value):>9} "
            f"vwap2m={f(self.long.value):>9} "
            f"sig={self.signal or '--':<11} "
            f"odd={100*self.odd/max(1,self.n):4.1f}% "
            f"zero={100*self.zero/max(1,self.n):4.1f}%"
        )


async def run(url, every=3000):
    st = State()

    async with websockets.connect(
        url,
        max_size=None
    ) as ws:
        print("[consumer] connected", flush=True)

        async for raw in ws:
            m = json.loads(raw)
            t = m["type"]

            if t == "trade":
                st.apply(m)

                if not st.warming and st.n % every == 0:
                    print(
                        f"[consumer] {st.line()}",
                        flush=True
                    )

            elif t == "replay_reset":
                print(
                    f"[consumer] RESET -> "
                    f"{m['target_timestamp_ms']} "
                    f"({m['warmup_events']} warmup, "
                    f"{m['purged_stale_events']} purged)",
                    flush=True
                )

                st = State()
                st.warming = True

            elif t == "warmup_complete":
                st.warming = False

                print(
                    f"[consumer] WARM DONE {st.line()}",
                    flush=True
                )

            elif t in (
                "session_completed",
                "session_stopped"
            ):
                print(
                    f"[consumer] {t.upper()} "
                    f"{st.line()}",
                    flush=True
                )
                break

            else:
                print(
                    f"[consumer] {t}",
                    flush=True
                )


if __name__ == "__main__":
    p = argparse.ArgumentParser()

    p.add_argument(
        "--url",
        required=True
    )

    p.add_argument(
        "--every",
        type=int,
        default=3000
    )

    a = p.parse_args()

    asyncio.run(
        run(a.url, a.every)
    )

The rolling VWAP windows are based on market timestamps, not on the number of trades. Every incoming trade enters both windows, and observations older than 30 seconds or two minutes are removed as replay time advances.

So the consumer state evolves incrementally:

The important point is that none of this state comes from the original TradeTape. The consumer only knows about events that have crossed the WebSocket.

That works cleanly while replay time moves forward. Seeking is where things become more difficult, because moving the replay cursor without resetting the consumer would leave it carrying state from the wrong point in the trading day.

Make Seeking State-Safe

Seeking isn't just a matter of moving the replay cursor. If the consumer has already built rolling state at one point in the trading day, jumping somewhere else without resetting that state would mix two different market histories.

Suppose the consumer has reached 14:00. Its two-minute VWAP still contains trades from roughly 13:58 onward. If we simply move the replay cursor back to 13:30 and continue emitting trades, those future observations remain in memory:

The replay therefore needs to reset the downstream state and rebuild it around the new timestamp before normal playback continues.

Reset and Warm Up the Consumer

The seek() method we added to ReplaySession already handles this sequence. The important part begins by stopping the current producer and locating the requested position in the tape:

was = self.state
await self._kill()

async with self._lock:
    idx = max(
        0,
        min(
            self.tape.index_at(target_ms),
            len(self.tape)-1
        )
    )

    self._epoch += 1
    self.cursor = idx
    self.state = State.PAUSED
    self.clock.pause()

Next, it calculates a warmup point two minutes before the target:

warm = max(0, self.tape.index_at(int(self.tape.ts[idx]) - self.warmup_ms))

We use two minutes because that matches the longest rolling window maintained by the consumer. Replaying that interval is enough to reconstruct both the 30-second and two-minute VWAPs at the new position.

Before sending those warmup trades, any normal trade messages still waiting in the session queue are removed:

purged = sum(1 for m in self._q if m.get("type") == "trade")
self._q = collections.deque(m for m in self._q if m.get("type") != "trade")

The session then sends an explicit replay_reset event:

self._ctrl(
    "replay_reset",
    reason="seek",
    target_timestamp_ms=int(self.tape.ts[idx]),
    warmup_from_ms=int(self.tape.ts[warm]),
    warmup_events=idx-warm,
    purged_stale_events=purged,
    epoch=self._epoch
)

The consumer responds by discarding its current state:

elif t == "replay_reset":
    st = State()
    st.warming = True

Now the session can send the historical trades immediately preceding the target:

for k in range(warm, idx):
    await self._put({
        **self.tape[k].to_wire(),
        "warmup": True
    })

self._ctrl(
    "warmup_complete",
    market_ts_ms=int(self.tape.ts[idx])
)

These trades pass through exactly the same State.apply() logic as normal replay events, but the consumer suppresses its regular output while warming is True.

The complete seek flow is therefore:

Check the Rebuilt State

In the full-session run, we paused the replay and sought to 13:30. The first actual event at or after that requested timestamp was 1784136600030.

The consumer received:

The old consumer state is gone, and 3,456 historical trades have rebuilt the two rolling VWAP windows around the new point in the session. Normal timed playback can now resume without carrying market state across the seek boundary.

Replay the Full AAPL Trading Day

All the pieces are now connected. The full-day tape can be controlled through FastAPI, while the separate consumer sees only the trade and control events arriving over the WebSocket.

Start the replay service in the first terminal:

python -m api.run --port 8765

For the final run, we’ll start at 10x, pause the market, switch to 50x, resume, pause again, seek to 13:30, rebuild the consumer state, and finally run toward the close at 400x.

Run the Full Replay

Save the following as a temporary demo.py in the project root. This script is only the driver for the demonstration. The replay engine and consumer remain in the packages we already built.

import asyncio, os, subprocess, sys
import httpx

BASE = "http://127.0.0.1:8765"
ROOT = os.getcwd()
SEEK_1330_MS = 1784136600000

async def demo():
    async with httpx.AsyncClient(base_url=BASE, timeout=120) as c:
        r = await c.post("/sessions", json={
            "symbol": "AAPL",
            "date": "2026-07-15",
            "tag": "fullday",
            "speed": 10.0
        })

        sid = r.json()["session_id"]

        consumer = subprocess.Popen([
            sys.executable,
            "-u",
            "-m",
            "consumer.consumer",
            "--url",
            f"ws://127.0.0.1:8765/sessions/{sid}/stream",
            "--every",
            "25000"
        ], cwd=ROOT)

        await asyncio.sleep(1.5)

        controls = [
            ("START @10.0x", f"/sessions/{sid}/start", None, 4),
            ("PAUSE", f"/sessions/{sid}/pause", None, 1.5),
            (
                "SPEED 50x while paused",
                f"/sessions/{sid}/speed",
                {"speed": 50.0},
                0.3
            ),
            ("RESUME", f"/sessions/{sid}/resume", None, 3),
            ("PAUSE", f"/sessions/{sid}/pause", None, 1),
            (
                "SEEK 13:30 while paused",
                f"/sessions/{sid}/seek",
                {"target_timestamp_ms": SEEK_1330_MS},
                3
            ),
            (
                "RESUME after seek",
                f"/sessions/{sid}/resume",
                None,
                3
            ),
            (
                "SPEED 400.0x to the close",
                f"/sessions/{sid}/speed",
                {"speed": 400.0},
                2
            )
        ]

        for label, path, payload, wait in controls:
            print(f"\n--- {label} ---")

            if payload is None:
                await c.post(path)
            else:
                await c.post(path, json=payload)

            await asyncio.sleep(wait)

        for _ in range(600):
            await asyncio.sleep(1)

            state = (
                await c.get(f"/sessions/{sid}")
            ).json()

            if state["state"] in ("completed", "stopped"):
                break

        print(
            f"\nfinal: {state['state']} "
            f"{state['cursor']:,}/{state['total_events']:,}"
        )

        print(
            "timing:",
            (
                await c.get(f"/sessions/{sid}/timing")
            ).json()
        )

        if consumer.poll() is None:
            consumer.terminate()

asyncio.run(demo())

Run it from a second terminal:

python demo.py

The consumer starts as its own process and attaches to the WebSocket before playback begins.

The actual run started like this:

The session can therefore be stopped, re-anchored at a different speed, and resumed without restarting the replay.

The next command moves directly to 13:30:

This is the state-safe seek from the previous section happening in the complete system. The consumer discards its old state, processes the 3,456 warmup events, and only then continues from the new market timestamp.

We can then accelerate the remainder of the session:

The consumer continues updating its state from the incoming events until the session reaches the end of the tape:

The two counts describe different things. The session cursor finishes at 1,032,409/1,032,409, meaning it has reached the end of the full-day tape. The consumer reports 306,343 events because its state was cleared during the seek and rebuilt from that new point onward. The seek also jumps over part of the historical tape rather than streaming every skipped trade in real time.

realized_speed is intentionally left unset for this run because the replay was re-anchored several times by pauses, speed changes, and the seek. A single end-to-end speed ratio wouldn't meaningfully describe a session that deliberately changed its clock along the way.

What matters here is that the same historical tape survives the complete control sequence, the consumer rebuilds its state after the seek, and playback continues through to the session close.

Test the Replay Engine

The full-day run shows that the system can make it through the complete control sequence, but terminal output alone doesn't tell us whether the replay stayed ordered, respected pause boundaries, or rebuilt the correct state after a seek.

We’ll test those behaviors against the smaller quiet15m tape created earlier:

market-time-machine/
└── tests/
    ├── __init__.py
    ├── conftest.py
    └── test_replay.py

The test suite covers four areas: event ordering, replay timing, pause/resume behavior, and state reconstruction after seeking.

Create tests/test_replay.py

Create tests/test_replay.py:

import asyncio
import numpy as np
import pytest

from replay import config
from replay.events import TradeTape
from replay.session import ReplaySession
from replay.clock import ReplayClock, replay_batches, new_stats, summarize

TAPE = sorted(config.PROCESSED.glob("*_quiet15m.npz"))[0]

@pytest.fixture
def tape():
    return TradeTape.load(TAPE)

async def collect(sess, seconds):
    out = []

    async def drain():
        async for m in sess.events():
            out.append(m)

    t = asyncio.create_task(drain())
    await asyncio.sleep(seconds)
    return out, t


@pytest.mark.asyncio
async def test_ordering(tape):
    s = ReplaySession(tape, speed=500)
    out, t = await collect(s, 0.1)

    await s.start()
    await asyncio.sleep(2)
    await s.stop()
    t.cancel()

    trades = [
        m for m in out
        if m["type"] == "trade"
    ]

    assert len(trades) > 1000

    keys = [
        (m["timestamp_ms"], m["sequence"])
        for m in trades
    ]

    assert keys == sorted(keys)
    assert len(set(keys)) == len(keys)


@pytest.mark.asyncio
@pytest.mark.parametrize("speed", [10, 50, 100])
async def test_timing(tape, speed):
    end = int(
        np.searchsorted(
            tape.ts,
            tape.ts[0] + 60_000,
            side="right"
        )
    )

    clock = ReplayClock(tape.ts[0], speed).start()
    st = new_stats(speed)

    async for i, j in replay_batches(tape, clock, 0, st):
        if j >= end:
            break

    r = summarize(st)

    assert abs(r["speed_error_pct"]) < 5
    assert r["lateness_p95_ms"] < 50


@pytest.mark.asyncio
async def test_pause_resume(tape):
    s = ReplaySession(tape, speed=100)
    out, t = await collect(s, 0.05)

    await s.start()
    await asyncio.sleep(1)

    await s.pause()

    n = len([
        m for m in out
        if m["type"] == "trade"
    ])

    await asyncio.sleep(1)

    assert len([
        m for m in out
        if m["type"] == "trade"
    ]) == n

    await s.resume()
    await asyncio.sleep(1)

    await s.stop()
    t.cancel()

    seqs = [
        m["sequence"]
        for m in out
        if m["type"] == "trade"
    ]

    assert seqs == sorted(seqs)
    assert len(set(seqs)) == len(seqs)


@pytest.mark.asyncio
async def test_pause_seek_resume(tape):
    s = ReplaySession(
        tape,
        speed=200,
        warmup_ms=120_000
    )

    out, t = await collect(s, 0.05)

    await s.start()
    await asyncio.sleep(0.5)
    await s.pause()

    target = int(tape.ts[0]) + 300_000
    await s.seek(target)

    assert s.info()["state"] == "paused"

    def past():
        return [
            m for m in out
            if m["type"] == "trade"
            and not m.get("warmup")
            and m["timestamp_ms"] >= target
        ]

    await asyncio.sleep(0.4)
    assert not past()

    await s.resume()
    await asyncio.sleep(1)

    got = past()

    await s.stop()
    t.cancel()

    assert got

    seqs = [m["sequence"] for m in got]

    assert seqs == sorted(seqs)
    assert len(set(seqs)) == len(seqs)


@pytest.mark.asyncio
async def test_seek_state_equivalence(tape):
    import sys

    sys.path.insert(0, str(config.ROOT))
    from consumer.consumer import State as ConsumerState

    s = ReplaySession(
        tape,
        speed=200,
        warmup_ms=120_000
    )

    live = ConsumerState()
    reset = None
    snap = None
    out = []

    async def drain():
        nonlocal live, reset, snap

        async for m in s.events():
            out.append(m)

            if m["type"] == "trade":
                live.apply(m)

            elif m["type"] == "replay_reset":
                reset = m
                live = ConsumerState()

            elif m["type"] == "warmup_complete":
                snap = (
                    live.n,
                    live.vol,
                    live.short.value,
                    live.long.value
                )

    t = asyncio.create_task(drain())

    await s.start()
    await asyncio.sleep(1)

    await s.seek(
        int(tape.ts[0]) + 600_000
    )

    for _ in range(100):
        if snap:
            break
        await asyncio.sleep(0.05)

    await s.stop()
    t.cancel()

    assert snap

    fresh = ConsumerState()

    lo = tape.index_at(
        reset["warmup_from_ms"]
    )

    hi = tape.index_at(
        reset["target_timestamp_ms"]
    )

    for k in range(lo, hi):
        fresh.apply(tape[k].to_wire())

    n, vol, short, long = snap

    assert n == fresh.n == reset["warmup_events"]
    assert vol == fresh.vol

    assert short == pytest.approx(
        fresh.short.value,
        rel=1e-12
    )

    assert long == pytest.approx(
        fresh.long.value,
        rel=1e-12
    )

    kinds = [m["type"] for m in out]

    seg = out[
        kinds.index("replay_reset") + 1:
        kinds.index("warmup_complete")
    ]

    assert not [
        m for m in seg
        if m["type"] == "trade"
        and not m.get("warmup")
    ]

test_ordering() checks that emitted trades remain sorted by (timestamp, sequence) and that the same event isn't emitted twice.

The timing test runs 60 seconds of historical market time at 10x, 50x, and 100x. It allows a small tolerance rather than expecting an event loop to behave like a hard real-time scheduler: realized speed must stay within 5% of the target, while 95th-percentile lateness must remain below 50 ms.

test_pause_resume() checks something different. Once pause() returns, the number of received trades should remain unchanged until playback resumes. After resuming, the resulting sequence must still be ordered and duplicate-free.

test_pause_seek_resume() covers the exact control pattern used in the full replay. The session pauses, moves five minutes into the tape, stays paused at the new position, and only begins releasing normal post-seek trades after resume().

Verify State Reconstruction Independently

The strongest test is test_seek_state_equivalence().

When the replay seeks, the consumer receives a reset followed by two minutes of warmup events. Rather than simply checking that a warmup_complete message appears, this test constructs a completely fresh ConsumerState and independently feeds it the same historical interval directly from the tape:

for k in range(lo, hi):
    fresh.apply(tape[k].to_wire())

The replay-built and independently rebuilt states must then agree on:

event count
cumulative volume
30-second VWAP
2-minute VWAP

The VWAP values are compared with a relative tolerance of 1e-12. The test also checks that no normal replay trades slip into the stream between replay_reset and warmup_complete.

Configure pytest

The asynchronous tests use pytest-asyncio. Create tests/conftest.py:

import pytest

def pytest_configure(config):
    config.addinivalue_line(
        "markers",
        "asyncio"
    )

Then add pytest.ini in the project root:

[pytest]
asyncio_mode = auto

Run the complete suite:

pytest tests/ -v

The recorded run produced:

The tests cover more than whether the replay eventually reaches the end of the tape. They check that historical ordering survives playback, accelerated timing remains within the expected tolerance, controls preserve the event sequence, and the state reconstructed after a seek matches an independent rebuild from the underlying historical data.

Conclusion

What I liked most about this build is how different the same historical dataset feels once we give it a clock again.

We started with a completed AAPL session from EODHD and ended with something that could move slowly, race ahead, pause in the middle, jump to another point in the day, and keep going while the consumer reacted only to what had reached it so far.

There's still plenty of room to take the project further. The replay could support multiple symbols, richer market state, several downstream consumers, persistent replay sessions, or even strategy and execution components that plug directly into the stream. The current version keeps those pieces out deliberately, but the core replay layer is now there to build on.

For me, that's the useful outcome of the project. EODHD gives us the historical events, but the replay layer lets another piece of software experience those events as a trading day rather than as a dataset that already knows how the day ends.

But there's an obvious problem. Stocks that fall 20% or more often rebound even when no insider buys anything. If we only measure what happened after CEO purchases, we may end up crediting the insider signal for a recovery that was already common among beaten-down stocks.

In this tutorial, we'll build a Python workflow to test that properly. We'll pull Form 4 transactions, isolate CEO purchases, collapse repeated filing rows into usable events, attach historical prices, calculate drawdowns and forward returns, and then compare the purchase episodes with similar no-purchase dates from the same stocks.

The interesting part isn't just the final return table. It's everything required to turn messy regulatory filings into a dataset that can support a fair comparison. Along the way, we'll deal with duplicate transaction rows, repeated purchases by the same CEO, trading-day alignment, incomplete price histories, and one-to-one control matching.

By the end, we'll have a full event-study workflow and a more useful answer than “the stock went up after the CEO bought.”

Table of Contents

• Prerequisites

• Import The Required Packages

• Build The Stock Universe

• Fetch CEO Purchases And Apply The Date Filter

• Turn Form 4 Rows Into Daily Purchase Events

• Add Historical Prices And Drawdowns

  • Calculate The Trailing High And Drawdown

  • Match Each Purchase With The Latest Available Price

• Convert Purchase Events Into Episodes

• Calculate Returns After CEO Purchases

  • Organize The Price History By Ticker

  • Find The Entry Date And Calculate Forward Returns

  • Summarize The Raw Returns

• Build The No-Purchase Control Group

  • Create The Control Candidates

  • Remove Dates Near CEO Purchases

  • Match Purchase Episodes With Controls

  • Build The Final Matched Dataset

• Compare CEO Purchases Against Similar No-Purchase Drawdowns

  • Calculate Forward Returns From Any Signal Date

  • Apply The Same Return Logic To Both Groups

  • Build The Final Comparison

• What The Case Study Found

• What This Test Can And Can't Say

Prerequisites

You don't need an advanced finance or quantitative background to follow this tutorial. A basic understanding of Python and pandas should be enough.

Before starting, make sure you have:

• Python installed locally, or access to a notebook environment such as Jupyter Notebook or Google Colab

• Basic familiarity with dataframes, functions, loops, and API requests

• An EODHD API key with access to the screener, Form 4 filings, and historical EOD endpoints

• Enough API credits to process the number of stocks you choose to analyze

The full case study uses Form 4 data from 500 securities. You can run the workflow on a smaller sample first if you want to understand the code without using as many API calls.

No prior knowledge of event studies or control matching is required. We'll build those parts step by step as they appear in the analysis.

Import The Required Packages

We only need a small set of packages for the full workflow. requests handles the API calls, pandas and NumPy do most of the data work, and SciPy gives us the one-to-one matching algorithm used later for the control group.

import json
import re
import numpy as np
import pandas as pd
import requests
from scipy.optimize import linear_sum_assignment

That's the full setup. We're only importing what the analysis actually needs, without adding extra libraries or unnecessary tooling. Make sure to install these packages using pip before importing them to your environment.

Build The Stock Universe

Before pulling insider filings, we need a list of companies to search.

Rather than starting with one market-cap segment, we'll build a mixed universe across micro-, small-, mid-, and large-cap stocks. This gives the analysis some variation instead of letting one part of the market dominate the sample.

The market-cap buckets are:

• micro_cap: $50 million to $300 million

• small_cap: $300 million to $2 billion

• mid_cap: $2 billion to $10 billion

• large_cap: $10 billion and above

For each bucket, we'll fetch 500 screener results, randomly select 250, and combine them into a 1,000-stock universe.

def fetch_stocks(filters, cap):
    api_key = 'YOUR EODHD API KEY'
    base_url = 'https://eodhd.com/api/screener'
    all_stocks = []
    for i in range(0,500,100):
        params = {
            "api_token": api_key,
            "filters": json.dumps(filters),
            "sort": "market_capitalization.desc",
            "limit": 100,
            "offset": i}
        resp = requests.get(base_url, params = params).json()
        stocks = list(pd.DataFrame(resp['data'])['code'])
        all_stocks.append(stocks)
    all_stocks = [item for sublist in all_stocks for item in sublist]
    df = pd.DataFrame(columns = ['ticker', f'cap'])
    df.ticker, df.cap = all_stocks, cap
    df = df.sample(n = 250, random_state = 42)
    return df

micro_filters = [
    ["exchange", "=", "us"],
    ["market_capitalization", ">=", 50_000_000],
    ["market_capitalization", "<", 300_000_000]
]

small_filters = [
    ["exchange", "=", "NYSE"],
    ["market_capitalization", ">=", 300_000_000],
    ["market_capitalization", "<", 2_000_000_000]
]

mid_filters = [
    ["exchange", "=", "NYSE"],
    ["market_capitalization", ">=", 2_000_000_000],
    ["market_capitalization", "<", 10_000_000_000]
]

large_filters = [
    ["exchange", "=", "NYSE"],
    ["market_capitalization", ">=", 10_000_000_000]
]

micro_stocks = fetch_stocks(micro_filters, 'micro_cap')
small_stocks = fetch_stocks(small_filters, 'small_cap')
mid_stocks = fetch_stocks(mid_filters, 'mid_cap')
large_stocks = fetch_stocks(large_filters, 'large_cap')

frames = [micro_stocks, small_stocks, mid_stocks, large_stocks]
stocks_1000 = pd.concat(frames, ignore_index = True)
stocks_1000 = stocks_1000.sample(frac = 1, random_state = 42).reset_index(drop = True)
stocks_1000

Note: Replace YOUR EODHD API KEY with your actual EODHD API key. If you don’t have one, you can obtain it by opening an EODHD developer account.

The screener returns at most 100 rows per request, so the loop moves through the first 500 results in five batches.

We then sample 250 tickers from those candidates. The fixed random seed makes the selection repeatable, so rerunning the cell produces the same sample. After that, we define the four market-cap filters and run the function for each one.

The final dataframe contains 1,000 tickers, with 250 from each bucket.

One caveat is worth stating now. The micro-cap filter uses the broader us exchange setting, while the other groups use NYSE. This is the screener sample used for the case study, but it shouldn't be treated as a perfectly representative sample of the entire US stock market.

Fetch CEO Purchases And Apply The Date Filter

With the stock universe ready, we can start searching the Form 4 filings for CEO purchases using EODHD’s Insider Transactions (SEC Form 4) API.

Form 4 data contains much more than straightforward insider buying. A filing can include sales, awards, option exercises, derivative transactions, and several rows belonging to the same trade. So we can't simply download every filing and treat every record as a buying signal.

For this analysis, a transaction must satisfy all of these conditions:

• appear under non-derivative transactions

• be reported by an officer

• have an officer title that identifies the person as a CEO

• use transaction code P

• represent acquired shares

• contain positive values for both shares and price

• refer to common stock

We also retain both the transaction date and filing date. The transaction date tells us when the CEO bought the shares, while the filing date tells us when outside investors could observe the purchase. Later in the analysis, the filing date will become the signal date.

The following block handles the complete extraction. It defines the filtering function, runs it across the first 500 stocks, and combines all qualifying rows into one dataframe.

def fetch_ceo_purchases(ticker):
    try:
        api_key = 'YOUR EODHD API KEY'

        all_form4 = []

        for i in range(0,1000,100):
            form4_url = f'https://eodhd.com/api/sec-filings/{ticker}/form4?api_token={api_key}&page[limit]=100&page[offset]={i}'
            resp = requests.get(form4_url).json()['data']
            all_form4.append(resp)

        all_form4 = [item for sublist in all_form4 for item in sublist]

        all_purchases = []

        ceo_pattern = re.compile(
            r'\bceo\b|chief executive officer|co-chief executive officer|co-ceo|chief exec officer',
            re.IGNORECASE
        )

        for filing in all_form4:
            footnote_map = {
                footnote['footnote_id']: footnote['text']
                for footnote in filing.get('footnotes', [])
            }

            for transaction in filing.get('non_derivative', []):
                officer_title = transaction.get('officer_title') or ''
                security_title = transaction.get('security_title') or ''
                shares_amount = transaction.get('shares_amount')
                price_per_share = transaction.get('price_per_share')

                is_ceo = bool(ceo_pattern.search(officer_title))

                is_purchase = (
                    transaction.get('is_officer') is True
                    and is_ceo
                    and transaction.get('transaction_code') == 'P'
                    and transaction.get('acquired_or_disposed') == 'A'
                    and shares_amount is not None
                    and shares_amount > 0
                    and price_per_share is not None
                    and price_per_share > 0
                    and 'common stock' in security_title.lower()
                )

                if not is_purchase:
                    continue

                linked_footnotes = ' '.join(
                    footnote_map.get(footnote_id, '')
                    for footnote_id in transaction.get('footnote_ids', [])
                )

                all_purchases.append({
                    'ticker': ticker,
                    'accession_number': filing['accession_number'],
                    'filed_at': filing['filed_at'],
                    'transaction_date': transaction['transaction_date'],
                    'reporting_owner_cik': transaction['reporting_owner_cik'],
                    'reporting_owner_name': transaction['reporting_owner_name'],
                    'officer_title': officer_title,
                    'security_title': security_title,
                    'shares_amount': shares_amount,
                    'price_per_share': price_per_share,
                    'total_value': transaction.get('total_value'),
                    'shares_owned_after': transaction.get('shares_owned_after'),
                    'footnotes': linked_footnotes
                })

        return all_purchases
    except:
        return None

all_ceo_purchases = []

for ticker in stocks_1000.ticker[:500]:
    ticker = ticker + '.US'
    ceo_purchases = fetch_ceo_purchases(ticker)
    if ceo_purchases:
        all_ceo_purchases.extend(ceo_purchases)
        print(f'{len(ceo_purchases)} ceo purchases found in {ticker}')
    else:
        print(f'no transaction found in {ticker}')
        
cp_df = pd.DataFrame(all_ceo_purchases)
cp_df.to_csv('ceo_purchases.csv')

The function requests the filings in batches of 100 and flattens the returned pages into one list. It then checks every non-derivative transaction against the CEO-purchase rules.

The CEO-title check uses a regular expression because filings don't use one perfectly consistent title. A CEO might appear as CEO, Chief Executive Officer, or Co-CEO, so matching only one exact string would miss valid records.

We also preserve the linked footnotes. Transaction code P is useful, but it doesn't tell the complete story by itself. A footnote may reveal that the purchase involved a trading plan, an offering, or another arrangement that deserves closer inspection.

I ran this step on the first 500 securities in the universe because the Form 4 endpoint can consume API credits quickly. The same loop can be extended to all 1,000 stocks for a larger sample.

Once the rows are collected, we restrict the dataset to filings submitted between the beginning of 2022 and the end of 2025.

cp_df = cp_df[cp_df["filed_at"].between("2022-01-01","2025-12-31")]

cp_df.tail()

These are still raw filing rows rather than independent CEO-buying signals. One purchase can be split across several rows when different blocks of shares were acquired at different prices. The next step is to collapse those fragments into daily purchase events.

Turn Form 4 Rows Into Daily Purchase Events

A CEO might buy shares at several prices on the same day. The filing can record each price block as a separate row, even though those rows belong to one broader purchase. If we treated every row as an independent signal, one busy purchase day could receive far more weight than another simply because it was split into more price levels.

So the next step is to group rows that share the same ticker, CEO, filing, and transaction date.

For each group, we'll add up the shares and total purchase value. We'll then calculate a weighted-average price:

weighted average price = total purchase value / total shares purchased

The following block performs the full aggregation and produces one row per daily CEO-purchase event.

cp_df['purchase_value'] = (
    cp_df['shares_amount'] * cp_df['price_per_share']
)

group_columns = [
    'ticker',
    'reporting_owner_cik',
    'reporting_owner_name',
    'officer_title',
    'accession_number',
    'filed_at',
    'transaction_date'
]

daily_events = (
    cp_df.groupby(
        group_columns,
        as_index=False,
        dropna=False
    )
    .agg(
        shares_purchased=('shares_amount', 'sum'),
        total_purchase_value=('purchase_value', 'sum'),
        transaction_rows=('shares_amount', 'size'),
        shares_owned_after=('shares_owned_after', 'max')
    )
)

daily_events['weighted_average_price'] = (
    daily_events['total_purchase_value']
    / daily_events['shares_purchased']
)

daily_events.filed_at = pd.to_datetime(daily_events.filed_at)
daily_events.to_csv('ceo_purchases_grouped.csv')
daily_events.tail()

The purchase_value column gives every raw row a dollar value before aggregation. Once the rows are grouped, those values can be summed without losing the effect of the different purchase prices.

The transaction_rows column is useful for checking how much collapsing happened. A value of 1 means the daily event already appeared as one row. A value of 5 means five separate filing rows were combined into one purchase event.

The aggregation reduced the dataset from 625 raw transaction rows to 535 daily purchase events.

That difference isn't just housekeeping. It changes the unit of analysis from “one reported price block” to “one CEO purchase day,” which is much closer to the economic event we're trying to study.

We're still not ready to calculate returns, though. A purchase event tells us that the CEO bought, but not whether the stock was near its high, down slightly, or already deep in a drawdown. Next, we'll attach the price context that was available when each filing became public.

Add Historical Prices And Drawdowns

A CEO purchase only becomes interesting in this test when we know where the stock was trading at the time.

A purchase made 5% below the yearly high is very different from one made after the stock has fallen 40%. So we now need to bring historical price data into the workflow and measure how far each stock was below its recent high when the filing became public.

We'll use adjusted close rather than the raw closing price because adjusted prices account for events such as stock splits and dividends. That gives us a more consistent series for comparing prices across time.

We pull prices from 2021 through 2025, even though the purchase analysis begins in 2022. The extra year is needed because the first 2022 observations still require enough earlier data to calculate a trailing one-year high.

The following block fetches the daily historical prices EODHD’s historical EOD endpoint for every ticker represented in the CEO-purchase dataset and combines them into one dataframe.

tickers = list(cp_df.ticker.unique())
historical_eod_entries = []

def fetch_historical_eod(ticker):
    api_key = 'YOUR EODHD API KEY'
    historical_url = f'https://eodhd.com/api/eod/{ticker}?from=2021-01-01&to=2025-12-31&period=d&api_token={api_key}&fmt=json'
    historical_resp = requests.get(historical_url).json()
    historical_filtered = []
    for item in historical_resp:
        item['ticker'] = ticker
        keys = ['ticker', 'date', 'adjusted_close']
        item = {key: item.get(key) for key in keys}
        historical_filtered.append(item)
    return historical_filtered

for ticker in tickers:
    try:
        historical_eod = fetch_historical_eod(ticker)
        historical_eod_entries.extend(historical_eod)
        print(f'{ticker} done')
    except:
        print(f'{ticker} error')

This code fetches the historical data and gives us one price row per ticker per trading day. Next, we calculate the rolling high and drawdown.

Calculate The Trailing High And Drawdown

For every trading day, we want to know the highest adjusted close reached during the previous 252 trading sessions. That's roughly one trading year.

The drawdown is then calculated as:

drawdown = adjusted close / trailing 252-day high - 1

A value of -0.20 means the stock is 20% below its trailing high. A value of -0.35 means it's 35% below that high.

The next block sorts each stock’s price history, calculates the rolling high, and converts the result into both decimal and percentage drawdown columns.

historical_df = pd.DataFrame(historical_eod_entries)
historical_df['date'] = pd.to_datetime(historical_df.date)

historical_df['rolling_high_252'] = (historical_df.groupby('ticker')['adjusted_close'].transform
                                     (lambda prices: prices.rolling(window=252, min_periods=200).max()))

historical_df['drawdown'] = (historical_df['adjusted_close']/ historical_df['rolling_high_252']- 1)
historical_df['drawdown_pct'] = (historical_df['drawdown'] * 100)

The min_periods=200 argument deserves a quick explanation.

A full rolling window contains 252 trading days, but requiring exactly 252 observations would remove many early rows. Allowing the calculation after 200 sessions gives us some flexibility while still requiring a substantial amount of historical data.

Rows without enough history remain missing rather than receiving a weak drawdown estimate.

Match Each Purchase With The Latest Available Price

Now we need to attach the price context to each CEO-purchase event.

The filing date is the signal date, but we shouldn't use the closing price from that same date. A Form 4 can be submitted before, during, or after the trading session, so the same-day close may not have been known when the filing appeared.

Instead, we use the latest completed trading day strictly before the filing date.

The next block performs that match with merge_asof(). Unlike a normal merge, it can match each filing with the nearest earlier price date rather than requiring both dates to be identical.

cp_df.filed_at = pd.to_datetime(cp_df.filed_at)

price_columns = ['ticker', 'date', 'adjusted_close', 'rolling_high_252', 'drawdown', 'drawdown_pct']

analysis_df = pd.merge_asof(
    cp_df.sort_values(['filed_at', 'ticker']),
    historical_df[price_columns].sort_values(['date', 'ticker']),
    by='ticker',
    left_on='filed_at',
    right_on='date',
    direction='backward',
    allow_exact_matches=False
)

analysis_df = analysis_df.rename(columns={'date': 'price_date'})

The key setting is allow_exact_matches=False. It prevents a filing dated March 10 from using the March 10 closing price. The merge will instead use the latest available trading day before March 10.

The merged dataframe now looks like this:

The merged dataframe now contains the purchase information alongside:

• price_date: the trading day used for the match

• adjusted_close: the stock price on that day

• rolling_high_252: the trailing one-year high

• drawdown: the decline expressed as a decimal

• drawdown_pct: the same decline expressed as a percentage

For example, a drawdown_pct value of -32.4 means the stock was 32.4% below its trailing 252-day high on the last completed trading day before the filing.

We now know both that the CEO bought and how beaten down the stock was when the purchase became public.

The next problem is repeated buying. A CEO who buys several times over a few weeks shouldn't automatically create several independent signals.

Convert Purchase Events Into Episodes

At this point, the dataset has one row per CEO purchase day. That's better than working with raw Form 4 rows, but it can still overcount the same underlying decision.

Imagine a CEO buys shares on Monday, again the following week, and once more two weeks later. Technically, those are three purchase events. Economically, they may be one sustained buying campaign.

Treating all three as independent signals would give frequent buyers more weight than CEOs who completed the same idea in one trade. So before calculating returns, we'll group nearby purchases into buying episodes.

The rule is simple: purchases by the same CEO in the same stock belong to one episode when consecutive filing dates are no more than 28 calendar days apart.

We'll first remove purchase events without a usable drawdown, sort the remaining rows by ticker, CEO, and filing date, and calculate the number of days since the previous purchase.

purchase_events = analysis_df.dropna(subset=["drawdown"])
purchase_events.filed_at = pd.to_datetime(purchase_events.filed_at)
purchase_events = purchase_events.sort_values(["ticker", "reporting_owner_cik", "filed_at"])
purchase_events["days_since_previous"] = purchase_events.groupby(["ticker", "reporting_owner_cik"])["filed_at"].diff().dt.days
purchase_events["new_episode"] = purchase_events["days_since_previous"].isna() | (purchase_events["days_since_previous"] > 28)
purchase_events["episode_id"] = purchase_events.groupby(["ticker", "reporting_owner_cik"])["new_episode"].cumsum()

The first purchase for every ticker-CEO combination automatically starts a new episode because there's no earlier filing to compare it with.

After that, a new episode begins only when the gap from the previous filing exceeds 28 days. Purchases separated by 28 days or less stay inside the same episode.

The cumulative sum of new_episode gives every buying sequence its own identifier.

Now that each purchase event belongs to an episode, we can collapse the events into one row per buying sequence.

For each episode, we keep the first and last filing dates, add up the shares and purchase value, count the purchase activity, and preserve the drawdown from the first filing date.

episodes = purchase_events.groupby(["ticker", "reporting_owner_cik", "reporting_owner_name", "episode_id"], 
                               as_index=False).agg(first_filing_date=("filed_at", "min"), 
                                                   last_filing_date=("filed_at", "max"),
                                                   first_transaction_date=("transaction_date", "min"),
                                                   total_shares=("shares_purchased", "sum"),
                                                   total_purchase_value=("total_purchase_value", "sum"),
                                                   purchase_days=("filed_at", "nunique"),
                                                   transaction_events=("filed_at", "size"),
                                                   initial_drawdown=("drawdown", "first"), 
                                                   initial_drawdown_pct=("drawdown_pct", "first"))

There are two activity counts here:

• purchase_days counts the number of distinct filing dates in the episode.

• transaction_events counts the daily purchase events that were grouped together.

The total shares and purchase value cover the entire episode. But the drawdown comes only from the first filing date because that's when the signal begins.

That detail matters for the 20% filter.

Suppose an episode starts when the stock is 18% below its high, then the CEO buys again after the drawdown reaches 24%. It would be misleading to call that an episode that began after a 20% decline.

So we group first, then apply the threshold using initial_drawdown.

episodes_20 = episodes[episodes["initial_drawdown"] <= -0.20].copy()

print("All purchase episodes:", len(episodes))
print("20% drawdown episodes:", len(episodes_20))
print("Tickers represented:", episodes_20["ticker"].nunique())

episodes_20

We have now reduced repeated purchases into 137 distinct CEO-buying episodes that started while the stock was at least 20% below its trailing high.

These episodes are the actual signals we'll follow through time. Next, we'll enter on the first trading day after the initial filing and measure what happened over one, three, six, and twelve months.

Calculate Returns After CEO Purchases

We now have 137 CEO-buying episodes that began while the stock was at least 20% below its trailing 252-day high.

The next question is straightforward: what happened after those filings became public?

We'll use the first trading day after the episode’s initial filing as the entry date. That keeps the test realistic. An outside investor couldn't act before the Form 4 appeared, and using the next trading session also handles filings submitted on weekends or market holidays.

The return horizons are:

• 1 Month: 21 trading days

• 3 Months: 63 trading days

• 6 Months: 126 trading days

• 12 Months: 252 trading days

Organize The Price History By Ticker

Before calculating returns, we'll create a separate, chronologically ordered price series for each stock.

This lets the return function find the correct entry date and then move forward by a fixed number of trading sessions without repeatedly filtering the full historical dataframe.

episodes_20 = episodes_20.reset_index(drop = True)
episodes_20['first_filing_date'] = pd.to_datetime(episodes_20['first_filing_date'])
historical_df['date'] = pd.to_datetime(historical_df['date'])

prices = historical_df[['ticker', 'date', 'adjusted_close']].dropna().sort_values(['ticker', 'date'])
price_map = {ticker: group.reset_index(drop=True) for ticker, group in prices.groupby('ticker')}

price_map is a dictionary where each ticker points to its own dataframe of dates and adjusted closing prices.

For example, price_map['AAT.US'] contains only the historical prices for AAT.US, already sorted from oldest to newest.

Find The Entry Date And Calculate Forward Returns

Now we can write a function that handles one purchase episode at a time.

The function will:

1. locate the stock’s price history

2. find the first trading day strictly after the filing date

3. save that day’s adjusted close as the entry price

4. move forward by 21, 63, 126, and 252 trading sessions

5. calculate the return at each horizon

def calculate_forward_returns(row):
    ticker_prices = price_map.get(row['ticker'])

    if ticker_prices is None:
        return pd.Series(dtype='object')

    dates = ticker_prices['date'].to_numpy(dtype='datetime64[ns]')
    entry_index = np.searchsorted(dates, np.datetime64(row['first_filing_date']), side='right')

    if entry_index >= len(ticker_prices):
        return pd.Series(dtype='object')

    entry_date = ticker_prices.loc[entry_index, 'date']
    entry_price = ticker_prices.loc[entry_index, 'adjusted_close']

    result = {'entry_date': entry_date, 'entry_price': entry_price}
    horizons = {'1m': 21, '3m': 63, '6m': 126, '12m': 252}

    for label, days in horizons.items():
        target_index = entry_index + days

        if target_index < len(ticker_prices):
            target_price = ticker_prices.loc[target_index, 'adjusted_close']
            result[f'date_{label}'] = ticker_prices.loc[target_index, 'date']
            result[f'return_{label}'] = target_price / entry_price - 1
        else:
            result[f'date_{label}'] = pd.NaT
            result[f'return_{label}'] = np.nan

    return pd.Series(result)

forward_returns = episodes_20.apply(calculate_forward_returns, axis=1)
episode_returns = pd.concat([episodes_20.reset_index(drop=True), forward_returns], axis=1)

episode_returns

The resulting dataframe contains the episode details, entry date, entry price, and forward returns at each horizon:

Summarize The Raw Returns

Looking at individual episodes is useful, but we also need a compact view of the full sample.

For each horizon, we'll calculate:

• the number of available observations

• the mean return

• the median return

• the percentage of returns above zero

summary = []

for horizon in ['1m', '3m', '6m', '12m']:
    returns = episode_returns[f'return_{horizon}'].dropna()

    summary.append({
        'horizon': horizon,
        'observations': len(returns),
        'mean_return': returns.mean(),
        'median_return': returns.median(),
        'positive_rate': (returns > 0).mean()
    })

summary_df = pd.DataFrame(summary)
summary_df

At first glance, the results look promising. The average return reached 11.9% after three months and 35.4% after twelve months. Most twelve-month observations were also positive.

But this is exactly where it's easy to jump to the wrong conclusion.

These numbers tell us what happened after CEOs bought. They don't tell us how much of that performance came from the CEO purchase itself.

The stocks were already down at least 20%. Some of them may have rebounded simply because beaten-down stocks sometimes recover.

To separate those two effects, we need a comparison group made from similar drawdown dates where no CEO purchase occurred nearby.

Build The No-Purchase Control Group

The raw return table looked encouraging, but it still gave CEO buying all the credit.

That's not a fair test. Every stock in the sample was already down at least 20%, and beaten-down stocks can rebound without any insider activity. We need to compare each CEO-purchase episode with another date where the same stock was under similar pressure but no CEO purchase happened nearby.

A valid control must satisfy six rules:

• same ticker

• same calendar year

• drawdown within five percentage points

• no more than 180 calendar days away

• no CEO purchase within 28 days before or after

• used only once

Each CEO-purchase episode is also matched only once.

Create The Control Candidates

We'll start by finding every trading day between 2022 and 2025 when a stock was at least 20% below its trailing high.

There's one problem, though. A stock can stay below that threshold for months. If we kept every trading day, one long decline could create hundreds of nearly identical control candidates.

To avoid that, we'll split each continuous drawdown period into 28-day blocks and keep one candidate from each block.

hist = historical_df[['ticker', 'date', 'adjusted_close', 'rolling_high_252', 'drawdown', 'drawdown_pct']].dropna(subset=['drawdown'])

hist['date'] = pd.to_datetime(hist['date'])
hist = hist[hist['date'].between('2022-01-01', '2025-12-31')].sort_values(['ticker', 'date'])

hist['below_20'] = hist['drawdown'] <= -0.20
hist['previous_below_20'] = hist.groupby('ticker')['below_20'].shift().fillna(False)
hist['new_state'] = hist['below_20'].ne(hist['previous_below_20'])
hist['drawdown_segment'] = hist.groupby('ticker')['new_state'].cumsum()

control_candidates = hist[hist['below_20']].copy()

control_candidates['segment_start'] = control_candidates.groupby(['ticker', 'drawdown_segment'])['date'].transform('min')
control_candidates['anchor_block'] = ((control_candidates['date'] - control_candidates['segment_start']).dt.days // 28)
control_candidates = control_candidates.sort_values(['ticker', 'date']).drop_duplicates(['ticker', 'drawdown_segment', 'anchor_block'])

control_candidates = control_candidates.reset_index(drop = True)
control_candidates

below_20 marks the dates that pass the drawdown threshold.

drawdown_segment then separates one continuous decline from another. If the stock recovers above the threshold and falls below it again later, that becomes a new segment.

Inside each segment, anchor_block counts 28-day windows from the day the drawdown began. Keeping one row per block gives us a manageable set of dates without treating every session in the same decline as a fresh event.

These dates are only potential controls. We still need to remove any that occurred close to CEO buying.

Remove Dates Near CEO Purchases

A no-purchase control should be genuinely separate from the insider signal.

For example, a drawdown date three days before a CEO filing would be a poor control. The transaction may already have happened, and the filing may simply not have appeared yet.

We therefore collect every CEO purchase filing date in the event dataset, not only the 137 episodes that passed the 20% threshold.

purchase_dates = analysis_df[['ticker', 'filed_at']].dropna().drop_duplicates()
purchase_dates['filed_at'] = pd.to_datetime(purchase_dates['filed_at'])
purchase_dates = purchase_dates.rename(columns={'filed_at': 'purchase_date'})
purchase_dates

For each candidate, we now need to find the closest purchase filing before it and the closest purchase filing after it.

Two merge_asof() operations handle that. The first searches backward, while the second searches forward.

control_candidates = pd.merge_asof(
    control_candidates.sort_values('date'),
    purchase_dates.sort_values('purchase_date'),
    by='ticker',
    left_on='date',
    right_on='purchase_date',
    direction='backward'
)

control_candidates = control_candidates.rename(columns={'purchase_date': 'previous_purchase_date'})

control_candidates = pd.merge_asof(
    control_candidates.sort_values('date'),
    purchase_dates.sort_values('purchase_date'),
    by='ticker',
    left_on='date',
    right_on='purchase_date',
    direction='forward'
)

control_candidates = control_candidates.rename(columns={'purchase_date': 'next_purchase_date'})

Each candidate now knows the nearest CEO purchase on either side.

We can calculate the distance from those filings and keep only dates that are more than 28 calendar days away from both.

days_from_previous = (control_candidates['date'] - control_candidates['previous_purchase_date']).dt.days
days_to_next = (control_candidates['next_purchase_date'] - control_candidates['date']).dt.days

far_from_previous = control_candidates['previous_purchase_date'].isna() | (days_from_previous > 28)
far_from_next = control_candidates['next_purchase_date'].isna() | (days_to_next > 28)

control_candidates = control_candidates[far_from_previous & far_from_next]
control_candidates

A missing previous or next filing is fine. It simply means there was no CEO purchase on that side of the candidate within the available dataset.

At this point, every remaining row represents a date when:

• the stock was down at least 20%

• the date wasn't part of the immediate neighborhood of a CEO purchase

• the stock had enough historical price data for the drawdown calculation

Match Purchase Episodes With Controls

Now comes the actual matching.

We first give every CEO-purchase episode and every control candidate a unique identifier. We also extract the calendar year because matches must come from the same stock and year.

purchase_pool = episodes_20.reset_index(drop = True)

purchase_pool['first_filing_date'] = pd.to_datetime(purchase_pool['first_filing_date'])
purchase_pool['year'] = purchase_pool['first_filing_date'].dt.year
purchase_pool['purchase_id'] = np.arange(len(purchase_pool))

control_candidates['year'] = control_candidates['date'].dt.year
control_candidates['control_id'] = np.arange(len(control_candidates))

The matching happens separately inside each ticker-year group.

Suppose a CEO bought when a stock was down 32%. We search for a no-purchase date in the same stock and year where the drawdown was close to 32%, while also keeping the dates no more than 180 calendar days apart.

A pair is valid only when:

• drawdown difference <= 0.05

• calendar distance <= 180 days

The next block builds the possible pairings and uses linear_sum_assignment() to select a one-to-one set of matches.

matches = []
max_drawdown_gap = 0.05
max_calendar_gap = 180

for (ticker, year), purchases in purchase_pool.groupby(['ticker', 'year']):
    controls = control_candidates[(control_candidates['ticker'] == ticker) & (control_candidates['year'] == year)].copy()

    if controls.empty:
        continue

    purchase_drawdowns = purchases['initial_drawdown'].to_numpy()[:, None]
    control_drawdowns = controls['drawdown'].to_numpy()[None, :]
    drawdown_cost = np.abs(purchase_drawdowns - control_drawdowns)

    purchase_dates = purchases['first_filing_date'].to_numpy(dtype='datetime64[D]')
    control_dates = controls['date'].to_numpy(dtype='datetime64[D]')
    calendar_gap = np.abs((purchase_dates[:, None] - control_dates[None, :]).astype('timedelta64[D]').astype(int))

    valid = (drawdown_cost <= max_drawdown_gap) & (calendar_gap <= max_calendar_gap)

    if not valid.any():
        continue

    cost = drawdown_cost + calendar_gap / 1000000
    cost[~valid] = 1000000

    row_indices, column_indices = linear_sum_assignment(cost)
    keep = cost[row_indices, column_indices] < 1000000

    selected = pd.DataFrame({
        'purchase_id': purchases.iloc[row_indices[keep]]['purchase_id'].to_numpy(),
        'control_id': controls.iloc[column_indices[keep]]['control_id'].to_numpy(),
        'drawdown_gap': drawdown_cost[row_indices[keep], column_indices[keep]],
        'calendar_gap_days': calendar_gap[row_indices[keep], column_indices[keep]]
    })

    matches.append(selected)

matched_pairs = pd.concat(matches, ignore_index=True)

The central idea is easier than the code first makes it look.

drawdown_cost measures how far apart the two drawdowns are. A purchase at -0.32 and a control at -0.34 have a difference of 0.02, or two percentage points.

The calendar distance is added as a very small tie-breaker. Drawdown similarity remains the main priority, but when two controls are almost equally close, the nearer date is preferred.

linear_sum_assignment() prevents the same control from being handed to several purchase episodes. It looks for a set of one-to-one matches that minimizes the combined cost across the group.

Build The Final Matched Dataset

The matching result currently contains only the purchase IDs, control IDs, and distance measures.

The final step is to merge the original purchase and control details back into those pairs so we can calculate returns for both sides.

selected_controls = control_candidates[['control_id', 'ticker', 'date', 'adjusted_close', 'drawdown', 'drawdown_pct']].rename(columns={'ticker': 'control_ticker',
                                                                                                                                       'date': 'control_date',
                                                                                                                                       'adjusted_close': 'control_signal_price',
                                                                                                                                       'drawdown': 'control_drawdown',
                                                                                                                                       'drawdown_pct': 'control_drawdown_pct'})

matched_sample = matched_pairs.merge(purchase_pool,on='purchase_id',how='left').merge(selected_controls,on='control_id',how='left')
matched_sample

Each row now contains one CEO-purchase episode and its matched no-purchase drawdown date:

We now have the two groups we actually wanted from the beginning: CEO purchases after large drawdowns and similar drawdowns in the same stocks without nearby CEO buying.

Compare CEO Purchases Against Similar No-Purchase Drawdowns

This is where the workflow finally earns its keep.

Each row in matched_sample contains two dates from the same stock:

• the first filing date of a CEO-buying episode

• a similar drawdown date with no nearby CEO purchase

From this point onward, both sides must be treated exactly the same. The CEO side enters on the first trading day after the filing date. The control side enters on the first trading day after the matched drawdown date. Both use the same adjusted prices and the same return horizons.

We'll first prepare the two signal-date columns and rebuild the ticker-level price map used earlier.

matched_sample['first_filing_date'] = pd.to_datetime(matched_sample['first_filing_date'])
matched_sample['control_date'] = pd.to_datetime(matched_sample['control_date'])
historical_df['date'] = pd.to_datetime(historical_df['date'])

prices = historical_df[['ticker', 'date', 'adjusted_close']].dropna().sort_values(['ticker', 'date'])
price_map = {ticker: group.reset_index(drop=True) for ticker, group in prices.groupby('ticker')}

The price map gives every ticker its own ordered history. That lets us use one return function for both the purchase and control dates instead of writing separate logic for each group.

Calculate Forward Returns From Any Signal Date

The next function takes only two inputs: a ticker and a signal date.

It finds the first trading session after that date, uses the adjusted close as the entry price, and calculates returns after 21, 63, 126, and 252 trading days.

def get_forward_returns(ticker, signal_date):
    result = {
        'entry_date': pd.NaT,
        'entry_price': np.nan,
        'return_1m': np.nan,
        'return_3m': np.nan,
        'return_6m': np.nan,
        'return_12m': np.nan
    }

    ticker_prices = price_map.get(ticker)

    if ticker_prices is None or pd.isna(signal_date):
        return pd.Series(result)

    dates = ticker_prices['date'].to_numpy(dtype='datetime64[ns]')
    entry_index = np.searchsorted(dates, np.datetime64(signal_date), side='right')

    if entry_index >= len(ticker_prices):
        return pd.Series(result)

    entry_price = ticker_prices.loc[entry_index, 'adjusted_close']
    result['entry_date'] = ticker_prices.loc[entry_index, 'date']
    result['entry_price'] = entry_price

    for label, days in {'1m': 21, '3m': 63, '6m': 126, '12m': 252}.items():
        target_index = entry_index + days

        if target_index < len(ticker_prices):
            target_price = ticker_prices.loc[target_index, 'adjusted_close']
            result[f'return_{label}'] = target_price / entry_price - 1

    return pd.Series(result)

The important detail is side='right'.

It prevents either group from entering on its signal date. The CEO-purchase return starts after the filing, and the control return starts after the matched drawdown date.

The function begins with missing values for every output. If a ticker is unavailable or there's not enough future price history for a horizon, that return simply stays as NaN.

Apply The Same Return Logic To Both Groups

Now we run the function twice for every matched pair.

The first pass uses the CEO-purchase ticker and filing date. The second uses the control ticker and control date. The returned columns are prefixed so the two sets remain easy to distinguish.

purchase_returns = matched_sample.apply(lambda row: get_forward_returns(row['ticker'], row['first_filing_date']), axis=1).add_prefix('purchase_')
control_returns = matched_sample.apply(lambda row: get_forward_returns(row['control_ticker'], row['control_date']), axis=1).add_prefix('control_')
matched_returns = pd.concat([
    matched_sample.reset_index(drop=True),
    purchase_returns.reset_index(drop=True),
    control_returns.reset_index(drop=True)], axis=1)

The resulting dataframe now places both outcomes side by side:

• CEO-purchase entry date and price

• control entry date and price

• CEO-purchase returns

• control returns

Not every pair survives at every horizon. A pair is usable only when both sides have enough future price history. That's why the number of observations falls as we move toward twelve months.

Build The Final Comparison

The last step is to compare the two return series at each horizon.

We 'll calculate:

• mean return for each group

• median return for each group

• mean return difference within the matched pairs

• median return difference within the matched pairs

• positive-return rate

• percentage of pairs where the CEO-purchase side beat the control

comparison = []

for horizon in ['1m', '3m', '6m', '12m']:
    purchase_col = f'purchase_return_{horizon}'
    control_col = f'control_return_{horizon}'
    valid = matched_returns[[purchase_col, control_col]].dropna()
    differences = valid[purchase_col] - valid[control_col]

    comparison.append({
        'horizon': horizon,
        'matched_pairs': len(valid),
        'purchase_mean': valid[purchase_col].mean(),
        'control_mean': valid[control_col].mean(),
        'mean_difference': differences.mean(),
        'purchase_median': valid[purchase_col].median(),
        'control_median': valid[control_col].median(),
        'median_difference': differences.median(),
        'purchase_positive_rate': (valid[purchase_col] > 0).mean(),
        'control_positive_rate': (valid[control_col] > 0).mean(),
        'purchase_win_rate': (valid[purchase_col] > valid[control_col]).mean()
    })

comparison_df = pd.DataFrame(comparison)
comparison_df

The paired statistics matter here.

median_difference is the median of:

CEO-purchase return - matched control return

for every pair. It's not simply the CEO median minus the control median.

The win rate asks an even more direct question: in what percentage of matched pairs did the CEO-purchase episode actually perform better?

The one-month result was weak. CEO-purchase episodes trailed the controls on the mean, median pair gap, positive-return rate, and win rate.

Three months was the one window where the result stayed consistent across the table. The CEO-purchase group returned 6.2 percentage points more on average, had a positive median paired advantage of 3.9 points, and won 59.6% of the matches.

The six- and twelve-month averages looked stronger than the typical pair. At twelve months, the CEO group returned nearly 10 percentage points more on average, yet it beat the control in only 37.9% of the comparisons.

That combination usually means a smaller number of large winners are pulling the average upward.

So the final answer is not that CEO buying always worked, or that it never mattered. The apparent edge depended heavily on the horizon, and three months was the only period where the different measures pointed in the same direction.

What The Case Study Found

The raw numbers made CEO buying look broadly bullish. After twelve months, the average return was 35.4%, and nearly two-thirds of the available observations were positive.

But once we added matched no-purchase drawdowns, the story became much narrower.

• One month showed no edge. CEO-purchase episodes underperformed their controls on the mean, median pair gap, positive-return rate, and win rate.

• Three months was the strongest window. The CEO group returned 6.2 percentage points more on average and beat its matched control in 59.6% of the pairs. This was the only horizon where the major measures pointed in the same direction.

• Six and twelve months were harder to trust. The averages were higher for the CEO-purchase group, but the median pair gaps were negative and most individual episodes lost to their controls.

• The drawdown itself explained a lot. Beaten-down stocks often rebounded even without CEO buying, so the raw post-purchase returns overstated the signal.

The most defensible conclusion is not that CEO buying predicts a long-term recovery. In this sample, it looked more like a possible three-month reversal signal, and even that result should be treated as exploratory rather than a trading rule.

What This Test Can And Can't Say

This was a 500-stock, screener-based sample, not the full market. The universe may carry survivorship bias, some code-P purchases may not have been fully discretionary, and matching similar drawdowns doesn't prove that CEO buying caused the returns. This is an exploratory case study, not a trading strategy.

The most useful part of the workflow was separating the insider signal from what beaten-down stocks already do on their own. Before adding controls, CEO purchases looked broadly bullish across several horizons. After adding them, the result became much narrower: a possible three-month edge, but no clean long-term guarantee.

That may feel less exciting than proving that CEO buying predicts a recovery. But it's a better answer. The workflow forced us to test the story we wanted to believe against a baseline, and the baseline changed the conclusion.

But a lot of teams still treat these numbers like something you glance at once. A screenshot in a deck. A one-off notebook cell. A quick check in a UI before a meeting.

That works until you need to answer basic questions that show up in real workflows:

What did TSLA's surface look like at 10:32? When did skew start steepening? Did the change come from the wings moving or the ATM shifting?

If you don't store the data as it arrives, you can't replay it, compare it, or audit it. You're stuck with whatever you happened to look at in the moment.

In this walkthrough, we'll build something small but practical: an internal database that continuously captures SpiderRock MLink's LiveImpliedQuote analytics for TSLA, stores each snapshot as queryable history, and also maintains a "latest view" table so you can pull the current surface state without scanning the full history.

The goal is not to build a trading system. It's to build a reliable internal dataset that you can monitor and query.

Note: SpiderRock MLink's LiveImpliedQuote analytics is a product offered for a fee, which includes exchange charges for the underlying market data used in its creation.

Table of Contents

• Prerequisites

• What Data We're Using

• Setup: Importing Packages

• Database Design

• Pulling LiveImpliedQuote

• Normalizing the Response Into Rows

• Writing To The Database

• Running a Short Polling Capture

• Analysis: Smile Reconstruction From the Database

  • Pick an Expiry with Good Coverage

  • Rebuild the Smile Across Snapshots

  • Zoom-In Around Spot

• Analysis: ATM IV and Skew Over Time

• Alert-Style Thresholds

• Wrapping Up

Prerequisites

Before running any of the code in this walkthrough, there are a few things you need to have in place.

On the API side, you need a SpiderRock MLink account with access to the LiveImpliedQuote feed. The examples use the REST interface, so no websocket setup is required, but you do need a valid API key. If you don't have one yet, you can reach out to SpiderRock directly to get access.

On the Python side, the environment is minimal. You need Python 3.10 or later for the tuple type hint syntax used in one of the function signatures. The external packages are requests, pandas, numpy, and matplotlib. Everything else – sqlite3, time, datetime – is part of the standard library. You can install the external dependencies with:

pip install requests pandas numpy matplotlib

No database setup is required beyond a writable local path. SQLite creates the file automatically on first run, so there's nothing to install or configure separately.

Finally, the walkthrough uses TSLA as the target symbol because it has a liquid and active options chain. If you want to swap in a different underlying, the only thing you need to change is the symbol variable in the config block.

What Data We're Using

This build is driven by one OptAnalytics message type from SpiderRock MLink: LiveImpliedQuote.

Each message represents an option contract and comes with the analytics you actually need for monitoring:

• the option identifier (symbol, expiry, strike, call or put)

• surface IV (sVol) and related surface fields

• Greeks (delta, gamma, theta, vega)

• context fields like underlying price (uPrc), time to expiry (years), and rate (rate)

• timestamps and calc source markers, which matter when you're turning a live feed into a database

We'll treat sVol as the main volatility field for the article and refer to it as surface IV. That keeps the workflow consistent when we rebuild smiles or compute skew proxies from stored history.

The demo uses TSLA because it has a rich and active options chain, which makes the database and queries more interesting even in a short capture window. The same pipeline works for any other underlying – the only thing you change is the symbol filter.

Setup: Importing Packages

Before touching the database or the API, we set up a small, repeatable environment. This section is intentionally minimal. We only import what we need for three things: making REST calls, storing data in SQLite, and doing basic analysis and plots.

import requests
import sqlite3
import pandas as pd
import numpy as np
import time
from datetime import datetime, timezone
import matplotlib.pyplot as plt
plt.style.use('ggplot')

• requests is used for calling MLink REST endpoints.

• sqlite3 gives us a lightweight database we can write to locally without extra setup.

• pandas and numpy are only for shaping and filtering the data once it comes back.

• time and datetime help us run a polling loop and timestamp each snapshot so the database becomes a real-time series.

Database Design

If the goal is to make live analytics queryable, the database design has to support two different needs.

First, you want an audit trail. Every snapshot should be preserved so you can reconstruct what the surface looked like at a specific time.

Second, you also want a fast way to answer "what does it look like right now" without scanning everything you've ever stored.

So we use two tables:

• implied_quote_history: Append-only. Every poll inserts a full snapshot.

• implied_quote_latest: One row per option contract. Each poll upserts into this table so it always reflects the most recent snapshot.

The core of both tables is a stable option identifier. In the feed, the option key is nested, so we normalize it into a single option_key string that includes symbol, expiry, strike, call or put, and venue fields. This becomes the primary key for the latest table and the main join key for queries.

#config
api_key = "YOUR SPIDERROCK API KEY"
mlink_url = "https://mlink-live.nms.saturn.spiderrockconnect.com/rest/json"

msg_type = "LiveImpliedQuote"

symbol = "TSLA"
poll_interval_s = 10
poll_duration_s = 120
limit = 2000

#create db connection
db_path = "/mnt/data/optanalytics_iv_greeks.db"

def get_conn(path: str = db_path):
    conn = sqlite3.connect(path)
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA synchronous=NORMAL;")
    return conn

#create db schema
def setup_db(path: str = db_path):
    conn = get_conn(path)
    cur = conn.cursor()

    cur.execute("""
    create table if not exists implied_quote_history (
        id integer primary key autoincrement,
        asof_ts text not null,

        option_key text not null,
        symbol text not null,
        expiry text not null,
        strike real not null,
        cp text not null,

        calc_source text,
        u_prc real,
        years real,
        rate real,

        s_vol real,
        atm_vol real,
        s_mark real,

        o_bid real,
        o_ask real,
        o_bid_iv real,
        o_ask_iv real,

        delta real,
        gamma real,
        theta real,
        vega real,

        src_ts text
    );
    """)

    cur.execute("""
    create index if not exists idx_hist_symbol_expiry_asof
    on implied_quote_history(symbol, expiry, asof_ts);
    """)

    cur.execute("""
    create index if not exists idx_hist_option_asof
    on implied_quote_history(option_key, asof_ts);
    """)

    cur.execute("""
    create table if not exists implied_quote_latest (
        option_key text primary key,

        last_asof_ts text not null,
        symbol text not null,
        expiry text not null,
        strike real not null,
        cp text not null,

        calc_source text,
        u_prc real,
        years real,
        rate real,

        s_vol real,
        atm_vol real,
        s_mark real,

        o_bid real,
        o_ask real,
        o_bid_iv real,
        o_ask_iv real,

        delta real,
        gamma real,
        theta real,
        vega real,

        src_ts text
    );
    """)

    cur.execute("""
    create index if not exists idx_latest_symbol_expiry
    on implied_quote_latest(symbol, expiry);
    """)

    conn.commit()
    conn.close()

setup_db()

This creates the SQLite database file and both tables. The history table is append-only and indexed for the two queries we'll run later: pulling snapshots by expiry and time, and pulling a specific option's timeline by option_key. The latest table is keyed by option_key, which lets us upsert and maintain a consistent "current view."

The columns we store are intentionally opinionated. We keep surface IV (s_vol), surface mark (s_mark), Greeks, and a few context fields. We also store timestamps so later we can reason about when a value was produced.

Pulling LiveImpliedQuote

Now we do the first live pull. The goal here is not to build a perfect filter. It's to confirm that we can retrieve a meaningful slice of TSLA option analytics and that the response structure is what we expect.

We request LiveImpliedQuote and filter by symbol using the where clause. The response is a list where most rows are actual LiveImpliedQuote messages, and one row at the end is a QueryResult summary.

def fetch_live_implied_quote(symbol: str, limit: int = 2000):
    where = f"okey.tk:eq:{symbol}"

    params = {
        "apiKey": api_key,
        "cmd": "getmsgs",
        "msgType": msg_type,
        "where": where,
        "limit": limit
    }

    r = requests.get(mlink_url, params=params)
    r.raise_for_status()
    return r.json()

raw = fetch_live_implied_quote(symbol, limit=limit)
print("raw messages:", len(raw))
print("first type:", raw[0].get("header", {}).get("mTyp") if raw else None)

This is a straight REST getmsgs call. We pass the API key, message type, and a simple symbol filter. The limit is important. It caps how many messages we get back in one poll, so for active underlyings, the returned set of strikes and expiries can vary between polls. That's fine for this tutorial, because the goal is to show the database pattern and the types of monitoring queries it enables.

This is the output you should see:

Normalizing the Response Into Rows

Right now, raw is a list of nested message objects. That format is fine for transport, but it's not something you can store or query directly. So now, we turn each LiveImpliedQuote message into one flat row with a consistent schema.

def make_option_key(okey: dict) -> str:
    return "|".join([
        str(okey.get("tk")),
        str(okey.get("dt")),
        str(okey.get("xx")),
        str(okey.get("cp")),
        str(okey.get("at")),
        str(okey.get("ts")),
    ])

def normalize_liq(raw: list, asof_ts: str, keep_calc_source: str = "Loop") -> pd.DataFrame:
    rows = []

    for row in raw:
        if row.get("header", {}).get("mTyp") != "LiveImpliedQuote":
            continue

        m = row.get("message", {})
        if keep_calc_source and m.get("calcSource") != keep_calc_source:
            continue

        pkey = m.get("pkey", {})
        okey = pkey.get("okey", {})
        if not okey:
            continue

        s_vol = m.get("sVol")
        if s_vol is None or s_vol == 0:
            continue

        o_bid = m.get("oBid", 0) or 0
        o_ask = m.get("oAsk", 0) or 0

        quote_ok = int(not (o_bid == 0 and o_ask == 0))

        rows.append({
            "asof_ts": asof_ts,
            "option_key": make_option_key(okey),

            "symbol": okey.get("tk"),
            "expiry": okey.get("dt"),
            "strike": okey.get("xx"),
            "cp": okey.get("cp"),

            "calc_source": m.get("calcSource"),
            "u_prc": m.get("uPrc"),
            "years": m.get("years"),
            "rate": m.get("rate"),

            "s_vol": s_vol,
            "atm_vol": m.get("atmVol"),
            "s_mark": m.get("sMark"),

            "o_bid": o_bid,
            "o_ask": o_ask,
            "o_bid_iv": m.get("oBidIv"),
            "o_ask_iv": m.get("oAskIv"),
            "quote_ok": quote_ok,

            "delta": m.get("de"),
            "gamma": m.get("ga"),
            "theta": m.get("th"),
            "vega": m.get("ve"),

            "src_ts": m.get("timestamp"),
        })

    df = pd.DataFrame(rows)
    if df.empty:
        return df

    df = (
        df.sort_values("src_ts")
          .drop_duplicates(subset=["option_key"], keep="last")
          .reset_index(drop=True)
    )
    return df

asof_ts = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
snapshot_df = normalize_liq(raw, asof_ts)

print("snapshot rows:", len(snapshot_df))
print("quote_ok distribution:", snapshot_df["quote_ok"].value_counts().to_dict() if not snapshot_df.empty else {})
snapshot_df.head()

There are three practical decisions baked into this normalization step:

• First, we build a stable option_key from the option identifier so we have a consistent primary key for the latest table.

• Second, we keep only calcSource="Loop". LiveImpliedQuote can include both Tick and Loop records. Loop records tend to be more consistent for snapshot-style analysis because the underlying reference price is stable across the surface.

• Third, we avoid aggressive filtering. In this dataset, the top-of-book bid and ask fields can be zero even when the analytics fields are populated. So instead of dropping those rows, we store a quote_ok flag and keep the record. That keeps the pipeline usable while still making it obvious later which rows had live quotes.

This is the output:

At this point, one row represents one option contract snapshot. The fact that quote_ok is 0 across the board simply means bid and ask are not populated in this slice, even though surface IV, Greeks, and other analytics fields are present. That's still useful for building a monitoring database, because the core idea here is tracking the evolution of analytics over time, not reconstructing executable markets.

Writing to the Database

Now that we have a clean snapshot DataFrame, the job is to persist it in two places.

History table: Append everything. This is the audit log. Latest table: Upsert by option_key. This is the fast "current view."

This separation is what makes the database useful. History lets you reconstruct any past snapshot. Latest lets you answer "what does the surface look like right now" without scanning time series.

def safe_add_column(table: str, col: str, col_type: str, path: str = db_path):
    conn = get_conn(path)
    cur = conn.cursor()
    existing = [r[1] for r in cur.execute(f"PRAGMA table_info({table});").fetchall()]
    if col not in existing:
        cur.execute(f"ALTER TABLE {table} ADD COLUMN {col} {col_type};")
    conn.commit()
    conn.close()

safe_add_column("implied_quote_history", "quote_ok", "INTEGER")
safe_add_column("implied_quote_latest", "quote_ok", "INTEGER")

def write_snapshot_to_db(df: pd.DataFrame, path: str = db_path) -> tuple[int, int]:
    if df.empty:
        return 0, 0

    conn = get_conn(path)
    cur = conn.cursor()

    cols = [
        "asof_ts",
        "option_key","symbol","expiry","strike","cp",
        "calc_source","u_prc","years","rate",
        "s_vol","atm_vol","s_mark",
        "o_bid","o_ask","o_bid_iv","o_ask_iv",
        "delta","gamma","theta","vega",
        "quote_ok","src_ts"
    ]

    for c in cols:
        if c not in df.columns:
            df[c] = None

    insert_df = df[cols].copy()

    cur.executemany(
        """
        insert into implied_quote_history (
            asof_ts,
            option_key, symbol, expiry, strike, cp,
            calc_source, u_prc, years, rate,
            s_vol, atm_vol, s_mark,
            o_bid, o_ask, o_bid_iv, o_ask_iv,
            delta, gamma, theta, vega,
            quote_ok, src_ts
        ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        insert_df.itertuples(index=False, name=None)
    )
    history_inserted = cur.rowcount

    cur.executemany(
        """
        insert into implied_quote_latest (
            option_key,
            last_asof_ts, symbol, expiry, strike, cp,
            calc_source, u_prc, years, rate,
            s_vol, atm_vol, s_mark,
            o_bid, o_ask, o_bid_iv, o_ask_iv,
            delta, gamma, theta, vega,
            quote_ok, src_ts
        ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        on conflict(option_key) do update set
            last_asof_ts=excluded.last_asof_ts,
            symbol=excluded.symbol,
            expiry=excluded.expiry,
            strike=excluded.strike,
            cp=excluded.cp,
            calc_source=excluded.calc_source,
            u_prc=excluded.u_prc,
            years=excluded.years,
            rate=excluded.rate,
            s_vol=excluded.s_vol,
            atm_vol=excluded.atm_vol,
            s_mark=excluded.s_mark,
            o_bid=excluded.o_bid,
            o_ask=excluded.o_ask,
            o_bid_iv=excluded.o_bid_iv,
            o_ask_iv=excluded.o_ask_iv,
            delta=excluded.delta,
            gamma=excluded.gamma,
            theta=excluded.theta,
            vega=excluded.vega,
            quote_ok=excluded.quote_ok,
            src_ts=excluded.src_ts
        """,
        insert_df[[
            "option_key","asof_ts","symbol","expiry","strike","cp",
            "calc_source","u_prc","years","rate",
            "s_vol","atm_vol","s_mark",
            "o_bid","o_ask","o_bid_iv","o_ask_iv",
            "delta","gamma","theta","vega",
            "quote_ok","src_ts"
        ]].itertuples(index=False, name=None)
    )
    latest_upserted = cur.rowcount

    conn.commit()
    conn.close()
    return history_inserted, latest_upserted

hist_n, latest_n = write_snapshot_to_db(snapshot_df)
print("history inserted:", hist_n)
print("latest upserted:", latest_n)

We batch write using executemany so inserts are fast even with thousands of option rows. The history insert is straightforward. The latest write uses a SQLite upsert keyed on option_key, which means if the contract already exists in the latest table, its fields are overwritten with the newest snapshot.

You should see:

After the first write, both tables have the same number of rows. That's expected, because there is only one snapshot in history so far. Once we start polling multiple snapshots, the history table will grow every cycle, while the latest table will stay roughly flat and continue updating in place.

Running a Short Polling Capture

At this point, the pipeline works end-to-end for a single snapshot. The whole point of the database, though, is to turn live analytics into a time series. So we run a short capture window and store multiple snapshots back-to-back.

This isn't meant to be a production scheduler. It's just a simple loop that runs for a couple of minutes, polls every few seconds, timestamps the snapshot, and writes it to both tables.

def poll_and_write(symbol: str, duration_s: int = poll_duration_s, interval_s: int = poll_interval_s):
    start = time.time()
    polls = 0
    total_hist = 0

    while time.time() - start < duration_s:
        asof_ts = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")

        raw = fetch_live_implied_quote(symbol, limit=limit)
        df = normalize_liq(raw, asof_ts)

        hist_n, latest_n = write_snapshot_to_db(df)
        polls += 1
        total_hist += hist_n

        print(f"[{polls}] {asof_ts} snapshot_rows={len(df)} history+={hist_n} latest_upsert={latest_n}")
        time.sleep(interval_s)

    print(f"done. polls={polls}, total_history_added={total_hist}")

poll_and_write(symbol, duration_s=120, interval_s=10)

Each loop iteration represents one snapshot. We generate a UTC timestamp (asof_ts), pull the latest batch from LiveImpliedQuote, normalize it into rows, then write it into the database. The history table accumulates every snapshot. The latest table overwrites by option_key, so it always represents the most recent view.

One practical detail is worth calling out. The API call is capped by limit, so you're not guaranteed to receive an identical set of strikes and expiries every poll. That's why snapshot_rows can vary between iterations.

In production, you usually stabilize the slice by pinning specific expiries and a strike band or by interpolating IV to fixed moneyness points. For this tutorial, we're keeping ingestion simple and focusing on the database pattern and the monitoring queries it enables.

You should see per-poll telemetry like this:

[1] 2026-04-14T18:09:29Z snapshot_rows=1454 history+=1454 latest_upsert=1454
...
done. polls=9, total_history_added=12806

This confirms the database is building a time series. Over nine polls, you stored 12,806 option rows in history. The latest table is updated each time, but it doesn't grow in the same way as history because it overwrites per contract key.

From the next section, we'll stop writing and start querying.

Analysis: Smile Reconstruction From the Database

Once the data is in implied_quote_history, the workflow flips. We stop thinking in terms of "API responses" and start thinking in terms of "queries." This section does two things. First, it picks an expiry that has enough rows to be representative. Then it reconstructs the call-side volatility smile for that expiry across a few timestamps.

Pick an Expiry with Good Coverage

If you pick an expiry that only appears sporadically in the captured snapshots, the smile plot will be misleading. So we start by looking at which expiries have the most rows in the history table.

conn = get_conn()

expiry_counts = pd.read_sql_query(
    """
    select expiry, count(*) as n
    from implied_quote_history
    where symbol = ?
    group by expiry
    order by n desc
    limit 10
    """,
    conn,
    params=(symbol,)
)

conn.close()
expiry_counts

This query scans only the history table, filters to TSLA, and counts how many option rows exist per expiry across the capture window. We keep the top 10 and pick the first one as the expiry we'll reconstruct.

The expiry date 2026-11-20 has the highest count.

Here, the count doesn't mean this expiry is "best" in any trading sense. It just means it showed up most consistently in the captured data. That makes it a practical choice for a clean smile comparison.

Rebuild the Smile Across Snapshots

Now we query the stored history for one expiry, keep only calls, and plot surface IV (s_vol) against strike for multiple snapshot timestamps.

chosen_expiry = "2026-11-20" 

conn = get_conn()
smile = pd.read_sql_query(
    """
    select asof_ts, strike, cp, s_vol, u_prc
    from implied_quote_history
    where symbol = ? and expiry = ?
    """,
    conn,
    params=(symbol, chosen_expiry)
)
conn.close()

smile_calls = smile[smile["cp"] == "Call"].copy()

ts_list = sorted(smile_calls["asof_ts"].unique())
pick = [ts_list[0], ts_list[len(ts_list)//2], ts_list[-1]]

plt.figure(figsize=(9,5))
for ts in pick:
    g = smile_calls[smile_calls["asof_ts"] == ts].sort_values("strike")
    plt.plot(g["strike"], g["s_vol"], label=ts)

plt.title(f"{symbol} Vol Smile (Calls) | Expiry {chosen_expiry} | 3 snapshots")
plt.xlabel("Strike")
plt.ylabel("Implied Vol (s_vol)")
plt.grid(True)
plt.legend()
plt.show()

We pull all rows for the chosen expiry from history, then filter to calls so we don't mix put and call shapes. To keep the plot readable, we only plot three snapshots. First, middle, and last.

Over a short capture window, the smiles often overlap heavily. That doesn't mean the system isn't working. It usually means the surface didn't move much in those two minutes. The important part is that we can reconstruct and compare it purely from stored history.

Zoom-In Around Spot

The full-range plot is useful for shape, but it can hide small shifts near the region people actually care about. So we zoom to a band around the underlying price.

s0 = float(smile_calls["u_prc"].dropna().median())
low, high = s0 * 0.6, s0 * 1.4

for ts in pick:
    g = smile_calls[smile_calls["asof_ts"] == ts].sort_values("strike")
    g = g[(g["strike"] >= low) & (g["strike"] <= high)]
    plt.plot(g["strike"], g["s_vol"], label=ts)

plt.title(f"{symbol} Vol Smile (Calls) | Expiry {chosen_expiry} | zoomed")
plt.xlabel("Strike")
plt.ylabel("Implied Vol (s_vol)")
plt.grid(True)
plt.legend(fontsize=8)
plt.show()

We take a robust spot proxy from the stored u_prc values and then keep strikes within a range around it. The goal is not precision. It's to make the chart readable and show whether the near-ATM region is drifting.

Here, even small changes become visible. This is also why storing history matters. If you only looked at one snapshot in isolation, these shifts would be easy to miss or dismiss.

Analysis: ATM IV and Skew Over Time

A full smile plot is useful, but it's not always the fastest way to monitor a surface. In practice, teams usually track a few summary numbers per expiry so they can spot changes quickly, then drill down only when something looks off.

Here we reduce each stored snapshot into two metrics for a single expiry.

• ATM IV: Surface IV at the strike closest to spot.

• Skew proxy: Surface IV at 0.9 times spot minus surface IV at 1.1 times spot, using the closest available strikes.

chosen_expiry = "2026-11-20"

conn = get_conn()
df = pd.read_sql_query(
    """
    select asof_ts, strike, s_vol, u_prc
    from implied_quote_history
    where symbol = ? and expiry = ? and cp = 'Call'
    """,
    conn,
    params=(symbol, chosen_expiry)
)
conn.close()

df["strike"] = df["strike"].astype(float)
df["s_vol"] = df["s_vol"].astype(float)

def closest_iv(grp: pd.DataFrame, target_strike: float):
    g = grp.iloc[(grp["strike"] - target_strike).abs().argsort()[:1]]
    return float(g["s_vol"].iloc[0]), float(g["strike"].iloc[0])

rows = []
for ts, grp in df.groupby("asof_ts"):
    spot = float(grp["u_prc"].dropna().median())
    atm_target = spot
    down_target = spot * 0.9
    up_target = spot * 1.1

    atm_iv, atm_k = closest_iv(grp, atm_target)
    down_iv, down_k = closest_iv(grp, down_target)
    up_iv, up_k = closest_iv(grp, up_target)

    rows.append({
        "asof_ts": ts,
        "spot": spot,
        "atm_strike": atm_k,
        "atm_iv": atm_iv,
        "k90": down_k,
        "iv_90": down_iv,
        "k110": up_k,
        "iv_110": up_iv,
        "skew_90_110": down_iv - up_iv
    })

metrics = pd.DataFrame(rows).sort_values("asof_ts").reset_index(drop=True)
metrics

We query the history table for one expiry and keep only calls, then group by snapshot timestamp. For each snapshot, we use the median u_prc as a spot proxy and pick the closest available strike to spot. That gives ATM IV. We repeat the same approach for 0.9 times spot and 1.1 times spot and compute a skew proxy as the difference.

The table also stores the actual strikes used (atm_strike, k90, k110). Options strikes are discrete, so the nearest strike can change between snapshots. Keeping the chosen strikes visible makes the metric explainable when it moves.

The output is a table with one row per snapshot timestamp and the computed metrics.

Now that we have a clean time series table, we can visualize the two metrics. First, ATM IV. Then, the skew proxy.

plt.plot(metrics["asof_ts"], metrics["atm_iv"])
plt.title(f"{symbol} ATM IV over time | Expiry {chosen_expiry}")
plt.xticks(rotation=30, ha="right")
plt.ylabel("ATM IV (s_vol)")
plt.grid(True)
plt.show()

plt.plot(metrics["asof_ts"], metrics["skew_90_110"])
plt.title(f"{symbol} Skew proxy (IV@0.9S - IV@1.1S) | Expiry {chosen_expiry}")
plt.xticks(rotation=30, ha="right")
plt.ylabel("Skew proxy")
plt.grid(True)
plt.show()

Here is the first chart, ATM IV over time.

ATM IV tends to move slowly over short windows unless there is a sharp repricing event. In this run, it stays fairly stable, which is a realistic outcome for a short capture. The value here is that the database turns "fairly stable" into something you can quantify and compare later, rather than a vague impression.

Here is the second chart, Skew proxy over time.

The skew proxy is more sensitive because it's based on wing points. If it changes, it usually means the downside is being repriced differently from the upside for that expiry. One nuance is that the nearest available strike can change between snapshots, which can create step-like moves even when the surface isn't moving dramatically. That's why we keep k90 and k110 in the metrics table. It keeps the skew plot explainable.

Alert-Style Thresholds

Once you have a metrics table per snapshot, adding a monitoring layer is straightforward. The idea isn't to generate trades. It's to flag when the surface moves enough that someone should look closer.

Here we do two checks:

• ATM IV change alert: Flag if ATM IV changes more than a small threshold between snapshots.

• Skew change alert: Flag if the skew proxy changes more than a threshold between snapshots.

alerts = metrics.copy()

alerts["atm_iv_change"] = alerts["atm_iv"].diff()
alerts["skew_change"] = alerts["skew_90_110"].diff()

atm_thresh = 0.002    
skew_thresh = 0.003   

alerts["atm_alert"] = alerts["atm_iv_change"].abs() >= atm_thresh
alerts["skew_alert"] = alerts["skew_change"].abs() >= skew_thresh

alerts[[
    "asof_ts",
    "atm_iv", "atm_iv_change", "atm_alert",
    "skew_90_110", "skew_change", "skew_alert",
    "atm_strike", "k90", "k110"
]]

We take the per-snapshot metrics table and compute first differences. Then we compare those changes to thresholds and store boolean flags. The output table keeps both the metrics and the strikes used for the calculations, so any alert is explainable rather than a black box.

In this run, the ATM IV alerts are all false, while the skew alert triggers once.

The skew alert fires because the skew proxy jumps by more than the threshold between two snapshots. This is explainable. If you see the table, you can see the strikes used for the proxy changed around the same time (k90 shifts from 340 to 315). Because strikes are discrete, nearest-strike metrics can step even when the surface is not moving dramatically.

To make this easier to read, we also plot the two series and mark alert points.

plt.plot(alerts["asof_ts"], alerts["atm_iv"])
for i, r in alerts[alerts["atm_alert"]].iterrows():
    plt.scatter(r["asof_ts"], r["atm_iv"],  s=30, edgecolors="r", alpha=0.6, linewidth=2)
plt.title(f"{symbol} ATM IV with alerts | Expiry {chosen_expiry}")
plt.xticks(rotation=30, ha="right")
plt.grid(True)
plt.show()

plt.plot(alerts["asof_ts"], alerts["skew_90_110"])
for i, r in alerts[alerts["skew_alert"]].iterrows():
    plt.scatter(r["asof_ts"], r["skew_90_110"], s=30, edgecolors="r", alpha=0.6, linewidth=2)
plt.title(f"{symbol} Skew proxy with alerts | Expiry {chosen_expiry}")
plt.xticks(rotation=30, ha="right")
plt.grid(True)
plt.show()

Both plots use the same pattern. Plot the metric as a line, then overlay a marker on any timestamp where the corresponding alert flag is true. This makes it obvious when something crossed the threshold.

This chart represents skew proxy with alerts.

This chart shows one alert marker, which matches what we saw in the table.

The ATM IV plot isn't featured since there are no alert points.

Wrapping Up

In this walkthrough, we used SpiderRock MLink's LiveImpliedQuote feed for TSLA and turned it into a small internal database you can query. We stored every snapshot in an append-only history table, maintained a latest view keyed by a stable option identifier, then used that stored data to rebuild a smile, track ATM surface IV and a simple skew proxy, and add a basic alert rule on top.

This fits well in B2B workflows because it turns live analytics into something operational: a dataset you can audit, replay, and monitor. The same pattern works whether you're building an internal dashboard, running routine surface checks for a desk, or doing a quick post-event review without relying on screenshots and one-off notebook runs.

If you want to extend it, the most practical next steps are longer capture windows, tracking multiple symbols, and moving from SQLite to Postgres once the data volume grows. If metric stability becomes important, you can also standardize the slice you track per poll or interpolate IV to fixed moneyness points so skew measures don't step when nearest strikes change.

With that being said, you've reached the end of the article. Hope you learned something new and useful.