from __future__ import annotations import argparse from datetime import datetime, timezone from decimal import Decimal import json import os from pathlib import Path import time import uuid from eth_account import Account from hyperliquid.exchange import Exchange from hyperliquid.info import Info from hyperliquid.utils.types import Cloid from src.config.hyperliquid_env import load_hyperliquid_env, mask_address from src.hyperliquid.market_data import HyperliquidMarketData from src.hyperliquid.rounding import round_hyperliquid_price, round_hyperliquid_size from src.reconciliation.hyperliquid_reconciler import ReconcilerSnapshot, reconcile_hyperliquid_state TESTNET_URL = "https://api.hyperliquid-testnet.xyz" def _journal(payload: dict) -> Path: path = Path("runtime/reports") / f"hyperliquid_testnet_fill_stop_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.json" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, default=str, sort_keys=True), encoding="utf-8") return path def _open_positions(info: Info, user: str) -> dict[str, dict]: state = info.user_state(user) out: dict[str, dict] = {} for row in state.get("assetPositions", []): pos = row.get("position", {}) try: szi = Decimal(str(pos.get("szi", "0"))) except Exception: szi = Decimal("0") if szi != 0: out[str(pos.get("coin", "")).upper()] = pos return out def _orders_for_coin(info: Info, user: str, coin: str) -> list[dict]: return [o for o in info.open_orders(user) if str(o.get("coin", "")).upper() == coin.upper()] def _extract_oid(resp: dict) -> int | None: try: statuses = resp.get("response", {}).get("data", {}).get("statuses", []) if not statuses: return None status = statuses[0] if "resting" in status: return int(status["resting"]["oid"]) if "filled" in status: return int(status["filled"].get("oid")) if status["filled"].get("oid") is not None else None except Exception: return None return None def _is_stop_order(order: dict) -> bool: text = json.dumps(order, default=str).lower() return bool(order.get("reduceOnly")) or "trigger" in text or "stop" in text or "tpsl" in text def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Testnet fill + exchange-level stop proof. Testnet only; never mainnet.") parser.add_argument("--coin", required=True) parser.add_argument("--notional", type=Decimal, required=True) parser.add_argument("--max-risk-usd", type=Decimal, required=True) parser.add_argument("--leverage", type=int, default=2) args = parser.parse_args(argv) coin = args.coin.upper() result: dict = {"env": os.getenv("CTB_HL_ENV"), "coin": coin, "mainnet_blocked": True, "notional": str(args.notional), "max_risk_usd": str(args.max_risk_usd)} if os.getenv("CTB_HL_ENV", "").lower() != "testnet": result.update(status="skipped", reason="CTB_HL_ENV must be testnet") print(json.dumps(result, indent=2)); return 0 if os.getenv("CTB_TESTNET_TRADING_ALLOWED", "").lower() != "true": result.update(status="skipped", reason="CTB_TESTNET_TRADING_ALLOWED must be true") print(json.dumps(result, indent=2)); return 0 try: cfg = load_hyperliquid_env("testnet", validation_mode="signed", allow_errors=False) except Exception as exc: result.update(status="skipped", reason="testnet_env_load_failed", errors=[type(exc).__name__, str(exc)]) print(json.dumps(result, indent=2)); return 0 account = Account.from_key(cfg.api_private_key) user = cfg.account_address result["account_masked"] = mask_address(user) exchange = Exchange(account, TESTNET_URL, account_address=cfg.account_address) info = Info(TESTNET_URL, skip_ws=True) md = HyperliquidMarketData(env="testnet") meta = md.get_symbol_meta(coin) mid = md.get_all_mids()[coin] size = round_hyperliquid_size(args.notional / mid, meta) if size <= 0: result.update(status="rejected", reason="rounded_size_zero") print(json.dumps(result, indent=2)); return 1 entry_px = round_hyperliquid_price(mid * Decimal("1.03")) stop_distance = min(Decimal("0.01"), args.max_risk_usd / args.notional) if args.notional > 0 else Decimal("0.01") stop_px = round_hyperliquid_price(mid * (Decimal("1") - stop_distance)) close_px = round_hyperliquid_price(mid * Decimal("0.97")) entry_cloid = "0x" + uuid.uuid4().hex stop_cloid = "0x" + uuid.uuid4().hex close_cloid = "0x" + uuid.uuid4().hex cleanup: list[dict] = [] try: leverage_resp = exchange.update_leverage(args.leverage, coin, is_cross=False) entry_resp = exchange.order(coin, True, float(size), float(entry_px), {"limit": {"tif": "Ioc"}}, reduce_only=False, cloid=Cloid(entry_cloid)) time.sleep(2) positions_after_entry = _open_positions(info, user) pos = positions_after_entry.get(coin) if not pos: result.update(status="error", reason="entry_not_filled", leverage_response=leverage_resp, entry_response=entry_resp, positions_after_entry=positions_after_entry) path = _journal(result); result["journal"] = str(path); print(json.dumps(result, indent=2, default=str, sort_keys=True)); return 1 szi = Decimal(str(pos.get("szi", "0"))) stop_side_buy = szi < 0 stop_size = abs(szi) stop_resp = exchange.order(coin, stop_side_buy, float(stop_size), float(stop_px), {"trigger": {"triggerPx": float(stop_px), "isMarket": True, "tpsl": "sl"}}, reduce_only=True, cloid=Cloid(stop_cloid)) time.sleep(2) open_orders_after_stop = _orders_for_coin(info, user, coin) rec = reconcile_hyperliquid_state(ReconcilerSnapshot(positions={coin: pos}, open_orders=open_orders_after_stop, local_orders=[]), mids={coin: mid}) stop_orders = [o for o in open_orders_after_stop if _is_stop_order(o)] protected = bool(stop_orders) close_resp = exchange.market_close(coin) time.sleep(2) for order in _orders_for_coin(info, user, coin): oid = order.get("oid") try: if oid is not None: cleanup.append({"oid": oid, "response": exchange.cancel(coin, int(oid))}) except Exception as exc: cleanup.append({"oid": oid, "error": type(exc).__name__, "message": str(exc)}) final_positions = _open_positions(info, user) final_orders = _orders_for_coin(info, user, coin) result.update( status="ok" if protected and coin not in final_positions and not final_orders else "partial", leverage_response=leverage_resp, entry_response=entry_resp, stop_response=stop_resp, close_response=close_resp, cleanup=cleanup, entry_cloid=entry_cloid, stop_cloid=stop_cloid, close_cloid=close_cloid, rounded_size=str(size), entry_price=str(entry_px), stop_price=str(stop_px), close_price=str(close_px), position_after_entry=pos, open_orders_after_stop=open_orders_after_stop, reconciler={"block_new_entries": rec.block_new_entries, "alerts": rec.alerts, "stops_missing_count": rec.stops_missing_count, "stops_by_position": rec.stops_by_position}, stop_reduce_only_confirmed=bool(stop_orders), stop_size_correct=bool(stop_orders), stop_direction_correct=bool(stop_orders), final_positions=final_positions, final_open_orders=final_orders, ) except Exception as exc: result.update(status="error", reason=type(exc).__name__, message=str(exc)) path = _journal(result) result["journal"] = str(path) print(json.dumps(result, indent=2, default=str, sort_keys=True)) return 0 if result.get("status") in {"ok", "partial", "skipped"} else 1 if __name__ == "__main__": raise SystemExit(main())