Build a Grok AI Trading Agent with Real-Time Social Sentiment

pip install openai httpx python-dotenv

XAI_API_KEY=xai-your-key-here
PURPLE_FLEA_KEY=pf_live_your_key_here
WATCHLIST=BTC,ETH,SOL,DOGE,PEPE,WIF

import os
from openai import OpenAI

# Grok is OpenAI-compatible — just change base_url and model
grok = OpenAI(
    api_key=os.getenv("XAI_API_KEY"),
    base_url="https://api.x.ai/v1",
)
GROK_MODEL = "grok-3"   # or grok-3-mini for lower cost

SENTIMENT_SYSTEM_PROMPT = """You are a crypto social sentiment analyst with access to
real-time X/Twitter data. You track narrative momentum for crypto tokens.

For each token, analyze:
1. MENTION VOLUME: Is it trending? (much higher than baseline = signal)
2. SENTIMENT SCORE: Net bullish/bearish/neutral across recent posts
3. INFLUENCER SIGNAL: Any high-follower accounts posting?
4. NARRATIVE: What story is driving the conversation?
5. LEAD SIGNAL: Based on past patterns, does this sentiment suggest
   a price move is likely in the next 1-2 hours?

Output strict JSON — no markdown:
{
  "token": "SOL",
  "mention_trend": "spike" | "elevated" | "normal" | "declining",
  "sentiment_score": -1.0 to 1.0,
  "bull_pct": 0.65,
  "bear_pct": 0.20,
  "neutral_pct": 0.15,
  "influencer_signal": true | false,
  "narrative": "2-3 sentence summary of dominant X narrative",
  "momentum_signal": "strong_buy" | "buy" | "hold" | "sell" | "strong_sell",
  "confidence": 0.0 to 1.0,
  "signal_horizon_minutes": 60
}"""

import os, json, time, logging
import httpx
from dotenv import load_dotenv
from grok_client import grok, GROK_MODEL
from prompts import SENTIMENT_SYSTEM_PROMPT

load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("grok-agent")

PF_KEY  = os.getenv("PURPLE_FLEA_KEY")
PF_BASE = "https://api.purpleflea.com/v1"
HEADERS = {"Authorization": f"Bearer {PF_KEY}", "Content-Type": "application/json"}
WATCHLIST = os.getenv("WATCHLIST", "BTC,ETH,SOL").split(",")

# Cooldown tracker: don't re-enter the same token within 90 minutes
last_trade: dict[str, float] = {}

def get_sentiment(token: str) -> dict:
    "Ask Grok for real-time X sentiment on a token."
    user_msg = f"""Analyze current X/Twitter sentiment for ${token} crypto token.
Look at posts from the last 2 hours. Identify any trending narratives,
influencer posts (>10k followers), and whether mention volume is above baseline.
Provide the JSON output as specified."""
    resp = grok.chat.completions.create(
        model=GROK_MODEL,
        messages=[
            {"role": "system", "content": SENTIMENT_SYSTEM_PROMPT},
            {"role": "user",   "content": user_msg},
        ],
        temperature=0.2,
        max_tokens=512,
    )
    content = resp.choices[0].message.content.strip()
    if content.startswith("```"):
        content = content.split("\n", 1)[1].rsplit("```", 1)[0]
    result = json.loads(content)
    log.info(
        f"${token} sentiment={result['sentiment_score']:+.2f} "
        f"signal={result['momentum_signal']} conf={result['confidence']:.0%}"
    )
    return result

def get_price(symbol: str) -> float:
    with httpx.Client() as http:
        r = http.get(f"{PF_BASE}/markets/{symbol}-PERP/ticker", headers=HEADERS)
    return r.json()["last_price"]

def place_momentum_trade(token: str, signal: dict) -> None:
    symbol = f"{token}-PERP"
    side   = "long" if signal["momentum_signal"] in ("strong_buy", "buy") else "short"
    # Scale position by confidence: 0.6 conf → $300, 0.9 conf → $750
    size_usd = int(500 * signal["confidence"] * 1.5)
    # Time-limited stop: sentiment signals decay — use 90-min horizon
    stop_pct = 0.02   # 2% stop
    tp_pct   = 0.045  # 4.5% take-profit → ~2.25R
    payload = {
        "symbol":     symbol,
        "side":       side,
        "size_usd":   size_usd,
        "leverage":   2,
        "stop_loss":  stop_pct,
        "take_profit": tp_pct,
        "reduce_only": False,
    }
    with httpx.Client() as http:
        r = http.post(f"{PF_BASE}/trading/order", json=payload, headers=HEADERS)
    r.raise_for_status()
    last_trade[token] = time.time()
    log.info(
        f"TRADE: {side} {symbol} ${size_usd} | narrative: {signal['narrative'][:80]}"
    )

def should_trade(token: str, signal: dict) -> bool:
    "Apply filters before executing a trade."
    # Cooldown: don't trade same token within 90 minutes
    cooldown = 90 * 60
    if time.time() - last_trade.get(token, 0) < cooldown:
        log.info(f"${token} on cooldown — skipping")
        return False
    # Require strong signal + high confidence
    if signal["momentum_signal"] not in ("strong_buy", "strong_sell", "buy", "sell"):
        return False
    if signal["confidence"] < 0.60:
        return False
    # Require mention spike — normal baseline rarely leads to moves
    if signal["mention_trend"] == "normal":
        return False
    return True

def run_agent(scan_interval: int = 600):
    "Scan all watchlist tokens every 10 minutes for sentiment signals."
    log.info(f"Grok sentiment agent starting | watching: {WATCHLIST}")
    while True:
        for token in WATCHLIST:
            try:
                signal = get_sentiment(token)
                if should_trade(token, signal):
                    place_momentum_trade(token, signal)
                time.sleep(3)   # rate limit spacing between tokens
            except Exception as e:
                log.error(f"Error scanning ${token}: {e}")
        log.info(f"Scan complete. Next scan in {scan_interval}s")
        time.sleep(scan_interval)

if __name__ == "__main__":
    run_agent()