Without the SDK: plain HTTP

"""Kairos wallet API (/v3) without the SDK: setup, order, sign in, read, cancel.

    pip install requests eth-account
    KAIROS_URL=https://<base url> KAIROS_PARTNER_ID=<id> PRIVATE_KEY=0x... python kairos_wallet.py hyperliquid:11492
"""
import json, os, sys, time

import requests
from eth_account import Account
from eth_account.messages import encode_defunct, encode_typed_data

BASE = os.environ["KAIROS_URL"].rstrip("/")
HEADERS = {
    "X-Kairos-Partner-Id": os.environ["KAIROS_PARTNER_ID"],
    "X-Kairos-Wallet-Api": "2",
}
account = Account.from_key(os.environ["PRIVATE_KEY"])
WALLET = account.address


def api(method, path, body=None, headers=None):
    r = requests.request(method, BASE + path, json=body, headers={**HEADERS, **(headers or {})}, timeout=60)
    data = r.json() if r.headers.get("content-type", "").startswith("application/json") else {"error": r.text}
    if r.status_code >= 400:
        raise RuntimeError(f"{method} {path}: {r.status_code} {data}")
    return data


def as_ints(types, type_name, value):
    """eth-account wants integers where the typed data carries uint/int as strings."""
    fields = {f["name"]: f["type"] for f in types.get(type_name, [])}
    out = {}
    for name, raw in value.items():
        kind = fields.get(name, "")
        base, is_list = kind.removesuffix("[]"), kind.endswith("[]")
        if base in types:
            out[name] = [as_ints(types, base, v) for v in raw] if is_list else as_ints(types, base, raw)
        elif base.startswith(("uint", "int")) and not is_list and isinstance(raw, str):
            out[name] = int(raw, 0)
        else:
            out[name] = raw
    return out


def sign(request):
    """Sign a step's EIP-1193 request with the local key. A real wallet does this via provider.request."""
    method, params = request["method"], request["params"]
    if method == "personal_sign":
        message = encode_defunct(hexstr=params[0])
    elif method == "eth_signTypedData_v4":
        typed = json.loads(params[1])  # the typed data is a JSON string
        typed["domain"] = as_ints(typed["types"], "EIP712Domain", typed["domain"])
        typed["message"] = as_ints(typed["types"], typed["primaryType"], typed["message"])
        message = encode_typed_data(full_message=typed)
    else:
        raise ValueError(f"not a signature request: {method}")
    return "0x" + account.sign_message(message).signature.hex().removeprefix("0x")


def send_calls(step):
    """wallet_sendCalls (Predict.fun approvals): send each fallback transaction, return the hashes."""
    from web3 import Web3  # pip install web3; only Predict.fun setup needs it

    w3 = Web3(Web3.HTTPProvider("https://bsc-dataseed.binance.org"))
    hashes = []
    for req in step["fallbackRequests"]:
        call = req["params"][0]
        tx = {"to": call["to"], "data": call["data"], "value": 0, "chainId": 56,
              "nonce": w3.eth.get_transaction_count(WALLET, "pending"), "gasPrice": w3.eth.gas_price}
        tx["gas"] = w3.eth.estimate_gas({**tx, "from": WALLET})
        sent = w3.eth.send_raw_transaction(account.sign_transaction(tx).raw_transaction)
        w3.eth.wait_for_transaction_receipt(sent)  # the next nonce needs this one mined
        hashes.append("0x" + sent.hex().removeprefix("0x"))
    return hashes


def run(action):
    """Start an action and answer every step until it is done."""
    step = api("POST", "/v3/actions", {"wallet": WALLET, "action": action})
    while True:
        status = step["status"]
        if status == "done":
            return step["result"]
        if status == "failed":
            raise RuntimeError(f"{step.get('errorCode') or 'FAILED'}: {step['error']}")
        if status == "needs_signature":
            print(f"  sign on {step.get('chainId') or 'any chain'}: {step['description']}")
            if step["request"]["method"] == "wallet_sendCalls":
                answer = {"result": send_calls(step)}
            else:
                answer = {"signature": sign(step["request"])}
            step = api("POST", f"/v3/actions/{step['id']}", answer)
        else:  # working: wait for the venue, or until a delayed market's hold ends
            hold = step.get("holdUntilMs")
            time.sleep(max(0.25, hold / 1000 - time.time()) if hold else 2)
            step = api("GET", f"/v3/actions/{step['id']}")


if __name__ == "__main__":
    market = sys.argv[1] if len(sys.argv) > 1 else "hyperliquid:11492"
    venue = market.split(":")[0]

    print("status", json.dumps(api("GET", f"/v3/wallets/{WALLET}")["venues"].get(venue)))
    print("setup", run({"type": "setup", "venues": [venue]}))

    order = run({
        "type": "order", "market": market, "outcome": "Yes", "side": "buy",
        "price": "0.01", "size": "1000", "timeInForce": "GTC", "postOnly": True,
    })
    print("order", order)

    token = run({"type": "login"})["token"]
    auth = {"Authorization": f"SIWE {token}"}
    open_orders = api("GET", f"/v3/wallets/{WALLET}/orders?open=true&venue={venue}", headers=auth)["orders"]
    print("open orders", [(o["orderId"], o["status"], o["price"], o["size"]) for o in open_orders])

    print("cancel", run({"type": "cancel", "orderIds": [order["orderId"]]}))
    positions = api("GET", f"/v3/wallets/{WALLET}/positions?venue={venue}", headers=auth)["positions"]
    print("positions", len(positions["positions"]), "resolved", len(positions["resolved"]))