NBBO consolidation: best bid/offer across fragmented venues

quotes = cu.make_quotes(
    symbols=["AAPL"], days=1, quotes_per_day=2_600, seed=11, base_prices={"AAPL": 320.0}
).to_pandas()
print(f"{len(quotes):,} rows x {quotes.shape[1]} columns")
quotes.head()

rng = np.random.default_rng(5)
TICK = 0.01

VENUES = {
    #        P(miss update), latency (ms), P(1 tick worse), P(1 tick better)
    "ARCA": dict(drop=0.04, lat_ms=3.0, worse=0.25, better=0.02),
    "EDGX": dict(drop=0.08, lat_ms=10.0, worse=0.35, better=0.005),
    "XNAS": dict(drop=0.02, lat_ms=1.5, worse=0.15, better=0.03),
}

parts = []
for venue, v in VENUES.items():
    d = quotes[rng.random(len(quotes)) >= v["drop"]].copy()
    n = len(d)
    d["ts"] = d["ts"] + pd.to_timedelta(rng.exponential(v["lat_ms"], n), unit="ms").round("us")
    d["bid"] = d["bid"] - TICK * (rng.random(n) < v["worse"]) + TICK * (rng.random(n) < v["better"])
    d["ask"] = d["ask"] + TICK * (rng.random(n) < v["worse"]) - TICK * (rng.random(n) < v["better"])
    d["venue"] = venue
    parts.append(d)

venue_quotes = (
    pd.concat(parts)[["ts", "venue", "bid", "ask", "bid_size", "ask_size"]]
    .assign(bid=lambda x: x["bid"].round(2), ask=lambda x: x["ask"].round(2))
    .sort_values(["ts", "venue"], kind="stable")
    .reset_index(drop=True)
)
venue_quotes.assign(spread=venue_quotes["ask"] - venue_quotes["bid"]).groupby("venue").agg(
    quotes=("ts", "count"), avg_spread=("spread", "mean")
).round(4)

VQ_SCHEMA = pa.schema(
    [
        pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False),
        pa.field("venue", pa.string()),
        pa.field("bid", pa.float64()),
        pa.field("ask", pa.float64()),
        pa.field("bid_size", pa.int64()),
        pa.field("ask_size", pa.int64()),
    ]
)
db.create_table("venue_quotes", VQ_SCHEMA, time_column="ts", sort_key=["ts", "venue"])
db.append("venue_quotes", pa.Table.from_pandas(venue_quotes, preserve_index=False).cast(VQ_SCHEMA),
          note="per-venue AAPL quotes, 1 session, 3 venues")

open_ts = venue_quotes["ts"].min().ceil("10s")
close_ts = pd.Timestamp("2026-06-01 20:00", tz="UTC")
grid_times = pd.date_range(open_ts, close_ts, freq="10s", inclusive="left")

grid = pd.DataFrame(
    {
        "ts": np.repeat(grid_times, len(VENUES)),
        "venue": np.tile(sorted(VENUES), len(grid_times)),
    }
)
GRID_SCHEMA = pa.schema(
    [pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False), pa.field("venue", pa.string())]
)
db.create_table("nbbo_grid", GRID_SCHEMA, time_column="ts", sort_key=["ts", "venue"])
db.append("nbbo_grid", pa.Table.from_pandas(grid, preserve_index=False).cast(GRID_SCHEMA))
print(f"grid: {len(grid_times)} instants x {len(VENUES)} venues = {len(grid):,} rows")

STALE_US = 60 * 1_000_000

# One frame, reused for every question below: each grid instant x venue gets
# the venue's prevailing quote, or NULL if the last one is older than STALE_US.
SNAPSHOT = db.table("nbbo_grid").join_asof(
    db.table("venue_quotes"), on="ts", by="venue",
    direction="backward", tolerance=STALE_US,
)

(
    SNAPSHOT.select("ts", "venue", "bid", "ask", quoted_at=col("ts_right"))
    .sort(["ts", "venue"])
    .limit(6)
    .to_pandas()
)

# Trust, but verify: one output row per grid row, and the SQL ASOF must agree
# with an independent pandas merge_asof on a venue's book.
state = SNAPSHOT.select("ts", "venue", "bid", "ask").sort(["ts", "venue"]).to_pandas()
assert len(state) == len(grid), f"ASOF join returned {len(state)} rows for {len(grid)} grid rows"

ref = pd.merge_asof(
    pd.DataFrame({"ts": grid_times}),
    venue_quotes[venue_quotes["venue"] == "XNAS"][["ts", "bid", "ask"]],
    on="ts", tolerance=pd.Timedelta("60s"),
)
chk = state[state["venue"] == "XNAS"].reset_index(drop=True)
assert chk["bid"].equals(ref["bid"]) and chk["ask"].equals(ref["ask"])
print(f"validated: {len(state):,} snapshot rows, XNAS column matches pandas merge_asof exactly")

nbbo = (
    SNAPSHOT.group_by("ts")
    .agg(nbb=col("bid").max(), nbo=col("ask").min(), venues_quoting=col("bid").count())
    .filter(col("nbb").is_not_null(), col("nbo").is_not_null())  # the HAVING
    .sort("ts")
    .to_arrow()
)

NBBO_SCHEMA = pa.schema(
    [
        pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False),
        pa.field("nbb", pa.float64()),
        pa.field("nbo", pa.float64()),
        pa.field("venues_quoting", pa.int64()),
    ]
)
db.create_table("nbbo_10s", NBBO_SCHEMA, time_column="ts")
db.append("nbbo_10s", nbbo.cast(NBBO_SCHEMA), note="10s-sampled NBBO from 3 venues")
db.table("nbbo_10s").limit(5).to_pandas()

QUOTING = SNAPSHOT.filter(col("bid").is_not_null(), col("ask").is_not_null())

share = lambda flag: (when(flag).then(lit(1.0)).otherwise(lit(0.0)).mean() * 100).round(1)

(
    QUOTING.with_columns(
        nbb=col("bid").max().over(partition_by="ts"),
        nbo=col("ask").min().over(partition_by="ts"),
    )
    .group_by("venue")
    .agg(
        pct_at_best_bid=share(col("bid") >= col("nbb")),
        pct_at_best_ask=share(col("ask") <= col("nbo")),
        avg_own_spread=(col("ask") - col("bid")).mean().round(4),
    )
    .sort("venue")
    .to_pandas()
)

count_if = lambda flag: when(flag).then(lit(1)).otherwise(lit(0)).sum()

(
    db.table("nbbo_10s")
    .select(
        instants=count_star(),
        locked=count_if(col("nbb") == col("nbo")),
        crossed=count_if(col("nbb") > col("nbo")),
        pct_locked_or_crossed=(
            when(col("nbb") >= col("nbo")).then(lit(1.0)).otherwise(lit(0.0)).mean() * 100
        ).round(2),
    )
    .to_pandas()
)

venue_spreads = QUOTING.group_by(col("venue").alias("book")).agg(
    avg_spread_usd=(col("ask") - col("bid")).mean().round(4)
)

db.sql(
    f"""
    {venue_spreads.sql()}
    UNION ALL
    SELECT 'NBBO', round(avg(nbo - nbb), 4) FROM nbbo_10s
    ORDER BY avg_spread_usd
    """
).to_pandas()

import matplotlib.pyplot as plt

MINUTE = time_bucket("1m", col("ts"))

venue_min = (
    QUOTING.group_by(MINUTE.alias("m"), "venue")
    .agg(spread=(col("ask") - col("bid")).mean())
    .sort("m")
    .to_pandas()
)
nbbo_min = (
    db.table("nbbo_10s")
    .group_by(MINUTE.alias("m"))
    .agg(spread=(col("nbo") - col("nbb")).mean())
    .sort("m")
    .to_pandas()
)

fig, ax = plt.subplots(figsize=(10, 4))
for venue, g in venue_min.groupby("venue"):
    ax.plot(g["m"], g["spread"] * 100, lw=0.8, alpha=0.7, label=venue)
ax.plot(nbbo_min["m"], nbbo_min["spread"] * 100, lw=1.6, color="black", label="NBBO")
ax.set_title("Quoted spread through the session: each venue vs consolidated NBBO")
ax.set_xlabel("time (UTC)")
ax.set_ylabel("spread (cents)")
ax.legend(ncols=4)
fig.tight_layout()