Sentiment & Social APIs — StockTwits, Reddit, Twitter

import requests

# StockTwits — no API key needed!
def get_stocktwits_sentiment(symbol):
    url = f"https://api.stocktwits.com/api/2/streams/symbol/{symbol}.json"
    data = requests.get(url).json()

    messages = data["messages"]
    bullish = sum(1 for m in messages if m.get("entities", {}).get("sentiment", {}).get("basic") == "Bullish")
    bearish = sum(1 for m in messages if m.get("entities", {}).get("sentiment", {}).get("basic") == "Bearish")
    total = bullish + bearish

    return {
        "symbol": symbol,
        "total_messages": len(messages),
        "bullish": bullish,
        "bearish": bearish,
        "bull_ratio": bullish / total if total > 0 else 0.5,
        "sample_message": messages[0]["body"]
    }

# Usage
sentiment = get_stocktwits_sentiment("AAPL")
print(f"AAPL: {sentiment['bullish']} bullish / {sentiment['bearish']} bearish ({sentiment['bull_ratio']:.0%})")

# Trending tickers
trending = requests.get("https://api.stocktwits.com/api/2/trending/symbols.json").json()
for t in trending["symbols"][:10]:
    print(f"  ${t['symbol']} — {t['title']}")

import praw
import re
from collections import Counter

# Setup Reddit API client
reddit = praw.Reddit(
    client_id="YOUR_CLIENT_ID",
    client_secret="YOUR_SECRET",
    user_agent="FinanceAPIs/1.0"
)

def scan_wsb_mentions(limit=100):
    """Scan r/wallstreetbets for ticker mentions."""
    ticker_pattern = re.compile(r'\$([A-Z]{1,5})\b|\b([A-Z]{2,5})\b')
    mentions = Counter()

    subreddit = reddit.subreddit("wallstreetbets")
    for post in subreddit.hot(limit=limit):
        text = post.title + " " + (post.selftext or "")
        tickers = ticker_pattern.findall(text)
        for match in tickers:
            ticker = match[0] or match[1]
            # Filter common words that look like tickers
            if ticker not in {"THE", "FOR", "AND", "ARE", "BUT", "NOT", "ALL", "CAN", "HAS", "HER", "WAS", "ONE", "OUR", "OUT", "DAY", "HAD", "HOT", "OIL", "OLD", "RED", "SIT", "TOP", "TWO", "WAR", "WIN", "YES", "BIG", "USE", "CEO", "IPO", "ETF", "IMO", "DD", "ITM", "OTM", "ATH"}:
                mentions[ticker] += 1

    return mentions.most_common(20)

# Top 20 most mentioned tickers on WSB right now
top_tickers = scan_wsb_mentions(200)
for ticker, count in top_tickers:
    print(f"  ${ticker}: {count} mentions")

import requests

# ApeWisdom — Top mentioned tickers (no API key!)
data = requests.get("https://apewisdom.io/api/v1.0/filter/all-stocks/page/1").json()

print("Top 10 most mentioned stocks on Reddit:")
for stock in data["results"][:10]:
    print(f"  #{stock['rank']} ${stock['ticker']} — {stock['mentions']} mentions ({stock['upvotes']} upvotes)")

# Filter by subreddit
wsb_data = requests.get("https://apewisdom.io/api/v1.0/filter/wallstreetbets/page/1").json()
for stock in wsb_data["results"][:5]:
    change = stock.get("mentions_24h_ago", 0)
    print(f"  ${stock['ticker']} — {stock['mentions']} mentions (24h change: {stock.get('rank_24h_ago', 'N/A')})")

import requests
from textblob import TextBlob  # pip install textblob

NEWS_KEY = "YOUR_NEWSAPI_KEY"

def get_news_sentiment(query, days=7):
    """Fetch news headlines and compute sentiment using TextBlob."""
    url = f"https://newsapi.org/v2/everything?q={query}&sortBy=publishedAt&language=en&pageSize=50&apiKey={NEWS_KEY}"
    articles = requests.get(url).json()["articles"]

    sentiments = []
    for a in articles:
        text = a["title"] + ". " + (a["description"] or "")
        blob = TextBlob(text)
        sentiments.append({
            "title": a["title"],
            "source": a["source"]["name"],
            "polarity": blob.sentiment.polarity,  # -1 to +1
            "subjectivity": blob.sentiment.subjectivity,  # 0 to 1
            "date": a["publishedAt"][:10]
        })

    avg_polarity = sum(s["polarity"] for s in sentiments) / len(sentiments)
    return {
        "query": query,
        "articles_analyzed": len(sentiments),
        "avg_sentiment": avg_polarity,
        "signal": "BULLISH" if avg_polarity > 0.1 else ("BEARISH" if avg_polarity < -0.1 else "NEUTRAL"),
        "top_positive": max(sentiments, key=lambda x: x["polarity"]),
        "top_negative": min(sentiments, key=lambda x: x["polarity"])
    }

result = get_news_sentiment("AAPL Apple")
print(f"Apple news sentiment: {result['signal']} ({result['avg_sentiment']:.3f})")

from pytrends.request import TrendReq
import pandas as pd

pytrends = TrendReq(hl='en-US')

# Search interest for "AAPL stock" over last 90 days
pytrends.build_payload(["AAPL stock"], timeframe="today 3-m")
interest = pytrends.interest_over_time()
print(interest.tail(10))

# Compare multiple tickers
pytrends.build_payload(["AAPL stock", "TSLA stock", "NVDA stock"], timeframe="today 3-m")
comparison = pytrends.interest_over_time()

# Related queries (what else people search)
pytrends.build_payload(["AAPL stock"])
related = pytrends.related_queries()
print("Rising queries:", related["AAPL stock"]["rising"])

# Detect search spikes (z-score > 2 = unusual interest)
import numpy as np
mean = interest["AAPL stock"].mean()
std = interest["AAPL stock"].std()
spikes = interest[interest["AAPL stock"] > mean + 2 * std]
print(f"Spike dates: {spikes.index.tolist()}")

import requests

YOUTUBE_KEY = "YOUR_YOUTUBE_API_KEY"

def count_youtube_videos(query, days=7):
    """Count YouTube videos about a stock in the last N days."""
    from datetime import datetime, timedelta
    after = (datetime.now() - timedelta(days=days)).isoformat("T") + "Z"

    url = f"https://www.googleapis.com/youtube/v3/search?part=snippet&q={query}+stock&type=video&publishedAfter={after}&maxResults=50&key={YOUTUBE_KEY}"
    data = requests.get(url).json()
    return data["pageInfo"]["totalResults"]

# Compare hype levels
for ticker in ["AAPL", "TSLA", "NVDA", "GME"]:
    count = count_youtube_videos(ticker)
    print(f"  {ticker}: {count} videos in last 7 days")

from dataclasses import dataclass
from typing import Dict, List
import numpy as np

@dataclass
class SentimentSignal:
    source: str
    score: float      # -1.0 (bearish) to +1.0 (bullish)
    confidence: float  # 0.0 to 1.0
    volume: int        # number of data points

class SentimentAggregator:
    """Aggregate sentiment from multiple sources with weighted scoring."""

    WEIGHTS = {
        "stocktwits": 0.20,
        "reddit": 0.15,
        "news": 0.30,
        "google_trends": 0.10,
        "options_flow": 0.25,
    }

    def aggregate(self, signals: List[SentimentSignal]) -> Dict:
        if not signals:
            return {"composite": 0, "signal": "NO_DATA"}

        # Weighted average, adjusted by confidence
        weighted_sum = 0
        weight_total = 0
        for s in signals:
            w = self.WEIGHTS.get(s.source, 0.1) * s.confidence
            weighted_sum += s.score * w
            weight_total += w

        composite = weighted_sum / weight_total if weight_total > 0 else 0

        # Determine signal strength
        if composite > 0.5:
            signal = "STRONG_BULL"
        elif composite > 0.2:
            signal = "BULLISH"
        elif composite > -0.2:
            signal = "NEUTRAL"
        elif composite > -0.5:
            signal = "BEARISH"
        else:
            signal = "STRONG_BEAR"

        # Check for convergence (all sources agree)
        signs = [np.sign(s.score) for s in signals if abs(s.score) > 0.1]
        convergence = abs(sum(signs)) / len(signs) if signs else 0

        return {
            "composite": round(composite, 3),
            "signal": signal,
            "convergence": round(convergence, 2),
            "sources": len(signals),
            "high_conviction": convergence > 0.8 and abs(composite) > 0.3
        }

# Example usage
agg = SentimentAggregator()
result = agg.aggregate([
    SentimentSignal("stocktwits", 0.65, 0.8, 150),
    SentimentSignal("reddit", 0.45, 0.6, 80),
    SentimentSignal("news", 0.30, 0.9, 25),
    SentimentSignal("google_trends", 0.55, 0.5, 1),
])
print(f"Composite: {result['composite']}, Signal: {result['signal']}, Convergence: {result['convergence']}")
print(f"High conviction: {result['high_conviction']}")