Spaces:
Runtime error
Runtime error
File size: 3,333 Bytes
f8c01d7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | # 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..
|