Spaces:

chigginrice
/

HypeAdvisor

Running on Zero

Daniel Alnawaz

undo entry price fix

12f14cb unverified 3 days ago

11.2 kB

"""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": <float>,
  "take_profit": <float>,
  "stop_loss": <float>,
  "risk_reward": <float>,
  "reasoning": "<3-5 sentences covering 4h trend, 1h confirmation, 15m entry rationale>",
  "key_risks": ["<risk1>", "<risk2>"]
}
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

	"""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": <float>,
	"take_profit": <float>,
	"stop_loss": <float>,
	"risk_reward": <float>,
	"reasoning": "<3-5 sentences covering 4h trend, 1h confirmation, 15m entry rationale>",
	"key_risks": ["<risk1>", "<risk2>"]
	}

	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