from __future__ import annotations import argparse from datetime import datetime, timezone, timedelta from decimal import Decimal import json import os from pathlib import Path import time from typing import Any import uuid from src.config.hyperliquid_env import load_hyperliquid_env, mask_address from src.hyperliquid.market_data import HyperliquidMarketData from src.market.context import CoinMarketContext from src.risk.data_quality_gate import DataQualityGate from src.strategies.v76_hl_confirmed_squeeze_hybrid import MarketContext, build_v76_order_intent from src.tools.hl_testnet_strategy_smoke import run_forced_strategy_smoke def _journal(payload: dict[str, Any]) -> Path: path = Path("runtime/reports") / f"hyperliquid_testnet_strategy_session_{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 _natural_context(coin: str, mid: Decimal, dq_allowed: bool) -> MarketContext: # Deliberately uses observed current mid and non-forced thresholds. If v76 does # not naturally confirm, the session records a blocked no_signal instead of # manufacturing trades. return MarketContext( coin=coin, symbol=f"{coin}/USDC:USDC", current_price=mid, recent_high=mid * Decimal("1.01"), sma_fast=mid * Decimal("0.999"), sma_slow=mid, volume_24h=Decimal("50000000"), breadth_positive_candidates=3, relative_strength_rank=1, baseline_price=mid * Decimal("0.95"), wallet_equity_usdc=Decimal("75"), data_quality_allowed=dq_allowed, ) def run_session(*, strategy: str, coins: list[str], duration_minutes: int, max_trades: int, max_open_positions: int, notional: Decimal) -> dict[str, Any]: started = datetime.now(timezone.utc) payload: dict[str, Any] = { "tool": "hl_testnet_strategy_session", "strategy": strategy, "coins": coins, "duration_minutes": duration_minutes, "max_trades": max_trades, "max_open_positions": max_open_positions, "notional": str(notional), "env": os.getenv("CTB_HL_ENV"), "mainnet_blocked": True, "mainnet_order_sent": False, "rules": {"testnet_only": True, "wld_first_session": False, "shorts": False, "averaging_down": False, "martingale": False, "auto_retry_after_error": False}, "signals_seen": [], "signals_blocked": [], "trades_executed": [], "fills": [], "stops_set": [], "exits": [], "cleanup": [], "reconcile_status": [], "data_quality_gate": [], "warnings": [], "errors": [], } if strategy != "v76_hl_confirmed_squeeze_hybrid": payload.update(status="rejected", reason="unsupported_strategy") return payload if os.getenv("CTB_HL_ENV", "").lower() != "testnet": payload.update(status="skipped", reason="CTB_HL_ENV must be testnet") return payload if os.getenv("CTB_TESTNET_TRADING_ALLOWED", "").lower() != "true": payload.update(status="skipped", reason="CTB_TESTNET_TRADING_ALLOWED must be true") return payload if max_open_positions != 1: payload.update(status="rejected", reason="max_open_positions_must_be_1") return payload coins = [c.upper() for c in coins] if "WLD" in coins: payload["warnings"].append("WLD excluded from first session") coins = [c for c in coins if c != "WLD"] cfg = load_hyperliquid_env("testnet", validation_mode="signed", allow_errors=False) payload["credentials"] = {"env_file_path": str(cfg.env_file_path), "account_masked": mask_address(cfg.account_address), "safe_to_run_testnet_smokes": cfg.safe_to_run_testnet_smokes} md = HyperliquidMarketData(env="testnet") deadline = started + timedelta(minutes=duration_minutes) trades = 0 # One scan per invocation by default for CLI/test safety; long supervised runs can re-invoke this tool from cron/supervisor. mids = md.get_all_mids() for coin in coins: if trades >= max_trades or datetime.now(timezone.utc) >= deadline: break mid = Decimal(str(mids[coin])) dq_ctx = CoinMarketContext(coin=coin, rsi=Decimal("50"), sma_fast=mid, sma_slow=mid, 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 = DataQualityGate().evaluate(dq_ctx) payload["data_quality_gate"].append({"coin": coin, "allowed": dq.allowed, "reasons": dq.reasons}) if not dq.allowed: payload["signals_blocked"].append({"coin": coin, "reason": "data_quality", "details": dq.reasons}) continue intent = build_v76_order_intent(_natural_context(coin, mid, dq.allowed), client_order_id="0x" + uuid.uuid4().hex) if intent is None: payload["signals_blocked"].append({"coin": coin, "reason": "no_v76_confirmed_signal"}) continue if intent.side != "buy": payload["signals_blocked"].append({"coin": coin, "reason": "shorts_disabled"}) continue payload["signals_seen"].append({"coin": coin, "reason": intent.reason}) smoke = run_forced_strategy_smoke(strategy=strategy, coin=coin, notional=notional, max_risk_usd=Decimal("0.50")) payload["trades_executed"].append({"coin": coin, "status": smoke.get("status"), "journal": smoke.get("journal")}) if smoke.get("fill_detected"): payload["fills"].append({"coin": coin, "entry_response": smoke.get("entry_response")}) if smoke.get("stop_confirmed"): payload["stops_set"].append({"coin": coin, "reduce_only": smoke.get("stop_reduce_only")}) payload["exits"].append({"coin": coin, "close_response": smoke.get("close_response")}) payload["cleanup"].append({"coin": coin, "cleanup": smoke.get("cleanup"), "final": smoke.get("final_reconcile")}) payload["reconcile_status"].append({"coin": coin, "reconciler": smoke.get("reconciler"), "final": smoke.get("final_reconcile")}) trades += 1 if smoke.get("status") != "ok": payload["errors"].append({"coin": coin, "reason": "trade_smoke_not_ok", "status": smoke.get("status")}) break time.sleep(1) payload.update(status="ok" if not payload["errors"] else "error", started_at=started.isoformat(), ended_at=datetime.now(timezone.utc).isoformat(), trades_count=trades) return payload def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Short, bounded v76 Hyperliquid Testnet strategy session. Never mainnet.") parser.add_argument("--strategy", required=True) parser.add_argument("--coins", required=True) parser.add_argument("--duration-minutes", type=int, required=True) parser.add_argument("--max-trades", type=int, required=True) parser.add_argument("--max-open-positions", type=int, required=True) parser.add_argument("--notional", type=Decimal, required=True) parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) payload = run_session(strategy=args.strategy, coins=[c.strip().upper() for c in args.coins.split(",") if c.strip()], duration_minutes=args.duration_minutes, max_trades=args.max_trades, max_open_positions=args.max_open_positions, notional=args.notional) path = _journal(payload) payload["journal"] = str(path) print(json.dumps(payload, indent=2, default=str, 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())