How to Automate X (Twitter) Trend Monitoring using the API

1#!/usr/bin/env python3
2import os
3import time
4import logging
5import requests
6from collections import deque
7
8logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
9
10BEARER = os.environ['TWITTER_BEARER_TOKEN']
11SLACK_WEBHOOK = os.environ['SLACK_WEBHOOK_URL']
12
13KEYWORDS = ['GPT', 'Claude AI', 'OpenAI', 'LLM']
14SPIKE_THRESHOLD = 3.0
15POLL_INTERVAL = 1800  # 30 minutes
16ALERT_COOLDOWN = 3600  # 1 hour per keyword
17
18headers = {'Authorization': f'Bearer {BEARER}'}
19hourly_history = {kw: deque(maxlen=24) for kw in KEYWORDS}
20alert_times = {kw: 0 for kw in KEYWORDS}
21
22def get_counts(query):
23    resp = requests.get(
24        'https://api.twitter.com/2/tweets/counts/recent',
25        headers=headers,
26        params={'query': query, 'granularity': 'hour'},
27        timeout=15
28    )
29    if resp.status_code == 429:
30        reset = int(resp.headers.get('x-rate-limit-reset', time.time() + 900))
31        wait = max(reset - time.time(), 0) + 5
32        logging.warning(f'Rate limited on counts. Waiting {wait:.0f}s')
33        time.sleep(wait)
34        return get_counts(query)
35    resp.raise_for_status()
36    return resp.json()['data']
37
38def get_context(query, n=3):
39    resp = requests.get(
40        'https://api.twitter.com/2/tweets/search/recent',
41        headers=headers,
42        params={
43            'query': query,
44            'tweet.fields': 'public_metrics,created_at',
45            'sort_order': 'relevancy',
46            'max_results': n
47        },
48        timeout=15
49    )
50    resp.raise_for_status()
51    tweets = resp.json().get('data', [])
52    logging.info(f'Context tweets fetched: {len(tweets)} (~${len(tweets)*0.005:.3f} cost)')
53    return tweets
54
55def send_alert(keyword, last_hour, baseline, ratio, context):
56    top = context[0]['text'][:140] if context else 'No context'
57    payload = {
58        'text': f'*X Trend Alert* \u2191 "{keyword}" spiking: {last_hour:,}/hr (avg {baseline:.0f}) = {ratio:.1f}x',
59        'attachments': [{
60            'color': '#ff4444',
61            'fields': [
62                {'title': 'Current (last hour)', 'value': f'{last_hour:,} tweets', 'short': True},
63                {'title': '24h Baseline', 'value': f'{baseline:.0f} tweets/hr', 'short': True},
64            ],
65            'text': f'Top tweet: {top}'
66        }]
67    }
68    requests.post(SLACK_WEBHOOK, json=payload, timeout=5)
69    logging.info(f'Alert sent: "{keyword}" at {ratio:.1f}x spike')
70
71def run_check():
72    for kw in KEYWORDS:
73        query = f'{kw} lang:en -is:retweet'
74        try:
75            counts = get_counts(query)
76            if len(counts) < 25:
77                continue
78            last_hour = counts[-1]['tweet_count']
79            baseline = sum(c['tweet_count'] for c in counts[-25:-1]) / 24
80            ratio = last_hour / baseline if baseline > 0 else 0
81            hourly_history[kw].append(last_hour)
82            logging.info(f'"{kw}": {last_hour}/hr, baseline {baseline:.0f}, ratio {ratio:.1f}x')
83            if ratio >= SPIKE_THRESHOLD and (time.time() - alert_times[kw]) >= ALERT_COOLDOWN:
84                context = get_context(query, n=3)
85                send_alert(kw, last_hour, baseline, ratio, context)
86                alert_times[kw] = time.time()
87            time.sleep(2)  # Small delay between keywords
88        except Exception as e:
89            logging.error(f'Error checking "{kw}": {e}')
90
91if __name__ == '__main__':
92    logging.info(f'Starting trend monitor for: {KEYWORDS}')
93    while True:
94        run_check()
95        logging.info(f'Sleeping {POLL_INTERVAL}s...')
96        time.sleep(POLL_INTERVAL)