Spaces:
Running on Zero
Running on Zero
| """LLM-powered trading advisor using HuggingFace Inference API.""" | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import re | |
| from dataclasses import dataclass | |
| from typing import Literal | |
| from huggingface_hub import InferenceClient | |
| from .hyperliquid import Interval, fetch_candles, fetch_mark_price | |
| from .indicators import IndicatorSnapshot, compute_indicators | |
| Direction = Literal["LONG", "SHORT", "NEUTRAL"] | |
| _SYSTEM_PROMPT = """You are a professional cryptocurrency swing-trading advisor specialised in HYPE on Hyperliquid. | |
| Your role is to analyse technical indicator data and give a clear, actionable trade recommendation: | |
| - LONG — enter a long position with the suggested TP/SL | |
| - SHORT — enter a short position with the suggested TP/SL | |
| - NEUTRAL — no trade; conditions are unclear or risky | |
| You think like a disciplined swing trader: | |
| - You favour 2:1 reward-to-risk setups minimum. | |
| - You respect the trend (EMA stack) and only fade it when confluent reversal signals exist. | |
| - You treat overbought/oversold RSI as *context*, not a standalone signal. | |
| - MACD histogram direction carries more weight than the histogram value itself. | |
| - Volume confirmation matters: a breakout on low volume is suspect. | |
| - You account for ATR-derived volatility when sizing TP/SL. | |
| - You NEVER suggest a trade if the setup is ambiguous. | |
| Always respond with valid JSON matching this exact schema: | |
| { | |
| "direction": "LONG" | "SHORT" | "NEUTRAL", | |
| "confidence": "LOW" | "MEDIUM" | "HIGH", | |
| "entry_price": <float>, | |
| "take_profit": <float>, | |
| "stop_loss": <float>, | |
| "risk_reward": <float>, | |
| "reasoning": "<2-4 sentence explanation>", | |
| "key_risks": ["<risk1>", "<risk2>"] | |
| } | |
| For NEUTRAL, set entry_price/take_profit/stop_loss/risk_reward to 0. | |
| Do NOT include any text outside the JSON block.""" | |
| class TradeAdvice: | |
| direction: Direction | |
| confidence: str | |
| entry_price: float | |
| take_profit: float | |
| stop_loss: float | |
| risk_reward: float | |
| reasoning: str | |
| key_risks: list[str] | |
| # Metadata | |
| coin: str | |
| interval: str | |
| indicators: IndicatorSnapshot | |
| def to_dict(self) -> dict: | |
| return { | |
| "direction": self.direction, | |
| "confidence": self.confidence, | |
| "entry_price": self.entry_price, | |
| "take_profit": self.take_profit, | |
| "stop_loss": self.stop_loss, | |
| "risk_reward": round(self.risk_reward, 2), | |
| "reasoning": self.reasoning, | |
| "key_risks": self.key_risks, | |
| "metadata": { | |
| "coin": self.coin, | |
| "interval": self.interval, | |
| "current_price": self.indicators.price, | |
| "rsi_14": round(self.indicators.rsi_14, 2), | |
| "ema_trend": self.indicators.ema_trend, | |
| "macd_hist": round(self.indicators.macd_hist, 4), | |
| "atr_14": round(self.indicators.atr_14, 4), | |
| }, | |
| } | |
| _DEFAULT_MODEL = "Qwen/Qwen2.5-7B-Instruct" | |
| def get_advice( | |
| coin: str = "HYPE", | |
| interval: Interval = "4h", | |
| lookback: int = 200, | |
| atr_tp_mult: float = 2.0, | |
| atr_sl_mult: float = 1.0, | |
| hf_token: str | None = None, | |
| model: str | None = None, | |
| ) -> TradeAdvice: | |
| """Fetch candles, compute indicators, call the HF Inference API, return a TradeAdvice.""" | |
| df = fetch_candles(coin=coin, interval=interval, lookback=lookback) | |
| snapshot = compute_indicators(df, atr_multiplier_tp=atr_tp_mult, atr_multiplier_sl=atr_sl_mult) | |
| user_message = f"""Analyse HYPE/{interval} and give a swing trade recommendation. | |
| ## Market Data | |
| {snapshot.to_prompt_block()} | |
| ## Last 10 candles (oldest first) | |
| {_format_recent_candles(df.tail(10))} | |
| Based on all of the above, provide your JSON recommendation now.""" | |
| token = hf_token or os.environ.get("HF_TOKEN") | |
| chosen_model = model or os.environ.get("HF_MODEL", _DEFAULT_MODEL) | |
| client = InferenceClient(model=chosen_model, token=token) | |
| response = client.chat_completion( | |
| messages=[ | |
| {"role": "system", "content": _SYSTEM_PROMPT}, | |
| {"role": "user", "content": user_message}, | |
| ], | |
| max_tokens=1024, | |
| temperature=0.1, | |
| ) | |
| raw_text = response.choices[0].message.content or "" | |
| raw_json = _extract_json_text(raw_text) | |
| data = json.loads(raw_json) | |
| return TradeAdvice( | |
| direction=data["direction"], | |
| confidence=data["confidence"], | |
| entry_price=float(data["entry_price"]), | |
| take_profit=float(data["take_profit"]), | |
| stop_loss=float(data["stop_loss"]), | |
| risk_reward=float(data["risk_reward"]), | |
| reasoning=data["reasoning"], | |
| key_risks=data.get("key_risks", []), | |
| coin=coin, | |
| interval=interval, | |
| indicators=snapshot, | |
| ) | |
| def _format_recent_candles(df) -> str: | |
| rows = ["time (UTC) | open | high | low | close | volume"] | |
| rows.append("-" * 72) | |
| for _, row in df.iterrows(): | |
| rows.append( | |
| f"{row['open_time'].strftime('%Y-%m-%d %H:%M')} | " | |
| f"{row['open']:7.4f} | {row['high']:7.4f} | " | |
| f"{row['low']:7.4f} | {row['close']:7.4f} | {row['volume']:,.0f}" | |
| ) | |
| return "\n".join(rows) | |
| def _extract_json_text(text: str) -> str: | |
| """Extract a JSON object from a raw LLM response string. | |
| Handles both bare JSON and markdown code fences. | |
| """ | |
| text = text.strip() | |
| # Strip ```json ... ``` or ``` ... ``` fences | |
| fence = re.match(r"^```(?:json)?\s*([\s\S]*?)\s*```$", text) | |
| if fence: | |
| return fence.group(1).strip() | |
| # Find the first { ... } block in case of preamble text | |
| brace_match = re.search(r"\{[\s\S]*\}", text) | |
| if brace_match: | |
| return brace_match.group(0) | |
| return text | |