from __future__ import annotations import argparse import json from dataclasses import dataclass from datetime import datetime, timezone from decimal import Decimal from pathlib import Path from typing import Any, Protocol import requests from src.hyperliquid.market_data import HyperliquidMarketData COINGECKO_MARKETS_URL = "https://api.coingecko.com/api/v3/coins/markets" COINGECKO_TRENDING_URL = "https://api.coingecko.com/api/v3/search/trending" class HttpTransport(Protocol): def get(self, url: str, *, params: dict[str, Any] | None = None, timeout: int = 10) -> Any: ... class RequestsTransport: def get(self, url: str, *, params: dict[str, Any] | None = None, timeout: int = 10) -> Any: response = requests.get(url, params=params or {}, timeout=timeout) response.raise_for_status() return response.json() def _d(value: Any, default: str = "0") -> Decimal: try: if value is None: return Decimal(default) return Decimal(str(value)) except Exception: return Decimal(default) def _load_seen(path: Path) -> set[str]: try: raw = json.loads(path.read_text(encoding="utf-8")) except Exception: return set() values = raw.get("seen_hyperliquid_symbols") if isinstance(raw, dict) else [] return {str(x).upper() for x in values} if isinstance(values, list) else set() def _save_seen(path: Path, symbols: set[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps({"updated_at": datetime.now(timezone.utc).isoformat(), "seen_hyperliquid_symbols": sorted(symbols)}, indent=2, sort_keys=True), encoding="utf-8") @dataclass(frozen=True) class RadarThresholds: min_volume_usd: Decimal = Decimal("10000000") pump_1h_pct: Decimal = Decimal("5") pump_24h_pct: Decimal = Decimal("15") dump_1h_pct: Decimal = Decimal("-5") dump_24h_pct: Decimal = Decimal("-15") max_rank: int = 300 def classify_market_rows(rows: list[dict[str, Any]], *, hl_symbols: set[str], trending_symbols: set[str], thresholds: RadarThresholds = RadarThresholds()) -> list[dict[str, Any]]: candidates: list[dict[str, Any]] = [] for row in rows: symbol = str(row.get("symbol") or "").upper() if not symbol or symbol not in hl_symbols: continue volume = _d(row.get("total_volume")) rank_raw = row.get("market_cap_rank") try: rank = int(rank_raw) if rank_raw is not None else 999999 except Exception: rank = 999999 change_1h = _d(row.get("price_change_percentage_1h_in_currency")) change_24h = _d(row.get("price_change_percentage_24h_in_currency", row.get("price_change_percentage_24h"))) reasons: list[str] = [] if volume < thresholds.min_volume_usd: reasons.append("low_volume_watch_only") if rank > thresholds.max_rank: reasons.append("low_rank_watch_only") if change_1h >= thresholds.pump_1h_pct: reasons.append("strong_1h_pump") if change_24h >= thresholds.pump_24h_pct: reasons.append("strong_24h_pump") if change_1h <= thresholds.dump_1h_pct: reasons.append("sharp_1h_selloff") if change_24h <= thresholds.dump_24h_pct: reasons.append("sharp_24h_selloff") if symbol in trending_symbols: reasons.append("coingecko_trending") if not any(reason in reasons for reason in ("strong_1h_pump", "strong_24h_pump", "sharp_1h_selloff", "sharp_24h_selloff", "coingecko_trending")): continue score = Decimal("50") if volume >= Decimal("50000000"): score += Decimal("15") elif volume >= thresholds.min_volume_usd: score += Decimal("8") if rank <= 100: score += Decimal("10") elif rank <= thresholds.max_rank: score += Decimal("5") if symbol in trending_symbols: score += Decimal("10") if abs(change_1h) >= Decimal("8") or abs(change_24h) >= Decimal("25"): score += Decimal("10") candidates.append({ "symbol": symbol, "coingecko_id": row.get("id"), "name": row.get("name"), "market_cap_rank": rank if rank != 999999 else None, "market_cap_usd": str(_d(row.get("market_cap"))), "total_volume_usd": str(volume), "price_change_1h_pct": str(change_1h), "price_change_24h_pct": str(change_24h), "reasons": reasons, "opportunity_score": str(min(score, Decimal("100"))), "trade_mode": "paper_or_shadow_only", "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False, }) return sorted(candidates, key=lambda item: Decimal(str(item["opportunity_score"])), reverse=True) def fetch_coingecko_markets(transport: HttpTransport, *, pages: int = 2, per_page: int = 250) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for page in range(1, pages + 1): raw = transport.get(COINGECKO_MARKETS_URL, params={ "vs_currency": "usd", "order": "market_cap_desc", "per_page": per_page, "page": page, "sparkline": "false", "price_change_percentage": "1h,24h,7d,30d", }, timeout=12) if isinstance(raw, list): rows.extend([row for row in raw if isinstance(row, dict)]) return rows def fetch_trending_symbols(transport: HttpTransport) -> set[str]: try: raw = transport.get(COINGECKO_TRENDING_URL, timeout=10) except Exception: return set() coins = raw.get("coins") if isinstance(raw, dict) else [] out: set[str] = set() if isinstance(coins, list): for entry in coins: item = entry.get("item") if isinstance(entry, dict) else None if isinstance(item, dict) and item.get("symbol"): out.add(str(item["symbol"]).upper()) return out def fetch_hyperliquid_symbols(env: str = "mainnet") -> set[str]: meta = HyperliquidMarketData(env=env).get_meta() raw_universe = meta.get("universe") if isinstance(meta, dict) else [] universe = raw_universe if isinstance(raw_universe, list) else [] return {str(row.get("name", "")).upper() for row in universe if isinstance(row, dict) and row.get("name")} def build_radar(*, env: str = "mainnet", state_path: str | Path = "runtime/research/coin_opportunity_radar_state.json", transport: HttpTransport | None = None) -> dict[str, Any]: transport = transport or RequestsTransport() state = Path(state_path) seen_before = _load_seen(state) try: hl_symbols = fetch_hyperliquid_symbols(env) except Exception: hl_symbols = set() markets = fetch_coingecko_markets(transport) trending = fetch_trending_symbols(transport) candidates = classify_market_rows(markets, hl_symbols=hl_symbols, trending_symbols=trending) new_hl_symbols = sorted(hl_symbols - seen_before) if seen_before else [] if hl_symbols: _save_seen(state, hl_symbols | seen_before) return { "timestamp": datetime.now(timezone.utc).isoformat(), "source": "coingecko_hyperliquid_opportunity_radar", "status": "loaded_research_only" if hl_symbols and markets else "partial_or_unavailable", "env": env, "hyperliquid_symbols_seen": len(hl_symbols), "new_hyperliquid_symbols": new_hl_symbols, "opportunities": candidates[:50], "research_only": True, "live_order_allowed": False, "mainnet_signed_action": False, } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Detect new Hyperliquid symbols and fast crypto movers; research-only.") parser.add_argument("--env", choices=["mainnet", "testnet"], default="mainnet") parser.add_argument("--output", default="runtime/research/coin_opportunity_radar_latest.json") parser.add_argument("--state", default="runtime/research/coin_opportunity_radar_state.json") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) payload = build_radar(env=args.env, state_path=args.state) output = Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") summary = {"status": payload["status"], "output": str(output), "opportunities": len(payload["opportunities"]), "new_hyperliquid_symbols": payload["new_hyperliquid_symbols"], "live_order_allowed": False, "mainnet_signed_action": False} print(json.dumps(payload if args.json else summary, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())