import streamlit as st import os import yfinance as yf import pandas as pd import numpy as np import torch from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split from sklearn.preprocessing import MinMaxScaler from datetime import datetime, timedelta # Configurazione iniziale di Streamlit st.set_page_config( page_title="Financial Prediction App", layout="wide", initial_sidebar_state="expanded" ) # Disabilita i warning di parallelismo os.environ["TOKENIZERS_PARALLELISM"] = "false" # Variabili globali per il modello tokenizer = None model = None @st.cache_resource(show_spinner="Caricamento del modello FinBERT...") def load_finbert(): global tokenizer, model try: from transformers import BertTokenizer, BertForSequenceClassification model_name = "yiyanghkust/finbert-tone" tokenizer = BertTokenizer.from_pretrained(model_name) model = BertForSequenceClassification.from_pretrained(model_name) model.eval() # Imposta il modello in modalità valutazione return tokenizer, model except Exception as e: st.error(f"Errore nel caricamento del modello: {str(e)}") return None, None def analyze_sentiment(text): global tokenizer, model if tokenizer is None or model is None: return "Modello non caricato", [0, 0, 0] try: inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True, max_length=512) with torch.no_grad(): outputs = model(**inputs) predictions = torch.nn.functional.softmax(outputs.logits, dim=-1) sentiment = torch.argmax(predictions).item() sentiment_map = {0: "Negative", 1: "Neutral", 2: "Positive"} return sentiment_map[sentiment], predictions[0].tolist() except Exception as e: st.error(f"Errore nell'analisi del sentiment: {str(e)}") return "Errore", [0, 0, 0] @st.cache_data(ttl=3600, show_spinner="Download dati storici...") def get_historical_data(symbol, period="1y"): try: data = yf.download(symbol, period=period, progress=False) if data.empty: st.warning(f"Nessun dato trovato per {symbol}") return None return data except Exception as e: st.error(f"Errore nel download dei dati: {str(e)}") return None def prepare_data(data, sentiment_score=None): # Feature engineering data['Return'] = data['Close'].pct_change() data['MA_7'] = data['Close'].rolling(window=7).mean() data['MA_30'] = data['Close'].rolling(window=30).mean() data['Volatility'] = data['Close'].rolling(window=7).std() # Aggiunta del sentiment se disponibile if sentiment_score is not None: data['Sentiment'] = sentiment_score # Rimozione dei valori NaN data.dropna(inplace=True) # Selezione delle feature features = ['Open', 'High', 'Low', 'Close', 'Volume', 'Return', 'MA_7', 'MA_30', 'Volatility'] if sentiment_score is not None: features.append('Sentiment') X = data[features] y = data['Close'].shift(-1) # Prevedere il prezzo di chiusura del giorno successivo # Rimuovi l'ultima riga senza target X = X[:-1] y = y[:-1] return X, y def train_and_predict(X, y, forecast_days=7): # Normalizzazione dei dati scaler = MinMaxScaler() X_scaled = scaler.fit_transform(X) # Divisione dei dati X_train, X_test, y_train, y_test = train_test_split(X_scaled, y, test_size=0.2, random_state=42) # Addestramento del modello model = RandomForestRegressor(n_estimators=100, random_state=42) model.fit(X_train, y_train) # Previsione per i prossimi giorni last_values = X.iloc[-1:].values last_values_scaled = scaler.transform(last_values) predictions = [] current_input = last_values_scaled.copy() for _ in range(forecast_days): pred = model.predict(current_input)[0] predictions.append(pred) # Aggiornamento dell'input per la previsione successiva new_row = current_input[0].copy() new_row[3] = pred # Aggiorna il prezzo di chiusura current_input = np.array([new_row]) return predictions def initialize_app(): st.title("📈 Financial Prediction App with FinBERT") st.markdown("### Previsioni finanziarie per azioni, forex, commodities, crypto e indici") # Carica il modello global tokenizer, model if tokenizer is None or model is None: with st.spinner("Caricamento del modello..."): tokenizer, model = load_finbert() if tokenizer is None or model is None: st.error("Impossibile caricare il modello. Riprova più tardi.") st.stop() return True def main(): # Inizializza l'app if not initialize_app(): return # Sidebar per le impostazioni st.sidebar.header("Impostazioni") asset_type = st.sidebar.selectbox( "Seleziona tipo di asset", ["Azioni", "Forex", "Commodities", "Crypto", "Indici"] ) # Mappatura dei simboli per tipo di asset symbol_map = { "Azioni": ["AAPL", "MSFT", "GOOGL", "AMZN", "TSLA"], "Forex": ["EURUSD=X", "GBPUSD=X", "USDJPY=X", "USDCHF=X"], "Commodities": ["GC=F", "SI=F", "CL=F", "NG=F"], "Crypto": ["BTC-USD", "ETH-USD", "BNB-USD", "SOL-USD"], "Indici": ["^GSPC", "^DJI", "^IXIC", "^FTSE"] } symbol = st.sidebar.selectbox("Seleziona simbolo", symbol_map[asset_type]) period = st.sidebar.selectbox("Periodo storico", ["1mo", "3mo", "6mo", "1y", "2y"]) forecast_days = st.sidebar.slider("Giorni di previsione", 1, 30, 7) # Input per l'analisi del sentiment st.subheader("Analisi del Sentiment con FinBERT") news_text = st.text_area("Inserisci notizie finanziarie (opzionale):", height=150) # Bottone per eseguire l'analisi if st.button("Genera Previsioni"): with st.spinner("Analisi in corso..."): # Analisi del sentiment sentiment = None sentiment_scores = None if news_text: sentiment, sentiment_scores = analyze_sentiment(news_text) st.success(f"Sentiment rilevato: {sentiment}") st.write("Score dettagliati:") st.write(f"Negativo: {sentiment_scores[0]:.4f}") st.write(f"Neutro: {sentiment_scores[1]:.4f}") st.write(f"Positivo: {sentiment_scores[2]:.4f}") # Conversione del sentiment in punteggio numerico sentiment_score = sentiment_scores[2] - sentiment_scores[0] # Positivo - Negativo else: st.info("Nessuna notizia inserita. Verrà utilizzato solo l'analisi tecnica.") sentiment_score = None # Download dei dati storici data = get_historical_data(symbol, period) if data is not None: # Preparazione dei dati X, y = prepare_data(data, sentiment_score) if len(X) > 10: # Verifica di avere dati sufficienti # Addestramento e previsione predictions = train_and_predict(X, y, forecast_days) # Creazione del dataframe delle previsioni last_date = data.index[-1] prediction_dates = [last_date + timedelta(days=i) for i in range(1, forecast_days+1)] predictions_df = pd.DataFrame({ "Data": prediction_dates, "Previsione": predictions }) # Visualizzazione dei risultati st.subheader(f"Previsioni per {symbol} ({asset_type})") st.write(predictions_df) # Grafico st.line_chart(data['Close'].tail(30)) st.line_chart(predictions_df.set_index("Data")) # Statistiche del modello st.subheader("Statistiche del Modello") st.write(f"Numero di campioni di training: {len(X)}") st.write(f"Feature utilizzate: {list(X.columns)}") if sentiment_score is not None: st.write(f"Punteggio di sentiment utilizzato: {sentiment_score:.4f}") else: st.error("Dati insufficienti per generare previsioni. Seleziona un periodo più lungo.") else: st.error("Impossibile recuperare i dati per il simbolo selezionato.") if __name__ == "__main__": main()