Build and Monetize a Market Data Replay System for Quantitative Trading

Turning historical tick data into a realistic, controllable stream is a valuable skill for quantitative traders, AI researchers, and developers who sell data‑driven tools. By creating a replay engine that mimics live market behavior, you can offer back‑testing environments, strategy‑validation services, or even a SaaS platform on marketplaces like Fiverr, Upwork, or Gumroad. This guide walks you through building a complete replay system in Python using FastAPI and WebSockets, then shows how to package and sell the solution.

Why a Replay Engine Matters for AI‑Driven Trading

Most machine‑learning models for Quantitative Trading assume they will see data tick‑by‑tick, with no peek into the future. When you train on a static CSV file, the model can inadvertently learn from future information, leading to over‑optimistic results. A replay system forces the model to consume events exactly as they occurred, preserving causality.

With a controllable clock, adjustable speed, pause/resume, and seek capabilities, you can:

• Run deterministic back‑tests that are repeatable across CI pipelines.
• Debug strategy logic by stepping through specific market moments.
• Generate synthetic live‑feeds for demo purposes or client presentations.
• Offer a “replay‑as‑a‑service” where users upload their own tick data and receive a WebSocket stream.

Project Overview

The system consists of three logical layers:

1. Data Loader – pulls a full day of AAPL tick data from EODHD, normalizes it into an immutable event tape.
2. Replay Clock & Session – drives the tape according to wall‑clock time, supports speed control, pause, resume, and seeking.
3. API Service – a FastAPI server exposing REST endpoints for control and a WebSocket endpoint that pushes each trade event to connected clients.

A lightweight consumer demonstrates how a strategy can compute rolling VWAP and market state solely from the incoming stream, proving that state rebuilds correctly after a seek operation.

Setting Up the Python Environment

Start with a fresh virtual environment to keep dependencies clean.

Configuration File

Create replay/config.py to centralize settings:

Downloading and Normalizing Market Data

EODHD provides a tick‑by‑tick endpoint for equities. A single trading day for AAPL can exceed one million rows. The goal is to turn this raw CSV into a deterministic tape where each row represents an event with a monotonic timestamp.

Loader Implementation

Create replay/loader.py:

import pandas as pd
import requests
from .config import EODHD_API_KEY, TAPE_PATH

def download_aapl_ticks(date_str: str) -> pd.DataFrame:
 url = f"https://eodhistoricaldata.com/api/intraday/AAPL.US?interval=1&fmt=json&api_token={EODHD_API_KEY}&from={date_str}"
 resp = requests.get(url, timeout=30)
 resp.raise_for_status()
 data = resp.json()
 df = pd.DataFrame(data)
 # Expected columns: timestamp, open, high, low, close, volume
 df.rename(columns={"timestamp": "ts"}, inplace=True)
 df["ts"] = pd.to_datetime(df["ts"])
 return df

def normalize_to_tape(df: pd.DataFrame) -> pd.DataFrame:
 # Ensure deterministic ordering
 df = df.sort_values("ts").reset_index(drop=True)
 # Add an event ID for replay consistency
 df["event_id"] = range(len(df))
 # Keep only fields needed by downstream consumers
 tape = df[["event_id", "ts", "open", "high", "low", "close", "volume"]]
 return tape

def build_tape(date_str: str):
 raw = download_aapl_ticks(date_str)
 tape = normalize_to_tape(raw)
 os.makedirs(os.path.dirname(TAPE_PATH), exist_ok=True)
 tape.to_parquet(TAPE_PATH, index=False)
 print(f"Tape saved to {TAPE_PATH} with {len(tape)} events")

Run the loader once to create the tape:

Building the Historical Replay Clock

The clock translates wall‑clock time into tape indices. It must support variable playback speed, pausing, and seeking.

Clock Core

Create replay/clock.py:

import time
import threading
from typing import Optional

class ReplayClock:
 def __init__(self, tape_length: int, base_speed: float = 1.0):
 self.tape_length = tape_length
 self.base_speed = base_speed
 self._speed = base_speed
 self._paused = False
 self._seek_request: Optional[int] = None
 self._start_wall = None # wall time when playback started
 self._start_tape = 0 # tape index at start_wall
 self._lock = threading.Lock()

 @property
 def speed(self) -> float:
 with self._lock:
 return self._speed

 @speed.setter
 def speed(self, value: float):
 with self._lock:
 if value <= 0:
 raise ValueError("Speed must be positive")
 # Adjust start points to keep current tape position unchanged
 current = self._current_tape_index()
 self._speed = value
 self._start_wall = time.time()
 self._start_tape = current

 def pause(self):
 with self._lock:
 self._paused = True
 self._start_wall = None

 def resume(self):
 with self._lock:
 if not self._paused:
 return
 self._paused = False
 self._start_wall = time.time()
 self._start_tape = self._current_tape_index()

 def seek(self, index: int):
 with self._lock:
 if not 0 <= index < self.tape_length:
 raise IndexError("Seek index out of bounds")
 self._seek_request = index
 self._start_wall = time.time()
 self._start_tape = index

 def _current_tape_index(self) -> int:
 if self._start_wall is None:
 return int(self._start_tape)
 elapsed = time.time() - self._start_wall
 advance = int(elapsed * self._speed)
 return int(self._start_tape) + advance

 def next_index(self) -> Optional[int]:
 with self._lock:
 if self._seek_request is not None:
 idx = self._seek_request
 self._seek_request = None
 return idx
 if self._paused:
 return None
 idx = self._current_tape_index()
 if idx >= self.tape_length:
 return None # end of tape
 return idx

The clock is thread‑safe, allowing the FastAPI endpoint to modify speed or pause while a background thread reads the next index.

Session Management

A session orchestrates the clock, loads the tape, and emits events.

Session Class

Create replay/session.py:

import pandas as pd
import threading
from .clock import ReplayClock
from .config import TAPE_PATH

class ReplaySession:
 def __init__(self):
 self.tape = pd.read_parquet(TAPE_PATH)
 self.clock = ReplayClock(len(self.tape))
 self._thread: Optional[threading.Thread] = None
 self._stop_event = threading.Event()
 self._latest_event = None
 self._listeners = [] # callbacks for each new event

 def add_listener(self, callback):
 self._listeners.append(callback)

 def _run_loop(self):
 while not self._stop_event.is_set():
 idx = self.clock.next_index()
 if idx is None:
 break # tape finished or stopped
 row = self.tape.iloc[idx]
 event = {
 "event_id": int(row["event_id"]),
 "ts": row["ts"].isoformat(),
 "open": float(row["open"]),
 "high": float(row["high"]),
 "low": float(row["low"]),
 "close": float(row["close"]),
 "volume": int(row["volume"])
 }
 self._latest_event = event
 for cb in self._listeners:
 cb(event)
 # Sleep to avoid busy‑waiting; actual timing is enforced by clock
 time.sleep(0.001)

 def start(self):
 if self._thread and self._thread.is_alive():
 return
 self._stop_event.clear()
 self._thread = threading.Thread(target=self._run_loop, daemon=True)
 self._thread.start()

 def pause(self):
 self.clock.pause()

 def resume(self):
 self.clock.resume()

 def seek(self, index: int):
 self.clock.seek(index)

 def set_speed(self, speed: float):
 self.clock.speed = speed

 def stop(self):
 self._stop_event.set()
 if self._thread:
 self._thread.join(timeout=2)

 def get_state(self):
 return {
 "tape_length": len(self.tape),
 "current_index": self.clock._current_tape_index() if not self._clock._paused else self.clock._start_tape,
 "speed": self.clock.speed,
 "paused": self.clock._paused,
 "latest_event": self._latest_event
 }

The session lets external components register a callback that receives each trade event as a dictionary. This design makes it easy to plug in a WebSocket broadcaster.

Exposing Controls

Now we build a thin HTTP layer that lets clients start, pause, seek, and adjust speed. The same server pushes events over a WebSocket.

API Server

Create api/server.py:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.responses import JSONResponse
import asyncio
from replay.session import ReplaySession

app = FastAPI(title="Market Data Replay Service")
session = ReplaySession()

# ---- REST CONTROLS ----
@app.post("/start")
def start_replay():
 session.start()
 return JSONResponse({"status": "started"})

@app.post("/pause")
def pause_replay():
 session.pause()
 return JSONResponse({"status": "paused"})

@app.post("/resume")
def resume_replay():
 session.resume()
 return JSONResponse({"status": "resumed"})

@app.post("/seek")
def seek_replay(index: int):
 try:
 session.seek(index)
 except IndexError as e:
 raise HTTPException(status_code=400, detail=str(e))
 return JSONResponse({"status": "seeked", "index": index})

@app.post("/set_speed")
def set_speed(speed: float):
 if speed <= 0:
 raise HTTPException(status_code=400, detail="Speed must be positive")
 session.set_speed(speed)
 return JSONResponse({"status": "speed_set", "speed": speed})

@app.get("/state")
def get_state():
 return JSONResponse(session.get_state())

# ---- WEBSOCKET STREAM ----
@app.websocket("/ws/trades")
async def trades_websocket(websocket: WebSocket):
 await websocket.accept()
 def on_event(event):
 # asyncio.create_task is safe because we are already in the event loop
 asyncio.create_task(websocket.send_json(event))
 session.add_listener(on_event)
 try:
 while True:
 # Keep connection alive; we could also receive ping/pong messages
 await websocket.receive_text()
 except WebSocketDisconnect:
 # Cleanup listener on disconnect
 session._listeners.remove(on_event)

Entry Point

Create api/run.py:

import uvicorn
from api.server import app

if __name__ == "__main__":
 uvicorn.run("api.server:app", host="0.0.0.0", port=8000, reload=False)

Start the service:

python api/run.py

You now have:

• REST endpoints at http://localhost:8000/start, /pause, /resume, /seek, /set_speed, and /state.
• A WebSocket at ws://localhost:8000/ws/trades that streams each trade as a JSON object.

Building a Stateful Consumer (Rolling VWAP Example)

To prove the replay works for real‑world quantitative logic, implement a consumer that calculates volume‑weighted average price (VWAP) over a sliding window using only the events it receives.

Consumer Implementation

Create consumer/consumer.py:

import asyncio
import websockets
import json
from collections import deque

class VWAPCalculator:
 def __init__(self, window_seconds: int = 60):
 self.window = window_seconds
 self.prices = deque() # each item: (timestamp, price, volume)
 self.volume_sum = 0.0
 self.price_volume_sum = 0.0

 def _purge_old(self, now):
 while self.prices and (now - self.prices[0][0]).total_seconds() > self.window:
 ts, price, vol = self.prices.popleft()
 self.volume_sum -= vol
 self.price_volume_sum -= price * vol

 def update(self, ts_iso, price, volume):
 ts = pd.to_datetime(ts_iso)
 self._purge_old(ts)
 self.prices.append((ts, price, volume))
 self.volume_sum += volume
 self.price_volume_sum += price * volume
 if self.volume_sum == 0:
 return 0.0
 return self.price_volume_sum / self.volume_sum

async def connect_and_compute(uri="ws://localhost:8000/ws/trades"):
 async with websockets.connect(uri) as ws:
 vwap = VWAPCalculator(window_seconds=30) # 30‑second VWAP
 async for message in ws:
 event = json.loads(message)
 vwap_val = vwap.update(event["ts"], event["close"], event["volume"])
 print(f"{event['ts']} | price={event['close']:.2f} | vol={event['volume']} | VWAP={vwap_val:.2f}")

if __name__ == "__main__":
 asyncio.run(connect_and_compute())

This consumer:

• Connects to the WebSocket stream.
• Maintains a rolling window of trades.
• Outputs the VWAP after each incoming tick.
• Relies solely on the data it receives, proving state can be rebuilt after a seek (the clock will resend events from the new index).

Testing and Validation

Automated tests ensure that seeking correctly rebuilds downstream state. Use pytest with an in‑memory tape for speed.

Test Suite

Create tests/test_replay.py:

import pytest
import pandas as pd
from replay.session import ReplaySession
from replay.clock import ReplayClock

@pytest.fixture
def small_tape(tmp_path):
 df = pd.DataFrame({
 "event_id": range(10),
 "ts": pd.date_range("2024-01-01", periods=10, freq="1s"),
 "open": [1]*10,
 "high": [1]*10,
 "low": [1]*10,
 "close": [1]*10,
 "volume": [10]*10
 })
 path = tmp_path / "tape.parquet"
 df.to_parquet(path)
 return path

def test_seek_rebuilds_state(monkeypatch, small_tape):
 # Monkey‑patch the session to use our tiny tape
 import replay.session as sess_mod
 original_init = sess_mod.ReplaySession.__init__
 def new_init(self):
 self.tape = pd.read_parquet(small_tape)
 self.clock = ReplayClock(len(self.tape))
 self._thread = None
 self._stop_event = None
 self._latest_event = None
 self._listeners = []
 monkeypatch.setattr(sess_mod.ReplaySession, "__init__", new_init)

 session = sess_mod.ReplaySession()
 received = []
 session.add_listener(lambda e: received.append(e["event_id"]))
 session.start()
 # Let first 5 events flow
 import time; time.sleep(0.2)
 session.pause()
 assert len(received) == 5
 # Seek to index 2 and resume
 session.seek(2)
 session.resume()
 time.sleep(0.2)
 # After seek we should see events 2,3,4,... again
 assert received[-5:] == [2,3,4,5,6]
 session.stop()

Run the tests:

pytest -q

A passing suite confirms that the clock, session, and any listener (like the VWAP consumer) correctly reconstruct state after a seek operation—a crucial property for trustworthy back‑testing and live‑simulation environments.

Monetizing Your Replay System

With a functional, test‑backed replay engine you can create several revenue streams:

1. Freelance Gigs – Offer to build custom replay servers for hedge funds or prop shops on Upwork or Fiverr. Highlight your ability to deliver deterministic market feeds, adjustable speed, and WebSocket integration.
2. Data‑as‑a‑Service – Host the FastAPI service on a cheap VPS (e.g., DigitalOcean, Hetzner). Charge clients a monthly subscription for access to a curated set of equity tapes (AAPL, MSFT, SPY) with adjustable playback. Use Gumroad to sell access tokens or API keys.
3. Educational Content – Record a walkthrough video showing how to download tick data, normalize it, and integrate the replay with a trading bot. Post the video on YouTube and link to your code repository (GitHub) and a paid “premium” version on Gumroad.
4. AI Model Marketplace – Package a pre‑trained reinforcement‑learning agent that expects a live‑like stream. Sell the agent together with the replay server as a bundle on platforms like AI Marketplace or your own Shopify store.

When pitching, emphasize the technical differentiators:

• Deterministic replay based on immutable Parquet tape.
• Sub‑millisecond timing control
• Horizontal scalability – multiple clients can connect to the same WebSocket stream without affecting playback.
• Easy integration – any Python or JavaScript client can consume JSON over WebSockets.

Next Steps and Enhancements

Once the core is stable, consider these upgrades to increase marketability:

1. **Multiple Symbols** – Extend the loader to accept a list of tickers and merge them into a single tape sorted by timestamp.
2. **Custom Event Types** – Add support for quotes, trades, and market‑depth (Level 2) events.
3. **Authentication** – Protect the WebSocket endpoint with JWT or API‑key verification to enable multi‑tenant SaaS.
4. **Dockerization** – Provide a Dockerfile and docker‑compose.yml for one‑click deployment.
5. **Monitoring** – Export Prometheus metrics (events per second, lag, active connections) to help clients assess performance.
6. **Extended Controls** – Add frame‑by‑frame stepping, loop mode, and the ability to export a segment of the tape as a new file.

Each of these features can be released as a paid upgrade or a consulting add‑on, further boosting your earnings potential.

Conclusion

Building a market data replay system in Python with FastAPI and WebSockets gives you a versatile tool that serves both quantitative developers and AI engineers. By delivering a deterministic, controllable stream of historical ticks, you enable reliable strategy testing, debugging, and demonstration—capabilities that clients are willing to pay for.

Follow the steps above to create the loader, clock, session, API layer, and a sample VWAP consumer. Test thoroughly with pytest to guarantee state safety after seeks. Then package the solution for freelance work, subscription‑based access, or educational content on platforms like Upwork, Fiverr, Gumroad, and YouTube. With a solid technical foundation and clear monetization paths, you can turn this replay engine into a profitable AI‑driven side business.