258 KiB
258 KiB
Лабораторная работа 4¶
Датасет - Цены на бриллианты https://www.kaggle.com/datasets/nancyalaswad90/diamonds-prices
- carat: Вес бриллианта в каратах
- cut: Качество огранки.
- color: Цвет бриллианта
- clarity: Чистота бриллианта
- depth: Процент глубины бриллианта
- table: Процент ширины бриллианта
- price: Цена бриллианта в долларах США
- x: Длина бриллианта в миллиметрах
- y: Ширина бриллианта в миллиметрах
- z: Глубина бриллианта в миллиметрах
Бизнес-цели:
- Прогнозирование цены бриллиантов на основании характеристик.
- Анализ частотности и сочетания характеристик бриллиантов, которые пользуются наибольшим спросом, чтобы лучше планировать запасы.
Загрузка набора данных¶
In [24]:
import pandas as pd
from sklearn import set_config
set_config(transform_output="pandas")
df = pd.read_csv("data/Diamonds.csv", index_col="id")
random_state=42
#считаем средний вес бриллиантов
average_carat = df['carat'].mean()
print(f"Среднее значение поля 'карат': {average_carat}")
#новый столбец, в котором 1 значит больше ср знач, 0 меньше
average_carat = df['carat'].mean()
df['above_average_carat'] = (df['carat'] > average_carat).astype(int)
df
Out[24]:
Разделение набора данных на обучающую и тестовые выборки (80/20) для задачи классификации¶
Целевой признак -- Cut
In [34]:
from typing import Tuple
import pandas as pd
from pandas import DataFrame
from sklearn.model_selection import train_test_split
def split_stratified_into_train_val_test(
df_input,
stratify_colname="y",
frac_train=0.6,
frac_val=0.15,
frac_test=0.25,
random_state=None,
) -> Tuple[DataFrame, DataFrame, DataFrame, DataFrame, DataFrame, DataFrame]:
"""
Splits a Pandas dataframe into three subsets (train, val, and test)
following fractional ratios provided by the user, where each subset is
stratified by the values in a specific column (that is, each subset has
the same relative frequency of the values in the column). It performs this
splitting by running train_test_split() twice.
Parameters
----------
df_input : Pandas dataframe
Input dataframe to be split.
stratify_colname : str
The name of the column that will be used for stratification. Usually
this column would be for the label.
frac_train : float
frac_val : float
frac_test : float
The ratios with which the dataframe will be split into train, val, and
test data. The values should be expressed as float fractions and should
sum to 1.0.
random_state : int, None, or RandomStateInstance
Value to be passed to train_test_split().
Returns
-------
df_train, df_val, df_test :
Dataframes containing the three splits.
"""
if frac_train + frac_val + frac_test != 1.0:
raise ValueError(
"fractions %f, %f, %f do not add up to 1.0"
% (frac_train, frac_val, frac_test)
)
if stratify_colname not in df_input.columns:
raise ValueError("%s is not a column in the dataframe" % (stratify_colname))
X = df_input # содержит все столбцы.
y = df_input[
[stratify_colname]
] # содержит столбец для стратификации.
X = df.drop(['above_average_carat'], axis=1)
y = df['above_average_carat']
# Первичное разбиение на обучающую и временную выборки.
df_train, df_temp, y_train, y_temp = train_test_split(
X, y, stratify=y, test_size=(1.0 - frac_train), random_state=random_state
)
if frac_val <= 0:
assert len(df_input) == len(df_train) + len(df_temp)
return df_train, pd.DataFrame(), df_temp, y_train, pd.DataFrame(), y_temp
# Вторичное разбиение на валидационную и тестовую выборки.
relative_frac_test = frac_test / (frac_val + frac_test)
df_val, df_test, y_val, y_test = train_test_split(
df_temp,
y_temp,
stratify=y_temp,
test_size=relative_frac_test,
random_state=random_state,
)
assert len(df_input) == len(df_train) + len(df_val) + len(df_test)
return df_train, df_val, df_test, y_train, y_val, y_test
#разбиение на 80% обучающей выборки и 20% тестовой выборки
#Стратификация выполняется по столбцу "above_average_carat"
X_train, X_val, X_test, y_train, y_val, y_test = split_stratified_into_train_val_test(
df, stratify_colname="above_average_carat", frac_train=0.80, frac_val=0, frac_test=0.20, random_state=random_state
)
display("X_train", X_train)
display("y_train", y_train)
display("X_test", X_test)
display("y_test", y_test)
Формирование конвейера для классификации данных¶
preprocessing_num -- конвейер для обработки числовых данных: заполнение пропущенных значений и стандартизация
preprocessing_cat -- конвейер для обработки категориальных данных: заполнение пропущенных данных и унитарное кодирование
features_preprocessing -- трансформер для предобработки признаков
features_engineering -- трансформер для конструирования признаков
drop_columns -- трансформер для удаления колонок
pipeline_end -- основной конвейер предобработки данных и конструирования признаков
In [40]:
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.compose import ColumnTransformer
from sklearn.discriminant_analysis import StandardScaler
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder
class DaimondFeatures(BaseEstimator, TransformerMixin):
def __init__(self):
pass
def fit(self, X, y=None):
return self
#добавляем новый столбец "Length_to_Width_Ratio" - отношение длины к ширине (столбцы x и y)
def transform(self, X, y=None):
X["Length_to_Width_Ratio"] = X["x"] / X["y"]
return X
#добавляем имя нового столбца к входным именам
def get_feature_names_out(self, features_in):
return np.append(features_in, ["Length_to_Width_Ratio"], axis=0)
#Список столбцов, которые будут удалены из данных
columns_to_drop = []
#Список числовых столбцов
num_columns = ["carat", "depth", "table", "x", "y", "z"]
#Список категориальных столбцов
cat_columns = ["cut", "color", "clarity"]
#Предобработка числовых столбцов
#Используется для заполнения пропущенных значений в числовых столбцах медианой
num_imputer = SimpleImputer(strategy="median")
#Стандартизирует числовые столбцы, приводя их к нулевому среднему и единичному стандартному отклонению
num_scaler = StandardScaler()
#Создается конвейер для последовательного применения SimpleImputer и StandardScaler
preprocessing_num = Pipeline(
[
("imputer", num_imputer),
("scaler", num_scaler),
]
)
#Предобработка категориальных столбцов
#Заполняет пропущенные значения в категориальных столбцов значением "unknown"
cat_imputer = SimpleImputer(strategy="constant", fill_value="unknown")
#Кодирует категориальные переменные с помощью one-hot кодирования
#handle_unknown=ignore позволяет игнорировать неизвестные категории
#drop=first исключает первую категорию для предотвращения мультиколлинеарности
cat_encoder = OneHotEncoder(handle_unknown="ignore", sparse_output=False, drop="first")
preprocessing_cat = Pipeline(
[
("imputer", cat_imputer),
("encoder", cat_encoder),
]
)
#Объединение этапов предобработки
#ColumnTransformer Позволяет применять разные преобразования к различным колонкам
features_preprocessing = ColumnTransformer(
verbose_feature_names_out=False,
#числовые и категориальные признаки обрабатываются разными конвейерами
transformers=[
("prepocessing_num", preprocessing_num, num_columns),
("prepocessing_cat", preprocessing_cat, cat_columns),
],
#все остальные столбцы, которые не указаны в трансформерах, будут переданы без изменений
remainder="passthrough"
)
#конструирование признаков
#создается еще один ColumnTransformer, который применяет класс DaimondFeatures к столбцам x и y
#Все остальные столбцы будут переданы без изменений
features_engineering = ColumnTransformer(
verbose_feature_names_out=False,
transformers=[
("add_features", DaimondFeatures(), ["x", "y"]),
],
remainder="passthrough",
)
#удаление столбцов, в данном случае ничего не будет удалено
drop_columns = ColumnTransformer(
verbose_feature_names_out=False,
transformers=[
("drop_columns", "drop", columns_to_drop),
],
remainder="passthrough",
)
#Постобработка столбцов
#Создается еще один ColumnTransformer который обрабатывает столбец Cabin_type с помощью prepocessing_cat
features_postprocessing = ColumnTransformer(
verbose_feature_names_out=False,
transformers=[
("prepocessing_cat", preprocessing_cat, ["Cabin_type"]),
],
remainder="passthrough",
)
#Финальный конвейер
pipeline_end = Pipeline(
[
("features_preprocessing", features_preprocessing),
("features_engineering", features_engineering),
("drop_columns", drop_columns),
]
)
Демонстрация работы конвейера для предобработки данных при классификации¶
In [41]:
#Обучает все трансформеры в конвейере на данных обучающей выборки
preprocessing_result = pipeline_end.fit_transform(X_train)
preprocessed_df = pd.DataFrame(
preprocessing_result,
#возвращает имена всех столбцов, которые были созданы в результате всех этапов предобработки в конвейере
columns=pipeline_end.get_feature_names_out(),
)
preprocessed_df
Out[41]:
Формирование набора моделей для классификации¶
In [42]:
from sklearn import ensemble, linear_model, naive_bayes, neighbors, neural_network, tree
#классификационные модели
class_models = {
#Логистическая регрессия
# от 0 до 1, принадлежит ли объект к классу
"logistic": {"model": linear_model.LogisticRegression()},
#гребневая регрессия
#Логическая, но с регуляризацией L2 (модель не так точно запоминает данные) и сбалансированными весами классов
"ridge": {"model": linear_model.LogisticRegression(penalty="l2", class_weight="balanced")},
#дерево решений с максимальной глубиной 7
#Деления данных на условия с помощью построения дерева
"decision_tree": {
"model": tree.DecisionTreeClassifier(max_depth=7, random_state=random_state)
},
#K-ближайших соседей с количеством соседей, равным 7
#Определяет ближайших объектов и находит и класс
"knn": {"model": neighbors.KNeighborsClassifier(n_neighbors=7)},
#Наивный байесовский классификатор - Вероятности для классификации
"naive_bayes": {"model": naive_bayes.GaussianNB()},
#Градиентный бустинг с 210 деревьями
#Постепенно улучшает предсказания с помощью слабых моделей
"gradient_boosting": {
"model": ensemble.GradientBoostingClassifier(n_estimators=210)
},
#Случайный лес с максимальной глубиной 11 и сбалансированными весами классов
"random_forest": {
"model": ensemble.RandomForestClassifier(
max_depth=11, class_weight="balanced", random_state=random_state
)
},
#Многослойный персептрон (нейронная сеть)с одним скрытым слоем
"mlp": {
"model": neural_network.MLPClassifier(
#содержажит 7 нейронов
hidden_layer_sizes=(7,),
#максимальное количеством итераций 500
max_iter=500,
#включение ранней остановки для предотвращения переобучения
early_stopping=True,
random_state=random_state,
)
},
}
Обучение моделей на обучающем наборе данных и оценка на тестовом¶
In [43]:
import numpy as np
from sklearn import metrics
#проходим по всем моделям
for model_name in class_models.keys():
#выводим названия
print(f"Model: {model_name}")
model = class_models[model_name]["model"]
#Получение модели и создание конвейера
model_pipeline = Pipeline([("pipeline", pipeline_end), ("model", model)])
model_pipeline = model_pipeline.fit(X_train, y_train.values.ravel())
#Предсказания на обучающей и тестовой выборке
y_train_predict = model_pipeline.predict(X_train)
y_test_probs = model_pipeline.predict_proba(X_test)[:, 1]
y_test_predict = np.where(y_test_probs > 0.5, 1, 0)
#Сохранение результатов в словаре моделей
class_models[model_name]["pipeline"] = model_pipeline
class_models[model_name]["probs"] = y_test_probs
class_models[model_name]["preds"] = y_test_predict
#Оценка производительности модели
#точность
class_models[model_name]["Precision_train"] = metrics.precision_score(
y_train, y_train_predict
)
class_models[model_name]["Precision_test"] = metrics.precision_score(
y_test, y_test_predict
)
#полнота
class_models[model_name]["Recall_train"] = metrics.recall_score(
y_train, y_train_predict
)
class_models[model_name]["Recall_test"] = metrics.recall_score(
y_test, y_test_predict
)
#аккуратность (верность) - Доля правильных предсказаний среди всех
class_models[model_name]["Accuracy_train"] = metrics.accuracy_score(
y_train, y_train_predict
)
class_models[model_name]["Accuracy_test"] = metrics.accuracy_score(
y_test, y_test_predict
)
#Площадь под кривой ROC, которая показывает качество классификации
class_models[model_name]["ROC_AUC_test"] = metrics.roc_auc_score(
y_test, y_test_probs
)
#Гармоническое среднее между точностью и полнотой
class_models[model_name]["F1_train"] = metrics.f1_score(y_train, y_train_predict)
class_models[model_name]["F1_test"] = metrics.f1_score(y_test, y_test_predict)
#Коэффициент корреляции Мэтьюса, который учитывает все возможные результаты классификации
class_models[model_name]["MCC_test"] = metrics.matthews_corrcoef(
y_test, y_test_predict
)
#Мера согласия между двумя классификаторами
class_models[model_name]["Cohen_kappa_test"] = metrics.cohen_kappa_score(
y_test, y_test_predict
)
#Матрица ошибок, показывающая количество истинных и ложных положительных и отрицательных предсказаний
class_models[model_name]["Confusion_matrix"] = metrics.confusion_matrix(
y_test, y_test_predict
)
Сводная таблица оценок качества для использованных моделей классификации¶
Матрица неточностей
In [44]:
from sklearn.metrics import ConfusionMatrixDisplay
import matplotlib.pyplot as plt
#Функция, создающая фигуру и набор подграфиков (axes)
_, ax = plt.subplots(int(len(class_models) / 2), 2, figsize=(12, 10), sharex=False, sharey=False)
for index, key in enumerate(class_models.keys()):
c_matrix = class_models[key]["Confusion_matrix"]
disp = ConfusionMatrixDisplay(
confusion_matrix=c_matrix, display_labels=["Less", "More"]
).plot(ax=ax.flat[index])
disp.ax_.set_title(key)
plt.subplots_adjust(top=1, bottom=0, hspace=0.4, wspace=0.1)
plt.show()
Точность, полнота, верность (аккуратность), F-мера
In [46]:
class_metrics = pd.DataFrame.from_dict(class_models, "index")[
[
"Precision_train",
"Precision_test",
"Recall_train",
"Recall_test",
"Accuracy_train",
"Accuracy_test",
"F1_train",
"F1_test",
]
]
#сортировка по столбцу Accuracy_test в порядке убывания
#чтобы увидеть модели с наивысшей точностью на тестовой выборке в начале таблицы
class_metrics.sort_values(
by="Accuracy_test", ascending=False
).style.background_gradient( #визуализация
cmap="plasma",
low=0.3,
high=1,
#Указывает, какие столбцы будут окрашены
subset=["Accuracy_train", "Accuracy_test", "F1_train", "F1_test"],
).background_gradient(
cmap="viridis",
low=1,
high=0.3,
subset=[
"Precision_train",
"Precision_test",
"Recall_train",
"Recall_test",
],
)
Out[46]:
ROC-кривая, каппа Коэна, коэффициент корреляции Мэтьюса
In [15]:
class_metrics = pd.DataFrame.from_dict(class_models, "index")[
[
"Accuracy_test",
"F1_test",
"ROC_AUC_test",
"Cohen_kappa_test",
"MCC_test",
]
]
class_metrics.sort_values(by="ROC_AUC_test", ascending=False).style.background_gradient(
cmap="plasma",
low=0.3,
high=1,
subset=[
"ROC_AUC_test",
"MCC_test",
"Cohen_kappa_test",
],
).background_gradient(
cmap="viridis",
low=1,
high=0.3,
subset=[
"Accuracy_test",
"F1_test",
],
)
Out[15]:
In [16]:
#выводим лучшую модель
best_model = str(class_metrics.sort_values(by="MCC_test", ascending=False).iloc[0].name)
display(best_model)
Вывод данных с ошибкой предсказания для оценки¶
In [18]:
#предобработка тестовой выборки
preprocessing_result = pipeline_end.transform(X_test)
preprocessed_df = pd.DataFrame(
preprocessing_result,
columns=pipeline_end.get_feature_names_out(),
)
#Получение предсказаний для лучшей модели
y_pred = class_models[best_model]["preds"]
#Определение индексов ошибок
#Сравнивает истинные значения из y_test (столбец above_average_carat) с предсказанными значениями y_pred
#Это дает булев массив, где True означает ошибку
error_index = y_test[y_test["above_average_carat"] != y_pred].index.tolist()
#выводим кол-во ошибок
display(f"Error items count: {len(error_index)}")
error_predicted = pd.Series(y_pred, index=y_test.index).loc[error_index]
error_df = X_test.loc[error_index].copy()
error_df.insert(loc=1, column="Predicted", value=error_predicted)
error_df.sort_index()
Out[18]:
Пример использования обученной модели (конвейера) для предсказания¶
In [19]:
model = class_models[best_model]["pipeline"]
example_id = 4500
test = pd.DataFrame(X_test.loc[example_id, :]).T
test_preprocessed = pd.DataFrame(preprocessed_df.loc[example_id, :]).T
display(test)
display(test_preprocessed)
#Получение вероятностей предсказания
result_proba = model.predict_proba(test)[0]
#Получение предсказания
result = model.predict(test)[0]
#Получение реального значения
real = int(y_test.loc[example_id].values[0])
display(f"predicted: {result} (proba: {result_proba})")
display(f"real: {real}")
Подбор гиперпараметров методом поиска по сетке¶
In [209]:
from sklearn.model_selection import GridSearchCV
#используем модель случайного леса
optimized_model_type = "random_forest"
random_forest_model = class_models[optimized_model_type]["pipeline"]
#Определение сетки гиперпараметров
param_grid = {
#Количество деревьев в лесу
"model__n_estimators": [10, 20, 30, 40, 50, 100, 150, 200, 250, 500],
#Количество столбцов
"model__max_features": ["sqrt", "log2", 2],
#Максимальная глубина дерева
"model__max_depth": [2, 3, 4, 5, 6, 7, 8, 9 ,10],
#Критерий оценки качества разбиения
"model__criterion": ["gini", "entropy", "log_loss"],
}
#Указываем модель, которую мы хотим оптимизировать
#Передаем сетку гиперпараметров для поиска
#Указываем, что хотим использовать все доступные процессоры для параллельной обработки
gs_optomizer = GridSearchCV(
estimator=random_forest_model, param_grid=param_grid, n_jobs=-1
)
#Метод fit запускает процесс поиска по сетке и обучает модель с различными комбинациями гиперпараметров
#метод ravel() для преобразования меток в одномерный массив
gs_optomizer.fit(X_train, y_train.values.ravel())
#выводим наилучшие значения гиперпараметров
gs_optomizer.best_params_
Out[209]:
Обучение модели с новыми гиперпараметрами
In [210]:
optimized_model = ensemble.RandomForestClassifier(
random_state=random_state,
criterion="gini",
max_depth=7,
max_features="sqrt",
n_estimators=30,
)
result = {}
result["pipeline"] = Pipeline([("pipeline", pipeline_end), ("model", optimized_model)]).fit(X_train, y_train.values.ravel())
result["train_preds"] = result["pipeline"].predict(X_train)
result["probs"] = result["pipeline"].predict_proba(X_test)[:, 1]
result["preds"] = np.where(result["probs"] > 0.5, 1, 0)
result["Precision_train"] = metrics.precision_score(y_train, result["train_preds"])
result["Precision_test"] = metrics.precision_score(y_test, result["preds"])
result["Recall_train"] = metrics.recall_score(y_train, result["train_preds"])
result["Recall_test"] = metrics.recall_score(y_test, result["preds"])
result["Accuracy_train"] = metrics.accuracy_score(y_train, result["train_preds"])
result["Accuracy_test"] = metrics.accuracy_score(y_test, result["preds"])
result["ROC_AUC_test"] = metrics.roc_auc_score(y_test, result["probs"])
result["F1_train"] = metrics.f1_score(y_train, result["train_preds"])
result["F1_test"] = metrics.f1_score(y_test, result["preds"])
result["MCC_test"] = metrics.matthews_corrcoef(y_test, result["preds"])
result["Cohen_kappa_test"] = metrics.cohen_kappa_score(y_test, result["preds"])
result["Confusion_matrix"] = metrics.confusion_matrix(y_test, result["preds"])
Формирование данных для оценки старой и новой версии модели
In [ ]:
optimized_metrics = pd.DataFrame(columns=list(result.keys()))
optimized_metrics.loc[len(optimized_metrics)] = pd.Series(
data=class_models[optimized_model_type]
)
optimized_metrics.loc[len(optimized_metrics)] = pd.Series(
data=result
)
optimized_metrics.insert(loc=0, column="Name", value=["Old", "New"])
optimized_metrics = optimized_metrics.set_index("Name")
Оценка параметров старой и новой модели
In [212]:
optimized_metrics[
[
"Precision_train",
"Precision_test",
"Recall_train",
"Recall_test",
"Accuracy_train",
"Accuracy_test",
"F1_train",
"F1_test",
]
].style.background_gradient(
cmap="plasma",
low=0.3,
high=1,
subset=["Accuracy_train", "Accuracy_test", "F1_train", "F1_test"],
).background_gradient(
cmap="viridis",
low=1,
high=0.3,
subset=[
"Precision_train",
"Precision_test",
"Recall_train",
"Recall_test",
],
)
Out[212]:
In [213]:
optimized_metrics[
[
"Accuracy_test",
"F1_test",
"ROC_AUC_test",
"Cohen_kappa_test",
"MCC_test",
]
].style.background_gradient(
cmap="plasma",
low=0.3,
high=1,
subset=[
"ROC_AUC_test",
"MCC_test",
"Cohen_kappa_test",
],
).background_gradient(
cmap="viridis",
low=1,
high=0.3,
subset=[
"Accuracy_test",
"F1_test",
],
)
Out[213]:
In [215]:
_, ax = plt.subplots(1, 2, figsize=(10, 4), sharex=False, sharey=False
)
for index in range(0, len(optimized_metrics)):
c_matrix = optimized_metrics.iloc[index]["Confusion_matrix"]
disp = ConfusionMatrixDisplay(
confusion_matrix=c_matrix, display_labels=["Less", "More"]
).plot(ax=ax.flat[index])
plt.subplots_adjust(top=1, bottom=0, hspace=0.4, wspace=0.3)
plt.show()
