VWAP, TWAP and execution benchmarks

quotes = cu.make_quotes(symbols=["AAPL", "MSFT", "NVDA"], days=3, seed=11)

qp = quotes.to_pandas()
rng = np.random.default_rng(17)
is_trade = rng.random(len(qp)) < 0.35
tr = qp.loc[is_trade, ["ts", "symbol", "bid", "ask"]].reset_index(drop=True)
n = len(tr)
buy = rng.random(n) < 0.5
at_mid = rng.random(n) < 0.10
px = np.where(buy, tr["ask"], tr["bid"])
px = np.where(at_mid, (tr["bid"] + tr["ask"]) / 2, px)
tr["price"] = np.round(px, 4)
tr["size"] = np.maximum(1, rng.lognormal(4.0, 1.2, n) // 100 * 100).astype("int64")
tr["side"] = np.where(buy, "B", "S")
tr["ts"] = (tr["ts"] + pd.to_timedelta(rng.uniform(0.2, 5.0, n), unit="ms")).dt.floor("us")
tr = tr.sort_values(["ts", "symbol"])[["ts", "symbol", "price", "size", "side"]]

ts_field = pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False)
trade_schema = pa.schema(
    [ts_field, pa.field("symbol", pa.string()), pa.field("price", pa.float64()),
     pa.field("size", pa.int64()), pa.field("side", pa.string())]
)
db.create_table("trades", trade_schema, time_column="ts", sort_key=["ts", "symbol"])
db.append("trades", pa.Table.from_pandas(tr, preserve_index=False).cast(trade_schema))

quote_schema = pa.schema(
    [ts_field, pa.field("symbol", 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("quotes", quote_schema, time_column="ts", sort_key=["ts", "symbol"])
db.append("quotes", quotes.cast(quote_schema))

print(f"{len(tr):,} trades derived from {len(qp):,} quotes")

day_vwap = db.sql(
    """
    SELECT time_bucket('1d', ts, 'America/New_York') AS session, symbol,
           vwap(price, size) AS day_vwap,
           sum(size)         AS volume
    FROM trades GROUP BY 1, 2 ORDER BY 1, 2
    """
).to_pandas()
day_vwap

ivwap = db.sql(
    """
    SELECT time_bucket('30m', ts) AS interval_start, symbol,
           vwap(price, size) AS interval_vwap,
           sum(size)         AS volume
    FROM trades GROUP BY 1, 2 ORDER BY 1, 2
    """
).to_pandas()
ivwap.head(6)

twap = db.sql(
    """
    WITH bars_1m AS (
        SELECT time_bucket('1m', ts) AS bar, symbol,
               last_value(price ORDER BY ts) AS close
        FROM trades GROUP BY 1, 2
    )
    SELECT time_bucket('1d', bar, 'America/New_York') AS session, symbol,
           avg(close) AS twap
    FROM bars_1m GROUP BY 1, 2 ORDER BY 1, 2
    """
).to_pandas()

bench = day_vwap.merge(twap, on=["session", "symbol"])
bench["vwap_minus_twap_bps"] = (bench["day_vwap"] / bench["twap"] - 1) * 1e4
bench[["session", "symbol", "day_vwap", "twap", "vwap_minus_twap_bps"]]

run = db.sql(
    """
    SELECT ts, price,
           vwap(price, size) OVER (ORDER BY ts) AS running_vwap
    FROM trades
    WHERE symbol = 'AAPL'
      AND ts >= '2026-06-02T00:00:00Z' AND ts < '2026-06-03T00:00:00Z'
    ORDER BY ts
    """
).to_pandas()
closes_1m = db.sql(
    """
    SELECT time_bucket('1m', ts) AS bar, last_value(price ORDER BY ts) AS close
    FROM trades
    WHERE symbol = 'AAPL'
      AND ts >= '2026-06-02T00:00:00Z' AND ts < '2026-06-03T00:00:00Z'
    GROUP BY 1 ORDER BY 1
    """
).to_pandas()
closes_1m["running_twap"] = closes_1m["close"].expanding().mean()

fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(run["ts"], run["price"], lw=0.3, color="0.7", label="trades")
ax.plot(run["ts"], run["running_vwap"], lw=1.4, label="running VWAP")
ax.plot(closes_1m["bar"], closes_1m["running_twap"], lw=1.4, ls="--", label="running TWAP")
ax.set_title("AAPL 2026-06-02: running VWAP vs running TWAP")
ax.set_xlabel("time (UTC)")
ax.set_ylabel("price")
ax.legend()
fig.tight_layout()

t0, t1 = pd.Timestamp("2026-06-02 14:00", tz="UTC"), pd.Timestamp("2026-06-02 15:00", tz="UTC")
win = tr[(tr["symbol"] == "AAPL") & (tr["ts"] >= t0) & (tr["ts"] < t1)]
liftable = win[win["side"] == "B"]  # prints where an aggressive buyer paid the offer

target = 50_000
pov = target / liftable["size"].sum()
fills = liftable[["ts", "symbol", "price"]].copy()
fills["size"] = np.maximum(1, (liftable["size"] * pov).round()).astype("int64")
fills.iloc[-1, fills.columns.get_loc("size")] += target - fills["size"].sum()

fill_schema = pa.schema(
    [ts_field, pa.field("symbol", pa.string()), pa.field("price", pa.float64()),
     pa.field("size", pa.int64())]
)
db.create_table("fills", fill_schema, time_column="ts", sort_key=["ts", "symbol"])
db.append("fills", pa.Table.from_pandas(fills, preserve_index=False).cast(fill_schema))
print(f"{len(fills):,} fills, {fills['size'].sum():,} shares, "
      f"~{pov:.0%} of aggressive buy volume")

qwin = db.sql(
    """
    SELECT * FROM quotes
    WHERE symbol = 'AAPL'
      AND ts >= '2026-06-02T13:50:00Z' AND ts < '2026-06-02T14:10:00Z'
    ORDER BY ts
    """
).to_arrow()
db.create_table("quotes_arrival", quote_schema, time_column="ts", sort_key=["ts", "symbol"])
db.append("quotes_arrival", qwin.cast(quote_schema))

parent = pa.table(
    {"ts": pa.array([t0], type=pa.timestamp("us", tz="UTC")), "symbol": ["AAPL"]}
).cast(pa.schema([ts_field, pa.field("symbol", pa.string())]))
db.create_table("parent", parent.schema, time_column="ts", sort_key=["ts", "symbol"])
db.append("parent", parent)

arrival = db.sql(
    """
    SELECT ts, symbol, ts_right AS quote_ts, (bid + ask) / 2 AS arrival_mid
    FROM asof_join('parent', 'quotes_arrival', 'ts', 'ts', 'symbol')
    """
).to_pandas()

direct = db.sql(
    """
    SELECT (bid + ask) / 2 AS mid FROM quotes
    WHERE symbol = 'AAPL' AND ts <= '2026-06-02T14:00:00Z'
    ORDER BY ts DESC LIMIT 1
    """
).to_pandas()["mid"][0]
assert np.isclose(arrival["arrival_mid"][0], direct)
arrival

fill_vwap = db.sql("SELECT vwap(price, size) AS v FROM fills").to_pandas()["v"][0]
mkt_vwap = db.sql(
    """
    SELECT vwap(price, size) AS v FROM trades
    WHERE symbol = 'AAPL'
      AND ts >= '2026-06-02T14:00:00Z' AND ts < '2026-06-02T15:00:00Z'
    """
).to_pandas()["v"][0]
arrival_mid = arrival["arrival_mid"][0]

print(f"arrival mid          : {arrival_mid:.4f}")
print(f"interval VWAP (14-15): {mkt_vwap:.4f}")
print(f"fill VWAP            : {fill_vwap:.4f}")
print(f"slippage vs interval VWAP: {(fill_vwap / mkt_vwap - 1) * 1e4:+.2f} bps")
print(f"slippage vs arrival mid  : {(fill_vwap / arrival_mid - 1) * 1e4:+.2f} bps")

fp = fills
mkt = db.sql(
    """
    SELECT ts, price, vwap(price, size) OVER (ORDER BY ts) AS running_vwap
    FROM trades
    WHERE symbol = 'AAPL'
      AND ts >= '2026-06-02T14:00:00Z' AND ts < '2026-06-02T15:00:00Z'
    ORDER BY ts
    """
).to_pandas()

fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(mkt["ts"], mkt["price"], lw=0.3, color="0.7", label="market trades")
ax.plot(mkt["ts"], mkt["running_vwap"], lw=1.4, label="running interval VWAP")
ax.scatter(fp["ts"], fp["price"], s=6, color="tab:red", label="our fills", zorder=3)
ax.axhline(arrival_mid, ls=":", color="tab:green", label="arrival mid")
ax.axhline(fill_vwap, ls="--", color="tab:red", label="fill VWAP")
ax.set_title("50k AAPL buy, 14:00-15:00 UTC: fills vs benchmarks")
ax.set_xlabel("time (UTC)")
ax.set_ylabel("price")
ax.legend(loc="best", fontsize=8)
fig.tight_layout()

eq = db.sql(
    "SELECT vwap(price, size) AS vwap, wavg(size, price) AS wavg FROM fills"
).to_pandas()
assert np.isclose(eq["vwap"][0], eq["wavg"][0])
eq