Threshold analyse

import pandas as pd
import numpy as np
import plotly.express as px
import plotly.graph_objects as go
from IPython.display import HTML, display

import nltk
from nltk.corpus import stopwords
import string
import re
import os

import optuna
from optuna.samplers import TPESampler

from sklearn import preprocessing
from sklearn.model_selection import train_test_split
from sklearn.feature_extraction.text import TfidfVectorizer

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_val_score
import pickle
from sklearn.metrics import f1_score, classification_report, confusion_matrix

import warnings  
from sklearn.exceptions import InconsistentVersionWarning  
  
warnings.filterwarnings("ignore", category=InconsistentVersionWarning)  

from tqdm import TqdmWarning  
warnings.filterwarnings("ignore", category=TqdmWarning)  

pd.set_option('display.max_colwidth', None)

df_afgerond_afgewezen = pd.read_csv('../data/TMS-10341-tabel-1.csv', lineterminator='\n')
df_afgerond_afgewezen = df_afgerond_afgewezen[df_afgerond_afgewezen["registratie"].isin(["BAG", "BGT", "BRT"])].reset_index(drop=True)
list_afgerond_afgewezen = df_afgerond_afgewezen['meldingsnummer'].tolist()

df_doorgestuurd = pd.read_csv('../data/TMS-10341-tabel-3.csv', lineterminator='\n')
df_doorgestuurd = df_doorgestuurd[df_doorgestuurd["registratie"].isin(["BAG", "BGT", "BRT"])].reset_index(drop=True)
df_doorgestuurd = df_doorgestuurd[df_doorgestuurd["doorgestuurd_naar_registratie"].isin(["BAG", "BGT", "BRT"])].reset_index(drop=True)
list_doorgestuurd = df_doorgestuurd['meldingsnummer'].to_list()

df_doorgestuurd = df_doorgestuurd.rename(columns={"registratie": "doorgestuurd_van_registratie", "doorgestuurd_naar_registratie": "registratie"})

df_afgerond_afgewezen = df_afgerond_afgewezen[~df_afgerond_afgewezen['meldingsnummer'].isin(list_doorgestuurd)].reset_index(drop=True)

df_combined = pd.concat([df_afgerond_afgewezen, df_doorgestuurd], ignore_index=True)
df_combined = df_combined[['meldingsnummer', 'omschrijving', 'registratie', 'doorgestuurd_van_registratie']].copy()
df_combined_copy = df_combined.copy()
df_combined_copy['omschrijving'] = df_combined_copy['omschrijving'].str.replace('Met vr. groet, Pieter', '')
df_combined_copy

# Pre processing
stop_words = stopwords.words('dutch')
stop_words = [word for word in stop_words if word not in ['niet', 'geen']]

def clean_text(x):
    x = re.sub(r'\S+@\S+', ' ', x)
    x = re.sub(r'http\S+|www\.\S+', ' ', x)
    custom_punct = string.punctuation + '—’–…“”²³'
    x = re.sub(f"[{re.escape(custom_punct)}]", " ", x)
    x = x.lower()
    x = re.sub(r'x{2,}', ' ', x)
    x = x.strip()
    x = nltk.word_tokenize(x, language='dutch')
    x = [token for token in x if not token in stop_words]
    x = [token for token in x if len(token) > 1]
    x = ' '.join(x)
    return x

df_combined.loc[:, 'clean'] = df_combined['omschrijving'].apply(lambda x: clean_text(x))
df_combined = df_combined[df_combined['clean']!=""].reset_index(drop=True)

# train-test split
X_raw = df_combined['clean']
labels = df_combined['registratie']
le = preprocessing.LabelEncoder()
le.fit(labels)
y = le.transform(labels)

X_train_raw, X_test_raw, y_train, y_test = train_test_split(
    X_raw, y, 
    test_size=0.2, 
    random_state=1
)

# Vectoriseren
vectorizer = TfidfVectorizer()
vectorizer.fit(X_train_raw)

X_train = vectorizer.transform(X_train_raw)
X_test = vectorizer.transform(X_test_raw)

target_names = [le.classes_[i] for i in np.unique(y_test)]

# Model laden
with open(r'../modellen/rf_text_classifier_combined_optuna.pkl', 'rb') as f:
    model = pickle.load(f)

y_pred = model.predict(X_test)

# Voorspelde percentages per registratie per melding
probs = model.predict_proba(X_test)

class_labels = le.inverse_transform(model.classes_)

df_probs = pd.DataFrame(
    probs,
    columns=class_labels,
    index=X_test_raw.index
)

df_probs['true_class'] = le.inverse_transform(y_test)
df_probs['pred_class'] = le.inverse_transform(y_pred)

# Voorspellingen toevoegen aan test df
df_combined.loc[X_test_raw.index, 'voorspelling'] = [le.classes_[i] for i in y_pred]
df_test = df_combined.loc[X_test_raw.index].copy()
df_test = df_test.join(df_probs[['BAG', 'BGT', 'BRT']])

# Alleen doorgestuurde meldingen
df_test_doorgestuurd = df_test[df_test['doorgestuurd_van_registratie'].notnull()].copy()

test_bag = len(df_test[df_test['registratie']=='BAG'])
test_bgt = len(df_test[df_test['registratie']=='BGT'])
test_brt = len(df_test[df_test['registratie']=='BRT'])

test_fout_bag = len(df_test_doorgestuurd[df_test_doorgestuurd['doorgestuurd_van_registratie']=='BAG'])
test_fout_bgt = len(df_test_doorgestuurd[df_test_doorgestuurd['doorgestuurd_van_registratie']=='BGT'])
test_fout_brt = len(df_test_doorgestuurd[df_test_doorgestuurd['doorgestuurd_van_registratie']=='BRT'])

human_error = {
    "BAG": test_fout_bag / test_bag * 100,
    "BGT": test_fout_bgt / test_bgt * 100,
    "BRT": test_fout_brt / test_brt * 100,
}

thresholds = np.linspace(0.7, 1.0, 31)

results = []

for cls in class_labels:
    df_cls = df_probs[df_probs["pred_class"] == cls]
    total_cls = len(df_cls)
    
    for t in thresholds:
        df_t = df_cls[df_cls[cls] >= t]
        
        total_predicted = len(df_t)

        if total_predicted == 0:
            results.append({
                "registratie": cls,
                "threshold": t,
                "dekkingspercentage": 0,
                "voorspeld": 0,
                "fout": 0,
                "foutpercentage": 0
            })
            continue

        # Percentage boven threshold van totaal per registratie
        coverage = total_predicted / total_cls * 100
        
        # Correct = true_class == deze klasse
        correct = (df_t["true_class"] == cls).sum()

        # Fout = predicted != true or true != class
        wrong = total_predicted - correct

        percentage = wrong / total_predicted * 100

        results.append({
            "registratie": cls,
            "threshold": t,
            "dekkingspercentage": coverage,
            "voorspeld": total_predicted,
            "fout": wrong,
            "foutpercentage": percentage
        })

df_results = pd.DataFrame(results)

def plot_interactive(df_results, metric, ylabel, is_percentage=False, human_error=None, subtitle=None, template="plotly_white"):
    fig = px.line(
        df_results,
        x="threshold",
        y=metric,
        color="registratie",
        markers=True,
        template=template,
        color_discrete_sequence=px.colors.qualitative.Vivid,
        hover_data={"threshold": False}
    )

    if subtitle:
        fig.update_layout(title={"text": f"{ylabel} per threshold per registratie<br><sup>{subtitle}</sup>"})
    else:
        fig.update_layout(title=f"{ylabel} per threshold per registratie")

    is_dark = template == "plotly_dark"

    fig.update_layout(
        xaxis_title="Threshold",
        yaxis_title=ylabel,
        legend_title="Registratie",
        hovermode="x unified",
        autosize=True,
        height=500,
        width=1000,
        hoverlabel=dict(
            bgcolor="#111111" if is_dark else "#ffffff",
            font=dict(
                color="#f8f9fa" if is_dark else "#111111"
            ),
            bordercolor="#444444" if is_dark else "#cccccc"
        )
    )

    fig.update_layout({
        'plot_bgcolor': 'rgba(0, 0, 0, 0)',
        'paper_bgcolor': 'rgba(0, 0, 0, 0)',
    })

    color_map = {
        trace.name: trace.line.color
        for trace in fig.data
        if trace.mode == "lines+markers"
    }

    if is_percentage:
        y_format = "%{y:.2f}%"
    else:
        y_format = "%{y:.0f}"

    for trace in fig.data:
        if trace.mode == "lines+markers":
            trace.hovertemplate = (
                "Registratie: %{fullData.name}<br>"
                "Threshold: %{x}<br>"
                f"{ylabel}: {y_format}<extra></extra>"
            )

    if human_error is not None:
        x_min = df_results["threshold"].min()
        x_max = df_results["threshold"].max()

        for registratie, value in human_error.items():
            fig.add_trace(
                go.Scatter(
                    x=[x_min, x_max],
                    y=[value, value],
                    mode="lines",
                    name=registratie,
                    line=dict(
                        color=color_map.get(registratie),
                        dash="dot",
                        width=2
                    ),
                    hovertemplate=(
                        f"Registratie: {registratie}<br>"
                        f"Menselijk foutpercentage: {value:.2f}%<extra></extra>"
                    )
                )
            )
    display(fig)

df_test_doorgestuurd[df_test_doorgestuurd['registratie']==df_test_doorgestuurd['voorspelling']][
    ['meldingsnummer', 'omschrijving', 'registratie', 'doorgestuurd_van_registratie', 'voorspelling', 'BAG', 'BGT', 'BRT']
].head()

df_test_doorgestuurd[df_test_doorgestuurd['doorgestuurd_van_registratie']==df_test_doorgestuurd['voorspelling']][
    ['omschrijving', 'registratie', 'doorgestuurd_van_registratie', 'voorspelling', 'BAG', 'BGT', 'BRT']
].tail()

results_doorgestuurd = []

for cls in class_labels:
    df_cls = df_test_doorgestuurd[df_test_doorgestuurd["voorspelling"] == cls]
    total_cls = len(df_cls)
    
    for t in thresholds:
        df_t = df_cls[df_cls[cls] >= t]
        
        total_predicted = len(df_t)

        if total_predicted == 0:
            results_doorgestuurd.append({
                "registratie": cls,
                "threshold": t,
                "dekkingspercentage": 0,
                "voorspeld": 0,
                "goed": 0,
                "fout": 0,
                "foutpercentage": 0
            })
            continue

        # Percentage boven threshold van totaal per registratie
        coverage = total_predicted / total_cls * 100
        
        # Correct = true_class == deze klasse
        correct = (df_t["registratie"] == cls).sum()

        # Fout = predicted != true or true != class
        wrong = total_predicted - correct

        percentage = wrong / total_predicted * 100

        results_doorgestuurd.append({
            "registratie": cls,
            "threshold": t,
            "dekkingspercentage": coverage,
            "voorspeld": total_predicted,
            "goed": correct,
            "fout": wrong,
            "foutpercentage": percentage
        })

df_results_doorgestuurd = pd.DataFrame(results_doorgestuurd)