> ## Documentation Index
> Fetch the complete documentation index at: https://docs.polaris.supply/llms.txt
> Use this file to discover all available pages before exploring further.

# Post-Trade Analysis

> Build participant-attributed post-trade analytics from raw Hyperliquid events, including wallet flow, liquidity roles, counterparties, execution cost, and forward markouts.

Use this workflow to turn raw Hyperliquid BTC events into participant-attributed
post-trade analytics. You will identify buyers and sellers, classify aggressor
and passive liquidity roles (taker and maker), measure execution cost, calculate
forward midpoint markouts, and summarize activity by wallet and counterparty.

Here, **L4** means wallet-attributed executed trades. It does not mean a complete
order lifecycle reconstruction because the public trade event does not contain
every order-state transition.

<Card title="Run the complete notebook" icon="github" href="https://github.com/polaris-data/notebooks/blob/main/notebooks/hyperliquid_l4_post_trade_analysis.ipynb" cta="Open on GitHub" arrow="true">
  Open the executed notebook when you want the same workflow with rendered
  tables and charts.
</Card>

## Set up the notebook

Clone the example repository, install its `uv` environment, and start
JupyterLab:

```bash theme={null}
git clone https://github.com/polaris-data/notebooks.git
cd notebooks
make install
make notebook
```

The example discovers a recent active interval from the open Catalog and does
not require a Polaris API key.

Start with the imports and helpers used throughout the analysis:

```python theme={null}
from itertools import islice
import warnings

from polaris_data import PolarisClient
import matplotlib.pyplot as plt
import pandas as pd

plt.style.use("seaborn-v0_8-darkgrid")
pd.set_option("display.max_columns", 40)
pd.set_option("display.max_colwidth", 100)
warnings.filterwarnings(
    "ignore",
    message=r".*snapshot coverage.*",
    category=UserWarning,
)


def as_utc(value):
    timestamp = pd.Timestamp(value)
    return (
        timestamp.tz_localize("UTC")
        if timestamp.tzinfo is None
        else timestamp.tz_convert("UTC")
    )


def accessible_bounds(market_info):
    """Return the no-key catalog interval for an open or preview market."""
    start = as_utc(market_info["start"])
    end = as_utc(market_info["end"])
    access = market_info.get("access") or {}
    cutoff = access.get("public_cutoff_date")
    if access.get("status") == "preview" and cutoff:
        public_day = as_utc(cutoff)
        start = max(start, public_day)
        end = min(end, public_day + pd.Timedelta(days=1))
    if start >= end:
        raise ValueError(
            "Catalog metadata does not expose a no-key interval for this market"
        )
    return start, end


def bounded_rows(iterator, limit):
    """Materialize at most limit rows and close a partially consumed iterator."""
    rows = list(islice(iterator, limit + 1))
    truncated = len(rows) > limit
    close = getattr(iterator, "close", None)
    if close is not None:
        close()
    return rows[:limit], truncated


def fetch_events(source, market, start, end, row_cap):
    with PolarisClient() as client:
        iterator = client.events(
            source=source,
            market=market,
            from_=start,
            to=end,
            allow_gaps=True,
            materialize_orderbooks=False,
        )
        events, truncated = bounded_rows(iterator, row_cap)
    if truncated:
        raise RuntimeError(
            f"The event query exceeded the {row_cap:,}-row notebook cap. "
            "Shorten window_length before continuing so the analysis is not partial."
        )
    return events


def raw_channel(event):
    return (event.get("raw") or {}).get("channel")


def short_wallet(value):
    if not isinstance(value, str) or len(value) < 13:
        return value
    return f"{value[:6]}…{value[-4:]}"
```

`bounded_rows` prevents the tutorial from silently materializing an unbounded
event stream. `fetch_events` raises if the query crosses the 100,000-row cap so
later summaries never describe a partial sample.

## Find a recent active window

The analysis uses 5 seconds of pre-trade reference data and 65 seconds after
the analysis window. This buffer lets every trade find an arrival midpoint and
5s, 30s, and 60s forward references within a 5-second tolerance.

```python theme={null}
source = "hyperliquid"
market = "BTC"
window_length = pd.Timedelta(minutes=15)
lookback = pd.Timedelta(seconds=5)
markout_horizons = (5, 30, 60)
reference_tolerance = pd.Timedelta(seconds=5)
forward_buffer = (
    pd.Timedelta(seconds=max(markout_horizons)) + reference_tolerance
)
row_cap = 100_000
search_steps = 24 * 60 // int(window_length / pd.Timedelta(minutes=1))
```

Catalog coverage can extend beyond the most recent locally available event
snapshot. Search backward in 15-minute chunks until you find both raw trades
and midpoint updates, then anchor the analysis to the latest observed exchange
timestamp:

```python theme={null}
with PolarisClient() as client:
    catalog = client.catalog(source=source, market=market)

market_info = catalog["markets"][0]
available_start, available_end = accessible_bounds(market_info)

probe_events = None
for offset in range(search_steps):
    probe_end = available_end - offset * window_length
    probe_start = max(available_start, probe_end - window_length)
    if probe_start >= probe_end:
        break
    candidate = fetch_events(
        source,
        market,
        probe_start,
        probe_end,
        row_cap,
    )
    channels = {raw_channel(event) for event in candidate}
    if "trades" in channels and "activeAssetCtx" in channels:
        probe_events = candidate
        break

if not probe_events:
    raise RuntimeError(
        f"No raw trades with midpoint context found for "
        f"{source}:{market} in the last 24 hours"
    )

latest_event_time = pd.to_datetime(
    max(
        event["exchange_timestamp"]
        for event in probe_events
        if event.get("exchange_timestamp") is not None
    ),
    unit="ms",
    utc=True,
)
window_end = latest_event_time - forward_buffer
window_start = window_end - window_length
fetch_start = max(available_start, window_start - lookback)
fetch_end = min(available_end, window_end + forward_buffer)

events = fetch_events(
    source,
    market,
    fetch_start,
    fetch_end,
    row_cap,
)

print(f"Catalog coverage: {available_start} -> {available_end}")
print(f"Analysis window: {window_start} -> {window_end}")
print(f"Fetched with reference buffers: {fetch_start} -> {fetch_end}")
print(f"Event rows: {len(events):,}")
display(pd.Series(market_info))
```

The selected interval moves as Catalog coverage advances. Record the printed
bounds with any saved result.

## Inspect normalized and raw trades

Polaris keeps the normalized `data` projection beside the original venue
message in `raw`. The normalized projection provides consistent price,
quantity, and side fields. Hyperliquid's venue-native
[`WsTrade`](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions)
payload adds the buyer and seller wallets, transaction hash, and trade ID
required for participant attribution.

```python theme={null}
trade_event_sample = next(
    event for event in events if raw_channel(event) == "trades"
)
raw_batch = trade_event_sample["raw"]["data"]

event_envelope = pd.Series(
    {
        "source": trade_event_sample.get("source"),
        "market": trade_event_sample.get("market"),
        "type": trade_event_sample.get("type"),
        "exchange_timestamp": pd.to_datetime(
            trade_event_sample.get("exchange_timestamp"),
            unit="ms",
            utc=True,
        ),
        "collector_timestamp": pd.to_datetime(
            trade_event_sample.get("collector_timestamp"),
            unit="ms",
            utc=True,
        ),
        "normalized_data": trade_event_sample.get("data"),
        "raw_channel": trade_event_sample["raw"].get("channel"),
        "raw_batch_size": len(raw_batch),
    }
)

display(event_envelope.to_frame("value"))
raw_trade_preview = dict(raw_batch[0])
raw_trade_preview["users"] = [
    short_wallet(wallet) for wallet in raw_trade_preview.get("users", [])
]
display(
    pd.Series(
        raw_trade_preview,
        name="first raw trade (wallets shortened)",
    ).to_frame()
)
```

Hyperliquid defines `users` as `[buyer, seller]` and `side` as the aggressing
side. The workflow preserves full wallet addresses in its DataFrames and
shortens them only for rendered summaries.

## Flatten and deduplicate raw trades

One Hyperliquid WebSocket message can contain a batch of trades. Polaris emits
one normalized event per trade while retaining the complete original batch on
each projection. Flattening every `raw.data` array therefore repeats records.

Use `(time, coin, tid)` as the composite trade identity. Do not deduplicate on
transaction hash because a zero hash can be valid for TWAP fills.

```python theme={null}
raw_trade_rows = [
    trade
    for event in events
    if raw_channel(event) == "trades"
    for trade in ((event.get("raw") or {}).get("data") or [])
]

raw_trades = pd.DataFrame(raw_trade_rows)
required_raw_columns = [
    "time",
    "coin",
    "tid",
    "px",
    "sz",
    "side",
    "hash",
    "users",
]
missing_columns = sorted(
    set(required_raw_columns) - set(raw_trades.columns)
)
if missing_columns:
    raise ValueError(
        f"Raw Hyperliquid trade payload is missing columns: {missing_columns}"
    )

identity_columns = ["time", "coin", "tid"]
unique_trades = raw_trades.drop_duplicates(identity_columns).copy()
unique_trades["participant_pair_valid"] = unique_trades["users"].map(
    lambda value: isinstance(value, (list, tuple)) and len(value) == 2
)
unique_trades["side_valid"] = unique_trades["side"].isin(["A", "B"])

integrity = pd.Series(
    {
        "event_rows": len(events),
        "normalized_trade_events": sum(
            event.get("type") == "trade" for event in events
        ),
        "flattened_raw_observations": len(raw_trades),
        "unique_raw_trades": len(unique_trades),
        "repeated_raw_observations_removed": (
            len(raw_trades) - len(unique_trades)
        ),
        "invalid_participant_pairs": int(
            (~unique_trades["participant_pair_valid"]).sum()
        ),
        "unsupported_sides": int((~unique_trades["side_valid"]).sum()),
        "duplicate_identities_after_dedup": int(
            unique_trades.duplicated(identity_columns).sum()
        ),
        "zero_hash_trades": int(
            (unique_trades["hash"] == "0x" + "0" * 64).sum()
        ),
    }
)
display(integrity.to_frame("count"))
```

Validate participant pairs and side values before calculating wallet roles:

```python theme={null}
valid_trades = unique_trades.loc[
    unique_trades["participant_pair_valid"]
    & unique_trades["side_valid"]
].copy()
if valid_trades.empty:
    raise RuntimeError(
        "No valid participant-attributed trades remain after raw-payload validation"
    )

valid_trades["timestamp"] = pd.to_datetime(
    valid_trades["time"],
    unit="ms",
    utc=True,
)
valid_trades["price"] = pd.to_numeric(valid_trades["px"], errors="raise")
valid_trades["quantity"] = pd.to_numeric(
    valid_trades["sz"],
    errors="raise",
)
valid_trades["notional_usd"] = (
    valid_trades["price"] * valid_trades["quantity"]
)
valid_trades["buyer"] = valid_trades["users"].str[0]
valid_trades["seller"] = valid_trades["users"].str[1]
valid_trades["aggressor_side"] = valid_trades["side"].map(
    {"B": "buy", "A": "sell"}
)
valid_trades["aggressor_sign"] = valid_trades["side"].map(
    {"B": 1.0, "A": -1.0}
)
valid_trades["aggressor_wallet"] = valid_trades["buyer"].where(
    valid_trades["side"] == "B",
    valid_trades["seller"],
)
valid_trades["passive_wallet"] = valid_trades["seller"].where(
    valid_trades["side"] == "B",
    valid_trades["buyer"],
)
valid_trades["trade_id"] = (
    valid_trades["time"].astype(str)
    + ":"
    + valid_trades["coin"]
    + ":"
    + valid_trades["tid"].astype(str)
)

trades_df = (
    valid_trades.loc[
        valid_trades["timestamp"].between(
            window_start,
            window_end,
            inclusive="left",
        ),
        [
            "trade_id",
            "timestamp",
            "coin",
            "tid",
            "hash",
            "price",
            "quantity",
            "notional_usd",
            "buyer",
            "seller",
            "aggressor_side",
            "aggressor_sign",
            "aggressor_wallet",
            "passive_wallet",
        ],
    ]
    .sort_values("timestamp")
    .reset_index(drop=True)
)

assert trades_df["trade_id"].is_unique
assert trades_df[["buyer", "seller"]].notna().all().all()
assert (trades_df["quantity"] > 0).all()

print(f"Analysis trades: {len(trades_df):,}")
trade_preview = trades_df.head().copy()
for wallet_column in [
    "buyer",
    "seller",
    "aggressor_wallet",
    "passive_wallet",
]:
    trade_preview[wallet_column] = trade_preview[wallet_column].map(
        short_wallet
    )
display(trade_preview)
```

## Attach arrival and forward midpoints

Hyperliquid's raw `activeAssetCtx` message contains `midPx`. Deduplicate repeated
midpoint projections by exchange timestamp and price before joining them to
trades.

For arrival cost, match backward to the latest midpoint observed within the
5-second lookback. For each markout, match forward from the exact horizon
target. Leave unmatched rows missing and report their coverage instead of
inventing a reference price.

```python theme={null}
midpoint_rows = []
for event in events:
    if raw_channel(event) != "activeAssetCtx":
        continue
    context = (((event.get("raw") or {}).get("data") or {}).get("ctx") or {})
    if context.get("midPx") is None or event.get("exchange_timestamp") is None:
        continue
    midpoint_rows.append(
        {
            "mid_timestamp": pd.to_datetime(
                event["exchange_timestamp"],
                unit="ms",
                utc=True,
            ),
            "mid_price": float(context["midPx"]),
        }
    )

midpoints_df = (
    pd.DataFrame(midpoint_rows)
    .drop_duplicates(["mid_timestamp", "mid_price"])
    .sort_values("mid_timestamp")
    .reset_index(drop=True)
)
if midpoints_df.empty:
    raise RuntimeError(
        "No raw activeAssetCtx midpoint observations were available"
    )

analysis_df = pd.merge_asof(
    trades_df.sort_values("timestamp"),
    midpoints_df,
    left_on="timestamp",
    right_on="mid_timestamp",
    direction="backward",
    tolerance=lookback,
).rename(
    columns={
        "mid_timestamp": "arrival_mid_timestamp",
        "mid_price": "arrival_mid",
    }
)

analysis_df["arrival_cost_bps"] = (
    analysis_df["aggressor_sign"]
    * (analysis_df["price"] - analysis_df["arrival_mid"])
    / analysis_df["arrival_mid"]
    * 10_000
)

coverage_rows = [
    {
        "reference": "arrival",
        "matched": int(analysis_df["arrival_mid"].notna().sum()),
        "unmatched": int(analysis_df["arrival_mid"].isna().sum()),
    }
]
```

Calculate the forward midpoint and aggressor-signed markout at each horizon:

```python theme={null}
for horizon in markout_horizons:
    target_column = f"target_{horizon}s"
    reference_time_column = f"mid_timestamp_{horizon}s"
    reference_price_column = f"mid_{horizon}s"
    markout_column = f"markout_{horizon}s_bps"

    targets = analysis_df[["trade_id", "timestamp"]].copy()
    targets[target_column] = (
        targets["timestamp"] + pd.Timedelta(seconds=horizon)
    ).astype("datetime64[ns, UTC]")
    targets = targets.sort_values(target_column)

    horizon_midpoints = midpoints_df.rename(
        columns={
            "mid_timestamp": reference_time_column,
            "mid_price": reference_price_column,
        }
    )
    horizon_midpoints[reference_time_column] = horizon_midpoints[
        reference_time_column
    ].astype("datetime64[ns, UTC]")

    joined = pd.merge_asof(
        targets,
        horizon_midpoints,
        left_on=target_column,
        right_on=reference_time_column,
        direction="forward",
        tolerance=reference_tolerance,
    )
    matched = joined[reference_time_column].notna()
    assert (
        joined.loc[matched, reference_time_column]
        >= joined.loc[matched, target_column]
    ).all()

    analysis_df = analysis_df.merge(
        joined[
            [
                "trade_id",
                target_column,
                reference_time_column,
                reference_price_column,
            ]
        ],
        on="trade_id",
        how="left",
        validate="one_to_one",
    )
    analysis_df[markout_column] = (
        analysis_df["aggressor_sign"]
        * (
            analysis_df[reference_price_column]
            - analysis_df["price"]
        )
        / analysis_df["price"]
        * 10_000
    )
    coverage_rows.append(
        {
            "reference": f"{horizon}s markout",
            "matched": int(
                analysis_df[reference_price_column].notna().sum()
            ),
            "unmatched": int(
                analysis_df[reference_price_column].isna().sum()
            ),
        }
    )

reference_coverage = pd.DataFrame(coverage_rows)
reference_coverage["coverage_pct"] = (
    reference_coverage["matched"]
    / (
        reference_coverage["matched"]
        + reference_coverage["unmatched"]
    )
    * 100
)
display(reference_coverage)
```

Positive arrival cost means the aggressor traded through the contemporaneous
midpoint. Positive markout means the midpoint subsequently moved in the
aggressor's direction.

## Summarize market-level activity

Weight execution cost and markouts by trade notional so a burst of small prints
does not dominate the aggregate result:

```python theme={null}
all_wallets = pd.concat(
    [analysis_df["buyer"], analysis_df["seller"]],
    ignore_index=True,
)
buy_notional = analysis_df.loc[
    analysis_df["aggressor_side"] == "buy",
    "notional_usd",
].sum()
sell_notional = analysis_df.loc[
    analysis_df["aggressor_side"] == "sell",
    "notional_usd",
].sum()
total_notional = analysis_df["notional_usd"].sum()

activity_summary = pd.Series(
    {
        "trades": len(analysis_df),
        "base_volume_btc": analysis_df["quantity"].sum(),
        "notional_usd": total_notional,
        "unique_wallets": all_wallets.nunique(),
        "aggressive_buy_notional_pct": (
            100 * buy_notional / total_notional
            if total_notional
            else float("nan")
        ),
        "aggressive_sell_notional_pct": (
            100 * sell_notional / total_notional
            if total_notional
            else float("nan")
        ),
        "net_aggressive_base_btc": (
            analysis_df["aggressor_sign"] * analysis_df["quantity"]
        ).sum(),
    }
)

markout_rows = []
metrics = [("arrival cost", "arrival_cost_bps")] + [
    (f"{horizon}s markout", f"markout_{horizon}s_bps")
    for horizon in markout_horizons
]
for label, column in metrics:
    valid = analysis_df[column].notna()
    weights = analysis_df.loc[valid, "notional_usd"]
    weighted_value = (
        (analysis_df.loc[valid, column] * weights).sum() / weights.sum()
        if weights.sum()
        else float("nan")
    )
    markout_rows.append(
        {
            "metric": label,
            "notional_weighted_bps": weighted_value,
            "median_bps": analysis_df.loc[valid, column].median(),
            "trades": int(valid.sum()),
        }
    )

markout_summary = pd.DataFrame(markout_rows)
display(activity_summary.to_frame("value"))
display(markout_summary)
```

## Attribute roles and flow to wallets

Each trade becomes two participant observations: the buyer receives positive
signed flow and the seller receives negative signed flow. The aggressing side
determines which wallet demanded liquidity and which wallet supplied it.

```python theme={null}
base_columns = [
    "trade_id",
    "timestamp",
    "quantity",
    "notional_usd",
    "arrival_mid",
    "price",
] + [f"mid_{horizon}s" for horizon in markout_horizons]

buyers = analysis_df[base_columns].copy()
buyers["wallet"] = analysis_df["buyer"]
buyers["participant_side"] = "buy"
buyers["participant_sign"] = 1.0
buyers["role"] = analysis_df["aggressor_side"].map(
    {"buy": "aggressor", "sell": "passive"}
)

sellers = analysis_df[base_columns].copy()
sellers["wallet"] = analysis_df["seller"]
sellers["participant_side"] = "sell"
sellers["participant_sign"] = -1.0
sellers["role"] = analysis_df["aggressor_side"].map(
    {"buy": "passive", "sell": "aggressor"}
)

participants_df = pd.concat([buyers, sellers], ignore_index=True)
participants_df["signed_base"] = (
    participants_df["participant_sign"] * participants_df["quantity"]
)
participants_df["signed_notional"] = (
    participants_df["participant_sign"]
    * participants_df["notional_usd"]
)
participants_df["aggressor_notional"] = participants_df[
    "notional_usd"
].where(participants_df["role"] == "aggressor", 0.0)
participants_df["passive_notional"] = participants_df[
    "notional_usd"
].where(participants_df["role"] == "passive", 0.0)
participants_df["participant_cost_bps"] = (
    participants_df["participant_sign"]
    * (participants_df["price"] - participants_df["arrival_mid"])
    / participants_df["arrival_mid"]
    * 10_000
)
```

Reconcile the participant expansion before aggregating it. Every trade must
produce exactly two rows, participant notional must equal twice the market
notional, and signed base flow must net to zero:

```python theme={null}
participant_rows_per_trade = participants_df.groupby("trade_id").size()
assert (participant_rows_per_trade == 2).all()

notional_reconciliation_error = abs(
    participants_df["notional_usd"].sum()
    - 2 * analysis_df["notional_usd"].sum()
)
assert notional_reconciliation_error < max(
    1e-6,
    analysis_df["notional_usd"].sum() * 1e-12,
)
assert abs(participants_df["signed_base"].sum()) < 1e-9

for horizon in markout_horizons:
    column = f"participant_markout_{horizon}s_bps"
    participants_df[column] = (
        participants_df["participant_sign"]
        * (
            participants_df[f"mid_{horizon}s"]
            - participants_df["price"]
        )
        / participants_df["price"]
        * 10_000
    )
    participants_df[f"weighted_markout_{horizon}s"] = (
        participants_df[column] * participants_df["notional_usd"]
    )
    participants_df[f"markout_weight_{horizon}s"] = participants_df[
        "notional_usd"
    ].where(participants_df[column].notna(), 0.0)
```

Aggregate wallet activity and rank buyer-seller counterparty pairs:

```python theme={null}
wallet_group = participants_df.groupby("wallet", sort=False)
participant_summary = wallet_group.agg(
    trade_count=("trade_id", "nunique"),
    total_notional=("notional_usd", "sum"),
    aggressor_notional=("aggressor_notional", "sum"),
    passive_notional=("passive_notional", "sum"),
    net_base=("signed_base", "sum"),
    net_notional=("signed_notional", "sum"),
).reset_index()
participant_summary["aggressor_share_pct"] = (
    participant_summary["aggressor_notional"]
    / participant_summary["total_notional"]
    * 100
)

for horizon in markout_horizons:
    numerator = wallet_group[f"weighted_markout_{horizon}s"].sum()
    denominator = wallet_group[f"markout_weight_{horizon}s"].sum()
    participant_summary = participant_summary.merge(
        (
            numerator / denominator.where(denominator != 0)
        ).rename(f"weighted_markout_{horizon}s_bps").reset_index(),
        on="wallet",
        how="left",
        validate="one_to_one",
    )

participant_summary["wallet_label"] = participant_summary["wallet"].map(
    short_wallet
)
participant_summary = participant_summary.sort_values(
    "total_notional",
    ascending=False,
).reset_index(drop=True)

counterparties = (
    analysis_df.groupby(["buyer", "seller"], as_index=False)
    .agg(
        trade_count=("trade_id", "nunique"),
        notional_usd=("notional_usd", "sum"),
    )
    .sort_values("notional_usd", ascending=False)
    .reset_index(drop=True)
)
counterparties["buyer_label"] = counterparties["buyer"].map(short_wallet)
counterparties["seller_label"] = counterparties["seller"].map(short_wallet)

top_wallet_share = (
    participant_summary.head(10)["total_notional"].sum()
    / participant_summary["total_notional"].sum()
    * 100
)
print(
    f"Top 10 wallet share of participant notional: "
    f"{top_wallet_share:.1f}%"
)
display(
    participant_summary[
        [
            "wallet_label",
            "trade_count",
            "total_notional",
            "aggressor_notional",
            "passive_notional",
            "net_base",
            "net_notional",
            "aggressor_share_pct",
            "weighted_markout_5s_bps",
            "weighted_markout_30s_bps",
            "weighted_markout_60s_bps",
        ]
    ].head(15)
)
display(
    counterparties[
        [
            "buyer_label",
            "seller_label",
            "trade_count",
            "notional_usd",
        ]
    ].head(15)
)
```

## Plot executions and midpoint

Plot raw executions over Hyperliquid's midpoint. Marker area scales with trade
notional and is capped at the 95th percentile so a single large print does not
hide the rest of the sample.

```python theme={null}
fig, ax = plt.subplots(figsize=(14, 6))
window_midpoints = midpoints_df.loc[
    midpoints_df["mid_timestamp"].between(
        window_start,
        window_end,
        inclusive="both",
    )
]
ax.plot(
    window_midpoints["mid_timestamp"],
    window_midpoints["mid_price"],
    color="#1f77b4",
    linewidth=1.4,
    label="Hyperliquid midPx",
)

size_cap = analysis_df["notional_usd"].quantile(0.95)
size_cap = size_cap if size_cap > 0 else 1.0
marker_sizes = 10 + 65 * (
    analysis_df["notional_usd"].clip(upper=size_cap) / size_cap
) ** 0.5
for side, color in [("buy", "#2ca02c"), ("sell", "#d62728")]:
    rows = analysis_df["aggressor_side"] == side
    ax.scatter(
        analysis_df.loc[rows, "timestamp"],
        analysis_df.loc[rows, "price"],
        s=marker_sizes.loc[rows],
        color=color,
        alpha=0.28,
        label=f"aggressive {side}",
    )

ax.set_title(f"{source}:{market} raw executions and midpoint")
ax.set_xlabel("Time (UTC)")
ax.set_ylabel("Price (USD)")
ax.legend(loc="best")
plt.tight_layout()
plt.show()
```

## Plot wallet liquidity roles

Compare the aggressor and passive notional attributed to the largest wallets:

```python theme={null}
top_participants = participant_summary.head(12).sort_values(
    "total_notional"
)
fig, ax = plt.subplots(figsize=(11, 7))
ax.barh(
    top_participants["wallet_label"],
    top_participants["aggressor_notional"] / 1_000_000,
    color="#ff7f0e",
    label="aggressor",
)
ax.barh(
    top_participants["wallet_label"],
    top_participants["passive_notional"] / 1_000_000,
    left=top_participants["aggressor_notional"] / 1_000_000,
    color="#4c78a8",
    label="passive",
)
ax.set_title("Top wallets by participant notional")
ax.set_xlabel("Participant notional (USD millions)")
ax.set_ylabel("Wallet")
ax.legend(loc="lower right")
plt.tight_layout()
plt.show()
```

## Plot the aggregate markout curve

Plot the notional-weighted markout for aggressors at each forward horizon:

```python theme={null}
markout_plot = markout_summary.loc[
    markout_summary["metric"].str.contains("markout")
].copy()
fig, ax = plt.subplots(figsize=(9, 5))
colors = [
    "#2ca02c" if value >= 0 else "#d62728"
    for value in markout_plot["notional_weighted_bps"]
]
bars = ax.bar(
    markout_plot["metric"],
    markout_plot["notional_weighted_bps"],
    color=colors,
)
ax.axhline(0, color="black", linewidth=0.8)
ax.bar_label(bars, fmt="%.3f bps", padding=3)
ax.set_title("Notional-weighted post-trade markout for aggressors")
ax.set_ylabel("Markout (bps; positive is favorable)")
ax.set_xlabel("Forward horizon")
plt.tight_layout()
plt.show()
```

## Interpret the results carefully

* Wallet addresses are public account identifiers, not verified trader
  identities. One entity can control many wallets, and one wallet can represent
  a vault or subaccount.
* The raw trade stream supports participant attribution after execution, but it
  does not expose complete order lifecycles, client order IDs, fees, positions,
  or realized P\&L.
* Markouts describe short-horizon price movement around an execution. They are
  not portfolio returns and do not include fees, funding, inventory, hedges, or
  activity in other markets.
* Zero transaction hashes can be valid for TWAP fills. Use `(time, coin, tid)`,
  not hash, as the deduplication key.
* Results cover a short, dynamically selected public interval. Treat them as a
  reproducible workflow example, not a persistent ranking of wallets.

## Related documentation

* [Events](/schemas/events) for mixed normalized event queries and raw-event behavior
* [Event envelope](/concepts/event-envelope) for collector and exchange timestamps
* [Catalog](/reference/catalog) for market identity, coverage, and access metadata
* [Perpetuals](/guides/perpetuals) for funding, trade-flow, liquidity, and order-book workflows
* [Liquidity / Microstructure](/guides/liquidity-microstructure) for cross-venue spread, slippage, and L2 depth analysis
* [Jupyter Notebook](/guides/jupyter-notebook-quickstart) for notebook setup and the curated example index