Spaces:
Running
Running
| #!/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__) | |
| 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)) | |
| def enabled(self) -> bool: | |
| # Always enabled because local advisor is always available. | |
| return True | |
| 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" | |
| def enabled(self) -> bool: | |
| return bool(self.hf_token) | |
| 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)) | |