"""Individual maker orders through the unmodified Hyperliquid Python SDK.
Install: pip install hyperliquid-python-sdk==0.24.0
Keys come from MM_PRIVATE_KEY; requests use MM_BASE_URL and MM_MASTER_ADDRESS.
"""
import argparse
import json
import os
import sys
import time
from eth_account import Account
from hyperliquid.exchange import Exchange
from hyperliquid.info import Info
from hyperliquid.utils.types import Cloid
def show(result):
print(json.dumps(result, indent=2))
if isinstance(result, dict):
statuses = result.get("response", {})
if isinstance(statuses, dict):
statuses = statuses.get("data", {}).get("statuses", [])
else:
statuses = []
if result.get("status") == "err" or any(
isinstance(status, dict) and "error" in status for status in statuses
):
raise SystemExit(1)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base-url", default=os.environ.get("MM_BASE_URL"))
parser.add_argument("--master", default=os.environ.get("MM_MASTER_ADDRESS"))
parser.add_argument("--coin", default="WETH/USDC")
commands = parser.add_subparsers(dest="command", required=True)
for name in ("metadata", "open-orders", "fills"):
commands.add_parser(name)
for name in ("order", "modify"):
command = commands.add_parser(name)
if name == "modify":
command.add_argument("oid", type=int)
command.add_argument("--side", choices=("buy", "sell"), required=True)
command.add_argument("--price", type=float, required=True)
command.add_argument("--size", type=float, required=True)
command.add_argument("--tif", choices=("Alo", "Gtc"), default="Alo")
command.add_argument("--cloid")
commands.add_parser("cancel").add_argument("oid", type=int)
commands.add_parser("cancel-by-cloid").add_argument("cloid")
commands.add_parser("schedule-cancel").add_argument(
"--after-seconds", type=int, help="Omit to disarm; minimum 5 seconds"
)
approve = commands.add_parser("approve-agent")
approve.add_argument("--name", default="quoting")
approve.add_argument("--key-file", required=True, help="New private dotenv file; must not exist")
args = parser.parse_args()
if not args.base_url:
parser.error("Set MM_BASE_URL to the adapter URL supplied by Spire")
if args.command == "metadata":
show(Info(args.base_url, skip_ws=True).spot_meta())
return
if args.command in ("open-orders", "fills"):
if not args.master:
parser.error("Set MM_MASTER_ADDRESS")
info = Info(args.base_url, skip_ws=True)
show(info.open_orders(args.master) if args.command == "open-orders" else info.user_fills(args.master))
return
key = os.environ.get("MM_PRIVATE_KEY")
if not key:
parser.error("Load the signing key into MM_PRIVATE_KEY")
wallet = Account.from_key(key)
exchange = Exchange(wallet, args.base_url, account_address=args.master or wallet.address)
if args.command == "approve-agent":
# Reserve the private output path before approving a key. Do not log the
# SDK's returned tuple: its second element is the generated private key.
descriptor = os.open(args.key_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "w") as secret_file:
result, agent_key = exchange.approve_agent(args.name)
if result.get("status") != "ok":
show(result)
secret_file.write(f"MM_PRIVATE_KEY={agent_key}\n")
secret_file.flush()
os.fsync(secret_file.fileno())
show(result)
print(f"API-wallet key saved in {args.key_file}", file=sys.stderr)
return
if args.command in ("order", "modify"):
kwargs = dict(name=args.coin, is_buy=args.side == "buy", sz=args.size,
limit_px=args.price, order_type={"limit": {"tif": args.tif}},
cloid=Cloid.from_str(args.cloid) if args.cloid else None)
result = exchange.order(**kwargs) if args.command == "order" else exchange.modify_order(args.oid, **kwargs)
elif args.command == "cancel":
result = exchange.cancel(args.coin, args.oid)
elif args.command == "cancel-by-cloid":
result = exchange.cancel_by_cloid(args.coin, Cloid.from_str(args.cloid))
else:
if args.after_seconds is not None and args.after_seconds < 5:
parser.error("Scheduled cancellation must be at least 5 seconds ahead")
result = exchange.schedule_cancel(
int(time.time() * 1000) + args.after_seconds * 1000 if args.after_seconds is not None else None
)
show(result)
if __name__ == "__main__":
main()
