Régression linéaire avec Keras

1. Configuartion et chargement des données

Importons d'abord les bibliothèques nécessaires.

Données

import numpy as np import pandas as pd

Apprentissage automatique

import keras

Visualisation

import plotly.express as px from plotly.subplots import make_subplots import plotly.graph_objects as go import seaborn as sns

Modèle (import personnalisé)

from Keras_memo import construire_modele


</details>Chargeons l'ensemble de données sur les taxis de Chicago.

<details><summary>Voir le code</summary>```
url_donnees = "https://download.mlcc.google.com/mledu-datasets/chicago_taxi_train.csv"
donnees_brutes = pd.read_csv(url_donnees)

print('Chargement des données terminé avec succès.') print('Nombre total de lignes : {0}\n\n'.format(len(donnees_entrainement.index))) donnees_entrainement.head(200)


</details>### 2. Entraînement, prédiction et visualisation

Définissons des fonctions pour visualiser les résultats. Note : la visualisation du modèle pour des caractéristiques bidimensionnelles peut présenter des limitations.

<details><summary>Fonctions de visualisation</summary>```
def generer_graphiques(df, noms_caracteristiques, nom_label, sorties_modele, taille_echantillon=200):
    echantillon = df.sample(n=taille_echantillon).copy().reset_index(drop=True)
    poids, biais, epoques, rmse = sorties_modele

    modele_plot_type = "scatter" if len(noms_caracteristiques) == 1 else "surface"
    fig = make_subplots(rows=1, cols=2,
                        subplot_titles=("Courbe de Perte", "Modèle Ajusté"),
                        specs=[[{"type": "scatter"}, {"type": modele_plot_type}]])

    tracer_donnees(echantillon, noms_caracteristiques, nom_label, fig)
    tracer_modele(echantillon, noms_caracteristiques, poids, biais, fig)
    tracer_courbe_perte(epoques, rmse, fig)

    fig.show()

def tracer_courbe_perte(epoques, rmse, fig):
    courbe = px.line(x=epoques, y=rmse)
    courbe.update_traces(line_color='#ff0000', line_width=3)
    fig.append_trace(courbe.data[0], row=1, col=1)
    fig.update_xaxes(title_text="Époque", row=1, col=1)
    fig.update_yaxes(title_text="Erreur Quadratique Moyenne Racine", row=1, col=1, range=[rmse.min()*0.8, rmse.max()])

def tracer_donnees(df, caracteristiques, label, fig):
    if len(caracteristiques) == 1:
        nuage = px.scatter(df, x=caracteristiques[0], y=label)
        fig.append_trace(nuage.data[0], row=1, col=2)
        fig.update_xaxes(title_text=caracteristiques[0], row=1, col=2)
        fig.update_yaxes(title_text=label, row=1, col=2)
    else:
        nuage = px.scatter_3d(df, x=caracteristiques[0], y=caracteristiques[1], z=label)
        fig.append_trace(nuage.data[0], row=1, col=2)
        fig.update_layout(scene1=dict(xaxis_title=caracteristiques[0], yaxis_title=caracteristiques[1], zaxis_title=label))

def tracer_modele(df, caracteristiques, poids, biais, fig):
    predictions_col = 'PREDICTION'
    df[predictions_col] = biais[0]
    for idx, carac in enumerate(caracteristiques):
        df[predictions_col] += poids[idx][0] * df[carac]

    if len(caracteristiques) == 1:
        ligne_modele = px.line(df, x=caracteristiques[0], y=predictions_col)
        ligne_modele.update_traces(line_color='#ff0000', line_width=3)
        fig.append_trace(ligne_modele.data[0], row=1, col=2)
    else:
        # Génération d'une surface pour 2 caractéristiques
        z_min = df[predictions_col].min()
        z_max = df[predictions_col].max()
        z_mid = (z_min + z_max) / 2
        y_min = df[caracteristiques[1]].min()
        y_max = df[caracteristiques[1]].max()
        y_mid = (y_min + y_max) / 2

        z_points = [z_min, z_mid, z_max]
        y_points = [y_min, y_mid, y_max]
        x_points = [(z_points[i] - poids[1][0] * y_points[i] - biais[0]) / poids[0][0] for i in range(3)]

        surface_data = pd.DataFrame({
            'x': x_points * 3,
            'y': np.repeat(y_points, 3),
            'z': np.tile(z_points, 3)
        })

        surface_fig = go.Figure(data=[go.Surface(
            x=surface_data['x'].values.reshape(3,3),
            y=surface_data['y'].values.reshape(3,3),
            z=surface_data['z'].values.reshape(3,3),
            colorscale=[[0, '#89CFF0'], [1, '#FFDB58']]
        )])
        fig.append_trace(surface_fig.data[0], row=1, col=2)

def afficher_infos_modele(noms_caracteristiques, nom_label, sorties):
    poids = sorties[0]
    biais = sorties[1]
    separateur = "-" * 40
    info = f"{separateur}\nINFO MODÈLE\n{separateur}\n"
    equation = f"{nom_label} = "

    for idx, carac in enumerate(noms_caracteristiques):
        info += f"Poids pour '{carac}': {poids[idx][0]:.3f}\n"
        equation += f"{poids[idx][0]:.3f} * {carac} + "

    info += f"Biais: {biais[0]:.3f}\n"
    equation += f"{biais[0]:.3f}"

    return info + "\n" + equation

poids_entraines = modele.get_weights()[0]
biais_entraine = modele.get_weights()[1]
liste_epoques = historique.epoch
rmse_par_epoque = pd.DataFrame(historique.history)["root_mean_squared_error"]

return poids_entraines, biais_entraine, liste_epoques, rmse_par_epoque

</details>La fonction principale pour lancer une expérience complète.

<details><summary>Expérimentation</summary>```
def lancer_experimentation(df, noms_caracteristiques, nom_label, taux_apprentissage, epoques, taille_lot):
    print(f'Démarrage de l\'expérience avec caractéristiques={noms_caracteristiques} et label={nom_label}\n')
    
    caracteristiques = df[noms_caracteristiques].values
    label = df[nom_label].values
    
    modele = construire_modele(taux_apprentissage, len(noms_caracteristiques))
    resultats = entrainer_modele(modele, df, caracteristiques, label, epoques, taille_lot)
    
    print('\nEntraînement terminé avec succès\n')
    print(afficher_infos_modele(noms_caracteristiques, nom_label, resultats))
    generer_graphiques(df, noms_caracteristiques, nom_label, resultats)
    
    return modele

Commençons avec une seule caractéristique.

Caractéristique et label

carac_selectionnees = ['TRIP_MILES'] label_cible = 'FARE'

modele_1 = lancer_experimentation(donnees_entrainement, carac_selectionnees, label_cible, taux_apprentissage, epoques, taille_lot)


</details>Modifiez le taux d'apprentissage (ex: 1 ou 0.0001) ou la taille du lot (ex: 500) pour observer l'impact des hyperparamètres.

Ensuite, utilisons deux caractéristiques. Il est préférable qu'elles soient à une échelle similaire. Convertissons donc les secondes en minutes.

<details><summary>Deux caractéristiques</summary>```
donnees_entrainement['TRIP_MINUTES'] = donnees_entrainement['TRIP_SECONDS'] / 60

carac_selectionnees = ['TRIP_MILES', 'TRIP_MINUTES']
modele_2 = lancer_experimentation(donnees_entrainement, carac_selectionnees, label_cible, taux_apprentissage, epoques, taille_lot)

def creer_lot(df, taille): echantillon = df.sample(n=taille).copy() echantillon.index = range(taille) return echantillon

def predire_tarif(modele, df, caracteristiques, label, taille_lot=50): lot = creer_lot(df, taille_lot) predictions = modele.predict_on_batch(x=lot[caracteristiques].values)

resultats = {
    "TARIF PREDIT": [],
    "TARIF OBSERVE": [],
    "PERTE L1": [],
    caracteristiques[0]: []
}
if len(caracteristiques) > 1:
    resultats[caracteristiques[1]] = []

for i in range(taille_lot):
    predit = predictions[i][0]
    observe = lot.iloc[i][label]
    resultats["TARIF PREDIT"].append(formater_montant(predit))
    resultats["TARIF OBSERVE"].append(formater_montant(observe))
    resultats["PERTE L1"].append(formater_montant(abs(observe - predit)))
    resultats[caracteristiques[0]].append(lot.iloc[i][caracteristiques[0]])
    if len(caracteristiques) > 1:
        resultats[caracteristiques[1]].append(f"{lot.iloc[i][caracteristiques[1]]:.2f}")

return pd.DataFrame(resultats)

def afficher_predictions(tableau): print("-" * 40) print("PRÉDICTIONS".center(40)) print("-" * 40) print(tableau.to_string())


</details>Effectuons les prédictinos.

predictions = predire_tarif(modele_2, donnees_entrainement, carac_selectionnees, label_cible) afficher_predictions(predictions)


#### 4. Références

Basé sur les ressources de Google Developers.

Étiquettes: Régression Linéaire Python Keras Pandas visualisation de données

Publié le 7 juillet à 01h18