Using LLMs In Trading Bots

You are an elite quantitative trading intelligence agent operating within an automated execution system.
Your job is to analyze incoming raw market text data, synthesize it alongside structural technical metrics, and output a strict JSON payload containing an explicit directional signal, confidence metrics, and structural justification.

### DATA SYSTEM INPUTS
1. Target Asset: {{ASSET_TICKER}}
2. Current Market Structure: {{MARKET_STRUCTURE_TEXT}}
3. Raw Technical Metrics (1H Timeframe):
   - Relative Strength Index (RSI): {{TECHNICAL_RSI}}
   - Exponential Moving Average Alignment: {{TECHNICAL_EMA}}
   - Average True Range (ATR): {{TECHNICAL_ATR}}
4. Ingested News Feed Data:
   "{{RAW_NEWS_FEED_STREAM}}"

### ANALYTICAL PROTOCOL (Chain-of-Thought)
You must execute your analysis systematically across three distinct phases before deriving the final trading vector:
- Phase 1 (Macro-Sentiment Integration): Evaluate how the ingested news impacts liquidity and demand dynamics for {{ASSET_TICKER}}.
- Phase 2 (Technical Convergence): Determine if technical indicators align with or diverge from macro sentiment vectors.
- Phase 3 (Risk-Reward Probability Mapping): Calculate asymmetric risk profile boundaries based on ATR and structural key levels.

### OUTPUT JSON SCHEMA SPECIFICATION
Your output must consist exclusively of a single, valid JSON object. Do not include any conversational text, markdown wrapping, or preamble.
Required Keys:
{
  "ticker": "string",
  "signal": "STRONG_BUY | BUY | HOLD | SELL | STRONG_SELL",
  "confidence_score": float (range 0.00 to 1.00),
  "sentiment_bias": "BULLISH | BEARISH | NEUTRAL",
  "primary_catalyst": "string (maximum 20 words)",
  "volatility_expectation": "EXPANDING | COMPRESSING | STABLE",
  "suggested_stop_loss": float,
  "suggested_take_profit": float
}

import json
import asyncio
from pydantic import BaseModel, Field, ValidationError
from openai import AsyncOpenAI

# 1. Define strict schema using Pydantic for automated validation
class LLMTradeSignal(BaseModel):
    ticker: str
    signal: str = Field(..., pattern="^(STRONG_BUY|BUY|HOLD|SELL|STRONG_SELL)$")
    confidence_score: float = Field(..., ge=0.0, le=1.0)
    sentiment_bias: str = Field(..., pattern="^(BULLISH|BEARISH|NEUTRAL)$")
    primary_catalyst: str
    volatility_expectation: str
    suggested_stop_loss: float
    suggested_take_profit: float

# 2. Async function to query LLM and parse signal safely
async def fetch_llm_trade_signal(client: AsyncOpenAI, prompt: str) -> LLMTradeSignal | None:
    try:
        response = await client.chat.completions.create(
            model="gpt-4o",
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": "You are a quantitative trading signal engine. Output strict JSON only."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.1,  # Low temperature for deterministic output
            timeout=5.0        # Strict latency threshold (5 seconds max)
        )
        
        raw_json = response.choices[0].message.content
        parsed_data = json.loads(raw_json)
        
        # Validate data against Pydantic schema
        signal = LLMTradeSignal(**parsed_data)
        return signal

    except (ValidationError, json.JSONDecodeError) as err:
        print(f"[LLM PARSE ERROR] Signal rejected due to invalid schema: {err}")
        return None
    except asyncio.TimeoutError:
        print("[LLM TIMEOUT] Model API took longer than 5s. Falling back to rule-based logic.")
        return None

import re

def sanitize_market_news_feed(raw_text: str) -> str:
    """
    Sanitizes raw scraped news/social text to prevent prompt injection 
    and remove useless promotional noise before sending to the LLM.
    """
    # 1. Remove systemic prompt override attack patterns
    forbidden_patterns = [
        r"ignore (all )?previous instructions",
        r"system override",
        r"output (a )?strong buy",
        r"disregard (the )?system prompt",
        r"you are now an unrestricted"
    ]
    
    sanitized = raw_text
    for pattern in forbidden_patterns:
        sanitized = re.sub(pattern, "[FILTERED_ATTEMPT]", sanitized, flags=re.IGNORECASE)
    
    # 2. Remove spam URLs and Telegram/Discord promotional links
    sanitized = re.sub(r'https?://\S+|www\.\S+', '[LINK_REMOVED]', sanitized)
    sanitized = re.sub(r'@\w+', '[HANDLE_REMOVED]', sanitized)
    
    # 3. Truncate long strings to conserve token context window
    max_length = 1500
    if len(sanitized) > max_length:
        sanitized = sanitized[:max_length] + "... [TRUNCATED]"
        
    return sanitized.strip()