相识大数据中的决策树 - ToB企服应用市场:ToB评测及商务社交产业平台

import numpy as np
from collections import Counter
class DecisionTree:
class Node:
def __init__(self):
self.value = None # 内部叶节点属性
self.feature_index = None
self.children = {}
def __str__(self):
if self.children:
s = '内部节点<%s>:\n' % self.feature_index
for fv, node in self.children.items():
ss = '[%s]-> %s' % (fv, node)
s += '\t' + ss.replace('\n', '\n\t') + '\n'
s = s[:-1]
else:
s = '叶节点(%s)' % self.value
return s
def __init__(self, gain_threshold=1e-2): # 信息增益阈值
self.gain_threshold = gain_threshold
def _entropy(self, y): # 熵: -sum(pi*log(pi))
c = np.bincount(y)
p = c[np.nonzero(c)] / y.size
return -np.sum(p * np.log2(p))
def _conditional_entropy(self, feature, y): # 条件熵
feature_values = np.unique(feature)
h = 0.
for v in feature_values:
y_sub = y[feature == v]
p = y_sub.size / y.size
h += p * self._entropy(y_sub)
return h
def _information_gain(self, feature, y): # 信息增益 = 经验熵 - 经验条件熵
return self._entropy(y) - self._conditional_entropy(feature, y)
def _select_feature(self, X, y, features_list): # 选择信息增益最大特征
if features_list:
gains = np.apply_along_axis(self._information_gain, 0, X[:, features_list], y)
index = np.argmax(gains)
if gains[index] > self.gain_threshold:
return index
return None
def _create_tree(self, X, y, features_list): # 创建节点
node = DecisionTree.Node()
labels_count = np.bincount(y)
node.value = np.argmax(labels_count) # 节点值总等于数据集中样本最多的类标记
if np.count_nonzero(labels_count) != 1: # 判断类标记是否全部一致
index = self._select_feature(X, y, features_list)
if index is not None:
node.feature_index = features_list.pop(index)
feature_values = np.unique(X[:, node.feature_index])
for v in feature_values:
idx = X[:, node.feature_index] == v
X_sub, y_sub = X[idx], y[idx]
node.children[v] = self._create_tree(X_sub, y_sub, features_list.copy())
return node
def train(self, X_train, y_train): # 训练决策树
_, n = X_train.shape
self.tree_ = self._create_tree(X_train, y_train, list(range(n)))
def predict(self, X_test): # 对每一个测试样本, 调用_predict_one, 将收集到的结果数组返回
return np.apply_along_axis(self._predict_one, axis=1, arr=X_test)
def _predict_one(self, x_test): # 搜索决策树, 对单个样本进行预测
node = self.tree_
while node.children:
child = node.children.get(x_test[node.feature_index])
if not child:
break
node = child
return node.value
def __str__(self):
if hasattr(self, 'tree_'):
return str(self.tree_)
return ''
# 使用示例数据集进行分类
# 数据集加载和数据预处理部分略...
# 假设X_train和y_train是已经准备好的训练数据集
# 假设X_test是待预测的测试数据集
tree = DecisionTree()
tree.train(X_train, y_train)
predictions = tree.predict(X_test)
print(predictions)

复制代码

import numpy as np
from collections import Counter
def calc_entropy(labels):
counter = Counter(labels)
probs = [counter[c] / len(labels) for c in counter]
entropy = -np.sum(probs * np.log2(probs))
return entropy
def calc_info_gain(data, labels, feature_idx):
feature_values = set(data[:, feature_idx])
entropy_before = calc_entropy(labels)
gain = entropy_before
for value in feature_values:
subset = data[data[:, feature_idx] == value]
subset_labels = labels[data[:, feature_idx] == value]
weight = len(subset_labels) / len(labels)
gain -= weight * calc_entropy(subset_labels)
return gain
def choose_best_feature(data, labels):
num_features = data.shape[1]
best_feature = None
best_gain = 0
for i in range(num_features):
info_gain = calc_info_gain(data, labels, i)
if info_gain > best_gain:
best_gain = info_gain
best_feature = i
return best_feature
def create_decision_tree(data, labels):
if len(set(labels)) == 1:
return labels[0]
if data.shape[1] == 0:
return Counter(labels).most_common(1)[0][0]
best_feature = choose_best_feature(data, labels)
tree = {best_feature: {}}
feature_values = set(data[:, best_feature])
for value in feature_values:
subset = data[data[:, best_feature] == value]
subset_labels = labels[data[:, best_feature] == value]
tree[best_feature][value] = create_decision_tree(subset, subset_labels)
return tree
# 使用示例数据集进行分类
data = np.array([[1, 1, 1], [1, 1, 0], [0, 1, 1], [0, 0, 1]])
labels = np.array([1, 1, 0, 0])
decision_tree = create_decision_tree(data, labels)
print(decision_tree)

复制代码

import numpy as np
import pickle
import matplotlib.pyplot as plt
from sklearn.datasets import load_boston
def loadDataSet(fileName):
dataMat = []
fr = open(fileName)
for line in fr.readlines():
curLine = line.strip().split('\t')
fltLine = list(map(float, curLine))
dataMat.append(fltLine)
return dataMat
def binSplitDataSet(dataSet, feature, value):
mat0 = dataSet[dataSet[:, feature] > value]
mat1 = dataSet[dataSet[:, feature] <= value]
return mat0, mat1
def regLeaf(dataSet):
return np.mean(dataSet[:, -1])
def regErr(dataSet):
return np.var(dataSet[:, -1]) * len(dataSet)
def chooseBestSplit(dataSet, leafType=regLeaf, errType=regErr, ops=(1, 4)):
tolS = ops[0]; tolN = ops[1]
if len(set(dataSet[:, -1].T.tolist()[0])) == 1:
return None, leafType(dataSet)
m, n = dataSet.shape
S = errType(dataSet)
bestS = float('inf'); bestIndex = 0; bestValue = 0
for featIndex in range(n - 1):
for splitVal in set(dataSet[:, featIndex].T.A.tolist()[0]):
mat0, mat1 = binSplitDataSet(dataSet,

复制代码