api_tweets_sentiment_analysis.py

# 1. importer toutes les librairies nécessaires

# pip install fastapi uvicorn requests transformers torch

import requests

from transformers import pipeline

from fastapi import FastAPI, HTTPException, Query

from collections import Counter

app = FastAPI(title="StockTwits Sentiment API")
 

BASE_URL = "https://api.stocktwits.com/api/2/streams/symbol/{ticker}.json"

HEADERS = {
    "User-Agent": ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                   "AppleWebKit/537.36 (KHTML, like Gecko) "
                   "Chrome/124.0.0.0 Safari/537.36"),
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "en-US,en;q=0.9,fr;q=0.8",
    "Referer": "https://stocktwits.com/",
    "Origin": "https://stocktwits.com",
}

def fetch_messages(ticker, max_pages=5):
    url = BASE_URL.format(ticker=ticker)
    messages = []
    max_id = None

    for page in range(max_pages):
        params = {"max": max_id} if max_id else {}
        data = requests.get(url, headers=HEADERS, params=params)
        data_json = data.json()

        messages += data_json["messages"]

        if not data_json["cursor"]["more"]:
            break
        max_id = data_json["cursor"]["max"] - 1

    
    return messages

# b)
def extract_filter(messages, min_chars: int = 15):
    messages_body = [m["body"] for m in messages]
    return [msg for msg in messages_body if len(msg) >= min_chars]

# c)
sentiment_model = pipeline(
    "sentiment-analysis",
    model="Neomac21/cryptobert_V3",
    truncation=True, ### J'ai ajouté les paramètres du modèle ###
    padding="max_length",
    max_length=512,
)

def classify(texts, score_threshold=0.6):
    results = sentiment_model(texts, truncation=True)
    filtered = [(text, result) for text, result in zip(texts, results) if result["score"] >= score_threshold]

    filtered_texts = [text for text, _ in filtered]  
    classification_results = [result for result in results if result["score"] >= score_threshold]
    
    return filtered_texts, classification_results


##################

# 4. servir les données en API avec FastAPI

# d)
# renvoi la distribution en %
def aggregate(labels):
    total = len(labels)
    counts = Counter(labels)
    return {label: round(n / total * 100, 1) for label, n in counts.items()}

@app.get("/sentiment")
def get_sentiment(ticker: str, max_pages: int = 5, min_chars: int = 15, min_score: float = 0.6):
    """Orchestre les étapes a→c et renvoie la distribution."""
    messages = fetch_messages(ticker, max_pages)

    if not messages:
        raise HTTPException(status_code=404, detail="No messages found for the given ticker.")  
    
    filtered_texts = extract_filter(messages, min_chars)
    filtered_messages, classified = classify(filtered_texts, min_score)

    if not classified:
        raise HTTPException(status_code=404, detail="No messages clear with the given criteria.")
    
    return {
        "messages" : filtered_messages, ### On peut aussi retourner les messages filtrés ###
        "labels": classified, ### On peut aussi retourner les labels de classification pour chaque message classifié. ###
        "ticker": ticker,
        "nb_of_messages": len(classified),
        "distribution": aggregate(classified), 
    }
  
# 5. lancer le serveur FastAPI et tester potentiels erreurs etc..