"""Multi-timeframe trading advisor using local transformers inference. Timeframe roles --------------- 4h — macro trend bias (EMA stack, MACD direction, volume trend) 1h — momentum confirmation (RSI, MACD histogram, BB position) 15m — entry timing (recent S/R, current price action, maker limit placement) """ from __future__ import annotations import json import logging import os import re from dataclasses import dataclass from typing import Literal import torch from transformers import AutoModelForCausalLM, AutoTokenizer from .hyperliquid import Interval, fetch_candles from .indicators import IndicatorSnapshot, compute_indicators log = logging.getLogger("hypeadvisor.advisor") Direction = Literal["LONG", "SHORT", "NEUTRAL"] _MODEL_ID = os.environ.get("HF_MODEL", "Qwen/Qwen2.5-7B-Instruct") # Candles per timeframe — enough history for all indicators to warm up _LOOKBACK: dict[Interval, int] = { "4h": 120, # ~20 days "1h": 120, # ~5 days "15m": 80, # ~20 hours } # Estimated seconds between model start and order placement. # Used to guide the model toward a limit price that will still be reachable. _MODEL_LATENCY_S = 15 _SYSTEM_PROMPT = """You are a professional cryptocurrency swing-trading advisor specialised in HYPE/USD perpetual futures on Hyperliquid. You receive a three-timeframe analysis (4h, 1h, 15m) and must output a single JSON trade recommendation. ## Timeframe roles - 4h → macro trend bias. Only trade in the direction the 4h trend permits. - 1h → momentum confirmation. Confirm the 4h bias with RSI, MACD histogram alignment. - 15m → entry timing. Locate a maker limit-order entry within the 15m structure. ## Entry price rules (CRITICAL — Hyperliquid-specific) - Always suggest a LIMIT order entry, not a market entry. - Maker limit orders rest in the book and earn a rebate on Hyperliquid (-0.01% fee). - A limit order that crosses the spread fills IMMEDIATELY as a taker: - LONG limit ABOVE current price → crosses the ask → fills at market (taker) ✗ - SHORT limit BELOW current price → crosses the bid → fills at market (taker) ✗ - To guarantee maker placement, the limit must rest on the passive side: - LONG maker entry: price BELOW current (order rests in book, waits for price to pull back to it) - SHORT maker entry: price ABOVE current (order rests in book, waits for price to rally to it) - Account for ~15 seconds of model-to-order latency: the price will drift slightly before the order is placed. - Keep the entry within ATR_15m × 0.25 of the current price so fill probability remains high. - If no clean 15m support/resistance zone exists within ATR_15m × 0.25, output NEUTRAL. ## Trade rules - Minimum 2:1 reward-to-risk ratio. - Do NOT trade if 4h and 1h momentum conflict (e.g. 4h bearish + 1h bullish divergence = NEUTRAL). - Do NOT trade if RSI is extreme (>75 or <25) on 4h without strong confluence. Always respond with valid JSON matching this exact schema — no text outside the JSON block: { "direction": "LONG" | "SHORT" | "NEUTRAL", "confidence": "LOW" | "MEDIUM" | "HIGH", "entry_price": , "take_profit": , "stop_loss": , "risk_reward": , "reasoning": "<3-5 sentences covering 4h trend, 1h confirmation, 15m entry rationale>", "key_risks": ["", ""] } For NEUTRAL set entry_price / take_profit / stop_loss / risk_reward to 0.""" @dataclass class TradeAdvice: direction: Direction confidence: str entry_price: float take_profit: float stop_loss: float risk_reward: float reasoning: str key_risks: list[str] coin: str tf_4h: IndicatorSnapshot tf_1h: IndicatorSnapshot tf_15m: 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, "entry_type": "LIMIT_MAKER", "metadata": { "coin": self.coin, "current_price": self.tf_15m.price, "4h": { "ema_trend": self.tf_4h.ema_trend, "rsi_14": round(self.tf_4h.rsi_14, 2), "macd_hist": round(self.tf_4h.macd_hist, 4), "atr_14": round(self.tf_4h.atr_14, 4), }, "1h": { "ema_trend": self.tf_1h.ema_trend, "rsi_14": round(self.tf_1h.rsi_14, 2), "macd_hist": round(self.tf_1h.macd_hist, 4), "bb_position": self.tf_1h.bb_position, }, "15m": { "rsi_14": round(self.tf_15m.rsi_14, 2), "atr_14": round(self.tf_15m.atr_14, 4), "bb_position": self.tf_15m.bb_position, }, }, } # --------------------------------------------------------------------------- # Model — loaded once at startup on CPU; ZeroGPU moves it to GPU per-request # --------------------------------------------------------------------------- log.info("Loading tokenizer for %s ...", _MODEL_ID) _tokenizer = AutoTokenizer.from_pretrained(_MODEL_ID) log.info("Loading model weights (CPU)...") _model = AutoModelForCausalLM.from_pretrained( _MODEL_ID, torch_dtype=torch.bfloat16, device_map="cpu", ) log.info("Model loaded.") # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def get_advice( coin: str = "HYPE", atr_tp_mult: float = 2.0, atr_sl_mult: float = 1.0, ) -> TradeAdvice: """Fetch three timeframes, compute indicators, run LLM, return TradeAdvice.""" log.info("Fetching candles for %s ...", coin) df_4h = fetch_candles(coin=coin, interval="4h", lookback=_LOOKBACK["4h"]) df_1h = fetch_candles(coin=coin, interval="1h", lookback=_LOOKBACK["1h"]) df_15m = fetch_candles(coin=coin, interval="15m", lookback=_LOOKBACK["15m"]) snap_4h = compute_indicators(df_4h, atr_multiplier_tp=atr_tp_mult, atr_multiplier_sl=atr_sl_mult) snap_1h = compute_indicators(df_1h, atr_multiplier_tp=atr_tp_mult, atr_multiplier_sl=atr_sl_mult) snap_15m = compute_indicators(df_15m, atr_multiplier_tp=atr_tp_mult, atr_multiplier_sl=atr_sl_mult) user_message = _build_prompt(coin, snap_4h, snap_1h, snap_15m, df_15m) messages = [ {"role": "system", "content": _SYSTEM_PROMPT}, {"role": "user", "content": user_message}, ] text = _tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, ) inputs = _tokenizer([text], return_tensors="pt").to("cuda") _model.to("cuda") with torch.no_grad(): output_ids = _model.generate( **inputs, max_new_tokens=512, temperature=0.1, do_sample=True, pad_token_id=_tokenizer.eos_token_id, ) new_tokens = output_ids[0][inputs["input_ids"].shape[1]:] raw_text = _tokenizer.decode(new_tokens, skip_special_tokens=True) log.info("Raw model output: %s", raw_text[:300]) data = json.loads(_extract_json_text(raw_text)) 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, tf_4h=snap_4h, tf_1h=snap_1h, tf_15m=snap_15m, ) # --------------------------------------------------------------------------- # Prompt builder # --------------------------------------------------------------------------- def _build_prompt( coin: str, s4h: IndicatorSnapshot, s1h: IndicatorSnapshot, s15m: IndicatorSnapshot, df_15m, ) -> str: maker_offset = s15m.atr_14 * 0.25 return f"""Analyse {coin}/USD and provide a multi-timeframe swing trade recommendation. ━━━ 4H — MACRO TREND ━━━ Price : {s4h.price:.4f} EMA trend : {s4h.ema_trend} (EMA20={s4h.ema_20:.4f} EMA50={s4h.ema_50:.4f} EMA200={s4h.ema_200:.4f}) RSI(14) : {s4h.rsi_14:.2f} [{s4h.rsi_zone}] MACD hist : {s4h.macd_hist:.4f} ({'positive ▲' if s4h.macd_hist > 0 else 'negative ▼'}) ATR(14) : {s4h.atr_14:.4f} Vol / 20-avg : {s4h.volume_vs_avg:.2f}x ━━━ 1H — MOMENTUM CONFIRMATION ━━━ Price : {s1h.price:.4f} EMA trend : {s1h.ema_trend} (EMA20={s1h.ema_20:.4f} EMA50={s1h.ema_50:.4f}) RSI(14) : {s1h.rsi_14:.2f} [{s1h.rsi_zone}] MACD hist : {s1h.macd_hist:.4f} ({'positive ▲' if s1h.macd_hist > 0 else 'negative ▼'}) BB position : {s1h.bb_position} (upper={s1h.bb_upper:.4f} mid={s1h.bb_mid:.4f} lower={s1h.bb_lower:.4f}) Vol / 20-avg : {s1h.volume_vs_avg:.2f}x ━━━ 15M — ENTRY ZONE ━━━ Price : {s15m.price:.4f} RSI(14) : {s15m.rsi_14:.2f} [{s15m.rsi_zone}] ATR(14) : {s15m.atr_14:.4f} BB position : {s15m.bb_position} (upper={s15m.bb_upper:.4f} lower={s15m.bb_lower:.4f}) Recent highs : {', '.join(f'{h:.4f}' for h in s15m.recent_highs[-5:])} Recent lows : {', '.join(f'{l:.4f}' for l in s15m.recent_lows[-5:])} ATR-based TP/SL (for reference): LONG → TP {s15m.suggested_long_tp:.4f} SL {s15m.suggested_long_sl:.4f} SHORT → TP {s15m.suggested_short_tp:.4f} SL {s15m.suggested_short_sl:.4f} ━━━ LAST 8 × 15M CANDLES ━━━ {_format_recent_candles(df_15m.tail(8))} ━━━ ENTRY CONSTRAINTS (Hyperliquid maker rules) ━━━ Model latency : ~{_MODEL_LATENCY_S}s before order reaches exchange Max entry offset from current price (maker fill probability): {maker_offset:.4f} (= ATR_15m × 0.25) LONG limit must be in range [{s15m.price - maker_offset:.4f}, {s15m.price:.4f}] ← BELOW current (pullback to support) SHORT limit must be in range [{s15m.price:.4f}, {s15m.price + maker_offset:.4f}] ← ABOVE current (rally to resistance) Provide your JSON recommendation now.""" 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: text = text.strip() fence = re.match(r"^```(?:json)?\s*([\s\S]*?)\s*```$", text) if fence: return fence.group(1).strip() brace_match = re.search(r"\{[\s\S]*\}", text) if brace_match: return brace_match.group(0) return text