xAI X (Twitter) Search

import os
from openai import OpenAI

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

# Simple X search
response = client.chat.completions.create(
    model="grok-4-1-fast",
    messages=[{
        "role": "user",
        "content": "Search X for what people are saying about Tesla stock today"
    }]
)
print(response.choices[0].message.content)

def search_x_topic(topic: str) -> str:
    """Search X for posts about a topic."""
    response = client.chat.completions.create(
        model="grok-4-1-fast",
        messages=[{
            "role": "user",
            "content": f"Search X for recent posts about {topic}. Summarize the main discussions and sentiment."
        }]
    )
    return response.choices[0].message.content

def search_stock_mentions(ticker: str) -> str:
    """Search X for stock ticker mentions."""
    response = client.chat.completions.create(
        model="grok-4-1-fast",
        messages=[{
            "role": "user",
            "content": f"""Search X for mentions of ${ticker} stock.
            Find:
            - Recent discussions
            - Sentiment (bullish/bearish)
            - Key influencer opinions
            - Breaking news mentions
            Return structured analysis."""
        }]
    )
    return response.choices[0].message.content

def monitor_account(handle: str) -> str:
    """Get recent posts from a specific X account."""
    response = client.chat.completions.create(
        model="grok-4-1-fast",
        messages=[{
            "role": "user",
            "content": f"Search X for the most recent posts from @{handle}. Summarize their latest activity."
        }]
    )
    return response.choices[0].message.content

def get_trending() -> str:
    """Get current trending topics on X."""
    response = client.chat.completions.create(
        model="grok-4-1-fast",
        messages=[{
            "role": "user",
            "content": "What are the current trending topics on X? List the top 10 with brief descriptions."
        }]
    )
    return response.choices[0].message.content

# Using Responses API with x_search tool
response = client.chat.completions.create(
    model="grok-4-1-fast",
    messages=[{
        "role": "user",
        "content": "Search X for posts about Bitcoin from the last 24 hours"
    }],
    tools=[{
        "type": "x_search",
        "x_search": {
            "enabled": True,
            "date_range": {
                "start": "2025-12-04",
                "end": "2025-12-05"
            }
        }
    }]
)

# Search only specific accounts
response = client.chat.completions.create(
    model="grok-4-1-fast",
    messages=[{
        "role": "user",
        "content": "What are these financial analysts saying about the market?"
    }],
    tools=[{
        "type": "x_search",
        "x_search": {
            "enabled": True,
            "allowed_x_handles": [
                "jimcramer",
                "elonmusk",
                "chaikinadx",
                "unusual_whales"
            ]
        }
    }]
)

def monitor_financial_news() -> dict:
    """Monitor financial news on X."""
    response = client.chat.completions.create(
        model="grok-4-1-fast",
        messages=[{
            "role": "user",
            "content": """Search X for breaking financial news in the last hour.
            Focus on:
            - Market-moving news
            - Earnings announcements
            - Fed/economic news
            - Major analyst calls

            Return as JSON:
            {
                "breaking_news": [...],
                "market_sentiment": "bullish/bearish/neutral",
                "key_events": [...]
            }"""
        }]
    )
    return response.choices[0].message.content

def track_earnings_reaction(ticker: str) -> str:
    """Track X reaction to earnings announcement."""
    response = client.chat.completions.create(
        model="grok-4-1-fast",
        messages=[{
            "role": "user",
            "content": f"""Search X for reaction to ${ticker} earnings.
            Analyze:
            - Overall sentiment
            - Key concerns raised
            - Positive highlights mentioned
            - Notable influencer reactions
            - Volume of discussion"""
        }]
    )
    return response.choices[0].message.content

def compare_sentiment(tickers: list) -> str:
    """Compare X sentiment across multiple stocks."""
    ticker_str = ", ".join([f"${t}" for t in tickers])
    response = client.chat.completions.create(
        model="grok-4-1-fast",
        messages=[{
            "role": "user",
            "content": f"""Compare X sentiment for: {ticker_str}
            For each, provide:
            - Current sentiment score (-1 to +1)
            - Key themes being discussed
            - Notable mentions
            Return as structured comparison."""
        }]
    )
    return response.choices[0].message.content

# Bad - too vague
"Search X for stocks"

# Good - specific query
"Search X for posts about $AAPL stock price movement today from verified financial accounts"

# Request JSON for easier parsing
content = """Search X for $NVDA sentiment. Return JSON:
{
    "sentiment": "bullish/bearish/neutral",
    "score": -1 to 1,
    "key_posts": [...],
    "influencer_opinions": [...]
}"""

# Filter to trusted sources
financial_handles = [
    "DeItaone",  # Breaking news
    "unusual_whales",  # Options flow
    "Fxhedgers",  # Market news
    "zaborsky"  # Analysis
]

# Combine X sentiment with price data
x_sentiment = search_stock_mentions("AAPL")
price_data = finnhub_client.get_quote("AAPL")
# Analyze together

def safe_x_search(query: str) -> dict:
    """X search with error handling."""
    try:
        response = client.chat.completions.create(
            model="grok-4-1-fast",
            messages=[{"role": "user", "content": query}],
            timeout=30
        )
        return {
            "success": True,
            "data": response.choices[0].message.content
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e)
        }