> ## 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.

# PropAMM Competitor Analysis

> Compare aligned PropAMM quote ladders at exact shared input sizes, including quote edge, curve impact, and observation coverage.

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.

<Card title="Run the complete notebook" icon="github" href="https://github.com/polaris-data/notebooks/blob/main/notebooks/propamm_quote_ladder_analysis.ipynb" cta="Open on GitHub" arrow="true">
  Open the executed notebook for multi-cohort summaries and the full quote-level
  and price-impact visualization.
</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 workflow uses the no-key preview interval reported by Catalog. Start with
the imports, comparison limits, and bounded-query helpers:

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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.

## Related documentation

* [PropAMMs](/guides/propamms) for source discovery, quote-ladder identity, and single-source analysis
* [PropAMM Quote Ladders](/schemas/propamm-quote-ladders) for the complete method signature and event fields
* [Catalog](/reference/catalog) for current source, coverage, and access metadata
* [Event envelope](/concepts/event-envelope) for collector timestamps and stored ordering
* [Jupyter Notebook](/guides/jupyter-notebook-quickstart) for notebook setup and the curated example index