Skip to content

ml spam detector

Jump to bottom
ghdrako edited this page Aug 28, 2026 · 11 revisions

Simple spam detector end2end

EDA

import pandas as pd
# Load spam dataset
emails = pd.read_csv("spam.csv")
# Display first records
print(emails.head())
# Display dataset dimensions
print(emails.shape)
# Display dataset information
print(emails.info())
# Check missing values
print(emails.isnull().sum())
# Display label counts
print(
    emails["label"].value_counts()
)
# Display percentages
print(
    emails["label"]
    .value_counts(normalize=True) * 100
)
import matplotlib.pyplot as plt
emails["label"].value_counts().plot(
    kind="bar"
)
plt.xlabel("Email Category")
plt.ylabel("Number of Emails")
plt.title("Spam and Ham Distribution")
plt.show()

Preprocesing

import re
def clean_text(text):
    text = text.lower()  # Converting Text to Lowercase
    text = re.sub(
        r"http\S+",
        "",
        text
    )
    text = re.sub(
        r"[^a-z\s]",
        "",
        text
    )
    # Removing Extra Spaces
    text = re.sub(
        r"\s+",
        " ",
        text
    ).strip()
    return text

emails["clean_message"] = (
    emails["message"]
    .apply(clean_text)
)
print(
    emails[
        ["message", "clean_message"]
    ].head()
)

Zbyt agresywne usuwanie cech spamu w preprocessingu:

  • Wyrażenie re.sub(r"http\S+", "", text) całkowicie usuwa linki. W wykrywaniu spamu sama obecność linku jest jedną z najsilniejszych cech. Zamiast go kasować, lepiej zamienić go na token, np. re.sub(r"http\S+", " httplink ", text).
  • Podobnie wyrażenie re.sub(r"[^a-z\s]", "", text) usuwa znaki walut ($, €, £) oraz wykrzykniki (!), które są typowymi wyznacznikami wiadomości phishingowych/spamowych.

Split

from sklearn.model_selection import train_test_split
X = emails["clean_message"]
y = emails["label"]
X_train, X_test, y_train, y_test = train_test_split(
    X,
    y,
    test_size=0.20,
    random_state=42,
    stratify=y
)

Feature extraction

from sklearn.feature_extraction.text import TfidfVectorizer
vectorizer = TfidfVectorizer(
    stop_words="english",
    max_features=5000
)
X_train_tfidf = vectorizer.fit_transform(
    X_train
)
X_test_tfidf = vectorizer.transform(
    X_test)

Treain

from sklearn.naive_bayes import MultinomialNB
model = MultinomialNB()
model.fit(
    X_train_tfidf,
    y_train
)
print("Training completed.")

Prediction

predictions = model.predict(
    X_test_tfidf
)
print(predictions[:10])

Evaluation

from sklearn.metrics import (
    accuracy_score,
    precision_score,
    recall_score,
    f1_score
)
print(
    "Accuracy:",
    accuracy_score(
        y_test,
        predictions
    )
)
print(
    "Precision:",
    precision_score(
        y_test,
        predictions,
        pos_label="spam"
    )
)
print(
    "Recall:",
    recall_score(
        y_test,
        predictions,
        pos_label="spam"
    )
)
print(
    "F1 Score:",
    f1_score(
        y_test,
        predictions,
        pos_label="spam"
    )
)


from sklearn.metrics import confusion_matrix
matrix = confusion_matrix(
    y_test,
    predictions,
    labels=[
        "ham",
        "spam"
    ]
)
print(matrix)


from sklearn.metrics import classification_report
print(
    classification_report(
        y_test,
        predictions
    )
)

# Although TF-IDF does not directly indicate feature importance, examining the vocabulary provides insight into the terms learned during training.
feature_names = (
    vectorizer.get_feature_names_out()
)
print(feature_names[:50])

Predicting New Emails

new_email = [
    "Congratulations! You have won a free vacation. Click here to claim your prize."
]
new_email_clean = [
    clean_text(text)
    for text in new_email
]
new_email_vector = vectorizer.transform(
    new_email_clean
)
prediction = model.predict(
    new_email_vector
)
probability = model.predict_proba(
    new_email_vector
)
print("Prediction:", prediction[0])
print(
    "Confidence:",
    probability.max()
)

Storing model and vectorizer to later use

import joblib
joblib.dump(
    model,
    "spam_model.pkl"
)
joblib.dump(
    vectorizer,
    "tfidf_vectorizer.pkl"
)

Building a Complete Pipeline

from sklearn.pipeline import Pipeline
pipeline = Pipeline([
    (
        "tfidf",
        TfidfVectorizer(
            preprocessor=clean_text,
            stop_words="english",
            max_features=5000
        )
    ),
    (
        "classifier",
        MultinomialNB()
    )
])
pipeline.fit(
    X_train,
    y_train
)
pipeline_prediction = pipeline.predict(
    new_email
)
print(pipeline_prediction)

Aplication

import re
import joblib
import pandas as pd
import streamlit as st

# 1. Ładowanie modelu i wektoryzatora
@st.cache_resource
def load_artifacts():
    model = joblib.load("spam_model.pkl")
    vectorizer = joblib.load("tfidf_vectorizer.pkl")
    return model, vectorizer

model, vectorizer = load_artifacts()

# 2. Funkcja czyszcząca
def clean_input(text):
    text = text.lower()
    text = re.sub(r"http\S+", "", text)
    text = re.sub(r"[^a-z\s]", "", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text

# 3. Interfejs predykcji
st.title("Spam Analyzer")
st.write("Classifying email using machine learning.")

email_input = st.text_area("Enter email text:")

if st.button("Analyze"):
    if email_input.strip() == "":
        st.warning("Please enter some text first.")
    else:
        cleaned_text = clean_input(email_input)
        vectorized_text = vectorizer.transform([cleaned_text])
        
        prediction = model.predict(vectorized_text)[0]
        probability = model.predict_proba(vectorized_text).max()
        
        if prediction == "spam":
            st.error(f"Prediction: **SPAM** (Confidence: {probability:.2%})")
        else:
            st.success(f"Prediction: **HAM** (Confidence: {probability:.2%})")

# 4. Sekcja podglądu danych
st.divider()

@st.cache_data
def load_data():
    return pd.read_csv("spam.csv")

try:
    emails = load_data()
    
    st.subheader("Dataset Overview")
    st.write(emails["label"].value_counts())
    
    st.subheader("Class Distribution")
    st.bar_chart(emails["label"].value_counts())
    
    st.subheader("Sample Messages")
    st.dataframe(emails[["message", "label"]].head(10))
except FileNotFoundError:
    st.info("Plik 'spam.csv' nie został znaleziony do wyświetlenia podglądu danych.")

lunch app

streamlit run app.py

Rozbudowa

Jeden plik model i transformer razem:

import joblib
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline

emails = pd.read_csv("spam.csv")

# Definiujemy czyszczenie bezpośrednio w TfidfVectorizer
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(
        preprocessor=clean_input, 
        stop_words="english", 
        max_features=5000
    )),
    ("classifier", MultinomialNB())
])

# Uczymy cały proces od surowego tekstu do etykiety
pipeline.fit(emails["message"], emails["label"])

# Zapisujemy tylko JEDEN artefakt
joblib.dump(pipeline, "spam_pipeline.pkl")

i uzycie

import streamlit as st
import joblib

# Ładujemy jeden plik
@st.cache_resource
def load_model():
    return joblib.load("spam_pipeline.pkl")

pipeline = load_model()

email_input = st.text_area("Enter email text:")

if st.button("Analyze") and email_input.strip():
    # Pipeline sam czyści tekst, tworzy TF-IDF i klasyfikuje:
    prediction = pipeline.predict([email_input])[0]
    confidence = pipeline.predict_proba([email_input]).max()

    st.write(f"Wynik: {prediction} ({confidence:.2%})")

Walidacja krzyzowa:

import re
import joblib
import pandas as pd
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report

# 1. Funkcja czyszcząca tekst
def clean_text(text):
    text = text.lower()
    text = re.sub(r"http\S+", " httplink ", text)
    text = re.sub(r"[^a-z\s]", "", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text

# 2. Wczytanie i podział danych
emails = pd.read_csv("spam.csv")
X_train, X_test, y_train, y_test = train_test_split(
    emails["message"],
    emails["label"],
    test_size=0.20,
    random_state=42,
    stratify=emails["label"]
)

# 3. Definicja bazowego Pipeline
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(preprocessor=clean_text)),
    ("classifier", MultinomialNB())
])

# 4. Siatka hiperparametrów (składnia: krok__parametr)
param_grid = {
    # Parametry TF-IDF
    "tfidf__max_features": [3000, 5000, 10000],
    "tfidf__ngram_range": [(1, 1), (1, 2)],        # pojedyncze słowa lub unigramy + bigramy
    "tfidf__stop_words": [None, "english"],
    
    # Parametry klasyfikatora Naive Bayes
    "classifier__alpha": [0.1, 0.5, 1.0]          # wygładzanie Laplace'a
}

# 5. Konfiguracja i uruchomienie GridSearchCV
grid_search = GridSearchCV(
    estimator=pipeline,
    param_grid=param_grid,
    cv=5,                          # 5-krotna stratyfikowana walidacja krzyżowa
    scoring="f1_macro",            # metryka dopasowana do niezbalansowanych klas
    n_jobs=-1,                     # równoległe obliczenia na wszystkich rdzeniach
    verbose=1
)

grid_search.fit(X_train, y_train)

# 6. Wyniki optymalizacji
print("Najlepsze parametry:", grid_search.best_params_)
print(f"Najlepszy wynik F1 (CV): {grid_search.best_score_:.4f}")

# 7. Ewaluacja na zbiorze testowym przy użyciu najlepszego modelu
best_pipeline = grid_search.best_estimator_
y_pred = best_pipeline.predict(X_test)
print("\nRaport klasyfikacji na zbiorze testowym:")
print(classification_report(y_test, y_pred))

# 8. Zapis zoptymalizowanego pipeline'u
joblib.dump(best_pipeline, "best_spam_pipeline.pkl")

Niestandardowy transformer:

Niestandardowy transformator (ang. custom transformer) w scikit-learn to własna klasa, która zachowuje się dokładnie tak jak wbudowane narzędzia biblioteki (np. TfidfVectorizer czy StandardScaler).

Tworzy się go, dziedzicząc po dwóch klasach bazowych:

  • BaseEstimator – dodaje obsługę parametrów, klonowanie oraz współpracę z GridSearchCV (get_params(), set_params()).
  • TransformerMixin – automatycznie dodaje metodę fit_transform(), więc wystarczy zaimplementować jedynie fit() i transform().

Dlaczego warto zamknąć czyszczenie w klasie zamiast zwykłej funkcji?

  • Możliwość strojenia w GridSearchCV: Możesz sprawdzić, czy model radzi sobie lepiej z usuwaniem liczb vs. zamianą ich na tokeny.
  • Hermetyzacja: Cała logika preprocessingu żyje wewnątrz pipeline'u, bez zewnętrznych luźnych funkcji.
  • Serializacja: Zapisując pipeline przez joblib, zapisujesz całą klasę ze stanem i parametrami.
import re
import joblib
import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline


class TextCleaner(BaseEstimator, TransformerMixin):
    def __init__(self, replace_links=True, remove_numbers=True, lowercase=True):
        # Parametry muszą być przypisane do self o dokładnie takich samych nazwach
        self.replace_links = replace_links
        self.remove_numbers = remove_numbers
        self.lowercase = lowercase

    def fit(self, X, y=None):
        # Transformator bezstanowy – niczego się nie uczy z danych, więc tylko zwraca self
        return self

    def _clean_single_text(self, text):
        if not isinstance(text, str):
            text = str(text)

        if self.lowercase:
            text = text.lower()

        if self.replace_links:
            # Zamiast usuwać, zamieniamy na słowo kluczowe
            text = re.sub(r"http\S+|www\.\S+", " httplink ", text)

        if self.remove_numbers:
            text = re.sub(r"\d+", " ", text)

        # Usunięcie znaków specjalnych poza spacjami i literami
        text = re.sub(r"[^a-z\s]", " ", text)
        
        # Redukcja wielokrotnych spacji
        text = re.sub(r"\s+", " ", text).strip()
        return text

    def transform(self, X):
        # X może być listą, serią pandas lub tablicą numpy
        if isinstance(X, pd.Series):
            return X.apply(self._clean_single_text)
        return [self._clean_single_text(text) for text in X]

Użycie w pełnym Pipeline i GridSearch

pipeline = Pipeline([
    ("cleaner", TextCleaner()),
    ("tfidf", TfidfVectorizer(stop_words="english")),
    ("classifier", MultinomialNB())
])

# W GridSearchCV możesz testować wpływ poszczególnych kroków czyszczenia:
param_grid = {
    "cleaner__replace_links": [True, False],
    "cleaner__remove_numbers": [True, False],
    "tfidf__max_features": [3000, 5000],
    "classifier__alpha": [0.1, 1.0]
}

# Całość przyjmuje surowy tekst bez wcześniejszego przygotowania
emails = pd.read_csv("spam.csv")
pipeline.fit(emails["message"], emails["label"])

# Zapis do pojedynczego pliku
joblib.dump(pipeline, "spam_custom_pipeline.pkl")

Test

Clone this wiki locally