from __future__ import annotations import argparse from dataclasses import asdict, is_dataclass from datetime import datetime, timezone from decimal import Decimal import json import os from pathlib import Path import time import uuid from typing import Any 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.execution.order_intent import OrderIntent from src.hyperliquid.market_data import HyperliquidMarketData from src.hyperliquid.rounding import round_hyperliquid_price, round_hyperliquid_size from src.market.context import CoinMarketContext from src.reconciliation.hyperliquid_reconciler import LocalOrderRecord, ReconcilerSnapshot, reconcile_hyperliquid_state from src.risk.data_quality_gate import DataQualityGate from src.risk.pretrade_risk_gate import PretradeRiskGate, RiskContext from src.strategies.v76_hl_confirmed_squeeze_hybrid import MarketContext, build_v76_order_intent TESTNET_URL = "https://api.hyperliquid-testnet.xyz" def _json_default(value: Any) -> Any: if isinstance(value, Decimal): return str(value) if is_dataclass(value) and not isinstance(value, type): return asdict(value) # type: ignore[arg-type] return str(value) def _journal(prefix: str, payload: dict[str, Any]) -> Path: path = Path("runtime/reports") / f"{prefix}_{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=_json_default, sort_keys=True), encoding="utf-8") return path def _open_positions(info: Info, user: str) -> dict[str, dict[str, Any]]: state = info.user_state(user) out: dict[str, dict[str, Any]] = {} 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[str, Any]]: return [o for o in info.open_orders(user) if str(o.get("coin", "")).upper() == coin.upper()] def _exchange_status_ok(resp: Any) -> bool: return isinstance(resp, dict) and resp.get("status") == "ok" def _forced_market_context(coin: str, mid: Decimal, notional: Decimal, data_quality_allowed: bool) -> MarketContext: # v76 breakout branch: current > recent_high and sma_fast >= sma_slow. # wallet_equity=notional*5 makes risk=0.25% equity and max_notional=20% equity == requested notional. wallet_equity = notional * Decimal("5") return MarketContext( coin=coin, symbol=f"{coin}/USDC:USDC", current_price=mid, recent_high=mid * Decimal("0.999"), sma_fast=mid * Decimal("0.998"), sma_slow=mid * Decimal("0.997"), volume_24h=Decimal("50000000"), breadth_positive_candidates=3, relative_strength_rank=1, baseline_price=mid * Decimal("0.95"), wallet_equity_usdc=wallet_equity, data_quality_allowed=data_quality_allowed, ) def _intent_to_json(intent: OrderIntent) -> dict[str, Any]: data = asdict(intent) return {k: (str(v) if isinstance(v, Decimal) else v) for k, v in data.items()} def run_forced_strategy_smoke(*, strategy: str, coin: str, notional: Decimal, max_risk_usd: Decimal) -> dict[str, Any]: coin = coin.upper() result: dict[str, Any] = { "tool": "hl_testnet_strategy_smoke", "strategy": strategy, "coin": coin, "mode": "forced-signal", "env": os.getenv("CTB_HL_ENV"), "notional": str(notional), "max_risk_usd": str(max_risk_usd), "mainnet_blocked": True, "mainnet_order_sent": False, "strategy_exchange_calls": 0, } if strategy != "v76_hl_confirmed_squeeze_hybrid": result.update(status="rejected", reason="unsupported_strategy") return result if os.getenv("CTB_HL_ENV", "").lower() != "testnet": result.update(status="skipped", reason="CTB_HL_ENV must be testnet") return result if os.getenv("CTB_TESTNET_TRADING_ALLOWED", "").lower() != "true": result.update(status="skipped", reason="CTB_TESTNET_TRADING_ALLOWED must be true") return result cfg = load_hyperliquid_env("testnet", validation_mode="signed", allow_errors=False) if cfg.will_use_mainnet_key or cfg.env_file_path.name != "Testnet.env": result.update(status="blocked", reason="testnet_env_file_not_isolated", env_file_path=str(cfg.env_file_path)) return result result["credentials"] = { "env_file_path": str(cfg.env_file_path), "account_address_masked": mask_address(cfg.account_address), "agent_wallet_derived_masked": mask_address(cfg.agent_wallet_address_derived), "agent_wallet_key_matches_private_key": cfg.agent_wallet_key_matches_private_key, "account_address_equals_agent_wallet_address": cfg.account_address_equals_agent_wallet_address, "safe_to_run_testnet_smokes": cfg.safe_to_run_testnet_smokes, "warnings": list(cfg.warnings), } md = HyperliquidMarketData(env="testnet") meta = md.get_symbol_meta(coin) mid = Decimal(str(md.get_all_mids()[coin])) dq_context = CoinMarketContext( coin=coin, rsi=Decimal("55"), sma_fast=mid * Decimal("0.998"), sma_slow=mid * Decimal("0.997"), atr=mid * Decimal("0.01"), funding=Decimal("0"), volume=Decimal("50000000"), mid=mid, spread_pct=Decimal("0.01"), timestamp=datetime.now(timezone.utc), reliability_score=Decimal("0.99"), stale_data=False, l2_available=True, ) dq_decision = DataQualityGate().evaluate(dq_context) result["data_quality_gate"] = {"allowed": dq_decision.allowed, "reasons": dq_decision.reasons} client_order_id = "0x" + uuid.uuid4().hex intent = build_v76_order_intent(_forced_market_context(coin, mid, notional, dq_decision.allowed), client_order_id=client_order_id) if intent is None: result.update(status="blocked", reason="strategy_returned_no_intent") return result # Force requested smoke notional while preserving v76's stop/risk semantics and full OrderIntent shape. rounded_size = round_hyperliquid_size(notional / mid, meta) intent = OrderIntent( strategy_id=intent.strategy_id, symbol=intent.symbol, coin=intent.coin, side=intent.side, reduce_only=intent.reduce_only, order_type=intent.order_type, tif=intent.tif, size=rounded_size, price=intent.price, trigger_price=intent.trigger_price, stop_loss=intent.stop_loss, take_profit=intent.take_profit, client_order_id=intent.client_order_id, reason=intent.reason, risk_usd=min(intent.risk_usd, max_risk_usd), estimated_notional_usd=notional, ) result["order_intent"] = _intent_to_json(intent) gate = PretradeRiskGate(min_order_notional_usd=Decimal("10"), max_parallel_positions=1, max_order_notional_usd=notional, max_risk_usd=max_risk_usd) risk = gate.evaluate(intent, RiskContext(open_positions=0, kill_switch_active=False, daily_loss_exceeded=False)) result["pretrade_risk_gate"] = {"allowed": risk.allowed, "reasons": risk.reasons, "checks": ["max_notional", "max_risk", "max_open_positions=1", "stop_loss", "data_quality_gate"]} if not risk.allowed or not dq_decision.allowed: result.update(status="blocked", reason="risk_or_data_quality_gate", journaled=True) return result account = Account.from_key(cfg.api_private_key) exchange = Exchange(account, TESTNET_URL, account_address=cfg.account_address) info = Info(TESTNET_URL, skip_ws=True) user = cfg.account_address cleanup: list[dict[str, Any]] = [] try: leverage_resp = exchange.update_leverage(2, coin, is_cross=False) entry_px = round_hyperliquid_price(mid * Decimal("1.03")) entry_resp = exchange.order(coin, True, float(intent.size), float(entry_px), {"limit": {"tif": "Ioc"}}, reduce_only=False, cloid=Cloid(intent.client_order_id)) 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) return result szi = Decimal(str(pos.get("szi", "0"))) stop_size = abs(szi) stop_side_buy = szi < 0 stop_px = round_hyperliquid_price(Decimal(str(intent.stop_loss))) stop_cloid = "0x" + uuid.uuid4().hex 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=[LocalOrderRecord(coin=coin, client_order_id=intent.client_order_id, order_role="entry", size=stop_size)]), mids={coin: mid}, max_position_notional_usd=notional * Decimal("2"), ) stop_orders = rec.stops_by_position.get(coin, []) stop_reduce_only = any(bool(o.get("reduceOnly") or o.get("reduce_only")) for o in stop_orders) stop_size_correct = any(abs(Decimal(str(o.get("sz") or o.get("origSz") or "0"))) == stop_size for o in stop_orders) expected_side = "A" if szi > 0 else "B" stop_direction_correct = any(str(o.get("side", "")).upper() == expected_side for o in 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)}) time.sleep(1) final_positions = _open_positions(info, user) final_open_orders = _orders_for_coin(info, user, coin) final_clean = coin not in final_positions and not final_open_orders result.update( status="ok" if _exchange_status_ok(entry_resp) and _exchange_status_ok(stop_resp) and stop_reduce_only and stop_size_correct and stop_direction_correct and final_clean else "partial", leverage_response=leverage_resp, entry_response=entry_resp, fill_detected=True, position_after_entry=pos, stop_response=stop_resp, stop_cloid=stop_cloid, open_orders_after_stop=open_orders_after_stop, stop_confirmed=bool(stop_orders), stop_reduce_only=stop_reduce_only, stop_size_correct=stop_size_correct, stop_direction_correct=stop_direction_correct, reconciler={"severity": rec.severity, "block_new_entries": rec.block_new_entries, "alerts": rec.alerts, "stops_missing_count": rec.stops_missing_count, "stops_by_position": rec.stops_by_position}, close_response=close_resp, cleanup=cleanup, final_positions=final_positions, final_open_orders=final_open_orders, final_reconcile={"positions_count": len(final_positions), "unexpected_open_orders_count": len(final_open_orders), "clean": final_clean}, order_intent_journaled=True, ) except Exception as exc: result.update(status="error", reason=type(exc).__name__, message=str(exc)) return result def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Forced-signal v76 strategy smoke through Hyperliquid Testnet execution. Never mainnet.") parser.add_argument("--strategy", required=True) 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("--mode", choices=["forced-signal"], required=True) parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) payload = run_forced_strategy_smoke(strategy=args.strategy, coin=args.coin, notional=args.notional, max_risk_usd=args.max_risk_usd) path = _journal("hyperliquid_testnet_strategy_smoke", payload) payload["journal"] = str(path) print(json.dumps(payload, indent=2, default=_json_default, sort_keys=True) if args.json else payload) return 0 if payload.get("status") in {"ok", "skipped"} else 1 if __name__ == "__main__": raise SystemExit(main())