PyBook 03: Assets & Chains

Rule #3, "Asset IDs are deterministic CAIP-19 and permanently stable." Zerion's docs say verbatim: "There is a non-zero probability that IDs may change in the future." We call that "disqualifying for anyone persisting portfolio history, and free to beat."

This book covers CAIP-2/CAIP-19 identity, the pinned deterministic asset id, the both-ways registry, groups with the decimals-equality law, and additive spam scoring.

from auradefi.assets.caip import canonical_caip19, parse_caip19

# USDC on Ethereum, pasted with its usual mixed-case (EIP-55) address:
mixed = "eip155:1/erc20:0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
parsed = parse_caip19(mixed)
assert parsed.chain_id == "eip155:1"
assert parsed.namespace == "erc20"
# Canonical form LOWERCASES EVM addresses: case variants converge.
assert parsed.reference == "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"
assert canonical_caip19(mixed) == canonical_caip19(mixed.lower())
assert canonical_caip19(canonical_caip19(mixed)) == canonical_caip19(mixed)  # idempotent
print("canonical:", canonical_caip19(mixed))

canonical: eip155:1/erc20:0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48

# Solana is the opposite: base58 mints are CASE-SENSITIVE, so canonical
# form PRESERVES case (docs/internal/DECISIONS.md: the asset-id pin depends on it).
SOL_CHAIN = "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp"
usdc_mint = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"
sol_usdc = f"{SOL_CHAIN}/token:{usdc_mint}"
assert parse_caip19(sol_usdc).reference == usdc_mint          # untouched
# A case-flipped mint is a DIFFERENT identifier, not the same asset:
assert canonical_caip19(sol_usdc) != canonical_caip19(f"{SOL_CHAIN}/token:{usdc_mint.lower()}")
print("Solana reference preserved:", parse_caip19(sol_usdc).reference)

Solana reference preserved: EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v

Anything malformed, wrong part count, unknown namespace, a bad reference, surrounding whitespace, raises CaipParseError (a ValidationError subclass). Identity strings are wire contracts; there is no lenient mode.

from auradefi.errors import CaipParseError

bad_inputs = [
    "eip155:1",                                   # no asset part at all
    "eip155:1/erc20:0xdeadbeef",                  # address too short
    "eip155:1/erc721:0x" + "a" * 40,              # namespace not seeded
    "eip155:1/slip44:007",                        # leading zeros not canonical
    f"{SOL_CHAIN}/token:0OIl",                    # 0, O, I, l are not base58
    " eip155:1/slip44:60",                        # whitespace is not identity
]
for value in bad_inputs:
    try:
        parse_caip19(value)
    except CaipParseError as exc:
        print("rejected:", repr(value))
    else:
        raise AssertionError(f"{value!r} must not parse")

rejected: 'eip155:1'
rejected: 'eip155:1/erc20:0xdeadbeef'
rejected: 'eip155:1/erc721:0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
rejected: 'eip155:1/slip44:007'
rejected: 'solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp/token:0OIl'
rejected: ' eip155:1/slip44:60'

Chains: CAIP-2 is the only key

The rule: "CAIP-2 chain ids mean no eth-mainnet (GoldRush) vs ethereum (Allium) vs 1 (Dune SIM) translation table." The registry refuses vendor names by design, and ships pre-seeded with the five seeded chains. native_decimals lives on the chain because "you cannot format an amount without knowing which chain it is on."

from auradefi.chains import evm
from auradefi.chains.registry import ChainRegistry
from auradefi.errors import UnknownChainError

registry = ChainRegistry()
ethereum = registry.get("eip155:1")
assert (ethereum.name, ethereum.native_symbol, ethereum.native_decimals) == ("Ethereum", "ETH", 18)
assert ethereum.native_caip19 == "eip155:1/slip44:60"

seeded = [chain.caip2 for chain in registry.chains()]
print("seed chains:", seeded)
assert len(seeded) == 5  # Ethereum, Polygon, Base + Bitcoin, Solana mainnets

try:
    registry.get("ethereum")  # the vendor name zoo, killed at the door
except UnknownChainError as exc:
    print("rejected:", exc)
else:
    raise AssertionError("vendor names must not resolve")

# EVM helpers round-trip numeric chain id <-> canonical CAIP-2.
assert evm.caip2_from_chain_id(8453) == "eip155:8453"
assert evm.chain_id_from_caip2("eip155:8453") == 8453
assert evm.normalize_address("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48").islower()

seed chains: ['bip122:000000000019d6689c085ae165831e93', 'eip155:1', 'eip155:137', 'eip155:8453', 'solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp']
rejected: unknown chain 'ethereum': CAIP-2 is the only key

Deterministic asset ids: compute one, watch it hold still

The pinned recipe (docs/internal/DECISIONS.md, rule #3):

"ast_" + sha256("\n".join(sorted(canonical_caip19s)).encode()).hexdigest()[:16]

Canonicalize, deduplicate, sort, hash. Input order never matters, EVM address case never matters, and the id below is asserted against a hardcoded literal: if the recipe ever drifted, this notebook (and the golden-vector tests) would fail.

import hashlib

from auradefi.assets.models import asset_id

USDC_ETH = "eip155:1/erc20:0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
USDC_POLY = "eip155:137/erc20:0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359"

usdc_id = asset_id([USDC_ETH, USDC_POLY])
print("asset id:", usdc_id)

assert usdc_id == "ast_a31ab4a449ad2399"                       # stability, pinned
assert asset_id([USDC_POLY, USDC_ETH]) == usdc_id              # order-independent
assert asset_id([USDC_ETH.lower(), USDC_POLY.lower()]) == usdc_id  # case-independent
assert asset_id([USDC_ETH, USDC_ETH, USDC_POLY]) == usdc_id    # dedup-stable

# Recompute from the pinned recipe by hand: same answer.
canonical = sorted({canonical_caip19(c) for c in (USDC_POLY, USDC_ETH)})
by_hand = "ast_" + hashlib.sha256("\n".join(canonical).encode()).hexdigest()[:16]
assert by_hand == usdc_id

asset id: ast_a31ab4a449ad2399

The registry: addressable both ways

Assets are addressable "both ways, on every filter: by canonical id or by chain:address. Zerion does this and it is why their filters are usable." Note decimals lives on the Implementation, never the Asset: bridged assets genuinely differ per chain.

from auradefi.assets.models import AssetClass, Implementation, make_asset
from auradefi.assets.registry import AssetRegistry
from auradefi.errors import UnknownAssetError

usdc = make_asset(
    symbol="USDC",
    name="USD Coin",
    implementations=[
        Implementation(caip19=USDC_ETH, chain_id="eip155:1", decimals=6),
        Implementation(caip19=USDC_POLY, chain_id="eip155:137", decimals=6),
    ],
    asset_class=AssetClass.STABLECOIN,
    external_ids=[("coingecko", "usd-coin")],
)
assert usdc.id == usdc_id  # make_asset computes the same pinned id

assets = AssetRegistry()
assets.register(usdc)
assets.register(usdc)  # identical re-registration is a no-op

# Way 1: by deterministic id.
assert assets.get_by_id(usdc_id) == usdc
# Way 2a: by CAIP-19: any case variant of an EVM address finds it.
shouty = "eip155:1/erc20:0x" + "A0B86991C6218B36C1D19D4A2E9EB0CE3606EB48"
assert assets.get_by_caip19(shouty) == usdc
# Way 2b: by (chain, address): the other half of both-ways addressing.
assert assets.get_by_chain_address("eip155:1", "0xA0B86991C6218B36C1D19D4A2E9EB0CE3606EB48") == usdc

try:
    assets.get_by_chain_address("eip155:1", "0x" + "0" * 40)
except UnknownAssetError as exc:
    print("unknown address:", type(exc).__name__)
else:
    raise AssertionError("expected UnknownAssetError")
print("both-ways lookup verified for", usdc.symbol)

unknown address: UnknownAssetError
both-ways lookup verified for USDC

Groups: the decimals-equality law and the `single` fallback

OneBalance's ob:usdc shape: "one 'USDC' row across seven chains … and an explicit single fallback bucket so nothing falls out of the model. Aggregation is only sound when implementations share decimals: enforce that. Zerion punts this to the client; doing it server-side is a real differentiator."

from auradefi.assets.groups import GroupKind, group_assets
from auradefi.errors import DecimalsMismatchError

weth = make_asset(
    symbol="WETH",
    name="Wrapped Ether",
    implementations=[
        Implementation(
            caip19="eip155:1/erc20:0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2",
            chain_id="eip155:1",
            decimals=18,
        ),
    ],
    asset_class=AssetClass.WRAPPED,
)

groups = group_assets([usdc, weth], {"USDC": [usdc.id]})
by_symbol = {group.symbol: group for group in groups}
assert by_symbol["USDC"].kind == GroupKind.GROUP     # explicit group
assert by_symbol["WETH"].kind == GroupKind.SINGLE    # the fallback bucket
# Nothing falls out: every asset lands in exactly one group.
assert sorted(aid for g in groups for aid in g.asset_ids) == sorted([usdc.id, weth.id])
print({g.symbol: (g.kind.value, g.id) for g in groups})

# The law: an explicit group mixing 6- and 18-decimal members is rejected
# server-side. Aggregating across scales would be meaningless.
try:
    group_assets([usdc, weth], {"MIXED": [usdc.id, weth.id]})
except DecimalsMismatchError as exc:
    print("rejected:", exc)
else:
    raise AssertionError("expected DecimalsMismatchError")

{'WETH': ('single', 'grp_346e08af2bef7b96'), 'USDC': ('group', 'grp_64e0f5fcda5ae8c6')}
rejected: explicit group 'MIXED' mixes decimals [6, 18]; aggregation requires a single value

Spam: the evidence, not a verdict

Rule #9. "Return the liquidity number, not just a spam boolean. The threshold is a product decision, not a vendor decision."

And rotki's scar: "a transient source failure once wiped previously-detected tokens. Make detection additive, never destructive." So an assessment carries score and the raw numbers, is_spam takes the threshold from the caller, and merge can only ever add.

from decimal import Decimal

from auradefi.assets.spam import SpamSignal, assess, is_spam

assessment = assess(
    [
        SpamSignal("airdropped_unrequested", 2.0),
        SpamSignal("symbol_impersonates_usdc", 1.5),
    ],
    liquidity_usd=Decimal("312.50"),   # a Decimal end-to-end, never a float
    holder_count=41,
)
assert assessment.score == 3.5
assert assessment.reasons == ("airdropped_unrequested", "symbol_impersonates_usdc")
assert assessment.liquidity_usd == Decimal("312.50")

# Same evidence, different products, different verdicts: by design.
assert is_spam(assessment, threshold=3.0) is True    # cautious wallet UI
assert is_spam(assessment, threshold=5.0) is False   # permissive analytics
print(f"score={assessment.score} liquidity_usd={assessment.liquidity_usd} "
      f"holders={assessment.holder_count}")

score=3.5 liquidity_usd=312.50 holders=41

from auradefi.assets.spam import merge

# A later scan finds a worse signal. Merge ADDS, keeps the old numbers.
later = assess([SpamSignal("honeypot_sell_reverts", 5.0)])
merged = merge(assessment, later)
assert merged.score == 5.0                                  # max, never sum-reset
assert merged.reasons == (
    "airdropped_unrequested", "symbol_impersonates_usdc", "honeypot_sell_reverts",
)
assert merged.liquidity_usd == Decimal("312.50")            # None never erases

# rotki's scar, prevented: an EMPTY assessment (transient source failure)
# changes NOTHING.
assert merge(merged, assess([])) == merged
print("additive merge holds:", merged.reasons)
print("\nassets & chains walkthrough complete: every assertion held")

additive merge holds: ('airdropped_unrequested', 'symbol_impersonates_usdc', 'honeypot_sell_reverts')

assets & chains walkthrough complete: every assertion held