Grok-powered agents with real-time data and financial execution

from openai import OpenAI
import requests
import json

# xAI uses OpenAI-compatible API — just swap the base_url
XAI_API_KEY = "xai-your-key-here"
PF_API_KEY  = "pf_live_your_key_here"

client = OpenAI(
    base_url = "https://api.x.ai/v1",
    api_key  = XAI_API_KEY
)

class GrokSentimentTrader:

    def analyze_x_sentiment(self, asset: str) -> dict:
        """Use Grok's live X access to analyze crypto sentiment."""
        response = client.chat.completions.create(
            model = "grok-3",
            messages = [
                {
                    "role": "system",
                    "content": (
                        "You are a crypto trading agent with live access to X. "
                        "Analyze recent posts about the given asset from the last 2 hours. "
                        "Look for: whale movements, developer announcements, exchange news, "
                        "notable influencer posts, and on-chain event references. "
                        "Respond only with valid JSON: {sentiment, confidence, direction, "
                        "signal_quality, key_accounts, reasoning}"
                    )
                },
                {
                    "role": "user",
                    "content": f"Analyze X sentiment for {asset} — last 2 hours."
                }
            ],
            temperature = 0.1
        )
        content = response.choices[0].message.content
        try:
            return json.loads(content)
        except:
            return {"sentiment": "neutral", "confidence": 0, "direction": "hold"}

    def verify_with_onchain(self, asset: str, signal: dict) -> bool:
        """Cross-reference signal against on-chain wallet data."""
        # In production: call Purple Flea wallet API to check
        # large address movements for the asset
        if signal["signal_quality"] == "high" and signal["confidence"] > 70:
            return True
        return signal["confidence"] > 85

    def execute_trade(self, market: str, direction: str,
                        size_usd: float) -> dict:
        """Open perpetual position on Purple Flea (275 markets)."""
        resp = requests.post(
            "https://purpleflea.com/api/v1/trading/perp/open",
            headers = {"X-API-Key": PF_API_KEY},
            json = {
                "market":    market,
                "side":      direction,
                "size_usd":  size_usd,
                "leverage":  3,
                "order_type": "market"
            }
        )
        return resp.json()

    def run_cycle(self, assets: list):
        """Run one full sentiment scan + execution cycle."""
        for asset in assets:
            print(f"[+] Grok analyzing X sentiment for {asset}...")
            signal = self.analyze_x_sentiment(asset)

            confirmed = self.verify_with_onchain(asset, signal)
            if not confirmed:
                print(f"    {asset}: signal not confirmed — skip")
                continue

            size = 50.0 if signal["confidence"] < 80 else 100.0
            result = self.execute_trade(
                market    = f"{asset}-PERP",
                direction = signal["direction"],
                size_usd  = size
            )
            print(f"    Trade: {result['status']} @ {result.get('entry_price','?')}")


if __name__ == "__main__":
    trader = GrokSentimentTrader()
    trader.run_cycle(["BTC", "ETH", "SOL", "DOGE"])