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.