Trading HIP-4 outcome markets on Hyperliquid

from hyperliquid.info import Info

info = Info("https://api.hyperliquid.xyz", skip_ws=True)
info.l2_snapshot("#10")  # KeyError: '#10' — coin not registered

encoding = 10 * outcome + side

def encode_coin(outcome_id: int, side: int) -> str:
    """Coin string for l2Book and order placement."""
    if side not in (0, 1):
        raise ValueError(f"side must be 0 or 1, got {side}")
    return f"#{10 * outcome_id + side}"


def encode_balance_coin(outcome_id: int, side: int) -> str:
    """Coin string for spotClearinghouseState balances."""
    if side not in (0, 1):
        raise ValueError(f"side must be 0 or 1, got {side}")
    return f"+{10 * outcome_id + side}"


def encode_asset_id(outcome_id: int, side: int) -> int:
    if side not in (0, 1):
        raise ValueError(f"side must be 0 or 1, got {side}")
    return 100_000_000 + 10 * outcome_id + side

import httpx

HYPERLIQUID_PUBLIC_URL = "https://api.hyperliquid.xyz"


def fetch_outcome_meta(base_url: str = HYPERLIQUID_PUBLIC_URL) -> dict:
    r = httpx.post(
        f"{base_url}/info",
        json={"type": "outcomeMeta"},
        timeout=10.0,
    )
    r.raise_for_status()
    return r.json()


data = fetch_outcome_meta()
for o in data["outcomes"]:
    print(f"outcome={o['outcome']} name={o['name']} desc={o['description']}")
for q in data.get("questions", []):
    print(f"question={q['question']} name={q['name']} legs={q['namedOutcomes']}")

{
  "outcomes": [
    {
      "outcome": 1,
      "name": "Recurring",
      "description": "class:priceBinary|underlying:BTC|expiry:20260504-0600|targetPrice:78213|period:1d",
      "sideSpecs": [{"name": "Yes"}, {"name": "No"}]
    }
  ],
  "questions": []
}

from dataclasses import dataclass


@dataclass
class RecurringSpec:
    cls: str
    underlying: str
    expiry: str
    target_price: float
    period: str


def parse_recurring_description(desc: str) -> RecurringSpec | None:
    if not desc.startswith("class:"):
        return None
    parts = dict(p.split(":", 1) for p in desc.split("|") if ":" in p)
    try:
        return RecurringSpec(
            cls=parts["class"],
            underlying=parts["underlying"],
            expiry=parts["expiry"],
            target_price=float(parts["targetPrice"]),
            period=parts["period"],
        )
    except KeyError:
        return None

from eth_account import Account
from hyperliquid.exchange import Exchange
from hyperliquid.info import Info


def make_clients(
    private_key: str,
    address: str,
    base_url: str = HYPERLIQUID_PUBLIC_URL,
) -> tuple[Info, Exchange]:
    account = Account.from_key(private_key)
    info = Info(base_url, skip_ws=True)
    exchange = Exchange(account, base_url, account_address=address)

    data = fetch_outcome_meta(base_url)
    for o in data.get("outcomes", []):
        for side in (0, 1):
            coin = encode_coin(o["outcome"], side)
            asset_id = encode_asset_id(o["outcome"], side)
            # Skip if the SDK ever ships native HIP-4 support and these are present.
            if coin in info.coin_to_asset:
                continue
            info.coin_to_asset[coin] = asset_id
            info.name_to_coin[coin] = coin
            exchange.info.coin_to_asset[coin] = asset_id
            exchange.info.name_to_coin[coin] = coin

    return info, exchange

Contract settles to YES if and only if
    markPrice0 + (settlementTime - t0) / (t1 - t0) * (markPrice1 - markPrice0) >= targetPrice

import time

def wait_for_settlement(
    info: Info,
    address: str,
    outcome_id: int,
    poll_interval: float = 5.0,
):
    while True:
        state = info.spot_user_state(address)
        balances = {b["coin"]: b for b in state.get("balances", [])}
        held = any(
            encode_balance_coin(outcome_id, side) in balances
            for side in (0, 1)
        )
        if not held:
            return
        time.sleep(poll_interval)

def fetch_book(base_url: str, coin: str) -> tuple[list, list]:
    r = httpx.post(
        f"{base_url}/info",
        json={"type": "l2Book", "coin": coin},
        timeout=10.0,
    )
    r.raise_for_status()
    levels = r.json().get("levels", [[], []])
    return levels[0], levels[1]


# Mainnet BTC daily, both sides
btc_yes_bids, btc_yes_asks = fetch_book(HYPERLIQUID_PUBLIC_URL, "#10")  # see encoding table
btc_no_bids, btc_no_asks = fetch_book(HYPERLIQUID_PUBLIC_URL, "#11")

info, exchange = make_clients(private_key, address)

result = exchange.order(
    "#10",                        # outcome 1, side 0 (YES) — see encoding table
    is_buy=True,
    sz=10,                        # integer size
    limit_px=0.42,                # 0.001 .. 0.999
    order_type={"limit": {"tif": "Gtc"}},
)
print(result)