Hyperliquid trailing stop loss

hyper.py

#!/usr/bin/env python3
"""
Ref: https://app.hyperliquid.xyz/join/DARKEREGO
"""
import argparse
import os
import signal
import sys
import time
from typing import Dict, Any, Optional, Tuple

from dotenv import load_dotenv
import eth_account
from eth_account.signers.local import LocalAccount
from hyperliquid.info import Info
from hyperliquid.exchange import Exchange
from hyperliquid.utils import constants


def load_credentials() -> Tuple[str, str]:
    """Load Hyperliquid credentials from environment variables."""
    load_dotenv()

    secret_key = os.getenv("HYPERLIQUID_SECRET_KEY")
    account_address = os.getenv("HYPERLIQUID_ACCOUNT_ADDRESS")

    if not secret_key:
        raise RuntimeError(
            "HYPERLIQUID_SECRET_KEY is not set. Add it to your environment or .env file."
        )
    if not account_address:
        raise RuntimeError(
            "HYPERLIQUID_ACCOUNT_ADDRESS is not set. Add it to your environment or .env file."
        )

    return secret_key, account_address


def init_clients(use_testnet: bool) -> Tuple[str, Info, Exchange]:
    """Initialize Info and Exchange clients for Hyperliquid."""
    secret_key, account_address = load_credentials()

    api_url = (
        constants.TESTNET_API_URL
        if use_testnet
        else constants.MAINNET_API_URL
    )

    account: LocalAccount = eth_account.Account.from_key(secret_key)

    info = Info(api_url, skip_ws=True)
    exchange = Exchange(account, api_url, account_address=account_address)

    return account_address, info, exchange


def find_open_position(
    info: Info, account_address: str, coin: str
) -> Optional[Dict[str, Any]]:
    """Return the position dict for an open perp on `coin`, or None if not found."""
    user_state = info.user_state(account_address)
    asset_positions = user_state.get("assetPositions", [])

    for asset_pos in asset_positions:
        position = asset_pos.get("position", {})
        if position.get("coin") != coin:
            continue
        try:
            size = float(position.get("szi", "0"))
        except (TypeError, ValueError):
            continue
        if size != 0.0:
            return position

    return None


def get_mid_price(info: Info, coin: str) -> float:
    """Fetch the current mid-price for a given coin."""
    mids = info.all_mids()
    if coin not in mids:
        raise RuntimeError(f"Coin {coin} not found in mid prices. Got keys: {list(mids.keys())}")
    try:
        return float(mids[coin])
    except (TypeError, ValueError) as exc:
        raise RuntimeError(f"Failed to parse mid price for {coin}: {mids[coin]}") from exc


def run_trailing_stop(
    coin: str,
    trail_pct: float,
    poll_interval: float,
    use_testnet: bool,
) -> None:
    """Main trailing-stop loop.

    Assumes you already have an open perp position on `coin`.
    The script starts trailing from the current mid-price when launched.
    """
    account_address, info, exchange = init_clients(use_testnet)

    position = find_open_position(info, account_address, coin)
    if position is None:
        print(f"[!] No open position found for {coin} on this account.")
        return

    size = float(position["szi"])
    side = "long" if size > 0 else "short"
    entry_px_str = position.get("entryPx", "N/A")

    print("============================================================")
    print(" Hyperliquid Trailing Stop Manager")
    print("============================================================")
    print(f"Account:         {account_address}")
    print(f"Network:         {'TESTNET' if use_testnet else 'MAINNET'}")
    print(f"Coin:            {coin}")
    print(f"Position size:   {size} ({side})")
    print(f"Entry price:     {entry_px_str}")
    print(f"Trail percent:   {trail_pct * 100:.4f}%")
    print(f"Poll interval:   {poll_interval:.2f} s")
    print("------------------------------------------------------------")

    current_price = get_mid_price(info, coin)

    if side == "long":
        highest_price = current_price
        stop_price = highest_price * (1.0 - trail_pct)
    else:
        lowest_price = current_price
        stop_price = lowest_price * (1.0 + trail_pct)

    print(f"Initial mid:     {current_price:.6f}")
    print(f"Initial stop:    {stop_price:.6f}")
    print("Press Ctrl+C to exit without closing the position.")
    print("============================================================")

    def handle_sigint(signum, frame):
        print("\n[!] Caught Ctrl+C, exiting without closing position.")
        sys.exit(0)

    signal.signal(signal.SIGINT, handle_sigint)
    highest_price = 0.0
    lowest_price = 0.0

    while True:
        time.sleep(poll_interval)
        try:
            price = get_mid_price(info, coin)
        except Exception as exc:
            print(f"[WARN] Failed to fetch price: {exc}. Retrying...")
            continue

        if side == "long":
            # Move stop up as new highs are made.
            if price > highest_price:
                highest_price = price
                stop_price = highest_price * (1.0 - trail_pct)
                print(
                    f"[LONG] New high: {highest_price:.6f}, moved stop to {stop_price:.6f}"
                )

            # Trigger if price falls to or below the stop.
            if price <= stop_price:
                print(
                    f"[LONG] Stop hit! Price {price:.6f} <= stop {stop_price:.6f}. Closing position..."
                )
                try:
                    result = exchange.market_close(coin)
                    print("[RESULT] market_close response:")
                    print(result)
                except Exception as exc:
                    print(f"[ERROR] Failed to close position: {exc}")
                break

        else:
            # side == "short"
            if price < lowest_price:
                lowest_price = price
                stop_price = lowest_price * (1.0 + trail_pct)
                print(
                    f"[SHORT] New low: {lowest_price:.6f}, moved stop to {stop_price:.6f}"
                )

            if price >= stop_price:
                print(
                    f"[SHORT] Stop hit! Price {price:.6f} >= stop {stop_price:.6f}. Closing position..."
                )
                try:
                    result = exchange.market_close(coin)
                    print("[RESULT] market_close response:")
                    print(result)
                except Exception as exc:
                    print(f"[ERROR] Failed to close position: {exc}")
                break

        print(
            f"[INFO] Coin: {coin} | Side: {side} | Last: {price:.6f} | Stop: {stop_price:.6f}"
        )


def parse_args(argv: Optional[list] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Simple CLI trailing stop manager for Hyperliquid perpetuals. "
            "Run it after opening a position; it will monitor price and "
            "close the position once the trailing stop is hit."
        )
    )

    parser.add_argument(
        "coin",
        type=str,
        help="Perpetual coin symbol, e.g. BTC, ETH, kPEPE.",
    )

    parser.add_argument(
        "--trail-pct",
        type=float,
        default=0.01,
        help=(
            "Trailing distance as a fraction :e.g. 0.01 = 1pct. "
            "For a long, stop = high * (1 - trail_pct). For a short, stop = low * (1 + trail_pct)."
        ),
    )

    parser.add_argument(
        "--poll-interval",
        type=float,
        default=2.0,
        help="Polling interval in seconds for fetching price. Default: 2.0.",
    )

    parser.add_argument(
        "--testnet",
        action="store_true",
        help="Use Hyperliquid testnet instead of mainnet.",
    )

    return parser.parse_args(argv)


def main() -> None:
    args = parse_args()
    if args.trail_pct <= 0.0:
        print("[ERROR] --trail-pct must be > 0.")
        sys.exit(1)

    if args.poll_interval <= 0.0:
        print("[ERROR] --poll-interval must be > 0.")
        sys.exit(1)

    try:
        run_trailing_stop(
            coin=args.coin.upper(),
            trail_pct=args.trail_pct,
            poll_interval=args.poll_interval,
            use_testnet=args.testnet,
        )
    except Exception as exc:
        print(f"[FATAL] {exc}")
        sys.exit(1)


if __name__ == "__main__":
    main()