wehub-resource-sync
866 lines
37 KiB
Python
866 lines
37 KiB
Python
"""Generate minimal synthetic test data for Ch03 microstructure notebooks.
|
|
|
|
Creates test-sized datasets that match the schemas expected by Ch03 notebooks
|
|
and related notebooks (Ch02 futures individual, Ch04 prediction markets).
|
|
|
|
Writes to ~/ml4t/test-data/data/ which serves as ML4T_DATA_PATH
|
|
in CI.
|
|
|
|
Usage:
|
|
uv run python tests/generate_test_microstructure.py
|
|
"""
|
|
|
|
from datetime import date, datetime, time, timedelta
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import polars as pl
|
|
|
|
# ── Output root ──────────────────────────────────────────────────────────────
|
|
TEST_DATA_ROOT = Path.home() / "ml4t" / "test-data" / "data"
|
|
|
|
# Seed for reproducibility
|
|
RNG = np.random.default_rng(42)
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# 1. ITCH Parsed Messages (for NB 02-10)
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# These go into the ITCH messages path that load_nasdaq_itch() resolves:
|
|
# ML4T_DATA_PATH / "equities" / "market" / "microstructure" / "nasdaq_itch" / "messages"
|
|
# Notebooks 02-10 read via utils.limit_orderbook.load_itch_messages(itch_dir, msg_type, symbol)
|
|
|
|
|
|
def _ns_timestamp(hour: int, minute: int, second: int = 0, micro: int = 0) -> datetime:
|
|
"""Create a nanosecond-precision datetime on the ITCH trading day (2020-01-30)."""
|
|
return datetime(2020, 1, 30, hour, minute, second, micro)
|
|
|
|
|
|
def generate_itch_messages() -> None:
|
|
"""Generate all ITCH message type parquet files."""
|
|
itch_dir = (
|
|
TEST_DATA_ROOT / "equities" / "market" / "microstructure" / "nasdaq_itch" / "messages"
|
|
)
|
|
|
|
# ── R (Stock Directory) ──────────────────────────────────────────────
|
|
r_dir = itch_dir / "R"
|
|
r_dir.mkdir(parents=True, exist_ok=True)
|
|
r_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([1, 2, 3], dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0, 0, 0], dtype=pl.UInt16),
|
|
"timestamp": [
|
|
_ns_timestamp(4, 0, 0),
|
|
_ns_timestamp(4, 0, 0),
|
|
_ns_timestamp(4, 0, 0),
|
|
],
|
|
"stock": ["AAPL", "MSFT", "NVDA"],
|
|
"market_category": ["Q", "Q", "Q"],
|
|
"financial_status": ["N", "N", "N"],
|
|
"round_lot_size": pl.Series([100, 100, 100], dtype=pl.UInt32),
|
|
"round_lots_only": ["N", "N", "N"],
|
|
"issue_classification": ["C", "C", "C"],
|
|
"issue_subtype": ["Z", "Z", "Z"],
|
|
"authenticity": ["P", "P", "P"],
|
|
"short_sale_threshold": ["N", "N", "N"],
|
|
"ipo_flag": ["N", "N", "N"],
|
|
"luld_reference_price_tier": ["1", "1", "1"],
|
|
"etp_flag": ["N", "N", "N"],
|
|
"etp_leverage_factor": pl.Series([0, 0, 0], dtype=pl.UInt32),
|
|
"inverse_indicator": ["N", "N", "N"],
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
r_df.write_parquet(r_dir / "part-000000.parquet")
|
|
|
|
# ── S (System Event) ─────────────────────────────────────────────────
|
|
s_dir = itch_dir / "S"
|
|
s_dir.mkdir(parents=True, exist_ok=True)
|
|
s_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([0, 0, 0, 0], dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0, 0, 0, 0], dtype=pl.UInt16),
|
|
"timestamp": [
|
|
_ns_timestamp(4, 0, 0),
|
|
_ns_timestamp(9, 30, 0),
|
|
_ns_timestamp(16, 0, 0),
|
|
_ns_timestamp(20, 0, 0),
|
|
],
|
|
"event_code": ["O", "Q", "M", "C"],
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
s_df.write_parquet(s_dir / "part-000000.parquet")
|
|
|
|
# ── A (Add Order) ────────────────────────────────────────────────────
|
|
# 20 orders for AAPL (stock_locate=1), spanning 10:00 to 15:00
|
|
a_dir = itch_dir / "A"
|
|
a_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
n_orders = 20
|
|
base_price_aapl = 320.0 # AAPL price circa Jan 2020
|
|
order_refs = list(range(1001, 1001 + n_orders))
|
|
sides = ["B" if i % 2 == 0 else "S" for i in range(n_orders)]
|
|
shares = [int(RNG.integers(100, 1001)) for _ in range(n_orders)]
|
|
# Prices: bids slightly below base, asks slightly above
|
|
prices = []
|
|
for i, side in enumerate(sides):
|
|
offset = RNG.uniform(0.01, 0.50)
|
|
if side == "B":
|
|
prices.append(round(base_price_aapl - offset, 4))
|
|
else:
|
|
prices.append(round(base_price_aapl + offset, 4))
|
|
|
|
# Timestamps spaced across 10:00-15:00 (300 minutes = 18000 seconds)
|
|
a_timestamps = [
|
|
_ns_timestamp(10, 0) + timedelta(seconds=int(i * 18000 / n_orders)) for i in range(n_orders)
|
|
]
|
|
|
|
# Prices stored as ITCH price4 integers (multiply by 10000) per spec
|
|
a_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([1] * n_orders, dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0] * n_orders, dtype=pl.UInt16),
|
|
"timestamp": a_timestamps,
|
|
"order_reference_number": pl.Series(order_refs, dtype=pl.UInt64),
|
|
"buy_sell_indicator": sides,
|
|
"shares": pl.Series(shares, dtype=pl.UInt32),
|
|
"stock": ["AAPL"] * n_orders,
|
|
"price": pl.Series([int(p * 10000) for p in prices], dtype=pl.UInt32),
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
a_df.write_parquet(a_dir / "part-000000.parquet")
|
|
|
|
# ── D (Order Delete) ─────────────────────────────────────────────────
|
|
d_dir = itch_dir / "D"
|
|
d_dir.mkdir(parents=True, exist_ok=True)
|
|
delete_refs = [1001, 1003, 1005, 1007, 1009]
|
|
d_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([1] * 5, dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0] * 5, dtype=pl.UInt16),
|
|
"timestamp": [
|
|
a_timestamps[0] + timedelta(seconds=30),
|
|
a_timestamps[2] + timedelta(seconds=30),
|
|
a_timestamps[4] + timedelta(seconds=30),
|
|
a_timestamps[6] + timedelta(seconds=30),
|
|
a_timestamps[8] + timedelta(seconds=30),
|
|
],
|
|
"order_reference_number": pl.Series(delete_refs, dtype=pl.UInt64),
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
d_df.write_parquet(d_dir / "part-000000.parquet")
|
|
|
|
# ── E (Order Executed) ───────────────────────────────────────────────
|
|
e_dir = itch_dir / "E"
|
|
e_dir.mkdir(parents=True, exist_ok=True)
|
|
exec_refs = [1002, 1004, 1006, 1008, 1010, 1012, 1014, 1016]
|
|
exec_shares = [min(shares[r - 1001] // 2, 200) for r in exec_refs]
|
|
e_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([1] * 8, dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0] * 8, dtype=pl.UInt16),
|
|
"timestamp": [a_timestamps[r - 1001] + timedelta(seconds=60) for r in exec_refs],
|
|
"order_reference_number": pl.Series(exec_refs, dtype=pl.UInt64),
|
|
"executed_shares": pl.Series(exec_shares, dtype=pl.UInt32),
|
|
"match_number": pl.Series(list(range(5001, 5009)), dtype=pl.UInt64),
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
e_df.write_parquet(e_dir / "part-000000.parquet")
|
|
|
|
# ── X (Order Cancel) ─────────────────────────────────────────────────
|
|
x_dir = itch_dir / "X"
|
|
x_dir.mkdir(parents=True, exist_ok=True)
|
|
cancel_refs = [1011, 1013, 1015]
|
|
cancel_shares = [shares[r - 1001] // 3 for r in cancel_refs]
|
|
x_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([1] * 3, dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0] * 3, dtype=pl.UInt16),
|
|
"timestamp": [a_timestamps[r - 1001] + timedelta(seconds=45) for r in cancel_refs],
|
|
"order_reference_number": pl.Series(cancel_refs, dtype=pl.UInt64),
|
|
"cancelled_shares": pl.Series(cancel_shares, dtype=pl.UInt32),
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
x_df.write_parquet(x_dir / "part-000000.parquet")
|
|
|
|
# ── C (Order Executed with Price) ────────────────────────────────────
|
|
c_dir = itch_dir / "C"
|
|
c_dir.mkdir(parents=True, exist_ok=True)
|
|
c_refs = [1017, 1018]
|
|
c_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([1] * 2, dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0] * 2, dtype=pl.UInt16),
|
|
"timestamp": [
|
|
a_timestamps[16] + timedelta(seconds=90),
|
|
a_timestamps[17] + timedelta(seconds=90),
|
|
],
|
|
"order_reference_number": pl.Series(c_refs, dtype=pl.UInt64),
|
|
"executed_shares": pl.Series([shares[16] // 4, shares[17] // 4], dtype=pl.UInt32),
|
|
"match_number": pl.Series([6001, 6002], dtype=pl.UInt64),
|
|
"printable": ["Y", "Y"],
|
|
"execution_price": pl.Series(
|
|
[int(prices[16] * 10000), int(prices[17] * 10000)], dtype=pl.UInt32
|
|
),
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
c_df.write_parquet(c_dir / "part-000000.parquet")
|
|
|
|
# ── P (Non-Cross Trade) ──────────────────────────────────────────────
|
|
p_dir = itch_dir / "P"
|
|
p_dir.mkdir(parents=True, exist_ok=True)
|
|
p_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([1, 1, 1], dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0, 0, 0], dtype=pl.UInt16),
|
|
"timestamp": [
|
|
_ns_timestamp(11, 30, 0),
|
|
_ns_timestamp(13, 0, 0),
|
|
_ns_timestamp(14, 30, 0),
|
|
],
|
|
"order_reference_number": pl.Series([2001, 2002, 2003], dtype=pl.UInt64),
|
|
"buy_sell_indicator": ["B", "S", "B"],
|
|
"shares": pl.Series([200, 150, 300], dtype=pl.UInt32),
|
|
"stock": ["AAPL", "AAPL", "AAPL"],
|
|
"price": pl.Series([int(base_price_aapl * 10000)] * 3, dtype=pl.UInt32),
|
|
"match_number": pl.Series([7001, 7002, 7003], dtype=pl.UInt64),
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
p_df.write_parquet(p_dir / "part-000000.parquet")
|
|
|
|
# ── U (Order Replace) ────────────────────────────────────────────────
|
|
u_dir = itch_dir / "U"
|
|
u_dir.mkdir(parents=True, exist_ok=True)
|
|
u_df = pl.DataFrame(
|
|
{
|
|
"stock_locate": pl.Series([1, 1], dtype=pl.UInt16),
|
|
"tracking_number": pl.Series([0, 0], dtype=pl.UInt16),
|
|
"timestamp": [
|
|
a_timestamps[18] + timedelta(seconds=20),
|
|
a_timestamps[19] + timedelta(seconds=20),
|
|
],
|
|
"original_order_reference_number": pl.Series([1019, 1020], dtype=pl.UInt64),
|
|
"new_order_reference_number": pl.Series([3001, 3002], dtype=pl.UInt64),
|
|
"shares": pl.Series([500, 600], dtype=pl.UInt32),
|
|
"price": pl.Series(
|
|
[int((base_price_aapl - 0.10) * 10000), int((base_price_aapl + 0.10) * 10000)],
|
|
dtype=pl.UInt32,
|
|
),
|
|
}
|
|
).cast({"timestamp": pl.Datetime("ns")})
|
|
u_df.write_parquet(u_dir / "part-000000.parquet")
|
|
|
|
print(f" ITCH messages written to {itch_dir}")
|
|
for sub in sorted(itch_dir.iterdir()):
|
|
if sub.is_dir() and sub.name != "enriched":
|
|
n = pl.scan_parquet(sub / "*.parquet").select(pl.len()).collect().item()
|
|
print(f" {sub.name}/: {n} rows")
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# 2. DataBento MBO (for NB 09-13)
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# Path: ML4T_DATA_PATH / "equities" / "market" / "microstructure" / "market_by_order" / "NVDA"
|
|
# File naming: xnas-itch-YYYYMMDD.mbo.dbn.parquet (DataBento convention)
|
|
# NB09, NB12 expect "timestamp" column. NB10, NB11, NB13 also need it.
|
|
# We provide BOTH ts_event and timestamp (same values) for compatibility.
|
|
|
|
|
|
def _generate_mbo_day(base_date: datetime, base_price_nano: int, start_order_id: int) -> list[dict]:
|
|
"""Generate one day of MBO messages with realistic bid/ask structure.
|
|
|
|
Returns a list of row dicts (not yet a DataFrame).
|
|
"""
|
|
rows: list[dict] = []
|
|
order_id = start_order_id
|
|
n_cycles = 50 # 50 cycles spread across 6.5 hours of trading
|
|
|
|
for cycle in range(n_cycles):
|
|
cycle_start_ms = cycle * 468_000 # ~7.8 min per cycle
|
|
|
|
# Phase 1: Adds (build book) - 15 orders per cycle
|
|
for i in range(15):
|
|
ts = base_date + timedelta(milliseconds=cycle_start_ms + i * 100)
|
|
side = "B" if i % 2 == 0 else "A"
|
|
if side == "B":
|
|
price_offset = -RNG.integers(1, 51) * 10_000_000
|
|
else:
|
|
price_offset = RNG.integers(1, 51) * 10_000_000
|
|
price = base_price_nano + price_offset
|
|
size = int(RNG.integers(1, 501))
|
|
rows.append(
|
|
{
|
|
"ts_event": ts,
|
|
"ts_recv": ts + timedelta(microseconds=int(RNG.integers(1, 100))),
|
|
"action": "A",
|
|
"side": side,
|
|
"price": price,
|
|
"size": size,
|
|
"order_id": order_id,
|
|
"flags": 0,
|
|
"publisher_id": 39,
|
|
}
|
|
)
|
|
order_id += 1
|
|
|
|
# Phase 2: Modifications - 3 per cycle
|
|
for i in range(3):
|
|
ts = base_date + timedelta(milliseconds=cycle_start_ms + 1500 + i * 200)
|
|
mod_order = order_id - 15 + i * 5
|
|
mod_side = "B" if i % 2 == 0 else "A"
|
|
if mod_side == "B":
|
|
price_offset = -RNG.integers(1, 31) * 10_000_000
|
|
else:
|
|
price_offset = RNG.integers(1, 31) * 10_000_000
|
|
rows.append(
|
|
{
|
|
"ts_event": ts,
|
|
"ts_recv": ts + timedelta(microseconds=int(RNG.integers(1, 100))),
|
|
"action": "M",
|
|
"side": mod_side,
|
|
"price": base_price_nano + price_offset,
|
|
"size": int(RNG.integers(1, 300)),
|
|
"order_id": mod_order,
|
|
"flags": 0,
|
|
"publisher_id": 39,
|
|
}
|
|
)
|
|
|
|
# Phase 3: Cancels - 3 per cycle
|
|
for i in range(3):
|
|
ts = base_date + timedelta(milliseconds=cycle_start_ms + 2100 + i * 200)
|
|
cancel_order = order_id - 14 + i * 5
|
|
rows.append(
|
|
{
|
|
"ts_event": ts,
|
|
"ts_recv": ts + timedelta(microseconds=int(RNG.integers(1, 100))),
|
|
"action": "C",
|
|
"side": "A" if i % 2 == 0 else "B",
|
|
"price": base_price_nano,
|
|
"size": 0,
|
|
"order_id": cancel_order,
|
|
"flags": 0,
|
|
"publisher_id": 39,
|
|
}
|
|
)
|
|
|
|
# Phase 4: Fills (F) and Trades (T) - 10 per cycle
|
|
for i in range(10):
|
|
ts = base_date + timedelta(milliseconds=cycle_start_ms + 2700 + i * 300)
|
|
fill_order = order_id - 13 + i
|
|
fill_size = int(RNG.integers(1, 200))
|
|
# Biased aggressor side for realistic imbalance runs.
|
|
# Runs of 10 consecutive cycles (~100 trades) with 95% bias,
|
|
# creating sustained imbalance that triggers bar boundaries.
|
|
# This mimics real institutional order flow patterns.
|
|
run_idx = cycle // 10
|
|
if run_idx % 2 == 0:
|
|
aggressor = "B" if RNG.random() < 0.95 else "A"
|
|
else:
|
|
aggressor = "A" if RNG.random() < 0.95 else "B"
|
|
trade_price = base_price_nano + RNG.integers(-5, 6) * 10_000_000
|
|
fill_side = "A" if aggressor == "B" else "B"
|
|
rows.append(
|
|
{
|
|
"ts_event": ts,
|
|
"ts_recv": ts + timedelta(microseconds=int(RNG.integers(1, 100))),
|
|
"action": "F",
|
|
"side": fill_side,
|
|
"price": trade_price,
|
|
"size": fill_size,
|
|
"order_id": fill_order,
|
|
"flags": 128,
|
|
"publisher_id": 39,
|
|
}
|
|
)
|
|
rows.append(
|
|
{
|
|
"ts_event": ts + timedelta(microseconds=1),
|
|
"ts_recv": ts + timedelta(microseconds=int(RNG.integers(2, 150))),
|
|
"action": "T",
|
|
"side": aggressor,
|
|
"price": trade_price,
|
|
"size": fill_size,
|
|
"order_id": fill_order,
|
|
"flags": 128,
|
|
"publisher_id": 39,
|
|
}
|
|
)
|
|
|
|
return rows
|
|
|
|
|
|
def generate_mbo_data() -> None:
|
|
"""Generate synthetic DataBento MBO tick data for NVDA.
|
|
|
|
Key schema requirements from notebooks:
|
|
- NB09 (lee_ready): expects "timestamp" column, reads parquet directly
|
|
- NB10 (information_bars): expects filename like xnas-itch-YYYYMMDD.mbo.dbn.parquet
|
|
- NB11 (lob_reconstruction): expects "ts_event" column, reads parquet directly
|
|
- NB12 (mbo_analysis): expects "timestamp" column, reads parquet directly
|
|
- NB13 (bar_sampling): expects "timestamp" column, filename like xnas-itch-*
|
|
|
|
We include both ts_event and timestamp columns, and use DataBento file naming.
|
|
We also generate enough data (spread across hours) for meaningful analysis.
|
|
"""
|
|
mbo_dir = TEST_DATA_ROOT / "equities" / "market" / "microstructure" / "market_by_order" / "NVDA"
|
|
mbo_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Remove old file if it exists (was named 20241104.parquet before)
|
|
old_file = mbo_dir / "20241104.parquet"
|
|
if old_file.exists():
|
|
old_file.unlink()
|
|
|
|
base_price_nano = 140_000_000_000 # $140 in nanodollars
|
|
|
|
# Generate 3 days of data. NB13 (bar_sampling) computes day-to-day CV which
|
|
# needs >= 2 days. NB10 (information_bars) also benefits from more trades.
|
|
trading_days = [
|
|
datetime(2024, 11, 4, 14, 30, 0), # Monday 9:30 AM ET in UTC
|
|
datetime(2024, 11, 5, 14, 30, 0), # Tuesday
|
|
datetime(2024, 11, 6, 14, 30, 0), # Wednesday
|
|
]
|
|
|
|
for day_idx, base_date in enumerate(trading_days):
|
|
rows = _generate_mbo_day(base_date, base_price_nano, 100_000 + day_idx * 10_000)
|
|
|
|
df = (
|
|
pl.DataFrame(rows)
|
|
.cast(
|
|
{
|
|
"ts_event": pl.Datetime("ns"),
|
|
"ts_recv": pl.Datetime("ns"),
|
|
"price": pl.Int64,
|
|
"size": pl.Int64,
|
|
"order_id": pl.Int64,
|
|
"flags": pl.Int64,
|
|
"publisher_id": pl.Int64,
|
|
}
|
|
)
|
|
.sort("ts_event")
|
|
)
|
|
|
|
# Add canonical "timestamp" column (same as ts_event) for notebooks that expect it.
|
|
# NB09, NB12, NB13 use "timestamp"; NB10, NB11 use "ts_event".
|
|
df = df.with_columns(pl.col("ts_event").alias("timestamp"))
|
|
|
|
# Write with DataBento filename convention: xnas-itch-YYYYMMDD.mbo.dbn.parquet
|
|
# NB10 and NB13 parse the filename: file_path.name.split("-")[2].split(".")[0]
|
|
date_str = base_date.strftime("%Y%m%d")
|
|
out_file = mbo_dir / f"xnas-itch-{date_str}.mbo.dbn.parquet"
|
|
df.write_parquet(out_file)
|
|
print(f" MBO day {date_str}: {len(df)} rows -> {out_file}")
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# 3. AlgoSeek TAQ (for NB 15-16)
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# Path: ML4T_DATA_PATH / "equities" / "market" / "microstructure" / "trade_and_quotes" / "symbol=AAPL" / "data.parquet"
|
|
# NB15 expects: TRADE, QUOTE BID, QUOTE ASK, QUOTE BID NB, QUOTE ASK NB event types
|
|
# NB15 does spread analysis using NBBO quotes and trade size distribution
|
|
|
|
|
|
def generate_taq_data() -> None:
|
|
"""Generate synthetic AlgoSeek TAQ tick data for AAPL on 2020-03-16.
|
|
|
|
Key schema requirements from notebooks:
|
|
- NB15 (taq_eda): Needs TRADE, QUOTE BID NB, QUOTE ASK NB event types
|
|
for spread analysis. Needs enough trades for size distribution.
|
|
- NB16 (taq_lob): Needs QUOTE BID/ASK for LOB reconstruction.
|
|
|
|
We generate ~600 events with realistic distributions.
|
|
"""
|
|
taq_dir = (
|
|
TEST_DATA_ROOT
|
|
/ "equities"
|
|
/ "market"
|
|
/ "microstructure"
|
|
/ "trade_and_quotes"
|
|
/ "symbol=AAPL"
|
|
)
|
|
taq_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# March 16, 2020: AAPL around $250, huge volatility day
|
|
base_date = datetime(2020, 3, 16)
|
|
base_price = 250.0
|
|
exchanges = ["Q", "N", "Z", "P", "K"]
|
|
|
|
rows = []
|
|
# Generate ~600 rows: mix of trades, exchange quotes, and NBBO quotes
|
|
# NB15 needs: TRADE events for trade analysis, QUOTE BID NB / QUOTE ASK NB for spread
|
|
for i in range(600):
|
|
# Random time between 9:30 and 16:00 (6.5 hours = 23400 seconds)
|
|
seconds_offset = int(RNG.integers(0, 23400))
|
|
ts = base_date + timedelta(
|
|
hours=9, minutes=30, seconds=seconds_offset, microseconds=int(RNG.integers(0, 999999))
|
|
)
|
|
|
|
# Event type distribution:
|
|
# ~15% trades, ~20% NBBO bids, ~20% NBBO asks, ~20% exchange bids, ~20% exchange asks
|
|
# We need QUOTE BID NB and QUOTE ASK NB for NB15's spread analysis
|
|
r = RNG.random()
|
|
if r < 0.15:
|
|
event_type = "TRADE"
|
|
price = round(base_price + RNG.normal(0, 5), 2)
|
|
quantity = int(RNG.integers(10, 10001))
|
|
elif r < 0.35:
|
|
event_type = "QUOTE BID NB"
|
|
price = round(base_price - abs(RNG.normal(0.03, 0.10)), 2)
|
|
quantity = int(RNG.integers(100, 5001))
|
|
elif r < 0.55:
|
|
event_type = "QUOTE ASK NB"
|
|
price = round(base_price + abs(RNG.normal(0.03, 0.10)), 2)
|
|
quantity = int(RNG.integers(100, 5001))
|
|
elif r < 0.75:
|
|
event_type = "QUOTE BID"
|
|
price = round(base_price - abs(RNG.normal(0.05, 0.20)), 2)
|
|
quantity = int(RNG.integers(100, 5001))
|
|
else:
|
|
event_type = "QUOTE ASK"
|
|
price = round(base_price + abs(RNG.normal(0.05, 0.20)), 2)
|
|
quantity = int(RNG.integers(100, 5001))
|
|
|
|
rows.append(
|
|
{
|
|
"timestamp": ts,
|
|
"event_type": event_type,
|
|
"price": price,
|
|
"quantity": quantity,
|
|
"exchange": exchanges[int(RNG.integers(0, len(exchanges)))],
|
|
"conditions": "00000000",
|
|
}
|
|
)
|
|
|
|
df = (
|
|
pl.DataFrame(rows)
|
|
.cast(
|
|
{
|
|
"timestamp": pl.Datetime("us"),
|
|
"price": pl.Float64,
|
|
"quantity": pl.Int64,
|
|
}
|
|
)
|
|
.sort("timestamp")
|
|
)
|
|
|
|
df.write_parquet(taq_dir / "data.parquet")
|
|
print(f" TAQ data: {len(df)} rows -> {taq_dir / 'data.parquet'}")
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# 4. IEX Parsed Data (for NB 14)
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# Path: ML4T_DATA_PATH / "equities" / "market" / "microstructure" / "iex" / "deep" / "parsed" / {type}/
|
|
|
|
|
|
def generate_iex_data() -> None:
|
|
"""Generate synthetic IEX DEEP parsed data."""
|
|
parsed_dir = (
|
|
TEST_DATA_ROOT / "equities" / "market" / "microstructure" / "iex" / "deep" / "parsed"
|
|
)
|
|
|
|
base_date = datetime(2025, 1, 15, 14, 30, 0) # 9:30 AM ET in UTC
|
|
base_price = 240.0 # AAPL-ish
|
|
|
|
# ── Quotes ───────────────────────────────────────────────────────────
|
|
quotes_dir = parsed_dir / "quotes"
|
|
quotes_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
quote_rows = []
|
|
for i in range(30):
|
|
ts = base_date + timedelta(seconds=i * 60)
|
|
spread = round(abs(RNG.normal(0.02, 0.01)), 4)
|
|
mid = base_price + RNG.normal(0, 0.5)
|
|
quote_rows.append(
|
|
{
|
|
"timestamp": ts,
|
|
"symbol": "AAPL",
|
|
"bid_price": round(mid - spread / 2, 2),
|
|
"bid_size": int(RNG.integers(100, 5001)),
|
|
"ask_price": round(mid + spread / 2, 2),
|
|
"ask_size": int(RNG.integers(100, 5001)),
|
|
}
|
|
)
|
|
|
|
quotes_df = pl.DataFrame(quote_rows).cast(
|
|
{
|
|
"timestamp": pl.Datetime("ns"),
|
|
"bid_price": pl.Float64,
|
|
"ask_price": pl.Float64,
|
|
"bid_size": pl.Int64,
|
|
"ask_size": pl.Int64,
|
|
}
|
|
)
|
|
quotes_df.write_parquet(quotes_dir / "data.parquet")
|
|
print(f" IEX quotes: {len(quotes_df)} rows -> {quotes_dir / 'data.parquet'}")
|
|
|
|
# ── Trades ───────────────────────────────────────────────────────────
|
|
trades_dir = parsed_dir / "trades"
|
|
trades_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
trade_rows = []
|
|
for i in range(20):
|
|
ts = base_date + timedelta(seconds=i * 90 + int(RNG.integers(0, 30)))
|
|
trade_rows.append(
|
|
{
|
|
"timestamp": ts,
|
|
"symbol": "AAPL",
|
|
"price": round(base_price + RNG.normal(0, 0.3), 2),
|
|
"size": int(RNG.integers(1, 501)),
|
|
}
|
|
)
|
|
|
|
trades_df = pl.DataFrame(trade_rows).cast(
|
|
{
|
|
"timestamp": pl.Datetime("ns"),
|
|
"price": pl.Float64,
|
|
"size": pl.Int64,
|
|
}
|
|
)
|
|
trades_df.write_parquet(trades_dir / "data.parquet")
|
|
print(f" IEX trades: {len(trades_df)} rows -> {trades_dir / 'data.parquet'}")
|
|
|
|
# ── Price Levels ─────────────────────────────────────────────────────
|
|
price_levels_dir = parsed_dir / "price_levels"
|
|
price_levels_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
pl_rows = []
|
|
for i in range(40):
|
|
ts = base_date + timedelta(seconds=i * 45)
|
|
side = "bid" if i % 2 == 0 else "ask"
|
|
offset = RNG.uniform(0.01, 0.50)
|
|
price = round(base_price - offset if side == "bid" else base_price + offset, 2)
|
|
pl_rows.append(
|
|
{
|
|
"timestamp": ts,
|
|
"symbol": "AAPL",
|
|
"side": side,
|
|
"price": price,
|
|
"size": int(RNG.integers(100, 3001)),
|
|
}
|
|
)
|
|
|
|
pl_df = pl.DataFrame(pl_rows).cast(
|
|
{
|
|
"timestamp": pl.Datetime("ns"),
|
|
"price": pl.Float64,
|
|
"size": pl.Int64,
|
|
}
|
|
)
|
|
pl_df.write_parquet(price_levels_dir / "data.parquet")
|
|
print(f" IEX price_levels: {len(pl_df)} rows -> {price_levels_dir / 'data.parquet'}")
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# 5. CME Individual Contracts (for Ch02 NB 04-06)
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# Path: ML4T_DATA_PATH / "futures" / "market" / "individual" / "{PRODUCT}" / "data.parquet"
|
|
# Schema matches what load_cme_futures(continuous=False) returns:
|
|
# timestamp (datetime[ns, UTC]), rtype, publisher_id, instrument_id,
|
|
# open, high, low, close, volume, product
|
|
#
|
|
# NB06 (futures_continuous) needs:
|
|
# - Multiple contracts with OVERLAPPING date ranges
|
|
# - Volume patterns that make front-month detection possible
|
|
# - Enough contracts for roll detection to produce adj_close
|
|
|
|
|
|
def generate_individual_futures() -> None:
|
|
"""Generate synthetic CME individual contract data for ES, NQ, CL.
|
|
|
|
Key requirements from NB06 (continuous construction):
|
|
- Contracts must overlap in time (concurrent trading)
|
|
- Front month should have highest volume (for volume-based roll detection)
|
|
- Need at least 3 contracts with clear roll transitions
|
|
- Need enough data points for roll gaps to produce adj_close
|
|
"""
|
|
individual_dir = TEST_DATA_ROOT / "futures" / "market" / "individual"
|
|
|
|
products = {
|
|
"ES": {"base_price": 4500.0, "tick": 0.25},
|
|
"NQ": {"base_price": 15500.0, "tick": 0.25},
|
|
"CL": {"base_price": 75.0, "tick": 0.01},
|
|
}
|
|
|
|
# Contract months: H=March, M=June, U=Sep, Z=Dec
|
|
# Instrument IDs encode contract month. Simulate 4 quarterly contracts
|
|
# overlapping across 2024, with volume-based rolls.
|
|
contract_specs = [
|
|
# (instrument_id, start_day_offset, end_day_offset, is_front_until_day)
|
|
# Contract 1 (H24): front month days 0-29, then rolls to contract 2
|
|
(49701, 0, 59, 29),
|
|
# Contract 2 (M24): front month days 30-89, then rolls to contract 3
|
|
(49702, 15, 119, 89),
|
|
# Contract 3 (U24): front month days 90-149, then rolls to contract 4
|
|
(49703, 75, 179, 149),
|
|
# Contract 4 (Z24): front month from day 150 onward
|
|
(49704, 135, 209, 209),
|
|
]
|
|
|
|
for product, cfg in products.items():
|
|
prod_dir = individual_dir / product
|
|
prod_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
rows = []
|
|
start = datetime(2024, 1, 2, 0, 0, 0)
|
|
|
|
for inst_id, start_day, end_day, front_until in contract_specs:
|
|
# Adjust instrument_id per product to be unique
|
|
if product == "NQ":
|
|
inst_id += 1000
|
|
elif product == "CL":
|
|
inst_id += 2000
|
|
|
|
for day_offset in range(start_day, end_day + 1):
|
|
# Generate one bar per day (24 hours apart for hourly-like data)
|
|
ts = start + timedelta(days=day_offset)
|
|
|
|
# Price drifts slightly
|
|
base = cfg["base_price"] + RNG.normal(0, cfg["base_price"] * 0.002)
|
|
o = round(base, 2)
|
|
h = round(base + abs(RNG.normal(0, cfg["base_price"] * 0.001)), 2)
|
|
l = round(base - abs(RNG.normal(0, cfg["base_price"] * 0.001)), 2)
|
|
c = round(base + RNG.normal(0, cfg["base_price"] * 0.0005), 2)
|
|
|
|
# Volume: high when front month, low when back month
|
|
if day_offset <= front_until:
|
|
vol = int(RNG.integers(10000, 50001)) # Front month: high volume
|
|
else:
|
|
vol = int(RNG.integers(100, 3001)) # Back month: low volume
|
|
|
|
rows.append(
|
|
{
|
|
"timestamp": ts,
|
|
"rtype": 35,
|
|
"publisher_id": 1,
|
|
"instrument_id": inst_id,
|
|
"open": o,
|
|
"high": h,
|
|
"low": l,
|
|
"close": c,
|
|
"volume": vol,
|
|
"product": product,
|
|
}
|
|
)
|
|
|
|
df = (
|
|
pl.DataFrame(rows)
|
|
.cast(
|
|
{
|
|
"timestamp": pl.Datetime("ns", time_zone="UTC"),
|
|
"rtype": pl.UInt8,
|
|
"publisher_id": pl.UInt16,
|
|
"instrument_id": pl.UInt32,
|
|
"open": pl.Float64,
|
|
"high": pl.Float64,
|
|
"low": pl.Float64,
|
|
"close": pl.Float64,
|
|
"volume": pl.UInt64,
|
|
}
|
|
)
|
|
.sort("timestamp")
|
|
)
|
|
|
|
df.write_parquet(prod_dir / "data.parquet")
|
|
print(f" Futures individual {product}: {len(df)} rows -> {prod_dir / 'data.parquet'}")
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# 6. Kalshi Events (for Ch04 NB 13)
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# Path: ML4T_DATA_PATH / "prediction_markets" / "kalshi_events.parquet"
|
|
# Schema: timestamp (Date), symbol (str), open/high/low/close (Float64), volume (Int64)
|
|
|
|
|
|
def generate_kalshi_data() -> None:
|
|
"""Generate synthetic Kalshi prediction market data."""
|
|
pm_dir = TEST_DATA_ROOT / "prediction_markets"
|
|
pm_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 5 contracts, ~10 days each = ~50 rows
|
|
contracts = [
|
|
"KXFED-27APR-T4.25",
|
|
"KXFED-27APR-T4.50",
|
|
"KXFED-27JUN-T4.00",
|
|
"KXINFL-27MAR-T3.0",
|
|
"KXGDP-27Q1-T2.0",
|
|
]
|
|
|
|
rows = []
|
|
base_date = date(2027, 3, 1)
|
|
|
|
for contract in contracts:
|
|
# Each contract gets a base probability and drifts
|
|
base_prob = RNG.uniform(0.2, 0.8)
|
|
for day in range(10):
|
|
d = base_date + timedelta(days=day)
|
|
# Random walk for probability
|
|
base_prob = max(0.01, min(0.99, base_prob + RNG.normal(0, 0.03)))
|
|
o = round(base_prob, 2)
|
|
h = round(min(0.99, base_prob + abs(RNG.normal(0, 0.02))), 2)
|
|
l = round(max(0.01, base_prob - abs(RNG.normal(0, 0.02))), 2)
|
|
c = round(max(0.01, min(0.99, base_prob + RNG.normal(0, 0.01))), 2)
|
|
vol = int(RNG.integers(50, 5001))
|
|
|
|
rows.append(
|
|
{
|
|
"timestamp": d,
|
|
"symbol": contract,
|
|
"open": o,
|
|
"high": h,
|
|
"low": l,
|
|
"close": c,
|
|
"volume": vol,
|
|
}
|
|
)
|
|
|
|
df = (
|
|
pl.DataFrame(rows)
|
|
.cast(
|
|
{
|
|
"timestamp": pl.Date,
|
|
"open": pl.Float64,
|
|
"high": pl.Float64,
|
|
"low": pl.Float64,
|
|
"close": pl.Float64,
|
|
"volume": pl.Int64,
|
|
}
|
|
)
|
|
.sort(["symbol", "timestamp"])
|
|
)
|
|
|
|
df.write_parquet(pm_dir / "kalshi_events.parquet")
|
|
print(f" Kalshi events: {len(df)} rows -> {pm_dir / 'kalshi_events.parquet'}")
|
|
|
|
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
# Main
|
|
# ═════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def main() -> None:
|
|
print(f"Generating test microstructure data in {TEST_DATA_ROOT}\n")
|
|
|
|
print("1. ITCH Parsed Messages")
|
|
generate_itch_messages()
|
|
|
|
print("\n2. DataBento MBO")
|
|
generate_mbo_data()
|
|
|
|
print("\n3. AlgoSeek TAQ")
|
|
generate_taq_data()
|
|
|
|
print("\n4. IEX Parsed Data")
|
|
generate_iex_data()
|
|
|
|
print("\n5. CME Individual Futures")
|
|
generate_individual_futures()
|
|
|
|
print("\n6. Kalshi Prediction Markets")
|
|
generate_kalshi_data()
|
|
|
|
print("\nDone.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|