PropAMM Competitor Analysis

Use this workflow to compare size-dependent quotes from multiple proprietary automated market makers. You will discover available sources, select a directed token pair shared by the largest number of sources, align one representative cohort, and compare quotes at exact common input sizes.

Run the complete notebook

Open the executed notebook for multi-cohort summaries and the full quote-level and price-impact visualization.

Set up the notebook

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

git clone https://github.com/polaris-data/notebooks.git
cd notebooks
make install
make notebook

The workflow uses the no-key preview interval reported by Catalog. Start with the imports, comparison limits, and bounded-query helpers:

from decimal import Decimal
from itertools import islice, product
import warnings

import pandas as pd
from polaris_data import PolarisClient

pd.set_option("display.max_columns", 30)
pd.set_option("display.max_colwidth", 100)
warnings.filterwarnings(
    "ignore",
    message=r".*snapshot coverage.*",
    category=UserWarning,
)

sources = [
    "fermiswap",
    "bopamm",
    "kipseli",
    "metric",
    "tempest",
    "taurusfi",
]
market = "ethereum"
window_length = pd.Timedelta(hours=1)
max_rows_per_source = 5_000
max_block_skew = 12
max_time_skew = pd.Timedelta(minutes=2)
baseline_input_target = Decimal("0.01")
representative_inputs = [Decimal("0.01"), Decimal("1"), Decimal("10")]


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 event_timestamp(row):
    """Support both legacy and v2 Polaris event envelopes."""
    value = row.get("collector_timestamp", row.get("timestamp"))
    return pd.to_datetime(value, unit="ms", utc=True)

Discover and fetch each source

Query up to the first hour of each source’s accessible interval. Keep every source in the coverage table so an empty short window is not confused with an unsupported schema.

rows_by_source = {}
coverage_records = []

with PolarisClient() as client:
    for source in sources:
        catalog = client.catalog(source=source, market=market)
        if not catalog.get("markets"):
            rows_by_source[source] = []
            coverage_records.append(
                {
                    "source": source,
                    "rows": 0,
                    "hit_cap": False,
                    "window": "not in catalog",
                }
            )
            continue

        market_info = catalog["markets"][0]
        start, accessible_end = accessible_bounds(market_info)
        end = min(start + window_length, accessible_end)
        rows, truncated = bounded_rows(
            client.propamm_quote_ladders(
                source=source,
                market=market,
                from_=start,
                to=end,
                allow_gaps=True,
            ),
            max_rows_per_source,
        )
        rows_by_source[source] = rows
        coverage_records.append(
            {
                "source": source,
                "rows": len(rows),
                "hit_cap": truncated,
                "window": f"{start} -> {end}",
            }
        )

coverage_df = pd.DataFrame(coverage_records)
display(coverage_df)

if coverage_df["hit_cap"].any():
    print(
        "At least one source hit the row cap. Shorten window_length or raise "
        "max_rows_per_source before treating the comparison as complete."
    )

allow_gaps=True permits a partial preview interval. Preserve the reported window and hit_cap value with any result.

Select a comparable directed pair

Flatten the event metadata without expanding the quote arrays. Keep Metric pools separate because two pools can produce different curves for the same token pair.

ladder_records = []
for source, rows in rows_by_source.items():
    for row in rows:
        values = (row.get("data") or {}).get("values") or {}
        pool = values.get("pool")
        ladder_records.append(
            {
                "source": source,
                "participant": (
                    source if not pool else f"{source}:{pool.lower()}"
                ),
                "timestamp": event_timestamp(row),
                "chain_id": values.get("chain_id"),
                "token_in": (values.get("token_in") or "").lower(),
                "token_out": (values.get("token_out") or "").lower(),
                "token_in_decimals": values.get("token_in_decimals"),
                "token_out_decimals": values.get("token_out_decimals"),
                "quotes": values.get("quotes") or [],
                "event_id": values.get("event_id"),
                "block_number": values.get("block_number"),
                "transaction_hash": values.get("transaction_hash"),
                "router": values.get("router"),
                "oracle": values.get("oracle"),
                "pool": pool,
            }
        )

ladders_df = pd.DataFrame(ladder_records)
if ladders_df.empty:
    raise ValueError(
        "No PropAMM quote ladders were found in the public sample windows"
    )

pair_coverage = (
    ladders_df.groupby(["token_in", "token_out"])
    .agg(
        sources=("source", "nunique"),
        ladders=("source", "size"),
        quote_points=("quotes", lambda values: sum(map(len, values))),
    )
    .sort_values(
        ["sources", "quote_points", "ladders"],
        ascending=False,
    )
)
selected_pair = pair_coverage.index[0]
print(f"Selected directed pair: {selected_pair[0]} -> {selected_pair[1]}")
display(pair_coverage.head(10))

Token direction is part of the comparison identity. A token_in -> token_out quote cannot be compared directly with the reverse direction.

Align one representative cohort

Use the least frequently observed participant as the cohort anchor. Search its observations from newest to oldest, then choose the nearest complete group that fits within both the 12-block and two-minute limits.

pair_ladders = ladders_df.loc[
    ladders_df["token_in"].eq(selected_pair[0])
    & ladders_df["token_out"].eq(selected_pair[1])
].dropna(subset=["block_number", "timestamp"])

participant_counts = pair_ladders.groupby("participant").size()
participants = sorted(participant_counts.index)
minimum_count = participant_counts.min()
anchor_participant = sorted(
    participant_counts[participant_counts.eq(minimum_count)].index
)[0]

selected_members = None
anchors = pair_ladders.loc[
    pair_ladders["participant"].eq(anchor_participant)
].sort_values("timestamp", ascending=False)

for _, anchor in anchors.iterrows():
    candidate_groups = []
    for participant in participants:
        if participant == anchor_participant:
            candidate_groups.append([anchor])
            continue

        candidates = pair_ladders.loc[
            pair_ladders["participant"].eq(participant)
        ]
        candidates = candidates.loc[
            (candidates["block_number"] - anchor["block_number"])
            .abs()
            .le(max_block_skew)
            & (candidates["timestamp"] - anchor["timestamp"])
            .abs()
            .le(max_time_skew)
        ]
        candidate_groups.append(
            [candidate for _, candidate in candidates.iterrows()]
        )

    if any(not candidates for candidates in candidate_groups):
        continue

    valid_combinations = []
    for combination in product(*candidate_groups):
        block_span = int(
            max(member["block_number"] for member in combination)
            - min(member["block_number"] for member in combination)
        )
        time_span = (
            max(member["timestamp"] for member in combination)
            - min(member["timestamp"] for member in combination)
        )
        if block_span > max_block_skew or time_span > max_time_skew:
            continue

        total_block_distance = sum(
            abs(member["block_number"] - anchor["block_number"])
            for member in combination
        )
        total_time_distance = sum(
            (
                abs(member["timestamp"] - anchor["timestamp"])
                for member in combination
            ),
            pd.Timedelta(0),
        )
        valid_combinations.append(
            (
                block_span,
                time_span,
                total_block_distance,
                total_time_distance,
                combination,
            )
        )

    if valid_combinations:
        selected_members = min(
            valid_combinations,
            key=lambda candidate: candidate[:4],
        )[-1]
        break

if selected_members is None:
    raise ValueError(
        "No complete quote-ladder cohort satisfies the configured skew limits"
    )

cohort_ladders = pd.DataFrame(
    [member.to_dict() for member in selected_members]
)
cohort_ladders["block_offset"] = (
    cohort_ladders["block_number"] - anchor["block_number"]
).astype(int)
cohort_ladders["time_offset_seconds"] = (
    cohort_ladders["timestamp"] - anchor["timestamp"]
).dt.total_seconds()

display(
    cohort_ladders[
        [
            "participant",
            "timestamp",
            "block_number",
            "block_offset",
            "time_offset_seconds",
            "transaction_hash",
            "router",
            "oracle",
            "pool",
            "event_id",
        ]
    ].sort_values("participant")
)

The offsets make observation skew visible. They do not make the quotes simultaneous.

Compare exact shared input sizes

Normalize onchain integer amounts with Decimal. Compare only input amounts present in every ladder, and choose the first shared amount at or above 0.01 as each participant’s curve baseline.

quote_records = []
probe_records = []

for _, ladder in cohort_ladders.iterrows():
    input_scale = Decimal(10) ** int(ladder["token_in_decimals"])
    output_scale = Decimal(10) ** int(ladder["token_out_decimals"])
    participant_quotes = []

    for quote in ladder["quotes"]:
        amount_in = Decimal(quote["amount_in"]) / input_scale
        amount_out = Decimal(quote["amount_out"]) / output_scale
        if amount_in <= 0:
            continue
        participant_quotes.append((amount_in, amount_out))
        quote_records.append(
            {
                "participant": ladder["participant"],
                "amount_in": amount_in,
                "amount_out": amount_out,
                "average_rate": amount_out / amount_in,
            }
        )

    if not participant_quotes:
        raise ValueError(
            f"{ladder['participant']} has no positive quote inputs"
        )
    probe_records.append(
        {
            "participant": ladder["participant"],
            "recorded_quote_points": len(participant_quotes),
            "recorded_min_probe": min(
                amount for amount, _ in participant_quotes
            ),
            "recorded_max_probe": max(
                amount for amount, _ in participant_quotes
            ),
        }
    )

normalized = pd.DataFrame(quote_records)
input_sets = [
    set(values["amount_in"])
    for _, values in normalized.groupby("participant")
]
exact_shared_inputs = sorted(set.intersection(*input_sets))
eligible_inputs = [
    amount
    for amount in exact_shared_inputs
    if amount >= baseline_input_target
]
if not eligible_inputs:
    raise ValueError(
        "No exact shared input meets the configured baseline target"
    )

baseline_input = eligible_inputs[0]
comparison = normalized.loc[
    normalized["amount_in"].isin(eligible_inputs)
].copy()
baseline_rates = (
    comparison.loc[comparison["amount_in"].eq(baseline_input)]
    .set_index("participant")["average_rate"]
)
best_rates = comparison.groupby("amount_in")["average_rate"].max()

comparison["baseline_rate"] = comparison["participant"].map(
    baseline_rates
)
comparison["best_rate"] = comparison["amount_in"].map(best_rates)
comparison["quote_edge_bps"] = comparison.apply(
    lambda row: (
        row["average_rate"] / row["best_rate"] - Decimal(1)
    )
    * Decimal(10_000),
    axis=1,
)
comparison["curve_impact_bps"] = comparison.apply(
    lambda row: (
        row["average_rate"] / row["baseline_rate"] - Decimal(1)
    )
    * Decimal(10_000),
    axis=1,
)

for column in ["amount_out", "quote_edge_bps", "curve_impact_bps"]:
    comparison[f"{column}_float"] = comparison[column].astype(float)

available_inputs = sorted(
    {
        amount
        for amount in representative_inputs
        if amount in set(comparison["amount_in"])
    }
    | {comparison["amount_in"].max()}
)
representative = comparison.loc[
    comparison["amount_in"].isin(available_inputs)
]
comparison_table = representative.pivot(
    index="amount_in",
    columns="participant",
    values=[
        "amount_out_float",
        "quote_edge_bps_float",
        "curve_impact_bps_float",
    ],
)

print(f"Exact shared inputs: {len(exact_shared_inputs):,}")
print(f"Curve baseline input: {baseline_input}")
display(pd.DataFrame(probe_records).set_index("participant"))
display(comparison_table)

At the same input size, a quote_edge_bps value of zero identifies the best observed output in the cohort; negative values trail that output. A negative curve_impact_bps value means a participant returned less output per input than it did at the common baseline size. The probe table describes the range that Polaris observed. Do not interpret recorded_max_probe as executable capacity.

Extend the comparison

The complete notebook repeats this alignment across every valid cohort. It reports quote-point win rate, median and lower-tail quote edge, median and worst curve impact, and plots average rate, quote edge, and curve impact across the latest cohort. When you extend the workflow, keep these controls explicit:

Compare the same chain_id, directed token pair, and exact normalized input amount.
Keep source and pool identity separate.
Record block and observation-time skew for every cohort.
Keep Decimal values through calculations and convert only for presentation.
Treat empty sources, capped queries, and incomplete cohorts as coverage results rather than silently dropping them.
Treat the dynamically selected public sample as a workflow example, not a persistent ranking of PropAMMs.

PropAMMs for source discovery, quote-ladder identity, and single-source analysis
PropAMM Quote Ladders for the complete method signature and event fields
Catalog for current source, coverage, and access metadata
Event envelope for collector timestamps and stored ordering
Jupyter Notebook for notebook setup and the curated example index

By Category

By Workflow

PropAMM Competitor Analysis

Run the complete notebook

Set up the notebook

Discover and fetch each source

Select a comparable directed pair

Align one representative cohort

Compare exact shared input sizes

Extend the comparison

Run the complete notebook

​Set up the notebook

​Discover and fetch each source

​Select a comparable directed pair

​Align one representative cohort

​Compare exact shared input sizes

​Extend the comparison

​Related documentation

Set up the notebook

Discover and fetch each source

Select a comparable directed pair

Align one representative cohort

Compare exact shared input sizes

Extend the comparison

Related documentation