6_XAI_LORE.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "f24a2d30-5ddc-4b3c-9706-70d367b5df98",
   "metadata": {},
   "source": [
    "# Spiegabilità: LORE - Estrazione ed elaborazione regole"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a4ebd60b-d7be-4377-8e68-3678ee83a9dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Caricamento dei dati di addestramento\n",
    "train = pd.read_csv(f\"XAI_Datasets/XAI_LORE_DoubleFeature.tsv\", sep='\\t')  \n",
    "\n",
    "# Salvataggio delle classi e del nome della colonna delle label in variabili\n",
    "y_train = np.array(train['Order_Label'])\n",
    "class_name = 'Order_Label'\n",
    "\n",
    "# Preparazione del set di addestramento per LORE \n",
    "res = prepare_dataset(train, class_name)\n",
    "X, feature_names, class_values, numeric_columns, rdf, real_feature_names, features_map = res\n",
    "\n",
    "# Preparazione dei dati di train e test\n",
    "y = np.array(X['Order_Label'])\n",
    "X = X.drop(labels=[\"Order_Label\"], axis=1)\n",
    "X_train, X_test, y_train, y_test = train_test_split(X, y, shuffle=False, test_size=0.3)\n",
    "\n",
    "# Definizione ed addestramento del modello Random Forest\n",
    "bb = RandomForestClassifier()\n",
    "bb.fit(X_train, y_train)\n",
    "\n",
    "# Funzione per effettuare predizioni delle classi \n",
    "def bb_predict(X):\n",
    "    return bb.predict(X)\n",
    "# Funzione per estrarre le probabilità predette associate alle classi di output\n",
    "def bb_predict_proba(X):\n",
    "    return bb.predict_proba(X)\n",
    "\n",
    "# Valutazione dell'accuratezza \n",
    "y_pred = bb_predict(X_test)\n",
    "# Stampa del classification report del Random Forest\n",
    "print('Accuracy %.3f' % accuracy_score(y_test, y_pred))\n",
    "print(classification_report(y_test, y_pred))\n",
    "\n",
    "# Calcolo del numero totale di campioni nel dataset\n",
    "n_samples = len(rdf)\n",
    "test_size = 0.3\n",
    "# Calcolo il numero di campioni nel training set in base alla dimensione del test set\n",
    "n_train = int(n_samples * (1 - test_size))\n",
    "random_state = 0\n",
    "# Crea un array di indici casuali per il training set utilizzando il seme specificato\n",
    "train_indices = np.random.RandomState(random_state).choice(n_samples, size=n_train, replace=False)\n",
    "# Seleziona le colonne desiderate dal dataset di training e ottiene i relativi valori \n",
    "K = rdf.iloc[train_indices][real_feature_names].values\n",
    "\n",
    "# Istanziazione di LORE per spiegare le istanze individuali\n",
    "lore_explainer = LOREM(K, bb_predict, feature_names, class_name, class_values, numeric_columns, features_map,\n",
    "                       neigh_type='geneticp', categorical_use_prob=True, continuous_fun_estimation=False,\n",
    "                       size=1000, ocr=0.1, random_state=42, ngen=10, bb_predict_proba=bb_predict_proba,\n",
    "                       verbose=True)\n",
    "\n",
    "# Definizione di una lista di indici per selezionare le istanze da spiegare (le prime 565)\n",
    "i_list = [i for i in range(565)]\n",
    "# Creazione di una lista di tuple (x, i) che rappresentano le istanze da spiegare\n",
    "x_list = []\n",
    "for i in i_list:\n",
    "    x_list.append((X_train.iloc[i], i))\n",
    "    \n",
    "# Definizione lista per memorizzare le spiegazioni generate\n",
    "exp_list_doubleFeature = []\n",
    "# Genera spiegazioni per ogni tupla (x, i) presente in x_list\n",
    "for x, i in x_list:\n",
    "    x_array = x.to_numpy()\n",
    "    # Genera una spiegazione utilizzando LORE\n",
    "    exp = lore_explainer.explain_instance(x_array, samples=300, use_weights=True, metric=neuclidean)\n",
    "    # Aggiunge la spiegazione all'elenco exp_list insieme all'indice corrispondente\n",
    "    exp_list_doubleFeature.append((exp, i))\n",
    " \n",
    "\n",
    " #Lista vuota per memorizzare le regole estratte dalle spiegazioni\n",
    "exp_list_rules=[]\n",
    "# Iterare attraverso la lista di tuple (exp, i)\n",
    "for exp, i in exp_list_doubleFeature:\n",
    "\n",
    "    # Stampa l'indice del record con annessa spiegazione\n",
    "    print(\"Explanation per il record: \" + str(i)) \n",
    "    print(exp)\n",
    "    # Aggiungo la regola estratta dalla spiegazione alla lista\n",
    "    exp_list_rules.append(str(exp.rule))\n",
    "    \n",
    "# Per salvare in un file rules generate \n",
    "#with open('LORE_Rules_DoubleFeature.json', 'w') as file:\n",
    "    #json.dump(exp_list_rules, file)\n",
    "# Per leggere da file le rules generate \n",
    "#with open('Lore_Rules_Json/LORE_Rules_DoubleFeature.json', 'r') as file:\n",
    "    #exp_list_rules = json.load(file)\n",
    "    \n",
    "# Funzione per estrarre gli intervalli dei valori delle feature \n",
    "def extract_feature_intervals(string):\n",
    "    \n",
    "    features = []\n",
    "    # Sostituisce '+' con '__' nella stringa per evitare errori di parsing\n",
    "    string = string.replace('+', '__')\n",
    "    # Estrae le informazioni sulle feature e la label dalla stringa\n",
    "    match = re.match(r'\\{(.+?)\\} --> \\{ Order_Label: (\\d) \\}', string)\n",
    "    features_str, label = match.groups()\n",
    "    label = int(label)\n",
    "    \n",
    "    # Definizione del pattern per estrarre le informazioni di ogni feature dalla stringa\n",
    "    pattern = r'(.+?)\\s*([<>]=?)\\s*(-?\\d+\\.\\d+)(?:,|$)'\n",
    "    # Trova tutte le corrispondenze dei pattern nella stringa di feature\n",
    "    feature_matches = re.findall(pattern, features_str)\n",
    "    \n",
    "    # Itera attraverso le corrispondenze delle feature\n",
    "    for match in feature_matches:\n",
    "        # Salva nelle variabili i nomi delle features, degli operatori di comparazione e il valore a destra del segno\n",
    "        feature_name = match[0].strip()\n",
    "        comparison_operator = match[1].strip()\n",
    "        number = match[2].strip()\n",
    "        \n",
    "        # Definisce l'intervallo come stringa contenente segno e valore\n",
    "        interval = f'{comparison_operator} {number}'\n",
    "        # Aggiunge le informazioni sulla feature e la label alla lista\n",
    "        features.append((feature_name, interval, label))\n",
    "        \n",
    "    return features\n",
    "\n",
    "# Funzione per calcolare quante volte le feature compaiono in ogni rule a seconda della classe\n",
    "def count_feature_occurrences(data):\n",
    "    \n",
    "    # Dizionario annidato per contare quante volte compare ciascuna feature per ogni label\n",
    "    feature_count = defaultdict(lambda: defaultdict(int))\n",
    "    \n",
    "    # Itera attraverso ogni stringa nel dataset\n",
    "    for string in data:\n",
    "        # Estrae gli intervalli delle feature dalla stringa usando la funzione extract_feature_intervals\n",
    "        feature_intervals = extract_feature_intervals(string)\n",
    "        # Itera attraverso ogni feature_interval estratto\n",
    "        for feature, interval, label in feature_intervals:\n",
    "            # Incrementa il conteggio della feature per la label corrispondente\n",
    "            feature_count[label][feature] += 1\n",
    "    \n",
    "    # Ordina i risultati in modo decrescente per count per ogni label\n",
    "    sorted_feature_count = {label: dict(sorted(features.items(), key=lambda x: x[1], reverse=True)) for label, features in feature_count.items()}\n",
    "\n",
    "    # Mostra le frequenze delle features in ordine decrescente per ogni classe\n",
    "    for label, features in sorted_feature_count.items():\n",
    "        print(f\"Classifica per la label {label}:\")\n",
    "        for feature, count in features.items():\n",
    "            print(f\"{feature}: {count} volte\")\n",
    "        print(\"\\n\")\n",
    "\n",
    "    return sorted_feature_count\n",
    "\n",
    "# Funzione per aggregare gli intervalli a seconda della feature e della label\n",
    "def aggregate_intervals(data):\n",
    "    \n",
    "    # Dizionario annidato per aggregare gli intervalli delle feature per ogni label\n",
    "    aggregated_intervals = defaultdict(lambda: defaultdict(list))\n",
    "    \n",
    "    # Itera attraverso ogni stringa nel dataset\n",
    "    for string in data:\n",
    "        # Estrae gli intervalli delle feature dalla stringa usando la funzione extract_feature_intervals\n",
    "        feature_intervals = extract_feature_intervals(string)\n",
    "        # Itera attraverso ogni feature_interval estratto\n",
    "        for feature, interval, label in feature_intervals:\n",
    "            # Aggiunge l'intervallo alla lista delle feature corrispondenti per la label\n",
    "            aggregated_intervals[label][feature].append(interval)\n",
    "    \n",
    "    # Per ogni label e feature in aggregated_intervals\n",
    "    for label, features in aggregated_intervals.items():  \n",
    "        print(f\"Intervalli per label {label}:\")\n",
    "        for feature, intervals in features.items():\n",
    "            # Chiamata alla funzione merge_intervals per unire gli intervalli sovrapposti\n",
    "            result_interval = merge_intervals(intervals)\n",
    "            print(f\"{feature}: {result_interval}\")\n",
    "        print(\"\\n\")\n",
    "        \n",
    "    return aggregated_intervals\n",
    "\n",
    "# Funzione per ottenere un unione degli intervalli in casi di sovrapposizione\n",
    "def merge_intervals(intervals):\n",
    "    # Inizializza lower e upper bounds con valori di default\n",
    "    lower_bound = float('-inf')\n",
    "    upper_bound = float('inf')\n",
    "\n",
    "    # Flag per verificare se ci sono operatori '<' o '<='\n",
    "    has_less_than = False \n",
    "\n",
    "    # Flag per verificare se ci sono operatori '>' o '>='\n",
    "    has_greater_than = False\n",
    "\n",
    "    # Lista per memorizzare i valori minimi e massimi\n",
    "    min_values = []\n",
    "    max_values = []\n",
    "\n",
    "    # Itera sugli intervalli e aggiorna i valori di lower e upper bounds\n",
    "    for interval in intervals:\n",
    "        # Separa l'operatore e il valore dall'intervallo\n",
    "        op, val = interval.split()\n",
    "        val = float(val)\n",
    "        \n",
    "        # Aggiorna le liste dei valori minimi e massimi in base all'operatore   \n",
    "        if op == '<':\n",
    "            # Imposta il flag per indicare la presenza di '<'\n",
    "            has_less_than = True \n",
    "            # Aggiunge il valore massimo alla lista\n",
    "            max_values.append(val)\n",
    "        \n",
    "        elif op == '<=':\n",
    "            # Imposta il flag per indicare la presenza di '<='\n",
    "            has_less_than = True\n",
    "            # Aggiunge il valore massimo alla lista\n",
    "            max_values.append(val)\n",
    "            \n",
    "        elif op == '>':\n",
    "            # Imposta il flag per indicare la presenza di '>'\n",
    "            has_greater_than = True\n",
    "            # Aggiunge il valore minimo alla lista\n",
    "            min_values.append(val)\n",
    "            \n",
    "        elif op == '>=':\n",
    "            # Imposta il flag per indicare la presenza di '>='\n",
    "            has_greater_than = True\n",
    "            # Aggiunge il valore minimo alla lista\n",
    "            min_values.append(val)\n",
    "\n",
    "    # Se ci sono solo operatori '<' o '<='\n",
    "    if has_less_than and not has_greater_than:\n",
    "        # Restituisce l'intervallo da '-∞' al massimo valore\n",
    "        result_interval = f'(-∞, {max(max_values)}]'\n",
    "        \n",
    "    # Se ci sono solo operatori '>' o '>='\n",
    "    elif has_greater_than and not has_less_than:\n",
    "        # Restituisce l'intervallo dal minimo valore a '+∞'\n",
    "        result_interval = f'[{min(min_values)}, ∞)'\n",
    "        \n",
    "    # Se ci sono entrambi gli operatori, calcola l'intervallo risultante\n",
    "    else:\n",
    "        # Imposta il limite inferiore\n",
    "        lower_bound = min(min_values) \n",
    "        # Imposta il limite superiore\n",
    "        upper_bound = max(max_values) \n",
    "\n",
    "        # Se il limite inferiore è minore o uguale al limite superiore\n",
    "        if lower_bound <= upper_bound:\n",
    "            # Crea la stringa rappresentante l'intervallo\n",
    "            result_interval = f'(-∞,∞), valore minore: {lower_bound}, valore maggiore: {upper_bound}'\n",
    "        else:\n",
    "            result_interval = 'Intervallo vuoto'\n",
    "\n",
    "    # Restituisce la stringa intervallo\n",
    "    return result_interval\n",
    "\n",
    "\n",
    "# Conta quante volte compare ciascuna feature per ogni label\n",
    "feature_count = count_feature_occurrences(exp_list_rules)\n",
    "# Aggrega gli intervalli per ciascuna feature\n",
    "aggregated_intervals = aggregate_intervals(exp_list_rules)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}