#!/usr/bin/env python3 """Advisor engine used by adaptive bot. Two advisor implementations: - HuggingFaceAdvisor: local heuristic + optional custom remote endpoint - CryptoSentimentAdvisor: Reddit r/hyperliquid + Fear & Greed Index → CryptoBERT (HF Inference API) Requires only HF_TOKEN (free HuggingFace account). Reddit and Fear & Greed need no API key. """ from __future__ import annotations import logging import math import time from dataclasses import dataclass from typing import Any, Dict, List, Optional import numpy as np import requests logger = logging.getLogger(__name__) @dataclass class AdvisorDecision: side: str # long | short | hold confidence: float p_long: Optional[float] = None source: str = "local" error: Optional[str] = None class HuggingFaceAdvisor: """Always-on advisor with local inference and optional remote override. Remote endpoint (optional) may return one of: - {"side": "long|short|hold", "confidence": 0..1} - {"p_long": 0..1, "confidence": 0..1} """ def __init__(self, endpoint_url: str, api_token: Optional[str] = None, timeout_seconds: float = 4.0): self.endpoint_url = (endpoint_url or "").strip() self.api_token = (api_token or "").strip() self.timeout_seconds = float(max(0.5, timeout_seconds)) @property def enabled(self) -> bool: # Always enabled because local advisor is always available. return True @property def remote_enabled(self) -> bool: return bool(self.endpoint_url) def _local_advise(self, features: list[float], candle: Dict[str, float]) -> AdvisorDecision: # Feature order: ret_1, ret_3, ret_8, macd, atr14, rsi, vol_z[, ret_16, bb_pct, candle_body] ret_1 = float(features[0]) if len(features) > 0 else 0.0 ret_3 = float(features[1]) if len(features) > 1 else 0.0 ret_8 = float(features[2]) if len(features) > 2 else 0.0 macd = float(features[3]) if len(features) > 3 else 0.0 atr14 = float(features[4]) if len(features) > 4 else 0.0 rsi = float(features[5]) if len(features) > 5 else 0.0 vol_z = float(features[6]) if len(features) > 6 else 0.0 ret_16 = float(features[7]) if len(features) > 7 else 0.0 # bb_pct: 0=lower band, 1=upper band; values >0.5 = above midline (bullish) bb_pct = float(features[8]) if len(features) > 8 else 0.5 bb_signal = (bb_pct - 0.5) * 2.0 # rescale to [-1, 1] # Small bounded logistic score; ATR/volume terms penalize noisy regimes. score = ( 2.1 * ret_1 + 2.6 * ret_3 + 1.9 * ret_8 + 1.5 * ret_16 + 1.8 * macd + 1.7 * rsi + 1.0 * bb_signal - 1.2 * min(max(atr14 * 100.0, 0.0), 4.0) - 0.25 * abs(vol_z) ) score = float(max(-6.0, min(6.0, score))) p_long = 1.0 / (1.0 + math.exp(-score)) # Confidence grows with distance from 0.5 and contracts in high-vol candles. range_frac = 0.0 close = float(candle.get("close", 0.0)) if close > 0: high = float(candle.get("high", close)) low = float(candle.get("low", close)) range_frac = max(0.0, (high - low) / close) confidence = min(1.0, abs(p_long - 0.5) * 2.0) confidence *= max(0.35, 1.0 - min(range_frac * 25.0, 0.65)) side = "hold" if p_long >= 0.53: side = "long" elif p_long <= 0.47: side = "short" return AdvisorDecision(side=side, confidence=float(confidence), p_long=float(p_long), source="local") def _parse_remote_response(self, data: Any) -> Optional[AdvisorDecision]: if not isinstance(data, dict): return None if "p_long" in data: p_long = float(max(0.0, min(1.0, float(data.get("p_long", 0.5))))) confidence = float(max(0.0, min(1.0, float(data.get("confidence", abs(p_long - 0.5) * 2.0))))) side = "long" if p_long >= 0.5 else "short" return AdvisorDecision(side=side, confidence=confidence, p_long=p_long, source="remote") side = str(data.get("side", "hold")).strip().lower() if side not in {"long", "short", "hold"}: side = "hold" confidence = float(max(0.0, min(1.0, float(data.get("confidence", 0.0))))) p_long = data.get("p_long") p_long_val = None if p_long is not None: p_long_val = float(max(0.0, min(1.0, float(p_long)))) return AdvisorDecision(side=side, confidence=confidence, p_long=p_long_val, source="remote") def advise( self, coin: str, interval: str, features: list[float], candle: Dict[str, float], timestamp_ms: int, ) -> Optional[AdvisorDecision]: local = self._local_advise(features=features, candle=candle) if not self.remote_enabled: return local payload = { "coin": coin, "interval": interval, "features": [float(v) for v in features], "candle": { "close": float(candle.get("close", 0.0)), "high": float(candle.get("high", 0.0)), "low": float(candle.get("low", 0.0)), }, "timestamp_ms": int(timestamp_ms), } headers = {"Content-Type": "application/json"} if self.api_token: headers["Authorization"] = f"Bearer {self.api_token}" try: response = requests.post( self.endpoint_url, json=payload, headers=headers, timeout=self.timeout_seconds, ) response.raise_for_status() remote = self._parse_remote_response(response.json()) return remote if remote is not None else local except Exception as exc: logger.warning("HF advisor request failed: %s", exc) local.error = str(exc) return local class CryptoSentimentAdvisor: """Sentiment advisor using Reddit r/hyperliquid + Crypto Fear & Greed Index. Two fully free, no-paid-key signal sources: - Reddit r/hyperliquid hot posts → CryptoBERT (HF Inference API) for text scoring - Alternative.me Fear & Greed Index → direct 0-100 sentiment score Combined: 65% Reddit/CryptoBERT + 35% Fear & Greed → p_long for advisor slot. Falls back gracefully if either source is unavailable. Required env var: HF_TOKEN (free HuggingFace account at huggingface.co) Reddit and Fear & Greed require no API key. Results are cached for cache_seconds to respect free-tier rate limits. """ _CRYPTOBERT_URL = "https://api-inference.huggingface.co/models/ElKulako/cryptobert" _REDDIT_URL = "https://www.reddit.com/r/hyperliquid/hot.json" _FNG_URL = "https://api.alternative.me/fng/" def __init__( self, hf_token: str, timeout_seconds: float = 8.0, cache_seconds: float = 300.0, ): self.hf_token = hf_token.strip() self.timeout_seconds = float(max(1.0, timeout_seconds)) self.cache_seconds = float(max(60.0, cache_seconds)) self._cache: Dict[str, Any] = {} # {coin: (ts, AdvisorDecision)} self._session = requests.Session() self._session.headers["User-Agent"] = "HyperliquidTradingBot/1.0" @property def enabled(self) -> bool: return bool(self.hf_token) @property def remote_enabled(self) -> bool: return self.enabled def _fetch_reddit_posts(self) -> List[str]: """Return up to 5 hot post titles from r/hyperliquid. No API key needed.""" try: r = self._session.get( self._REDDIT_URL, params={"limit": 8, "raw_json": 1}, timeout=self.timeout_seconds, ) r.raise_for_status() titles: List[str] = [] for post in r.json().get("data", {}).get("children", []): t = post.get("data", {}).get("title", "").strip() if t: titles.append(t) if len(titles) >= 5: break return titles except Exception as exc: logger.debug("Reddit fetch failed: %s", exc) return [] def _fetch_fear_greed(self) -> Optional[float]: """Return Crypto Fear & Greed Index (0–100) as a p_long in [0, 1]. No API key needed.""" try: r = self._session.get(self._FNG_URL, params={"limit": 1}, timeout=self.timeout_seconds) r.raise_for_status() value = int(r.json()["data"][0]["value"]) return float(value) / 100.0 except Exception as exc: logger.debug("Fear & Greed fetch failed: %s", exc) return None def _score_with_cryptobert(self, texts: List[str]) -> Optional[float]: """Score texts with CryptoBERT, return average p_long = bullish/(bullish+bearish).""" headers = { "Authorization": f"Bearer {self.hf_token}", "Content-Type": "application/json", } p_longs: List[float] = [] for text in texts: try: r = self._session.post( self._CRYPTOBERT_URL, headers=headers, json={"inputs": text}, timeout=self.timeout_seconds, ) if r.status_code == 503: logger.debug("CryptoBERT still loading, skipping text") continue r.raise_for_status() result = r.json() items = result[0] if (isinstance(result, list) and result and isinstance(result[0], list)) else result scores = {item["label"]: float(item["score"]) for item in items if "label" in item} bullish = scores.get("Bullish", 0.0) bearish = scores.get("Bearish", 0.0) total = bullish + bearish if total > 1e-6: p_longs.append(bullish / total) except Exception as exc: logger.debug("CryptoBERT scoring failed: %s", exc) return float(np.mean(p_longs)) if p_longs else None def advise( self, coin: str, interval: str, features: list, candle: Dict[str, float], timestamp_ms: int, ) -> Optional[AdvisorDecision]: if not self.enabled: return None cached = self._cache.get(coin) if cached is not None and (time.time() - cached[0]) < self.cache_seconds: return cached[1] try: reddit_posts = self._fetch_reddit_posts() fng_score = self._fetch_fear_greed() reddit_score = self._score_with_cryptobert(reddit_posts) if reddit_posts else None # Combine sources: 65% text sentiment, 35% Fear & Greed. # Degrade gracefully when one source is unavailable. if reddit_score is not None and fng_score is not None: p_long = 0.65 * reddit_score + 0.35 * fng_score elif reddit_score is not None: p_long = reddit_score elif fng_score is not None: p_long = fng_score else: return None p_long = float(np.clip(p_long, 0.01, 0.99)) confidence = float(np.clip(abs(p_long - 0.5) * 2.0, 0.0, 1.0)) side = "long" if p_long >= 0.53 else ("short" if p_long <= 0.47 else "hold") decision = AdvisorDecision( side=side, confidence=confidence, p_long=p_long, source="remote", ) logger.info( "CryptoSentimentAdvisor: reddit=%.3f fng=%.2f combined_p_long=%.3f side=%s posts=%d", reddit_score if reddit_score is not None else float("nan"), fng_score if fng_score is not None else float("nan"), p_long, side, len(reddit_posts), ) self._cache[coin] = (time.time(), decision) return decision except Exception as exc: logger.warning("CryptoSentimentAdvisor error for %s: %s", coin, exc) return AdvisorDecision(side="hold", confidence=0.0, p_long=0.5, source="remote", error=str(exc))