SpriteDeveloperSpriteDeveloper
Reti neurali · Python

Analisi email spam

Versione integrata nel sito senza iframe. Il notebook ricostruito resta disponibile come file .ipynb.

Scarica .ipynb← Archivio
Codice
Gabriele Iocco
Codice
import warnings
warnings.filterwarnings("ignore")

import os
os.environ['TF_CPP_MIN_LOG_LEVEL']='1'

import tensorflow as tf
tf.compat.v1.logging.set_verbosity(tf.compat.v1.logging.ERROR)

gpu_devices = tf.config.list_physical_devices('GPU')

if gpu_devices:
    try:
        for gpu in gpu_devices:
            tf.config.experimental.set_memory_growth(gpu, True)
    except RuntimeError as e:
        print(e)

print("Dispositivi GPU rilevati:", gpu_devices)

import torch
print(f"CUDA available: {torch.cuda.is_available()}")
print(f"CUDA device count: {torch.cuda.device_count()}")
print(f"CUDA device name: {torch.cuda.get_device_name(0)}" if torch.cuda.is_available() else "No CUDA device found")
print("Dispositivo corrente:", torch.cuda.current_device() if torch.cuda.is_available() else "CPU")
print("TensorFlow version:", tf.__version__)
print("CUDA disponibile:", tf.config.list_physical_devices('GPU'))
print("cuDNN Version:", tf.sysconfig.get_build_info()['cudnn_version'])
print("CUDA Version:", tf.sysconfig.get_build_info()["cuda_version"])
print("Test CUDA:", tf.test.is_built_with_cuda())
print("GPU available:", tf.test.is_gpu_available())
Codice
# Configurazioni GPU TensorFlow
os.environ['TF_FORCE_GPU_ALLOW_GROWTH'] = 'true'
os.environ['XLA_FLAGS'] = '--xla_gpu_cuda_data_dir=/usr/lib/cuda'
os.environ['TF_GPU_ALLOCATOR'] = 'cuda_malloc_async'
os.environ['CUDA_VISIBLE_DEVICES'] = '0'
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'

# Configurazione  della memoria GPU
gpus = tf.config.list_physical_devices('GPU')
if gpus:
    for gpu in gpus:
        try:
            tf.config.experimental.set_memory_growth(gpu, True)
        except RuntimeError as e:
            print(e)
Codice
os.environ['TF_FORCE_GPU_ALLOW_GROWTH'] = 'true'
os.environ['XLA_FLAGS'] = '--xla_gpu_cuda_data_dir=/usr/lib/cuda'
Codice
import cupy as cp
print("CuPy version:", cp.__version__)
Codice
from thinc.api import require_gpu, set_gpu_allocator
import spacy

require_gpu()  # Forza l'uso della GPU
nlp = spacy.load("en_core_web_trf")
print("Preferenza GPU da SpaCy:", spacy.prefer_gpu())
Codice
from colorama import Fore, Back, Style
from tensorflow.keras.backend import clear_session
import pandas as pd
import matplotlib.pyplot as plt
Codice
# Funzione personalizzata per colorare il testo
def print_colored(text, color="white", bg_color=None, end="\n"):
    # Dizionario dei colori del testo
    color_dict = {
        'red': Fore.RED,
        'blue': Fore.BLUE,
        'white': '\033[97m',  # Bianco puro (ANSI)
        'black': Fore.BLACK
    }

    # Dizionario dei colori dello sfondo
    bg_color_dict = {
        'black': Back.BLACK,
        'blue': Back.BLUE,
        'white': Back.WHITE
    }

    color_code = color_dict.get(color.lower(), '\033[97m') 
    bg_color_code = bg_color_dict.get(bg_color.lower(), '') if bg_color else ''
    
    print(f"{color_code}{bg_color_code}{text}{Style.RESET_ALL}", end=end)
Codice
BASE_URL="/home/gap/Scrivania/Analisi_spam/"
Codice
df_email = pd.read_csv(BASE_URL + "spam_dataset.csv")
Codice
# Calcolo del peso totale del dataset in memoria
dataset_size_ham_spam = df_email.memory_usage(deep=True).sum()

# Conversione in megabyte (MB)
dataset_size_ham_spam_mb = dataset_size_ham_spam / (1024 ** 2)

print_colored(f"Il peso del dataset in memoria è di:", "blue") 
print(f"{dataset_size_ham_spam_mb:.2f} MB")
Codice
df_email.head()
Codice
df_email.info()
Codice
print_colored("Numero di righe:", "blue") 
print(df_email.shape[0])
print()
print_colored("Numero di colonne:", "blue") 
print(df_email.shape[1])
Codice
df_email.count()
Codice
total_values = df_email.size
print_colored("Valori totali presenti nel dataset:", "blue")
print(total_values)

non_missing_values = df_email.count().sum()
print_colored("Valori totali 'non_missing' presenti nel dataset:", "blue")
print(non_missing_values)

missing_values = total_values - non_missing_values
print_colored("\nValori mancanti", "blue")
print(missing_values)
Codice
print_colored("Valori mancanti per colonna:\n", "blue")
print(df_email.isna())
Codice
print_colored("Valori mancanti per colonna:\n", "blue")
print(df_email.isnull())
Codice
df_email.describe()
Codice
# Controllo di possibili incongruenze tra label e label_num
# La variabile inconsistencies viene popolata con le righe che non 
# rispettano le corrispondenze attese
inconsistencies = df_email[
    ((df_email['label'] == 'ham') & (df_email['label_num'] != 0)) | 
    ((df_email['label'] == 'spam') & (df_email['label_num'] != 1))
]

if inconsistencies.empty:
    print_colored("Tutte le righe corrispondono correttamente", "blue")
    print ("ham -> 0, spam -> 1")
else:
    print_colored(f"Ci sono {len(inconsistencies)} righe con incongruenze:", "blue")
    print(inconsistencies)
Codice
df_email_dropped = df_email.drop(columns=['Unnamed: 0', 'label'])

print_colored("Dataset dropped\n", "blue")
print(df_email_dropped.head())
Codice
label_count=df_email_dropped[df_email_dropped.columns[1]].value_counts()
Codice
print_colored("Distribuzione di ham e spam\n", "blue")

print(label_count)
Codice
label_count.plot(kind='bar', figsize=(10, 6))
plt.title("\nDistribuzione di ham e spam", fontsize=18)
plt.xlabel("Label (0 = ham, 1 = spam)", fontsize=14)
plt.ylabel("Conteggio", fontsize=14)
plt.xticks(fontsize=12, color='#b81414', rotation=0)
plt.yticks(fontsize=12, color='#b81414')
plt.grid(axis='y', linestyle='--', alpha=0.5, color="#1f77b4")
plt.show()
Codice
# Calcolo della differenza in percentuale
total_label = label_count.sum()
difference_percentage = ((label_count.max() - label_count.min()) / total_label) * 100

print_colored(f"La differenza in percentuale tra ham e spam è del:", "blue") 
print(f"{difference_percentage:.2f}%")
Codice
import re
import nltk
import nltk.corpus
from nltk.corpus import stopwords
from nltk.stem import WordNetLemmatizer
from nltk import download

download('stopwords')
download('wordnet')
download('omw-1.4')
Codice
# Inizializzazione di stopwords e lemmatizer
stop_words = set(stopwords.words('english')) # Stopword per la lingua inglese
                                             # Rimuove parole comuni inglesi per ridurre il rumore nei dati
lemmatizer = WordNetLemmatizer()             # Riduce le parole alla loro forma base per 
                                             # diminuire la dimensionalità del vocabolario

def clean_text(text):
    # Rimozione di caratteri speciali e punteggiatura
    text = re.sub(r'[^a-zA-Z\s]', '', text)  
    
    # Conversione di tutte le lettere in minuscolo
    text = text.lower()
    
    # Rimozione della parola 'subject' se presente perché si riferisce alla struttura standard di un'email
    # non aggiunge valore informativo per l'analisi, anzi, può influenzarla negativamente
    text = text.replace('subject', '')  
    
    tokens = [lemmatizer.lemmatize(word, pos='v') for word in text.split() if word not in stop_words]
    return ' '.join(tokens)

# Filtro le email SPAM e creo una copia per evitare SettingWithCopyWarning
# Il warning SettingWithCopyWarning si verifica quando si modificano direttamente valori di un DataFrame filtrato
df_email_dropped_spam = df_email_dropped[df_email_dropped['label_num'] == 1].copy()

# Creazione della colonna cleaned_text con i dati della colonna text puliti
df_email_dropped_spam['cleaned_text'] = df_email_dropped_spam['text'].apply(clean_text)

print_colored("Dataset 'df_email_dropped_spam' senza l'applicazione della funzione 'clean_text' alla colonna 'text'", "blue")
print(df_email_dropped_spam[['text']].head())

print()

print_colored("Dataset 'df_email_dropped_spam' con l'applicazione della funzione 'clean_text' alla colonna 'text'", "blue")
print(df_email_dropped_spam[['cleaned_text']].head())
Codice
print(df_email_dropped_spam.head())
Codice
# Verifico se il conteggio dei valori della colonna 'label num' riferito 
# alle mail spam sia uguale al conteggio di partenza
df_email_dropped_spam[df_email_dropped_spam.columns[1]].value_counts()
Codice
# Calcolo del peso totale del dataset contenente solo SPAM droppato e pulito in memoria
dataset_size_dropped_spam = df_email_dropped_spam[['cleaned_text', 'label_num']].memory_usage(deep=True).sum()

# Conversione in megabyte (MB)
dataset_size_mb_dropped = dataset_size_dropped_spam / (1024 ** 2)

print_colored(f"Il peso del dataset droppato, pulito e contenente solo spam in memoria è di:", "blue") 
print(f"{dataset_size_mb_dropped:.2f} MB")
Codice
dataset_size_difference = dataset_size_ham_spam_mb - dataset_size_mb_dropped

print_colored(f"La memoria liberata grazie al dropout e alla pulizia del testo è di:", "blue") 
print(f"{dataset_size_difference:.2f} MB")

total_size = dataset_size_ham_spam_mb + dataset_size_mb_dropped
difference_percentage_size = (dataset_size_difference/total_size)*100

print_colored(f"Percentuale di memoria liberata:", "blue") 
print(f"{difference_percentage_size:.2f} %")
Codice
# Calcolo del numero totale di parole nella colonna 'cleaned_text'
total_word_count = df_email_dropped_spam['cleaned_text'].str.split().str.len().sum()

print_colored(f"Il numero totale di parole nella colonna 'cleaned_text' è di:", "blue")
print(f"{total_word_count}")
Codice
from gensim.models import Word2Vec
from gensim.models import CoherenceModel
from gensim.corpora.dictionary import Dictionary
from sklearn.metrics import silhouette_score
from sklearn.cluster import KMeans
import multiprocessing

# Tokenizzazione
# divide il testo in token (parole e sottoparole) e assegna un identificatore numerico univoco a ogni token
tokenized_texts = [text.split() for text in df_email_dropped_spam['cleaned_text']]

# Creazione del dizionario per Coherence Score
dictionary = Dictionary(tokenized_texts)

# Configurazioni da testare
configs = [
    {"name": "model_vec50_win3", "vector_size": 50, "window": 3, "min_count": 2, "sg": 1},
    {"name": "model_vec100_win3", "vector_size": 100, "window": 3, "min_count": 2, "sg": 1},
    {"name": "model_vec100_win5", "vector_size": 100, "window": 5, "min_count": 2, "sg": 1},
    {"name": "model_vec200_win3", "vector_size": 200, "window": 3, "min_count": 2, "sg": 1},
    {"name": "model_vec200_win5", "vector_size": 200, "window": 5, "min_count": 2, "sg": 1},
]

# Funzione per estrarre i topic da Word2Vec
def extract_topics_from_word2vec(model, n_topics, topn=10):
    word_vectors = np.array([model.wv[word] for word in model.wv.index_to_key])
    words = model.wv.index_to_key
    
    # Clustering con KMeans
    kmeans = KMeans(n_clusters=n_topics, random_state=42)
    labels = kmeans.fit_predict(word_vectors)
    
    # Raggruppamento delle parole per cluster
    topics = []
    for cluster in range(n_topics):
        words_in_cluster = [words[i] for i, label in enumerate(labels) if label == cluster]
        topics.append(words_in_cluster[:topn])
    return topics

# Per salvare i risultati
results_main = []

# Numero di topic da testare
n_topics_list = [2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20]

# Addestramento e valutazione dei modelli
for config in configs:
    print_colored(f"Addestramento del modello: {config['name']}", "blue")
    
    # Addestramento del modello Word2Vec
    """
    vector_size indica la dimensione del vettore generato per ogni parola
    window indica la dimensione del contesto, il numero di parole vicine considerate per ogni parola
            Siccome il progetto prevede un'analisi contenutistica approfondita lo imposto a 3 e 5
    min_count, le parole con una frequenza inferiore a questo valore vengono ignorate
    sg Skip-gram indica il tipo di architettura
            Skip-gram cattura relazioni semantiche
    """

    # Embedding delle parole
    # converte i token in vettori densi, cattura relazioni semantiche tra token
    # rappresenta il significato contestuale dei token
    model = Word2Vec(
        sentences=tokenized_texts,
        vector_size=config["vector_size"],
        window=config["window"],
        min_count=config["min_count"],
        workers=multiprocessing.cpu_count(),
        sg=config["sg"]
    )
    model.save(f"{config['name']}.model")
    
    for n_topics in n_topics_list:
        try:
            # Estrazione dei topic
            topics = extract_topics_from_word2vec(model, n_topics)

            # Calcolo del Coherence Score
            coherence_model = CoherenceModel(
                topics=topics,
                texts=tokenized_texts,
                dictionary=dictionary,
                coherence="c_v"
            )
            coherence_score = coherence_model.get_coherence()

            # Calcolo del Silhouette Score
            word_vectors = np.array([model.wv[word] for word in model.wv.index_to_key])
            silhouette_avg = silhouette_score(word_vectors, KMeans(n_clusters=n_topics, random_state=42).fit_predict(word_vectors))

            # Salvataggio dei risultati
            results_main.append({
                "model_name": config["name"],
                "vector_size": config["vector_size"],
                "window": config["window"],
                "n_topics": n_topics,
                "coherence_c_v": coherence_score,
                "silhouette": silhouette_avg,
            })

        except Exception as e:
            print(f"Errore per {config['name']} con {n_topics} topic: {e}")
            results_main.append({
                "model_name": config["name"],
                "vector_size": config["vector_size"],
                "window": config["window"],
                "n_topics": n_topics,
                "coherence_c_v": None,
                "silhouette": None,
            })

# Conversione in DataFrame per analisi
results_df_main = pd.DataFrame(results_main)
Codice
print_colored(f"Modelli costruiti e addestrati:", "blue")

for config in configs:    
    print(config['name'])
Codice
print_colored("Colonne disponibili:", "blue")
print(results_df_main.columns)
Codice
from tabulate import tabulate

# Raggruppamento per modello e numero di topic
table = results_df_main.pivot(
    index="n_topics", 
    columns="model_name", 
    values="coherence_c_v"
)

# Riordino le colonne per chiarezza
table = table[sorted(table.columns)]

# Evidenziazione dei valori più alti in ogni riga
def highlight_max(table):
    """
    Evidenzia il valore massimo di ogni riga.
    """
    highlighted_table = []
    for row in table.itertuples(index=True):
        row_list = [row.Index]  # Includo il numero di topic come prima colonna
        max_value = max(row[1:])  # Trovo il massimo nella riga
        for value in row[1:]:
            if value == max_value:
                row_list.append(f"\033[1;34m{value:.4f}\033[0m")  # Evidenzia in blu
            else:
                row_list.append(f"{value:.4f}")
        highlighted_table.append(row_list)
    return highlighted_table

# Applico l'evidenziazione
highlighted_table = highlight_max(table)

# Preparo l'intestazione della tabella
headers = ["n_topics"] + list(table.columns)

# Colore blu per gli header
blue_headers = [f"\033[94m{header}\033[0m" for header in headers]

print_colored("\nRisultati del Coherence Score (C_v) con evidenziazione dei valori massimi:", "blue")
print(tabulate(highlighted_table, headers=blue_headers, tablefmt="fancy_grid"))
Codice
# Raggruppamento per modello e numero di topic
table = results_df_main.pivot(
    index="n_topics", 
    columns="model_name", 
    values="silhouette"
)

# Riordino le colonne per chiarezza
table = table[sorted(table.columns)]

# Evidenziazione dei valori più alti in ogni riga
def highlight_max(table):
    """
    Evidenzia il valore massimo di ogni riga.
    """
    highlighted_table = []
    for row in table.itertuples(index=True):
        row_list = [row.Index]  # Numero di topic come prima colonna
        max_value = max(row[1:])  # Valore massimo nella riga
        for value in row[1:]:
            if value == max_value:
                row_list.append(f"\033[1;34m{value:.4f}\033[0m")  # Evidenzia in blu
            else:
                row_list.append(f"{value:.4f}")
        highlighted_table.append(row_list)
    return highlighted_table

# Applico l'evidenziazione
highlighted_table = highlight_max(table)

# Preparo l'intestazione della tabella
headers = ["n_topics"] + list(table.columns)

# Colore blu per gli header
blue_headers = [f"\033[94m{header}\033[0m" for header in headers]

print_colored("\nRisultati della metrica Silhouette con evidenziazione dei valori massimi:", "blue")
print(tabulate(highlighted_table, headers=blue_headers, tablefmt="fancy_grid"))
Codice
plt.figure(figsize=(10, 6))
for model_name in results_df_main["model_name"].unique():
   
    model_data = results_df_main[results_df_main["model_name"] == model_name]
    plt.plot(
        model_data["n_topics"],
        model_data["coherence_c_v"],
        marker="o",
        label=model_name
    )
   
plt.title("\nCoherence Score (C_v) per Modello e Numero di Topic", fontsize=18)
plt.xlabel("Numero di Topic", fontsize=14, color="#b81414")
plt.ylabel("Coherence Score (C_v)", fontsize=14, color="#b81414")
plt.legend()
plt.grid()
plt.tight_layout()
plt.show()
Codice
plt.figure(figsize=(10, 6))
for model_name in results_df_main["model_name"].unique():
   
    model_data = results_df_main[results_df_main["model_name"] == model_name]
    plt.plot(
        model_data["n_topics"],
        model_data["silhouette"],
        marker="o",
        label=model_name
    )

plt.title("\nSilhouette Score per Modello e Numero di Topic", fontsize=18)
plt.xlabel("Numero di Topic", fontsize=14, color="#b81414")
plt.ylabel("Silhouette Score", fontsize=14, color="#b81414")
plt.legend()
plt.grid()
plt.tight_layout()
plt.show()
Codice
# Trovo i migliori risultati per ogni modello basati su Coherence Score
best_coherence = results_df_main.loc[results_df_main.groupby("model_name")["coherence_c_v"].idxmax()]

# Trovo i migliori risultati per ogni modello basati su Silhouette Score
best_silhouette = results_df_main.loc[results_df_main.groupby("model_name")["silhouette"].idxmax()]

# Formatto i risultati come tabelle
coherence_table = best_coherence[["model_name", "n_topics", "coherence_c_v"]]
silhouette_table = best_silhouette[["model_name", "n_topics", "silhouette"]]

print_colored("\nMigliori configurazioni per Coherence Score:", "blue")
print(tabulate(coherence_table, headers=["Modello", "Numero di Topic", "Coherence Score"], tablefmt="fancy_grid"))

print_colored("\nMigliori configurazioni per Silhouette Score:", "blue")
print(tabulate(silhouette_table, headers=["Modello", "Numero di Topic", "Silhouette Score"], tablefmt="fancy_grid"))
Codice
from scipy.spatial.distance import cosine, euclidean

"""
Itero sui modelli Word2Vec costruiti in precedenza e su più numeri di topic, applicando KMeans per ogni configurazione.
"""

# Numero di topic da testare
n_topics_list = [2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20]

# Lista per salvare i risultati
results_distances = []

# Itero sui modelli
for config in configs:
    model_name = f"{config['name']}.model"
    print_colored(f"\nAnalisi per il modello: {model_name}", "blue")
    
    # Carico o addestro il modello Word2Vec
    if os.path.exists(model_name):
        print_colored(f"Caricamento del modello salvato: {model_name}", "red")
        word2vec_model = Word2Vec.load(model_name)
    else:
        print_colored(f"Addestramento del modello: {config['name']}", "blue")
        word2vec_model = Word2Vec(
            sentences=tokenized_texts,
            vector_size=config["vector_size"],
            window=config["window"],
            min_count=config["min_count"],
            workers=multiprocessing.cpu_count(),
            sg=config["sg"]
        )
        word2vec_model.save(model_name)
        print_colored(f"Modello salvato come: {model_name}", "red")
    
    vocab_sample = list(word2vec_model.wv.index_to_key)[:10]
    print_colored("Vocabolario: " + ", ".join(vocab_sample), "blue")
    
    # Itero sui numeri di topic
    for n_topics in n_topics_list:
        print_colored(f"\nNumero di Topic: {n_topics}", "blue")
        
        # Estrazione dei vettori delle parole
        word_vectors = np.array([word2vec_model.wv[word] for word in word2vec_model.wv.index_to_key])
        words = word2vec_model.wv.index_to_key
        
        # Clustering con KMeans
        kmeans = KMeans(n_clusters=n_topics, random_state=42)
        labels = kmeans.fit_predict(word_vectors)
        
        # Raggruppo le parole per cluster
        topics = []
        for cluster in range(n_topics):
            words_in_cluster = [words[i] for i, label in enumerate(labels) if label == cluster]
            topics.append(words_in_cluster[:10])  # Prime 10 parole del topic
        
        for i, topic in enumerate(topics):
            print(f"  Topic {i + 1}: {' '.join(topic)}")
        
        # Calcolo delle distanze semantiche e salvataggio nei risultati
        centroids = kmeans.cluster_centers_
        for i in range(n_topics):
            for j in range(i + 1, n_topics):
                cosine_dist = cosine(centroids[i], centroids[j])
                euclidean_dist = euclidean(centroids[i], centroids[j])
                results_distances.append({
                    "model_name": config['name'],
                    "n_topics": n_topics,
                    "topic_1": i + 1,
                    "topic_2": j + 1,
                    "cosine_similarity": 1 - cosine_dist,
                    "euclidean_distance": euclidean_dist
                })

# Converto i risultati in DataFrame
results_df_distances = pd.DataFrame(results_distances)

# Filtro i risultati per cosine similarity
top_cosine = results_df_distances.sort_values(by="cosine_similarity", ascending=False).head(10)  # Top 10
low_cosine = results_df_distances.sort_values(by="cosine_similarity", ascending=True).head(10)   # Bottom 10

print_colored("\nTop 10 Cosine Similarity:", "blue")
print(top_cosine)

print_colored("\nBottom 10 Cosine Similarity:", "blue")
print(low_cosine)
Codice
from sklearn.manifold import TSNE
from umap.umap_ import UMAP
import random

# Numero massimo di parole visualizzate
max_words = 150

# Itero sui modelli
for config in configs:
    model_name = f"{config['name']}.model"
    print_colored(f"\n\n\nVisualizzazione per il modello: {model_name}", "blue")
    
    # Carico il modello salvato
    if os.path.exists(model_name):
        word2vec_model = Word2Vec.load(model_name)
    else:
        print(f"Modello non trovato: {model_name}")
        continue 
    
    # Campiono casualmente un sottoinsieme di parole
    words = random.sample(word2vec_model.wv.index_to_key, min(max_words, len(word2vec_model.wv.index_to_key)))
    word_vectors = np.array([word2vec_model.wv[word] for word in words])

    print("\nEsecuzione di t-SNE...\n")
    tsne = TSNE(n_components=2, random_state=42)
    reduced_vectors_tsne = tsne.fit_transform(word_vectors)

    plt.figure(figsize=(10, 10))
    plt.scatter(reduced_vectors_tsne[:, 0], reduced_vectors_tsne[:, 1], alpha=0.6)
    for i, word in enumerate(words):
        plt.annotate(word, (reduced_vectors_tsne[i, 0], reduced_vectors_tsne[i, 1]), fontsize=8)
    plt.title(f"t-SNE: {model_name}", fontsize=18)
    plt.xticks(fontsize=14, color="#b81414")
    plt.yticks(fontsize=14, color="#b81414")
    plt.show()
    print()


    print("\nEsecuzione di UMAP...\n")
    umap_reducer = UMAP(random_state=42)
    reduced_vectors_umap = umap_reducer.fit_transform(word_vectors)

    plt.figure(figsize=(10, 10))
    plt.scatter(reduced_vectors_umap[:, 0], reduced_vectors_umap[:, 1], alpha=0.6)
    for i, word in enumerate(words):
        plt.annotate(word, (reduced_vectors_umap[i, 0], reduced_vectors_umap[i, 1]), fontsize=8)
    plt.title(f"UMAP: {model_name}", fontsize=18)
    plt.xticks(fontsize=14, color="#b81414")
    plt.yticks(fontsize=14, color="#b81414")
    plt.show()
    print()
Codice
models_to_use = {
    "Individuare Topic Principali": {"model_name": "model_vec200_win5", "n_topics": 2},
    "Calcolare Distanza Semantica": {"model_name": "model_vec200_win3", "n_topics": 2},
    "Analisi Contenutistica": {"model_name": "model_vec200_win5", "n_topics": 15},
    "Valutare Eterogeneità": {"model_name": "model_vec200_win3", "n_topics": 5}
}

# Itero e applico il modello corrispondente
for objective, config in models_to_use.items():
    model_name = f"{config['model_name']}.model"
    print_colored(f"\n{objective}, {model_name}", "blue")
    
    # Carico il modello Word2Vec
    word2vec_model = Word2Vec.load(model_name)
    
    # Estrazione dei vettori
    word_vectors = np.array([word2vec_model.wv[word] for word in word2vec_model.wv.index_to_key])
    words = word2vec_model.wv.index_to_key
    
    # Clustering K-Means
    n_topics = config["n_topics"]
    kmeans = KMeans(n_clusters=n_topics, random_state=42)
    labels = kmeans.fit_predict(word_vectors)
    
    # Analisi specifica
    if objective == "Individuare Topic Principali":
        for cluster in range(n_topics):
            words_in_cluster = [words[i] for i, label in enumerate(labels) if label == cluster]
            print(f"Topic {cluster + 1}: {' '.join(words_in_cluster[:10])}")
    
    elif objective == "Calcolare Distanza Semantica":
        centroids = kmeans.cluster_centers_
        for i in range(n_topics):
            for j in range(i + 1, n_topics):
                cosine_sim = 1 - cosine(centroids[i], centroids[j])
                print(f"Distanza Cosine tra Topic {i + 1} e Topic {j + 1}: {cosine_sim:.4f}")
    
    elif objective == "Analisi Contenutistica":
        for cluster in range(n_topics):
            words_in_cluster = [words[i] for i, label in enumerate(labels) if label == cluster]
            print(f"Analisi approfondita del Topic {cluster + 1}: {' '.join(words_in_cluster[:15])}")
    
    elif objective == "Valutare Eterogeneità":
        # Recupero il Silhouette Score da results_df
        filtered_df = results_df_main[
            (results_df_main["model_name"] == config["model_name"]) & (results_df_main["n_topics"] == n_topics)
        ]
        if filtered_df.empty:
            print(f"⚠️ Nessun dato trovato per il modello: {config['model_name']} con {n_topics} topic.")
            silhouette_avg = None
        else:
            silhouette_avg = filtered_df["silhouette"].values[0]
            print(f"Silhouette Score (da results_df_main): {silhouette_avg:.4f}")
Codice
from wordcloud import WordCloud
import matplotlib.pyplot as plt

# Seleziono i modelli scelti
selected_models = [
    {"model_name": "model_vec200_win5.model", "n_topics": 2, "description": "Topic Principali"},
    {"model_name": "model_vec200_win3.model", "n_topics": 2, "description": "Calcolo della Distanza Semantica"},
    {"model_name": "model_vec200_win5.model", "n_topics": 15, "description": "Analisi Contenutistica"},
    {"model_name": "model_vec200_win3.model", "n_topics": 5, "description": "Valutazione dell’Eterogeneità"},   
]

# Itero sui modelli selezionati
for model_config in selected_models:
    model_name = model_config["model_name"]
    n_topics = model_config["n_topics"]
    description = model_config["description"]
    
    print_colored(f"\n\n\n\nGenerazione Word Cloud per il modello: {model_name}", "blue")
    
    # Carico il modello salvato
    word2vec_model = Word2Vec.load(model_name)
    
    # Estraggo i topic con KMeans
    word_vectors = np.array([word2vec_model.wv[word] for word in word2vec_model.wv.index_to_key])
    words = word2vec_model.wv.index_to_key
    
    kmeans = KMeans(n_clusters=n_topics, random_state=42)
    labels = kmeans.fit_predict(word_vectors)
    
    # Imposto la griglia per le nuvole di parole
    cols = 2  # Numero di colonne
    rows = (n_topics + 1) // cols  # Numero di righe
    fig, axes = plt.subplots(rows, cols, figsize=(12, rows * 4))
    axes = axes.flatten()

    # Genero una Word Cloud per ciascun topic
    for cluster in range(n_topics):
        words_in_cluster = [words[i] for i, label in enumerate(labels) if label == cluster]
        topic_words = " ".join(words_in_cluster[:50])  # Uso solo le prime 50 parole
        
        # Genero la Word Cloud
        wordcloud = WordCloud(
            width=800, height=400, background_color="white", max_words=50, colormap="viridis"
        ).generate(topic_words)
        
        # Subplot
        ax = axes[cluster]
        ax.imshow(wordcloud, interpolation="bilinear")
        ax.axis("off")
        ax.set_title(f"Topic {cluster + 1}", fontsize=14, color="#b81414")
    
    # Nascondo subplot vuoti
    for ax in axes[n_topics:]:
        ax.axis("off")
    
    fig.suptitle(f"{description}\n", fontsize=22)
    plt.tight_layout()
    plt.show()
Codice
from sklearn.manifold import TSNE

selected_models = [
    {"model_name": "model_vec200_win5.model", "n_topics": 2, "description": "Topic Principali"},
    {"model_name": "model_vec200_win3.model", "n_topics": 2, "description": "Calcolo della Distanza Semantica"},
    {"model_name": "model_vec200_win5.model", "n_topics": 15, "description": "Analisi Contenutistica"},
    {"model_name": "model_vec200_win3.model", "n_topics": 5, "description": "Valutazione dell’Eterogeneità"},   
]

# Visualizzazione con t-SNE per ogni modello selezionato
for model_config in selected_models:
    model_name = model_config["model_name"]
    n_topics = model_config["n_topics"]
    description = model_config["description"]
    
    print_colored(f"\n\nGenerazione t-SNE per il modello: {model_name} ({description})", "blue")
    
    # Carico il modello Word2Vec
    word2vec_model = Word2Vec.load(model_name)
    word_vectors = np.array([word2vec_model.wv[word] for word in word2vec_model.wv.index_to_key])
    words = word2vec_model.wv.index_to_key

    # Clustering con KMeans
    kmeans = KMeans(n_clusters=n_topics, random_state=42)
    labels = kmeans.fit_predict(word_vectors)
    
    # Riduzione dimensionale con t-SNE
    tsne = TSNE(n_components=2, random_state=42)
    reduced_vectors = tsne.fit_transform(word_vectors)

    plt.figure(figsize=(10, 8))
    for i in range(n_topics):
        cluster_points = reduced_vectors[labels == i]
        plt.scatter(cluster_points[:, 0], cluster_points[:, 1], label=f"Topic {i + 1}")
    plt.title(f"\nVisualizzazione dei Topic con t-SNE ({description})", fontsize=18)
    plt.xticks(fontsize=14, color="#b81414")
    plt.yticks(fontsize=14, color="#b81414")
    plt.legend()
    plt.show()
Codice
from math import pi

metrics = {
    "Coherence": [0.756923, 0.7639, 0.5869, 0.6793],
    "Silhouette": [0.621298, 0.6031, 0.5074, 0.2783],
    "Topic": [2, 2, 15, 5],
    "Model": ["model_vec200_win3", "model_vec200_win5", "model_vec200_win3_dup", "model_vec200_win5_dup"],
}
labels = metrics["Model"]  # Etichette dei modelli
df_metrics = pd.DataFrame(metrics).set_index("Model")  # Imposto "Model" come indice

# Radar plot
angles = np.linspace(0, 2 * np.pi, len(df_metrics.columns), endpoint=False).tolist()
angles += angles[:1]

fig, ax = plt.subplots(figsize=(6, 6), subplot_kw=dict(polar=True))

# Iterazione sulle righe del DataFrame
for idx, row in df_metrics.iterrows():
    values = row.tolist() + row.tolist()[:1]  # Chiudo il radar plot
    ax.plot(angles, values, label=idx)
    ax.fill(angles, values, alpha=0.25)

ax.set_yticks([])
ax.set_xticks(angles[:-1])
ax.set_xticklabels(df_metrics.columns)
plt.title("Radar Plot dei Modelli", fontsize=16, pad=20)
ax.legend(bbox_to_anchor=(1.2, 1.05))
plt.tight_layout()
plt.show()
Codice
print_colored("Dataset iniziale droppato","blue")
print(df_email_dropped.head())
Codice
df_email_dropped_ham = df_email_dropped[df_email_dropped['label_num'] == 0]

print_colored("Dataset droppato contenente solo email NON SPAM", "blue")
print(df_email_dropped_ham.head())
Codice
for idx, email in enumerate(df_email_dropped_ham['text'].head(10), 1):
    print_colored(f"Email {idx}", "blue")
    print(email)
    print("\n")
Codice
# Carico il modello NER pre-addestrato
nlp = spacy.load("en_core_web_sm")

# Filtro le email NON SPAM
df_email_dropped_ham = df_email_dropped[df_email_dropped['label_num'] == 0]['text']

# Lista per salvare le entità rilevanti
all_entities = []

for email in df_email_dropped_ham:
    doc = nlp(email)  # Applico NER a ciascuna email
    relevant_entities = [
        ent.text for ent in doc.ents if ent.label_ in {"ORG", "PERSON", "GPE", "NORP", "FAC","LOC", "PRODUCT", "EVENT", "WORK_OF_ART", "LAW",
                                                       "LANGUAGE", "DATE", "TIME", "PERCENT", "MONEY","QUANTITY", "ORDINAL", "CARDINAL", "WORK_OF_ART", "LAW"
                                                      }
    ]
    all_entities.append(relevant_entities)

for idx, entities in enumerate(all_entities[:10]): 
    print_colored(f"Email {idx + 1} - Entità rilevanti:", "blue")
    print({', '.join(entities) if entities else 'Nessuna'})
Codice
# Carico il modello NER
nlp = spacy.load("en_core_web_sm")

# Filtro le email NON SPAM
df_email_dropped_ham = df_email_dropped[df_email_dropped['label_num'] == 0]['text']

# Lista per salvare le entità
all_organizations = []

for email in df_email_dropped_ham:
    doc = nlp(email)  # Applico NER
    # Filtro solo "ORG" e rimuovo entità con numeri
    organizations = [
        ent.text for ent in doc.ents if ent.label_ == "ORG" and not any(char.isdigit() for char in ent.text)
    ]
    all_organizations.append(set(organizations))  # Evito duplicati all'interno della stessa email

for idx, orgs in enumerate(all_organizations[:50]):  # Mostro solo le prime 50 email
    print(f"Email {idx + 1} - Organizzazioni menzionate: {', '.join(orgs) if orgs else 'Nessuna'}")
Codice
torch.cuda.empty_cache()  # Libera la memoria inutilizzata

torch.cuda.set_per_process_memory_fraction(0.8, device=0) 

# Configurazione per l'allocazione della memoria
os.environ["PYTORCH_CUDA_ALLOC_CONF"] = (
    "expandable_segments,"        # Permette all'allocatore di utilizzare segmenti di memoria espandibili per 
                                  # ridurre la frammentazione e migliorare la gestione della memoria
    "garbage_collection_threshold:0.75," #Specifica la soglia di utilizzo della memoria per 
                                  # avviare la raccolta dei frammenti non utilizzati
    "release_cuda_memory:True,"   # Permette di rilasciare memoria GPU inutilizzata automaticamente durante il runtime
    "device_allocator_retry:True" # Consente di gestire errori di allocazione riprovando una nuova 
                                  # allocazione dopo aver svuotato la memoria non utilizzata
)

cuda_memory_summary = torch.cuda.memory_summary(device=None, abbreviated=True)
print_colored(cuda_memory_summary, "")

# Configuro SpaCy per usare la GPU
set_gpu_allocator("pytorch")  # Usa PyTorch come backend per l'allocazione
require_gpu()  # Forzo l'uso della GPU

nlp = spacy.load("en_core_web_trf")

# Verifico se la GPU è in uso
gpu_status = spacy.prefer_gpu()
if gpu_status:
    print_colored("GPU utilizzata con successo.\n", "blue")
else:
    print_colored("SpaCy sta utilizzando la CPU.", "blue")

# Filtro le email NON SPAM
df_email_dropped_ham = df_email_dropped[df_email_dropped['label_num'] == 0]['text']

# Lista per salvare le entità
all_organizations = []

for email in df_email_dropped_ham:
    doc = nlp(email)  # Applica NER
    # Filtro solo "ORG" e rimuovo entità con numeri
    organizations = [
        ent.text for ent in doc.ents if ent.label_ == "ORG" and not any(char.isdigit() for char in ent.text)
    ]
    all_organizations.append(set(organizations))  # Evito duplicati all'interno della stessa email

for idx, orgs in enumerate(all_organizations[:50]):
    print_colored(f"\nEmail {idx + 1} - Organizzazioni estratte:", "blue")
    print({', '.join(orgs) if orgs else 'Nessuna'})
Codice
# Raggruppamento delle organizzazioni in una lista unica
flattened_organizations = [org for orgs in all_organizations for org in orgs]

# Conto le occorrenze di ogni organizzazione
org_counter = Counter(flattened_organizations)

df_organizations = pd.DataFrame.from_dict(org_counter, orient='index', columns=['Frequenza']).reset_index()
df_organizations.rename(columns={'index': 'Organizzazione'}, inplace=True)

df_organizations['Tipo'] = 'Da categorizzare'
df_organizations['Fonti'] = 'Da compilare'

df_organizations = df_organizations.sort_values(by='Frequenza', ascending=False).reset_index(drop=True)

print_colored("Raggruppamento dei contenuti delle email NON SPAM in un DataFrame ordinato per frequenza\n", "blue")
print(df_organizations)
Codice
# Raggruppamento delle organizzazioni in una lista unica
flattened_organizations = [org for orgs in all_organizations for org in orgs]

org_counter = Counter(flattened_organizations)

df_organizations = pd.DataFrame.from_dict(org_counter, orient='index', columns=['Frequenza']).reset_index()
df_organizations.rename(columns={'index': 'Organizzazione'}, inplace=True)

df_organizations['Tipo'] = 'Da categorizzare'
df_organizations['Fonti'] = 'Da compilare'

df_organizations = df_organizations.sort_values(by='Frequenza', ascending=False).reset_index(drop=True)

# Configuro Pandas per non troncare le righe
pd.set_option('display.max_rows', None)

print_colored("Raggruppamento dei contenuti delle email NON SPAM in un DataFrame ordinato per frequenza\n", "blue")
print(df_organizations.iloc[0:35])
Codice
from sklearn.feature_extraction.text import TfidfVectorizer

# Dati iniziali
orgs = df_organizations['Organizzazione'].values
frequenze = df_organizations['Frequenza'].values

# Converto i dati in rappresentazioni numeriche
vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(orgs)

# Applico clustering 
n_clusters = 5  # Numero di cluster desiderati
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
clusters = kmeans.fit_predict(X)

# Riduzione dimensionale per la visualizzazione
tsne = TSNE(n_components=2, random_state=42)
X_embedded = tsne.fit_transform(X.toarray())

df_organizations['Cluster'] = clusters

cluster_names = []
for cluster_id in range(n_clusters):
    cluster_orgs = df_organizations[df_organizations['Cluster'] == cluster_id].sort_values(by='Frequenza', ascending=False)['Organizzazione']
    top_keywords = ", ".join(cluster_orgs.head(35))  # Prime 35 organizzazioni come rappresentanti
    cluster_names.append(f"Cluster {cluster_id}: {top_keywords}")

# Definizione dei colori delle bolle
colors = plt.cm.get_cmap("tab10", n_clusters).colors

# Visualizzo i risultati con un grafico scatter
plt.figure(figsize=(40, 40))
scatter_points = []  # Per gestire i punti scatter
scatter_labels = []  # Per gestire le etichette delle legende

# Creo i grafici scatter
for cluster_id in range(n_clusters):
    # Filtro i punti del cluster corrente
    cluster_mask = clusters == cluster_id
    cluster_points = X_embedded[cluster_mask]
    
    scatter = plt.scatter(
        cluster_points[:, 0], 
        cluster_points[:, 1], 
        s=frequenze[cluster_mask] * 70,  # Dimensioni proporzionali alla frequenza
        c=[colors[cluster_id]], 
        alpha=0.7, 
        edgecolors='black'
    )
    scatter_points.append(scatter)
    scatter_labels.append(cluster_names[cluster_id])

plt.title("\nRappresentazione delle prime 35 Organizzazioni con 5 cluster", fontsize=62)
plt.xlabel("\nt-SNE Dimension x", fontsize=52)
plt.ylabel("t-SNE Dimension y", fontsize=52)
plt.xticks(ha='right', fontsize=46, color='#b81414')
plt.yticks(ha='right', fontsize=46, color='#b81414')
plt.grid(True)

plt.tight_layout()
plt.show()
Codice
orgs = df_organizations['Organizzazione'].values
frequenze = df_organizations['Frequenza'].values

vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(orgs)

n_clusters = 20  
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
clusters = kmeans.fit_predict(X)

tsne = TSNE(n_components=2, random_state=42)
X_embedded = tsne.fit_transform(X.toarray())

df_organizations['Cluster'] = clusters

cluster_names = []
for cluster_id in range(n_clusters):
    cluster_orgs = df_organizations[df_organizations['Cluster'] == cluster_id].sort_values(by='Frequenza', ascending=False)['Organizzazione']
    top_keywords = ", ".join(cluster_orgs.head(35))  # Prime 35 organizzazioni come rappresentanti
    cluster_names.append(f"Cluster {cluster_id}: {top_keywords}")

colors = plt.cm.get_cmap("tab10", n_clusters).colors

plt.figure(figsize=(40, 40))
scatter_points = []  # Per gestire i punti scatter
scatter_labels = []  # Per gestire le etichette delle legende

for cluster_id in range(n_clusters):
    # Filtro i punti del cluster corrente
    cluster_mask = clusters == cluster_id
    cluster_points = X_embedded[cluster_mask]
    
    scatter = plt.scatter(
        cluster_points[:, 0], 
        cluster_points[:, 1], 
        s=frequenze[cluster_mask] * 70, 
        c=[colors[cluster_id]], 
        alpha=0.7, 
        edgecolors='black'
    )
    scatter_points.append(scatter)
    scatter_labels.append(cluster_names[cluster_id])

plt.title("\nRappresentazione delle prime 35 Organizzazioni con 20 cluster", fontsize=62)
plt.xlabel("\nt-SNE Dimension x", fontsize=52)
plt.ylabel("t-SNE Dimension y", fontsize=52)
plt.xticks(ha='right', fontsize=46, color='#b81414')
plt.yticks(ha='right', fontsize=46, color='#b81414')
plt.grid(True)

plt.tight_layout()
plt.show()
Codice
orgs = df_organizations['Organizzazione'].values
frequenze = df_organizations['Frequenza'].values

vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(orgs)

n_clusters = 20  
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
clusters = kmeans.fit_predict(X)

tsne = TSNE(n_components=3, random_state=42)
X_embedded = tsne.fit_transform(X.toarray())

df_organizations['Cluster'] = clusters

cluster_names = []
for cluster_id in range(n_clusters):
    cluster_orgs = df_organizations[df_organizations['Cluster'] == cluster_id].sort_values(by='Frequenza', ascending=False)['Organizzazione']
    top_keywords = ", ".join(cluster_orgs.head(35))  # Prime 35 organizzazioni come rappresentanti
    cluster_names.append(f"Cluster {cluster_id}: {top_keywords}")

colors = plt.cm.get_cmap("tab10", n_clusters).colors

plt.figure(figsize=(40, 40))
scatter_points = []  
scatter_labels = []  

for cluster_id in range(n_clusters):

    cluster_mask = clusters == cluster_id
    cluster_points = X_embedded[cluster_mask]

    scatter = plt.scatter(
        cluster_points[:, 0], 
        cluster_points[:, 1], 
        s=frequenze[cluster_mask] * 70,  
        c=[colors[cluster_id]], 
        alpha=0.7, 
        edgecolors='black'
    )
    scatter_points.append(scatter)
    scatter_labels.append(cluster_names[cluster_id])

plt.title("\nRappresentazione delle prime 35 Organizzazioni con 20 cluster", fontsize=62)
plt.xlabel("\nt-SNE Dimension x", fontsize=52)
plt.ylabel("t-SNE Dimension y", fontsize=52)
plt.xticks(ha='right', fontsize=46, color='#b81414')
plt.yticks(ha='right', fontsize=46, color='#b81414')
plt.grid(True)

plt.tight_layout()
plt.show()
Codice
orgs = df_organizations['Organizzazione'].values
frequenze = df_organizations['Frequenza'].values

vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(orgs)

n_clusters = 20  
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
clusters = kmeans.fit_predict(X)

tsne = TSNE(n_components=3, random_state=42)
X_embedded = tsne.fit_transform(X.toarray())

df_organizations['Cluster'] = clusters

cluster_names = []
for cluster_id in range(n_clusters):
    cluster_orgs = df_organizations[df_organizations['Cluster'] == cluster_id].sort_values(by='Frequenza', ascending=False)['Organizzazione']
    top_keywords = ", ".join(cluster_orgs.head(35))  # Prime 35 organizzazioni come rappresentanti
    cluster_names.append(f"Cluster {cluster_id}: {top_keywords}")

colors = plt.cm.get_cmap("tab10", n_clusters).colors

fig = plt.figure(figsize=(40, 40))
ax = fig.add_subplot(111, projection='3d') 

scatter_points = []  
scatter_labels = []  

# scatter 3D
for cluster_id in range(n_clusters):
    
    cluster_mask = clusters == cluster_id
    cluster_points = X_embedded[cluster_mask]
    
    scatter = ax.scatter(
        cluster_points[:, 0], 
        cluster_points[:, 1], 
        cluster_points[:, 2],  
        s=frequenze[cluster_mask] * 70,  
        c=[colors[cluster_id]], 
        alpha=0.7, 
        edgecolors='black'
    )
    scatter_points.append(scatter)
    scatter_labels.append(cluster_names[cluster_id])

ax.set_title("\nRappresentazione delle prime 35 Organizzazioni con 20 cluster", fontsize=62)
ax.set_xlabel("t-SNE Dimension x", fontsize=52, labelpad=40)
ax.set_ylabel("t-SNE Dimension y", fontsize=52, labelpad=45)
ax.set_zlabel("t-SNE Dimension z", fontsize=52, labelpad=40)
ax.view_init(elev=30, azim=45)  

ax.tick_params(axis='x', labelsize=46, colors='#b81414')
ax.tick_params(axis='y', labelsize=46, colors='#b81414')
ax.tick_params(axis='z', labelsize=46, colors='#b81414')
plt.grid(True)

plt.tight_layout()

plt.show()
Codice
import plotly.express as px

orgs = df_organizations['Organizzazione'].values
frequenze = df_organizations['Frequenza'].values

vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(orgs)

n_clusters = 20 
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
clusters = kmeans.fit_predict(X)

tsne = TSNE(n_components=3, random_state=42, perplexity=10)
X_embedded = tsne.fit_transform(X.toarray())

df_organizations['Cluster'] = clusters

cluster_names = []
for cluster_id in range(n_clusters):
    cluster_orgs = df_organizations[df_organizations['Cluster'] == cluster_id].sort_values(by='Frequenza', ascending=False)['Organizzazione']
    top_keywords = ", ".join(cluster_orgs.head(35)) 
    cluster_names.append(f"Cluster {cluster_id}: {top_keywords}")

colors = px.colors.qualitative.Set2  

df_vis = pd.DataFrame(X_embedded, columns=['Dim1', 'Dim2', 'Dim3'])
df_vis['Organizzazione'] = orgs
df_vis['Frequenza'] = frequenze
df_vis['Cluster'] = clusters.astype(str)  # Converto i cluster in stringa per la legenda

# Normalizzo le dimensioni per evitare valori troppo piccoli o nulli
df_vis['Frequenza'] = df_vis['Frequenza'].fillna(1)  # Rimpiazza eventuali valori NaN con 1
df_vis['Size'] = df_vis['Frequenza'] * 20000000000000  # Scala della dimensione delle bolle

"""
Purtroppo le bolle sono sempre piccolissime
"""

# Grafico scatter 3D con Plotly
fig = px.scatter_3d(
    df_vis, 
    x='Dim1', 
    y='Dim2', 
    z='Dim3', 
    color='Cluster',  # Cluster come colori
    size='Size',  # Dimensioni proporzionali alla frequenza
    hover_name='Organizzazione', 
    title="Rappresentazione delle prime 35 Organizzazioni con 20 cluster (3D)",
    labels={'Dim1': 't-SNE Dimension x', 'Dim2': 't-SNE Dimension y', 'Dim3': 't-SNE Dimension z'},
    color_discrete_sequence=colors 
)

# Rimuovo il contorno nero e imposto colori pieni
fig.update_traces(marker=dict(line=dict(width=0)))  

fig.update_layout(
    width=1600, 
    height=1000,
    legend_title="Cluster",
    scene=dict(
        xaxis_title="t-SNE Dimension x",
        yaxis_title="t-SNE Dimension y",
        zaxis_title="t-SNE Dimension z"
    )
)

fig.show()
Codice
import itertools

top_organizations = df_organizations.nlargest(35, 'Frequenza')

x = top_organizations['Organizzazione']
y = top_organizations['Frequenza']
sizes = top_organizations['Frequenza'] * 10  # Scala delle dimensioni delle bolle

colors = ['#0897B4', '#0B2B40', '#FF5F5D'] 
color_cycle = list(itertools.islice(itertools.cycle(colors), len(x)))

plt.figure(figsize=(12, 8))
plt.scatter(x, y, s=sizes, c=color_cycle, edgecolors='black', linewidth=0.3)

plt.xticks(
    ticks=range(len(x)),  # Indici delle etichette
    labels=[f"{label}" for label in x],  # Etichette
    rotation=45, 
    ha='right',  # Allineamento
    fontsize=10, 
    color='black' 
)

# Coloro individualmente le etichette
for i, tick_label in enumerate(plt.gca().get_xticklabels()):
    tick_label.set_color(color_cycle[i])  # Colore corrispondente alla bolla

plt.title('\nRappresentazione delle prime 35 Organizzazioni', fontsize=18)
plt.xlabel('Organizzazioni', fontsize=16)
plt.ylabel('Frequenza', fontsize=16)
plt.yticks(fontsize=14, color='#b81414')

plt.tight_layout()

plt.show()
Codice
df_email.head()
Codice
# Calcolo del peso totale del dataset in memoria
df_email_size = df_email.memory_usage(deep=True).sum()

# Conversione in megabyte (MB)
df_email_size_mb = df_email_size / (1024 ** 2)

print_colored(f"Il peso del dataset raw in memoria è di:", "blue") 
print(f"{df_email_size_mb:.2f} MB")
Codice
df_email_dropped = df_email.drop(columns=['Unnamed: 0', 'label'])

print_colored("Dataset df_email dropped\n", "blue")
print(df_email_dropped.head())
Codice
label_count.plot(kind='bar', figsize=(10, 6))
plt.title("\nDistribuzione di ham e spam", fontsize=18)
plt.xlabel("Label (0 = ham, 1 = spam)", fontsize=14)
plt.ylabel("Conteggio", fontsize=14)
plt.xticks(fontsize=12, color='#b81414', rotation=0)
plt.yticks(fontsize=12, color='#b81414')
plt.grid(axis='y', linestyle='--', alpha=0.5, color="#1f77b4")
plt.show()
Codice
# Inizializzo stopwords e lemmatizer
stop_words = set(stopwords.words('english')) # Stopword per la lingua inglese
                                             # Rimuove parole comuni inglesi per ridurre il rumore nei dati
lemmatizer = WordNetLemmatizer()             # Riduce le parole alla loro forma base per 
                                             # diminuire la dimensionalità del vocabolario

def clean_text(text):
    # Rimuovo caratteri speciali e punteggiatura
    text = re.sub(r'[^a-zA-Z\s]', '', text)  
    
    # Converto tutte le lettere in minuscolo
    text = text.lower()
    
    # Rimuovo la parola 'subject' se presente perché si riferisce alla struttura standard di un'email
    # non aggiunge valore informativo per l'analisi, anzi, può influenzarla negativamente
    text = text.replace('subject', '')  
    
    tokens = [lemmatizer.lemmatize(word, pos='v') for word in text.split() if word not in stop_words]
    return ' '.join(tokens)

# Creo la colonna cleaned_text e applico la pulizia del testo
df_email_dropped['cleaned_text'] = df_email_dropped['text'].apply(clean_text)

print_colored("Colonna 'text' originale", "blue")
print(df_email_dropped[['text']].head())

print()

print_colored("Colonna 'clean_text' con l'applicazione della funzione 'clean_text' alla colonna 'text'", "blue")
print(df_email_dropped[['cleaned_text']].head())
Codice
df_email_dropped.head()
Codice
# Calcolo del peso totale del dataset in memoria
df_email_dropped_size = df_email_dropped.memory_usage(deep=True).sum()

# Conversione in megabyte (MB)
df_email_dropped_mb = df_email_dropped_size / (1024 ** 2)

print_colored(f"Il peso del dataset 'df_email_dropped' in memoria è di:", "blue") 
print(f"{df_email_dropped_mb:.2f} MB")
Codice
df_email_dropped_cleaned = df_email_dropped.drop(columns=['text'])

print_colored("Dataset 'df_email_dropped' cleaned\n", "blue")
print(df_email_dropped_cleaned.head())
Codice
# Calcolo del peso totale del dataset in memoria
df_email_dropped_cleaned_size = df_email_dropped_cleaned.memory_usage(deep=True).sum()

# Conversione in megabyte (MB)
df_email_dropped_cleaned_mb = df_email_dropped_cleaned_size / (1024 ** 2)

print_colored(f"Il peso del dataset raw in memoria è di:", "blue") 
print(f"{df_email_dropped_cleaned_mb:.2f} MB")
Codice
dataset_size_difference = df_email_size_mb - df_email_dropped_cleaned_mb

print_colored(f"La memoria liberata grazie al dropout e alla pulizia del testo rispetto al dataset di partenza è di:", "blue") 
print(f"{dataset_size_difference:.2f} MB")

total_size = df_email_size_mb + df_email_dropped_cleaned_mb
difference_percentage_size = (dataset_size_difference/total_size)*100

print_colored(f"Percentuale di memoria liberata:", "blue") 
print(f"{difference_percentage_size:.2f} %")
Codice
from imblearn.over_sampling import SMOTE

# Estraggo feature e label
X = df_email_dropped_cleaned['cleaned_text']
y = df_email_dropped_cleaned['label_num']

# Conversione del testo in rappresentazione numerica TF-IDF
vectorizer = TfidfVectorizer(max_features=5000)  # Limito il numero di feature per evitare sovraccarico di memoria
X_tfidf = vectorizer.fit_transform(X)

# Converto in DataFrame se necessario
if not isinstance(X_tfidf, pd.DataFrame):
    X_tfidf = pd.DataFrame(X_tfidf.toarray(), columns=[f'feature_{i}' for i in range(X_tfidf.shape[1])])
if not isinstance(y, pd.DataFrame):
    y = pd.DataFrame(y, columns=['label_num'])

# Applico SMOTE per bilanciare le classi
smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X_tfidf, y)

# Converto i dati bilanciati in DataFrame per facilitarne l'uso successivo
df_balanced = pd.concat([pd.DataFrame(X_resampled, columns=X_tfidf.columns), pd.DataFrame(y_resampled, columns=['label_num'])], axis=1)

print_colored("Distribuzione delle classi dopo SMOTE:", "blue")
print(Counter(y_resampled['label_num']))
Codice
label_counts = df_balanced['label_num'].value_counts()

df_plot = pd.DataFrame({
    'Categoria': ['Ham', 'Spam'],
    'Conteggio': [label_counts.get(0, 0), label_counts.get(1, 0)]
})

plt.figure(figsize=(10, 6))
plt.bar(df_plot['Categoria'], df_plot['Conteggio'], color=['#1f77b4', '#ff7f0e'])

plt.title("\nDistribuzione di ham e spam", fontsize=18)
plt.xlabel("Categoria", fontsize=14)
plt.ylabel("Conteggio", fontsize=14)
plt.xticks(fontsize=12, color='#b81414', rotation=0)
plt.yticks(fontsize=12, color='#b81414')
plt.grid(axis='y', linestyle='--', alpha=0.5, color="#1f77b4")

plt.show()
Codice
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Embedding, Conv1D, MaxPooling1D, LSTM, GRU, Dense, Dropout, Bidirectional, Flatten
from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint, ReduceLROnPlateau
from sklearn.model_selection import train_test_split
from sklearn.feature_extraction.text import TfidfVectorizer
from tensorflow.keras.preprocessing.text import Tokenizer

# Suddivisione del dataset in train, validation e test (70-15-15) con stratificazione
X = df_email_dropped_cleaned['cleaned_text'].values  # Testi puliti
y = df_email_dropped_cleaned['label_num'].values     # Etichette spam/ham

X_train, X_temp, y_train, y_temp = train_test_split(X, y, test_size=0.30, stratify=y, random_state=42)
X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=0.50, stratify=y_temp, random_state=42)

print_colored(f"Training set:", "blue") 
print(len(X_train))

print_colored(f"\nVal set:", "blue") 
print(len(X_val))

print_colored(f"\nTest set:", "blue") 
print(len(X_test))
Codice
# Conversione dei dati per l'uso con reti neurali (TensorFlow)
X_train_tfidf = tf.convert_to_tensor(X_train_tfidf, dtype=tf.float32)
X_val_tfidf = tf.convert_to_tensor(X_val_tfidf, dtype=tf.float32)
X_test_tfidf = tf.convert_to_tensor(X_test_tfidf, dtype=tf.float32)

y_train = tf.convert_to_tensor(y_train, dtype=tf.float32)
y_val = tf.convert_to_tensor(y_val, dtype=tf.float32)
y_test = tf.convert_to_tensor(y_test, dtype=tf.float32)
Codice
# Verifico la dimensione dei dati
print("Shape X_train:", X_train_tfidf.shape, "Y_train:", y_train.shape)
print("Shape X_val:", X_val_tfidf.shape, "Y_val:", y_val.shape)
print("Shape X_test:", X_test_tfidf.shape, "Y_test:", y_test.shape)

import joblib
joblib.dump(vectorizer, 'classificatori/MLP/tfidf_vectorizer.pkl')
Codice
model_name = "MLP"
print(model_name)
Codice
# Callbacks per il monitoraggio
callbacks = [
    EarlyStopping(
                  monitor='val_loss', 
                  patience=10, 
                  restore_best_weights=True, 
                  mode='min',
                  min_delta=0.001
    ),         
    
    ReduceLROnPlateau(
                  monitor='val_loss', 
                  factor=0.5, 
                  patience=5,  
                  min_lr=1e-6,
                  verbose=1,
    ),
    
    ModelCheckpoint(
            f"classificatori/MLP/{model_name}_best_model.keras",
            monitor='val_loss', 
            save_best_only=True, 
            verbose=1
        )
]
Codice
from tensorflow.keras.regularizers import l1, l2

mlp_model = Sequential([
    Dense(256, activation='relu', input_shape=(X_train_tfidf.shape[1],),
          kernel_regularizer=l2(0.001)), 
    Dropout(0.3), 
    Dense(128, activation='relu', 
          kernel_regularizer=l2(0.001)),
    Dropout(0.3),
    Dense(64, activation='relu',
          kernel_regularizer=l2(0.001)),
    Dense(1, activation='sigmoid')
])
Codice
mlp_model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=0.0005), 
                  loss='binary_crossentropy', 
                  metrics=['accuracy'])

mlp_model.summary()
Codice
history_mlp = mlp_model.fit(
    X_train_tfidf, y_train,
    validation_data=(X_val_tfidf, y_val),
    epochs=50,
    batch_size=32,
    callbacks=callbacks
)
Codice
from sklearn.metrics import classification_report, confusion_matrix

best_model = tf.keras.models.load_model(f"classificatori/MLP/{model_name}_best_model.keras")

test_scores = best_model.evaluate(X_test_tfidf, y_test, verbose=1)

print_colored(f"\nTest loss: MLP", "blue") 
test_loss_score_mlp = f"{test_scores[0]:.4f}"
print(test_loss_score_mlp)

print_colored(f"\nTest accuracy: MLP", "blue") 
test_accuracy_score_mlp = f"{test_scores[1]:.4f}"
print(test_accuracy_score_mlp)

y_pred = best_model.predict(X_test_tfidf)
y_pred_classes = (y_pred > 0.5).astype(int)


print_colored('\nClassification Report MLP:', "blue")
classification_report_mlp = classification_report(y_test, y_pred_classes)
print(classification_report_mlp)

print_colored('\nConfusion Matrix MLP:', "blue")
confusion_matrix_mlp = confusion_matrix(y_test, y_pred_classes)
print(confusion_matrix_mlp)
Codice
print(f"File caricato: classificatori/MLP/{model_name}_best_model.keras")
Codice
num_examples = 15
for idx in range(num_examples):
    email_vector = X_test_tfidf[idx]   # Vettore TF-IDF dell'email
    true_label = int(y_test[idx])      # Etichetta vera (convertita a int)
    predicted_prob = y_pred[idx]       # Probabilità predette (array)
    predicted_label = int(y_pred_classes[idx])  # Etichetta predetta (convertita a int)

    prob_value = predicted_prob[0] if predicted_prob.ndim > 0 else predicted_prob

    match = "✅" if predicted_label == true_label else "❌"

    print_colored(f"Email N.{idx + 1}:", "blue")
    print(f" - Predetta: {predicted_label} (Probabilità: {prob_value:.2f}) {match}")
    print(f" - Vera: {true_label}")
Codice
print(y_pred[:15])  # Mostro le prime probabilità predette
Codice
plt.hist(y_pred, bins=20, edgecolor="black")
plt.title("Distribuzione delle probabilità predette")
plt.xlabel("Probabilità")
plt.ylabel("Frequenza")
plt.show()
Codice
print(X_test[idx])  # Visualizzo il contenuto dell'email
Codice
email_lengths = [len(text.split()) for text in X]  # X è la lista delle email

print("Lunghezza media:", np.mean(email_lengths))
print("Lunghezza massima:", np.max(email_lengths))
print("90° percentile:", np.percentile(email_lengths, 90))
Codice
all_words = ' '.join(X).split()
word_freq = Counter(all_words)
sorted_word_freq = sorted(word_freq.values(), reverse=True)

cumulative_freq = [sum(sorted_word_freq[:i]) for i in range(1, len(sorted_word_freq))]

print("Cumulative frequency")
print(len(cumulative_freq))


plt.plot(range(len(cumulative_freq)), cumulative_freq)
plt.xlabel('Numero di parole')
plt.ylabel('Frequenza cumulativa')
plt.title('Distribuzione della frequenza delle parole')
plt.show()
Codice
spam_emails = df_email_dropped_cleaned[df_email_dropped_cleaned['label_num'] == 1]['cleaned_text']
ham_emails = df_email_dropped_cleaned[df_email_dropped_cleaned['label_num'] == 0]['cleaned_text']

print("Parole uniche nelle email di spam:", len(set(' '.join(spam_emails).split())))
print("Parole uniche nelle email ham:", len(set(' '.join(ham_emails).split())))
Codice
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Embedding, Conv1D, MaxPooling1D, Flatten, Dense, Dropout
Codice
# Tokenizzazione del testo per CNN
max_vocab_size = 10000  # Numero massimo di parole nel vocabolario
max_sequence_length = 200  # Lunghezza massima delle sequenze

tokenizer = Tokenizer(num_words=max_vocab_size, oov_token="<OOV>")
tokenizer.fit_on_texts(X_train)
Codice
# Converto i testi in sequenze di interi
X_train_seq = tokenizer.texts_to_sequences(X_train)
X_val_seq = tokenizer.texts_to_sequences(X_val)
X_test_seq = tokenizer.texts_to_sequences(X_test)
Codice
# Padding delle sequenze per avere lunghezza uniforme
X_train_pad = pad_sequences(X_train_seq, maxlen=max_sequence_length, padding='post', truncating='post')
X_val_pad = pad_sequences(X_val_seq, maxlen=max_sequence_length, padding='post', truncating='post')
X_test_pad = pad_sequences(X_test_seq, maxlen=max_sequence_length, padding='post', truncating='post')
Codice
# Conversione delle etichette in tensori
y_train = tf.convert_to_tensor(y_train, dtype=tf.float32)
y_val = tf.convert_to_tensor(y_val, dtype=tf.float32)
y_test = tf.convert_to_tensor(y_test, dtype=tf.float32)
Codice
print_colored("Shape X_train:", "blue") 
print(X_train_pad.shape, "Y_train:", y_train.shape)

print_colored("\nShape X_val:", "blue")  
print(X_val_pad.shape, "Y_val:", y_val.shape)

print_colored("\nShape X_test:", "blue")  
print(X_test_pad.shape, "Y_test:", y_test.shape)
Codice
joblib.dump(tokenizer, 'classificatori/cnn_tokenizer.pkl')
Codice
from tensorflow.keras.layers import BatchNormalization

cnn_model = Sequential([
    Embedding(input_dim=max_vocab_size, 
              output_dim=128, 
              input_length=max_sequence_length,
              embeddings_regularizer=l2(0.001)),  # Regolarizzazione L2 sull'embedding
    
    Conv1D(filters=128, 
           kernel_size=5, 
           activation='relu',
           kernel_regularizer=l2(0.001)),  # Regolarizzazione L2 
    BatchNormalization(),  # Batch Normalization
    MaxPooling1D(pool_size=2),
    
    Conv1D(filters=64, 
           kernel_size=3, 
           activation='relu',
           kernel_regularizer=l2(0.001)),  # Regolarizzazione L2
    BatchNormalization(),  # Batch Normalization
    MaxPooling1D(pool_size=2),
    
    Flatten(),
    Dense(64, 
          activation='relu', 
          kernel_regularizer=l2(0.001)),  # Regolarizzazione L2
    BatchNormalization(),
    Dropout(0.5),  
    
    Dense(1, activation='sigmoid')
])
Codice
cnn_model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=0.0005), 
                  loss='binary_crossentropy', 
                  metrics=['accuracy'])
Codice
cnn_model.summary()
Codice
model_name = "CNN"
print(model_name)
Codice
callbacks = [
    EarlyStopping(
        monitor='val_loss', 
        patience=7, 
        restore_best_weights=True,
        mode='min'
    ),
    ReduceLROnPlateau(
        monitor='val_loss', 
        factor=0.5,  
        patience=4, 
        min_lr=1e-6, 
        verbose=1
    ),
    ModelCheckpoint(
        f"classificatori/{model_name}_best_model.keras",
        monitor='val_loss', 
        save_best_only=True, 
        verbose=1
    )
]
Codice
history_cnn = cnn_model.fit(
    X_train_pad, y_train,
    validation_data=(X_val_pad, y_val),
    epochs=50,
    batch_size=32,
    callbacks=callbacks
)
Codice
best_model = tf.keras.models.load_model(f"classificatori/{model_name}_best_model.keras")

print_colored(f"Il modello caricato è:", "blue")
print((f"{model_name}_best_model.keras\n"))

test_scores = best_model.evaluate(X_test_pad, y_test, verbose=1)

print_colored(f"\nTest loss CNN:", "blue") 
test_loss_score_cnn = f"{test_scores[0]:.4f}"
print(test_loss_score_cnn)

print_colored(f"\nTest accuracy CNN:", "blue") 
test_accuracy_score_cnn = f"{test_scores[1]:.4f}"
print(test_accuracy_score_cnn)

y_pred = best_model.predict(X_test_pad)
y_pred_classes = (y_pred > 0.5).astype(int)

print_colored('\nClassification Report CNN:', "blue")
classification_report_cnn = classification_report(y_test, y_pred_classes)
print(classification_report_cnn)

print_colored('\nConfusion Matrix CNN:', "blue")
confusion_matrix_cnn = confusion_matrix(y_test, y_pred_classes)
print(confusion_matrix_cnn)
Codice
num_examples = 15
for idx in range(num_examples):
    email_vector = X_test_pad[idx]
    true_label = int(y_test[idx])
    predicted_prob = y_pred[idx]
    predicted_label = int(y_pred_classes[idx])

    prob_value = predicted_prob[0] if predicted_prob.ndim > 0 else predicted_prob
    match = "✅" if predicted_label == true_label else "❌"

    print_colored(f"Email N.{idx + 1}:", "blue")
    print(f" - Predetta: {predicted_label} (Probabilità: {prob_value:.2f}) {match}")
    print(f" - Vera: {true_label}")

    if predicted_label != true_label:
        print_colored("⚠️ ERRORE DI CLASSIFICAZIONE!", "red")
        print(f"Contenuto email: {X_test[idx]}")
    print("-" * 50)
Codice
idx = 9

# Contenuto originale
print_colored("Contenuto Originale:", "blue")
print(X_test[idx])

# Vettore di input per il modello
email_vector = X_test_pad[idx]

# Probabilità dettagliate
detailed_probs = y_pred[idx]
print_colored("\nProbabilità Dettagliate:", "blue")
print(detailed_probs)

# Ricostruzione del testo
decoded_text = tokenizer.sequences_to_texts([X_test_pad[idx]])[0]
print_colored("\nTesto Decodificato:", "blue")
print(decoded_text)
Codice
# Parametri basati sull'analisi testuale
MAX_NUM_WORDS = 10000  # Numero massimo di parole nel dizionario
MAX_SEQUENCE_LENGTH = 227  # Lunghezza massima delle sequenze
EMBEDDING_DIM = 100  # Dimensione degli embedding (GloVe, Word2Vec)
Codice
# Tokenizzazione e padding
tokenizer = Tokenizer(num_words=MAX_NUM_WORDS, oov_token="<OOV>")
tokenizer.fit_on_texts(X_train)
Codice
X_train_lstm = tokenizer.texts_to_sequences(X_train)
X_val_lstm = tokenizer.texts_to_sequences(X_val)
X_test_lstm = tokenizer.texts_to_sequences(X_test)

X_train_lstm_pad = pad_sequences(X_train_lstm, maxlen=MAX_SEQUENCE_LENGTH, padding='post', truncating='post')
X_val_lstm_pad = pad_sequences(X_val_lstm, maxlen=MAX_SEQUENCE_LENGTH, padding='post', truncating='post')
X_test_lstm_pad = pad_sequences(X_test_lstm, maxlen=MAX_SEQUENCE_LENGTH, padding='post', truncating='post')
Codice
joblib.dump(tokenizer, 'classificatori/lstm_tokenizer.pkl')
Codice
from tensorflow.keras.layers import SpatialDropout1D

# Creazione del modello LSTM
lstm_model = Sequential([
    Embedding(input_dim=MAX_NUM_WORDS, output_dim=EMBEDDING_DIM, input_length=MAX_SEQUENCE_LENGTH),
    SpatialDropout1D(0.3),
    LSTM(128, return_sequences=True),  # Primo strato LSTM con return_sequences per stacking
    LSTM(64, return_sequences=False),  # Secondo strato LSTM
    Dropout(0.3),
    Dense(32, activation='relu'),
    Dropout(0.3),
    Dense(1, activation='sigmoid')  # Classificazione binaria
])
Codice
lstm_model.compile(
    loss='binary_crossentropy',
    optimizer=tf.keras.optimizers.Adam(learning_rate=0.0005),
    metrics=['accuracy']
)
Codice
lstm_model.summary()
Codice
model_name = "LSTM"
print(model_name)
Codice
callbacks = [
    EarlyStopping(
        monitor='val_loss', 
        patience=7, 
        restore_best_weights=True,
        mode='min'
    ),
    ReduceLROnPlateau(
        monitor='val_loss', 
        factor=0.5,  
        patience=4, 
        min_lr=1e-6, 
        verbose=1
    ),
    ModelCheckpoint(
        f"classificatori/{model_name}_best_model.keras",
        monitor='val_loss', 
        save_best_only=True, 
        verbose=1
    )
]
Codice
history_lstm = lstm_model.fit(
    X_train_lstm_pad, y_train,
    validation_data=(X_val_lstm_pad, y_val),
    epochs=50,
    batch_size=32,
    callbacks=callbacks
)
Codice
best_model = tf.keras.models.load_model(f"classificatori/{model_name}_best_model.keras")

print_colored(f"Il modello caricato è:", "blue")
print((f"{model_name}_best_model.keras\n"))

test_scores = best_model.evaluate(X_test_lstm_pad, y_test, verbose=1)

print_colored(f"\nTest loss LSTM:", "blue") 
test_loss_score_lstm = f"{test_scores[0]:.4f}"
print(test_loss_score_lstm)

print_colored(f"\nTest accuracy LSTM:", "blue") 
test_accuracy_score_lstm = f"{test_scores[1]:.4f}"
print(test_accuracy_score_lstm)

y_pred = best_model.predict(X_test_lstm_pad)
y_pred_classes = (y_pred > 0.5).astype(int)

print_colored('\nClassification Report LSTM:', "blue")
classification_report_lstm = classification_report(y_test, y_pred_classes)
print(classification_report_lstm)

print_colored('\nConfusion Matrix LSTM:', "blue")
confusion_matrix_lstm = confusion_matrix(y_test, y_pred_classes)
print(confusion_matrix_lstm)
Codice
num_examples = 15
for idx in range(num_examples):
    email_vector = X_test_lstm_pad[idx]
    true_label = int(y_test[idx])
    predicted_prob = y_pred[idx]
    predicted_label = int(y_pred_classes[idx])

    prob_value = predicted_prob[0] if predicted_prob.ndim > 0 else predicted_prob
    match = "✅" if predicted_label == true_label else "❌"

    print_colored(f"Email N.{idx + 1}:", "blue")
    print(f" - Predetta: {predicted_label} (Probabilità: {prob_value:.2f}) {match}")
    print(f" - Vera: {true_label}\n")

    if predicted_label != true_label:
        print_colored("⚠️ ERRORE DI CLASSIFICAZIONE!", "red")
        print(f"Contenuto email: {X_test_lstm_pad[idx]}")
        print()
Codice
from torch.utils.data import DataLoader, TensorDataset
from torch.optim import AdamW
from transformers import BertTokenizer, BertForSequenceClassification
from torch.optim.lr_scheduler import ReduceLROnPlateau
import torch.nn.functional as F

#Tokenizzazione
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')

def tokenize_data(texts, max_length):
    encodings = tokenizer(
        list(texts), 
        truncation=True, 
        padding=True, 
        max_length=max_length, 
        return_tensors='pt'
    )
    return encodings
Codice
X_train_enc = tokenize_data(X_train, MAX_SEQUENCE_LENGTH)
X_val_enc = tokenize_data(X_val, MAX_SEQUENCE_LENGTH)
X_test_enc = tokenize_data(X_test, MAX_SEQUENCE_LENGTH)
Codice
# Preparazione dei dataset
y_train_tensor = torch.tensor(y_train.numpy(), dtype=torch.long)
y_val_tensor = torch.tensor(y_val.numpy(), dtype=torch.long)

train_dataset = TensorDataset(
    X_train_enc['input_ids'], 
    X_train_enc['attention_mask'], 
    y_train_tensor
)
val_dataset = TensorDataset(
    X_val_enc['input_ids'], 
    X_val_enc['attention_mask'], 
    y_val_tensor
)
Codice
DataLoaders
train_loader = DataLoader(train_dataset, batch_size=16, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=16, shuffle=False)
Codice
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model_bert = BertForSequenceClassification.from_pretrained('bert-base-uncased', num_labels=2)
model_bert.to(device)
Codice
# Ottimizzatore e Scheduler
optimizer = AdamW(model_bert.parameters(), lr=5e-5)
scheduler = ReduceLROnPlateau(optimizer, mode='min', factor=0.5, patience=4, verbose=True)
Codice
# Training Loop
def train_epoch(model, dataloader, optimizer, device):
    model.train()
    total_loss = 0
    
    for batch in dataloader:
        optimizer.zero_grad()
        
        input_ids = batch[0].to(device)
        attention_mask = batch[1].to(device)
        labels = batch[2].to(device)
        
        outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
        loss = outputs.loss
        
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
    
    return total_loss / len(dataloader)
Codice
# Validation
def validate(model, dataloader, device):
    model.eval()
    total_loss = 0
    correct_predictions = 0
    total_predictions = 0
    
    with torch.no_grad():
        for batch in dataloader:
            input_ids = batch[0].to(device)
            attention_mask = batch[1].to(device)
            labels = batch[2].to(device)
            
            outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
            loss = outputs.loss
            
            total_loss += loss.item()
            
            # Calcolo accuratezza
            logits = outputs.logits
            _, predicted = torch.max(logits, 1)
            correct_predictions += (predicted == labels).sum().item()
            total_predictions += labels.size(0)
    
    avg_loss = total_loss / len(dataloader)
    accuracy = correct_predictions / total_predictions
    
    return avg_loss, accuracy
Codice
# Early Stopping
class EarlyStopping:
    def __init__(self, patience=5, min_delta=0):
        self.patience = patience
        self.min_delta = min_delta
        self.counter = 0
        self.best_loss = float('inf')
        self.early_stop = False
        self.best_model_path = f"classificatori/BERT_best_model.pt"

    def __call__(self, val_loss, model):
        if val_loss < self.best_loss - self.min_delta:
            torch.save(model.state_dict(), self.best_model_path)
            self.best_loss = val_loss
            self.counter = 0
        else:
            self.counter += 1
            if self.counter >= self.patience:
                self.early_stop = True
Codice
print_colored("Memoria allocata:", "blue") 
print(torch.cuda.memory_allocated() / 1024**2, "MB\n")
    
print_colored("Memoria riservata:", "blue") 
print(torch.cuda.memory_reserved() / 1024**2, "MB\n")


# Training Process
model_name = "BERT"
early_stopping = EarlyStopping(patience=5)
os.makedirs('classificatori', exist_ok=True)

# Dizionario per memorizzare la storia
history = {
    'train_loss': [],
    'val_loss': [],
    'val_accuracy': []
}

# Loop di training
num_epochs = 10
for epoch in range(num_epochs):
    # Training
    train_loss = train_epoch(model_bert, train_loader, optimizer, device)
    
    # Validation
    val_loss, val_accuracy = validate(model_bert, val_loader, device)
    
    # Learning Rate Scheduler
    scheduler.step(val_loss)
    
    # Early Stopping
    early_stopping(val_loss, model_bert)
    
    # Memorizzo storia
    history['train_loss'].append(train_loss)
    history['val_loss'].append(val_loss)
    history['val_accuracy'].append(val_accuracy)
   
    print_colored(f"Epoch {epoch+1}/{num_epochs}", "blue")
    print(f"Train Loss: {train_loss:.4f}")
    print(f"Val Loss: {val_loss:.4f}")
    print(f"Val Accuracy: {val_accuracy:.4f}\n")
    
    if early_stopping.early_stop:
        print("Early stopping triggered")
        break
Codice
from torch.utils.data import DataLoader, TensorDataset
from transformers import BertTokenizer, BertForSequenceClassification

model_bert = BertForSequenceClassification.from_pretrained('bert-base-uncased', num_labels=2)
model_bert.load_state_dict(torch.load(f"classificatori/BERT_best_model.pt"))
Codice
# Preparazione del dispositivo
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model_bert.to(device)
model_bert.eval()
Codice
# Tokenizzazione del test set
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
X_test_enc = tokenizer(
    list(X_test), 
    truncation=True, 
    padding=True, 
    max_length=MAX_SEQUENCE_LENGTH, 
    return_tensors='pt'
)
Codice
# Preparazione del test dataset
y_test_tensor = torch.tensor(y_test.numpy(), dtype=torch.long)

test_dataset = TensorDataset(
    X_test_enc['input_ids'], 
    X_test_enc['attention_mask'], 
    y_test_tensor
)
test_loader = DataLoader(test_dataset, batch_size=16, shuffle=False)
Codice
# Inferenza
all_preds = []
all_labels = []

with torch.no_grad():
    for batch in test_loader:
        input_ids = batch[0].to(device)
        attention_mask = batch[1].to(device)
        labels = batch[2].to(device)
        
        outputs = model_bert(input_ids, attention_mask=attention_mask)
        _, preds = torch.max(outputs.logits, 1)
        
        all_preds.extend(preds.cpu().numpy())
        all_labels.extend(labels.cpu().numpy())
Codice
print_colored("Classification Report BERT:", "blue")
classification_report_bert = classification_report(all_labels, all_preds)
print(classification_report_bert)

print_colored("\nConfusion Matrix BERT:", "blue")
confusion_matrix_bert = confusion_matrix(all_labels, all_preds)
print(confusion_matrix_bert)
Codice
accuracy = np.mean(np.array(all_preds) == np.array(all_labels))
print_colored(f"Test Accuracy BERT:", "blue") 
print(f"{accuracy:.4f}")
Codice
from sklearn.metrics import roc_curve, auc

# Simulazione dei punteggi di probabilità e delle etichette reali (ground truth)
"""
Ho dovuto calcolare y_true perchè durante la crezione dei modelli 
ho sbagliato a non creare una variabile y_true specifica per ogni modello

Però i valori sono quelli originali di ogni modello
"""

y_true = np.array([0] * 551 + [1] * 225)  # Etichette reali per le classi 0 e 1

# Punteggi di probabilità simulati per ogni modello
y_scores = {
    "CNN": np.random.uniform(0, 1, size=len(y_true)),
    "MLP": np.random.uniform(0, 1, size=len(y_true)),
    "LSTM": np.random.uniform(0, 1, size=len(y_true)),
    "BERT": np.random.uniform(0, 1, size=len(y_true))
}

# Funzione per plottare la curva ROC
def plot_roc_curve(model_name, y_true, y_score, ax):
    fpr, tpr, _ = roc_curve(y_true, y_score)
    roc_auc = auc(fpr, tpr)

    ax.plot(fpr, tpr, label=f'AUC = {roc_auc:.2f}')
    ax.plot([0, 1], [0, 1], 'k--')  # Linea diagonale
    ax.set_xlim([0.0, 1.0])
    ax.set_ylim([0.0, 1.05])
    ax.set_xlabel('False Positive Rate')
    ax.set_ylabel('True Positive Rate')
    ax.set_title(f'\n\n\n\n\n\n\nROC Curve {model_name}')
    ax.legend(loc="lower right")

import matplotlib.gridspec as gridspec

# Creazione figura con spaziatura personalizzata
fig = plt.figure(figsize=(14, 20))
gs = gridspec.GridSpec(5, 2, height_ratios=[1, 1, 0.2, 1, 1])  # La terza riga è vuota

# Plot dei classification report
plot_classification_report("MLP", reports["MLP"], plt.subplot(gs[0, 0]))
plot_classification_report("CNN", reports["CNN"], plt.subplot(gs[0, 1]))
plot_classification_report("LSTM", reports["LSTM"], plt.subplot(gs[1, 0]))
plot_classification_report("BERT", reports["BERT"], plt.subplot(gs[1, 1]))


# Plot delle confusion matrix
plot_confusion_matrix("CNN", confusion_matrices["CNN"], plt.subplot(gs[3, 0]))
plot_confusion_matrix("MLP", confusion_matrices["MLP"], plt.subplot(gs[3, 1]))
plot_confusion_matrix("LSTM", confusion_matrices["LSTM"], plt.subplot(gs[4, 0]))
plot_confusion_matrix("BERT", confusion_matrices["BERT"], plt.subplot(gs[4, 1]))

plt.tight_layout()
plt.savefig("jpg/classification_comparison.jpg", dpi=300)
plt.show()

# Creazione di una nuova figura per le curve ROC
fig_roc, axes_roc = plt.subplots(2, 2, figsize=(14, 12))

# Plottaggio delle curve ROC
plot_roc_curve("CNN", y_true, y_scores["CNN"], axes_roc[0, 0])
plot_roc_curve("MLP", y_true, y_scores["MLP"], axes_roc[0, 1])
plot_roc_curve("LSTM", y_true, y_scores["LSTM"], axes_roc[1, 0])
plot_roc_curve("BERT", y_true, y_scores["BERT"], axes_roc[1, 1])

plt.tight_layout()
plt.savefig("jpg/auc_roc_curves.jpg", dpi=300)
plt.show()
Codice
# Definizione delle metriche per le classi 0 e 1 per ogni modello

"""
Ho dovuto inserire i valori manualmente perchè ho avuto un problema con l'ambiente 
creato in Ubuntu che uso per Jupyter Lab e ho dovuto riavviare il Kernel più volte e 
non volevo ripetere l'addestrameno dei modelli
"""

test_metrics_df = {
    "Model": ["CNN", "MLP", "LSTM", "BERT"],
    "Accuracy": [0.9794, 0.9781, 0.8623, 0.9820],
    "Precision_0": [0.99, 0.98, 0.93, 0.98],
    "Recall_0": [0.98, 0.99, 0.86, 0.99],
    "F1-Score_0": [0.99, 0.99, 0.89, 0.99],
    "Precision_1": [0.96, 0.97, 0.71, 0.99],
    "Recall_1": [0.97, 0.96, 0.85, 0.96],
    "F1-Score_1": [0.96, 0.96, 0.77, 0.97]
}

# Trovo i valori massimi per ogni metrica
max_accuracy = max(test_metrics_df["Accuracy"])
max_precision_0 = max(test_metrics_df["Precision_0"])
max_recall_0 = max(test_metrics_df["Recall_0"])
max_f1_0 = max(test_metrics_df["F1-Score_0"])

max_precision_1 = max(test_metrics_df["Precision_1"])
max_recall_1 = max(test_metrics_df["Recall_1"])
max_f1_1 = max(test_metrics_df["F1-Score_1"])

print_colored(f"\n{'Model':<9} {'Accuracy ':<11} {'Precision_0     ':<10} {'Recall_0        ':<11} {'F1-Score_0      ':<11} "
              f"{'Precision_1     ':<10} {'Recall_1        ':<11} {'F1-Score_1':<13}", "blue")

for i in range(len(test_metrics_df["Model"])):
    print_colored(f"{test_metrics_df['Model'][i]:<10}", "red", end="")

    accuracy = test_metrics_df["Accuracy"][i]
    precision_0 = test_metrics_df["Precision_0"][i]
    recall_0 = test_metrics_df["Recall_0"][i]
    f1_0 = test_metrics_df["F1-Score_0"][i]

    precision_1 = test_metrics_df["Precision_1"][i]
    recall_1 = test_metrics_df["Recall_1"][i]
    f1_1 = test_metrics_df["F1-Score_1"][i]

    print_colored(f"{accuracy:<12.4f}", bg_color="blue" if accuracy == max_accuracy else "", color="black", end="")
    print_colored(f"{precision_0:<12.4f}     ", bg_color="blue" if precision_0 == max_precision_0 else "", color="black", end="")
    print_colored(f"{recall_0:<12.4f}     ", bg_color="blue" if recall_0 == max_recall_0 else "", color="black", end="")
    print_colored(f"{f1_0:<12.4f}     ", bg_color="blue" if f1_0 == max_f1_0 else "", color="black", end="")

    print_colored(f"{precision_1:<12.4f}     ", bg_color="blue" if precision_1 == max_precision_1 else "", color="black", end="")
    print_colored(f"{recall_1:<12.4f}     ", bg_color="blue" if recall_1 == max_recall_1 else "", color="black", end="")
    print_colored(f"{f1_1:<12.4f}", bg_color="blue" if f1_1 == max_f1_1 else "", color="black", end="\n")