做市商指南¶

from decimal import Decimal
import json, time, uuid
import requests
from eth_account import Account
from eth_account.messages import encode_typed_data

API     = "https://api.sera.cx/api/v1"
DOMAIN  = {
    "name": "Sera",
    "version": "1",
    "chainId": 1,                                                  # Mainnet
    "verifyingContract": "0xB5C50C5D5f038404F85970b7f5B7259C4AC0E198",
}

PRIVATE_KEY = "0x...你的錢包私鑰..."                                # 簽名者
API_KEY     = "sera_..."                                           # 見 身分驗證
API_SECRET  = "..."
OWNER       = Account.from_key(PRIVATE_KEY).address

AUTH = {"Authorization": f"Bearer {API_KEY}:{API_SECRET}"}

ORDER_TYPES = {
    "Order": [
        {"name": "user",                 "type": "address"},
        {"name": "expiration",           "type": "uint48"},
        {"name": "feeBps",               "type": "uint48"},
        {"name": "recipient",            "type": "address"},
        {"name": "fromToken",            "type": "address"},
        {"name": "toToken",              "type": "address"},
        {"name": "fromAmount",           "type": "uint256"},
        {"name": "toAmount",             "type": "uint256"},
        {"name": "initialDepositAmount", "type": "uint256"},
        {"name": "uuid",                 "type": "uint256"},
    ]
}

CANCEL_ORDER_TYPES = {
    "CancelOrder": [
        {"name": "owner",   "type": "address"},
        {"name": "orderId", "type": "uint256"},
    ]
}

def compose_uuid(order_id: str, executor_id: int = 0,
                 leg_id: int = 0, group_uuid: str | None = None) -> int:
    """把 UUID4 字串打包成鏈上複合 uint256。"""
    oid    = int(uuid.UUID(order_id))
    gsrc   = int(uuid.UUID(group_uuid)) if group_uuid else oid
    gid    = gsrc >> 16                                            # 128 位元中的最高 112 位元
    return (executor_id << 252) | (oid << 124) | (gid << 12) | leg_id

def new_order_id() -> str:
    return str(uuid.uuid4())

USDC = {"address": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", "decimals": 6}
EURC = {"address": "0x1aBaEA1f7C830bD89Acc67eC4af516284b1bC33c", "decimals": 6}

def raw(amount: Decimal, decimals: int) -> int:
    return int(amount * (Decimal(10) ** decimals))

def place_order(side: str, base: dict, quote: dict,
                amount: Decimal, price: Decimal,
                expiration: int) -> dict:
    order_id = new_order_id()
    uuid_int = compose_uuid(order_id)                              # executor_id=0

    # 對的自然 raw
    base_raw  = raw(amount, base["decimals"])
    quote_raw = raw(amount * price, quote["decimals"])

    # EIP-712 結構: fromToken 是簽名者支出的幣。
    if side == "bid":
        from_token, to_token = quote["address"], base["address"]
        from_amount, to_amount = quote_raw, base_raw
    else:
        from_token, to_token = base["address"], quote["address"]
        from_amount, to_amount = base_raw, quote_raw

    order_struct = {
        "user":                 OWNER,
        "expiration":           expiration,
        "feeBps":                0,
        "recipient":            "0x0000000000000000000000000000000000000000",
        "fromToken":            from_token,
        "toToken":              to_token,
        "fromAmount":           from_amount,
        "toAmount":             to_amount,
        "initialDepositAmount":  0,
        "uuid":                 uuid_int,
    }
    signable = encode_typed_data(DOMAIN, ORDER_TYPES, order_struct)
    signature = Account.from_key(PRIVATE_KEY).sign_message(signable).signature.hex()

    wire = {
        "owner_address": OWNER,
        "side":          side,
        "amount":        str(amount),
        "price":         str(price),
        "order_type":    "limit",
        "from_address":  base["address"],                           # base，不是 spend
        "to_address":    quote["address"],                          # quote，不是 receive
        "order_id":      order_id,
        "uuid_int":      str(uuid_int),
        "signature":     "0x" + signature.lstrip("0x"),
        "expiration":    expiration,
    }
    r = requests.post(f"{API}/orders", headers={**AUTH, "Content-Type": "application/json"},
                      data=json.dumps(wire), timeout=10)
    if r.status_code != 201:
        envelope = r.json().get("detail", {})
        raise RuntimeError(f"{envelope.get('error_code')}: {envelope.get('detail')}")
    return {"order_id": order_id, "uuid_int": str(uuid_int)}

placed = place_order("bid", EURC, USDC, Decimal("1000"),
                     Decimal("1.085"), int(time.time()) + 3600)
print("placed:", placed)

def cancel_order(order_id: str, uuid_int: int) -> None:
    signable = encode_typed_data(
        DOMAIN, CANCEL_ORDER_TYPES,
        {"owner": OWNER, "orderId": uuid_int},
    )
    signature = Account.from_key(PRIVATE_KEY).sign_message(signable).signature.hex()
    r = requests.post(
        f"{API}/orders/cancel",
        headers={**AUTH, "Content-Type": "application/json"},
        data=json.dumps({
            "owner_address": OWNER,
            "order_id":      order_id,
            "uuid_int":      str(uuid_int),
            "signature":     "0x" + signature.lstrip("0x"),
        }),
        timeout=10,
    )
    if r.status_code == 429:
        raise RuntimeError("撤單冷卻中 — 稍後重試")
    r.raise_for_status()

cancel_order(placed["order_id"], int(placed["uuid_int"]))

def get_mid(pair: str) -> Decimal:
    """回傳 `pair` 當前的 mid 價。

    實作要對接你自己的資料源 — 內部定價服務、Bloomberg、
    某個聚合器 REST API 等。回傳值是以 quote-per-base 表示的
    Decimal 價格（例如 EURC/USDC = 1.085）。
    """
    raise NotImplementedError("接入你的報價來源。")

SPREAD_BPS = Decimal("5")     # 單側 5 bps
DRIFT_BPS  = Decimal("3")     # 僅當 mid 移動 > 3 bps 才重新掛單
POLL_S     = 2.0
QTY        = Decimal("1000")  # 每側的 base 數量

open_bid = open_ask = None
last_mid = None

while True:
    try:
        mid = get_mid("EURC/USDC")
        if last_mid is not None and \
           abs(mid - last_mid) / last_mid * 10_000 <= DRIFT_BPS:
            time.sleep(POLL_S); continue

        # 先撤再掛 — 避免跨週期自成交。
        if open_bid: cancel_order(**open_bid)
        if open_ask: cancel_order(**open_ask)
        open_bid = open_ask = None

        half   = SPREAD_BPS / Decimal(10_000)
        bid_px = (mid * (Decimal(1) - half)).quantize(Decimal("0.0001"))
        ask_px = (mid * (Decimal(1) + half)).quantize(Decimal("0.0001"))
        expiry = int(time.time()) + 3600

        placed_bid = place_order("bid", EURC, USDC, QTY, bid_px, expiry)
        placed_ask = place_order("ask", EURC, USDC, QTY, ask_px, expiry)
        open_bid   = {"order_id": placed_bid["order_id"], "uuid_int": int(placed_bid["uuid_int"])}
        open_ask   = {"order_id": placed_ask["order_id"], "uuid_int": int(placed_ask["uuid_int"])}
        last_mid   = mid
    except RuntimeError as e:
        # 我們的型別化 error_code: INSUFFICIENT_EQUITY、STP_BLOCKED 等。
        print("place/cancel error:", e)
    except requests.RequestException as e:
        print("network error:", e)
    time.sleep(POLL_S)

def place_vl_batch(legs: list[dict]) -> list[dict]:
    """legs: 每個為 {side, base, quote, amount, price, expiration}。"""
    leg_0_uuid = new_order_id()                                    # group_id 的來源
    signed_legs = []
    for i, leg in enumerate(legs):
        order_id = leg_0_uuid if i == 0 else new_order_id()
        uuid_int = compose_uuid(order_id, executor_id=0,
                                leg_id=i, group_uuid=leg_0_uuid)

        base, quote = leg["base"], leg["quote"]
        base_raw    = raw(leg["amount"], base["decimals"])
        quote_raw   = raw(leg["amount"] * leg["price"], quote["decimals"])
        if leg["side"] == "bid":
            ft, tt, fa, ta = quote["address"], base["address"], quote_raw, base_raw
        else:
            ft, tt, fa, ta = base["address"], quote["address"], base_raw, quote_raw

        struct = {
            "user": OWNER, "expiration": leg["expiration"],
            "feeBps": 0,
            "recipient": "0x0000000000000000000000000000000000000000",
            "fromToken": ft, "toToken": tt,
            "fromAmount": fa, "toAmount": ta,
            "initialDepositAmount": 0, "uuid": uuid_int,
        }
        signable = encode_typed_data(DOMAIN, ORDER_TYPES, struct)
        sig = Account.from_key(PRIVATE_KEY).sign_message(signable).signature.hex()
        signed_legs.append({
            "owner_address": OWNER,
            "side":          leg["side"],
            "amount":        str(leg["amount"]),
            "price":         str(leg["price"]),
            "order_type":    "limit",
            "from_address":  base["address"],
            "to_address":    quote["address"],
            "order_id":      order_id,
            "uuid_int":      str(uuid_int),
            "signature":     "0x" + sig.lstrip("0x"),
            "expiration":    leg["expiration"],
        })

    r = requests.post(
        f"{API}/orders/vl/batch",
        headers={**AUTH, "Content-Type": "application/json"},
        data=json.dumps({"orders": signed_legs}),
        timeout=10,
    )
    r.raise_for_status()
    return r.json()["order_ids"]