/
Ricoshet
/
ML6
Обзор
Документация
Войти
/
Ricoshet
/
ML6
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
tree_code.py
265 строк
10 KB
Ricoshet
upload files
08 янв 2025, 21:39
08 янв 2025, 21:39
dc87af6
Код
Авторство
О чём код?
from typing import Dict import numpy as np from collections import Counter def find_best_split( feature_vector: np.ndarray, target_vector: np.ndarray ) -> (np.ndarray, np.ndarray, float, float): """ Находит оптимальный порог для разбиения вектора признака по критерию Джини. Критерий Джини определяется следующим образом: .. math:: Q(R) = -\\frac {|R_l|}{|R|}H(R_l) -\\frac {|R_r|}{|R|}H(R_r), где: * :math:`R` — множество всех объектов, * :math:`R_l` и :math:`R_r` — объекты, попавшие в левое и правое поддерево соответственно. Функция энтропии :math:`H(R)`: .. math:: H(R) = 1 - p_1^2 - p_0^2, где: * :math:`p_1` и :math:`p_0` — доля объектов класса 1 и 0 соответственно. Указания: - Пороги, приводящие к попаданию в одно из поддеревьев пустого множества объектов, не рассматриваются. - В качестве порогов, нужно брать среднее двух соседних (при сортировке) значений признака. - Поведение функции в случае константного признака может быть любым. - При одинаковых приростах Джини нужно выбирать минимальный сплит. - Для оптимизации рекомендуется использовать векторизацию вместо циклов. Parameters ---------- feature_vector : np.ndarray Вектор вещественнозначных значений признака. target_vector : np.ndarray Вектор классов объектов (0 или 1), длина `feature_vector` равна длине `target_vector`. Returns ------- thresholds : np.ndarray Отсортированный по возрастанию вектор со всеми возможными порогами, по которым объекты можно разделить на два различных поддерева. ginis : np.ndarray Вектор со значениями критерия Джини для каждого порога в `thresholds`. threshold_best : float Оптимальный порог для разбиения. gini_best : float Оптимальное значение критерия Джини. """ sorted_indices = np.argsort(feature_vector) feature_vector = feature_vector[sorted_indices] target_vector = target_vector[sorted_indices] unique_values = np.unique(feature_vector) if len(unique_values) == 1: return np.array([]), np.array([]), None, None thresholds = (unique_values[:-1] + unique_values[1:]) / 2 def gini(groups, classes): total_instances = float(sum([len(group) for group in groups])) gini = 0.0 for group in groups: size = float(len(group)) if size == 0: continue score = 0.0 for class_val in classes: p = (group == class_val).sum() / size score += p * p gini += (1.0 - score) * (size / total_instances) return gini classes = np.unique(target_vector) best_gini = float("inf") best_threshold = None ginis = [] for threshold in thresholds: left_group = target_vector[feature_vector <= threshold] right_group = target_vector[feature_vector > threshold] gini_score = gini([left_group, right_group], classes) ginis.append(gini_score) if gini_score < best_gini: best_gini = gini_score best_threshold = threshold return thresholds, np.array(ginis), best_threshold, best_gini class DecisionTree: def __init__( self, feature_types, max_depth=None, min_samples_split=None, min_samples_leaf=None, ): if any(ft not in {"real", "categorical"} for ft in feature_types): raise ValueError("There is unknown feature type") self._tree = {} self._feature_types = feature_types self._max_depth = max_depth self._min_samples_split = min_samples_split self._min_samples_leaf = min_samples_leaf def _fit_node(self, sub_X: np.ndarray, sub_y: np.ndarray, node: Dict): """ Обучение узла дерева решений. Если все элементы в подвыборке принадлежат одному классу, узел становится терминальным. Parameters ---------- sub_X : np.ndarray Подвыборка признаков. sub_y : np.ndarray Подвыборка меток классов. node : dict Узел дерева, который будет заполнен информацией о разбиении. """ # Stopping criterion: all objects are of the same class or the selection cannot be split. if np.all(sub_y == sub_y[0]) or sub_X.shape[0] <= 1: node["type"] = "terminal" node["class"] = sub_y[0] return feature_best, threshold_best, gini_best, split = None, None, None, None for feature in range(sub_X.shape[1]): feature_type = self._feature_types[feature] if feature_type == "real": feature_vector = sub_X[:, feature] elif feature_type == "categorical": counts = Counter(sub_X[:, feature]) clicks = Counter(sub_X[sub_y == 1, feature]) ratio = { key: clicks.get(key, 0) / count for key, count in counts.items() } sorted_categories = sorted(ratio, key=ratio.get) categories_map = { category: i for i, category in enumerate(sorted_categories) } feature_vector = np.vectorize(categories_map.get)(sub_X[:, feature]) else: raise ValueError("Некорректный тип признака") if len(np.unique(feature_vector)) <= 1: continue _, _, threshold, gini = find_best_split(feature_vector, sub_y) if gini_best is None or gini < gini_best: feature_best = feature gini_best = gini split = feature_vector < threshold if feature_type == "real": threshold_best = threshold elif feature_type == "categorical": threshold_best = [ k for k, v in categories_map.items() if v < threshold ] if feature_best is None: node["type"] = "terminal" node["class"] = Counter(sub_y).most_common(1)[0][0] return node["type"] = "nonterminal" node["feature_split"] = feature_best if self._feature_types[feature_best] == "real": node["threshold"] = threshold_best elif self._feature_types[feature_best] == "categorical": node["categories_split"] = threshold_best else: raise ValueError("Некорректный тип признака") node["left_child"], node["right_child"] = {}, {} self._fit_node(sub_X[split], sub_y[split], node["left_child"]) self._fit_node(sub_X[~split], sub_y[~split], node["right_child"]) def _predict_node(self, x: np.ndarray, node: Dict) -> int: """ Рекурсивное предсказание класса для одного объекта по узлу дерева решений. Если узел терминальный, возвращается предсказанный класс. Если узел не терминальный, выборка передается в соответствующее поддерево для дальнейшего предсказания. Parameters ---------- x : np.ndarray Вектор признаков одного объекта. node : dict Узел дерева решений. Returns ------- int Предсказанный класс объекта. """ if node["type"] == "terminal": return node["class"] if self._feature_types[node["feature_split"]] == "real": if x[node["feature_split"]] < node["threshold"]: return self._predict_node(x, node["left_child"]) else: return self._predict_node(x, node["right_child"]) elif self._feature_types[node["feature_split"]] == "categorical": if x[node["feature_split"]] in node["categories_split"]: return self._predict_node(x, node["left_child"]) else: return self._predict_node(x, node["right_child"]) else: raise ValueError("Incorrect feature type") def fit(self, X, y): self._fit_node(X, y, self._tree) def predict(self, X): predicted = [] for x in X: predicted.append(self._predict_node(x, self._tree)) return np.array(predicted) def _get_depth(self, node): """ Рекурсивно вычисляет глубину дерева. Parameters ---------- node : dict Узел дерева решений. Returns ------- int Глубина узла. """ if node["type"] == "terminal": return 0 else: left_depth = self._get_depth(node["left_child"]) right_depth = self._get_depth(node["right_child"]) return 1 + max(left_depth, right_depth) def get_depth(self): """ Вычисляет глубину дерева. Returns ------- int Глубина дерева. """ return self._get_depth(self._tree)