<jupyter_start><jupyter_text>数据集来源: http://archive.ics.uci.edu/ml/datasets/Heart+Disease - 选择这个数据集的原因在于特征既有离散型和连续型两种类型<jupyter_code>df = pd.read_csv("processed.cleveland.data",header=None,names=["age","sex","cp","trestbps","chol","fbs","restectg","thalach","exang","oldpeak","slope","ca","thal","num"])
df.head()
# numerical or categorical
for index in df:
    print(index,len(set(df[index])))
# drop ? values
print(len(df))
index = ["?" not in row for row in df.values]
# convert to np.float
dataset = df.values[index].astype(np.float32)
print(len(dataset))
# split to X,y
X,y = dataset[:,:-1],dataset[:,-1]
y = y.astype(np.int64)
print(X.shape)
print(y.shape)
kinds = ["categorical" if len(set(col))<6 else "numerical"for col in X.T]
print(len(kinds))
from sklearn.model_selection import train_test_split
from collections import Counter
X_train,X_test,y_train,y_test = train_test_split(X,y)
def entropy(y):
    precs = np.array(list(Counter(y).values()))/len(y)
    ent = np.sum(-1 * precs * np.log2(precs))
    return ent
# 选择哪个特征进行分裂
def decide_feature(X,y,fas,kinds):
    # fas refers to feature_available;if one feature can be splitted,this feature_available is True,else False
    (n_samples,n_features) = X.shape
    ers = np.zeros(n_features)
    bestfvs = np.zeros(n_features)
    for fi in range(n_features):
        if not fas[fi]:
            continue
        
        if kinds[fi] == "categorical":
            I,H = entropy(y),0
            for fv,c in Counter(X[:,fi]).items():
                splity = y[X[:,fi] == fv]
                proc = c/n_samples
                I -= proc * entropy(splity)
                H += -proc * np.log2(proc)
            ers[fi] = I/H
        else:
            for fv in list(sorted(set(X[:,fi])))[:-1]:
                splity_less = y[X[:,fi] <= fv]
                proc_less = len(splity_less) / n_samples
                splity_more = y[X[:,fi] > fv]
                proc_more = len(splity_more) / n_samples
                I = -proc_less * entropy(splity_less) - proc_more * entropy(splity_more) + entropy(y)
                H = -1*proc_less * np.log2(proc_less) - proc_more * np.log2(proc_more)
                if I/H > ers[fi]:
                    ers[fi] = I/H
                    bestfvs[fi] = fv
    return ers,bestfvs
fas = np.array([True]*len(y_train))
decide_feature(X_train,y_train,fas,kinds)
def build_tree(X,y,fas,kinds):
    counts = dict(Counter(y))
    if len(counts) == 1 or (fas==False).all():
        result = max(counts,key=counts.get)
        return {"counts":counts,"result":result}
    ers,bestfvs = decide_feature(X,y,fas,kinds)
    next_ = {}
    bestfi = np.argmax(ers)
    if kinds[bestfi] == "categorical":
        fas[bestfi] = False
        for fv in set(X[:,bestfi]):
            index = (X[:,bestfi] == fv) 
            next_["{}{}".format("==",fv)] = build_tree(X[index],y[index],fas,kinds)
    else:
        bestfv = bestfvs[bestfi]
        index_less = X[:,bestfi] <= bestfv
        fas_less = fas.copy()
        if len(set(X[index_less][:,bestfi])) == 1:
            fas_less[bestfi] = False
        next_["{}{}".format("<=",bestfv)] = build_tree(X[index_less],y[index_less],fas_less,kinds)
        index_more = X[:,bestfi] > bestfv
        fas_more = fas.copy()
        if len(set(X[index_more][:,bestfi])) == 1:
            fas_more[bestfi] = False
        next_["{}{}".format(">",bestfv)] = build_tree(X[index_more],y[index_more],fas_more,kinds)
    return {"fi":bestfi,"counts":counts,"result":None,"next":next_}
fas = np.array([True]*X_train.shape[-1])
tree = build_tree(X_train,y_train,fas,kinds)
tree
# build_tree 里面有两个错误，一是在计算信息增益率时除数不能为0，二是终止条件需要改进
def predict_one(x,kinds,tree):
    while tree["result"] == None:
        fi = tree["fi"]
        fv = x[fi]
        flag = False
        for condition in tree["next"]:
            if eval(str(fv)+condition):
                tree = tree["next"][condition]
                flag = True
                break
        if not flag:
            counts = tree["counts"]
            return max(counts,key=counts.get)
    return tree["result"]
class C45DecisionTree:
    @staticmethod
    def entropy(y):
        precs = np.array(list(Counter(y).values()))/len(y)
        ent = np.sum(-1 * precs * np.log2(precs))
        return ent

    # 选择哪个特征进行分裂
    def decide_feature(self,X,y,fas,kinds):
        # fas refers to feature_available;if one feature can be splitted,this feature_available is True,else False
        (n_samples,n_features) = X.shape
        ers = np.ones(n_features) * -1
        bestfvs = np.zeros(n_features)
        for fi in range(n_features):
            if not fas[fi]:
                continue

            if kinds[fi] == "categorical":
                I,H = self.entropy(y),0
                for fv,c in Counter(X[:,fi]).items():
                    splity = y[X[:,fi] == fv]
                    proc = c/n_samples
                    I -= proc * self.entropy(splity)
                    H += -proc * np.log2(proc)
                ers[fi] = I/(H+1e-7)
            else:
#                 print(set(X[:,fi]))
                for fv in list(sorted(set(X[:,fi])))[:-1]:
                    splity_less = y[X[:,fi] <= fv]
                    proc_less = len(splity_less) / n_samples
                    splity_more = y[X[:,fi] > fv]
                    proc_more = len(splity_more) / n_samples
                    I = -proc_less * self.entropy(splity_less) - proc_more * self.entropy(splity_more) + self.entropy(y)
                    H = -1*proc_less * np.log2(proc_less) - proc_more * np.log2(proc_more)
                    if I/(H+1e-7) > ers[fi]:
                        ers[fi] = I/(H+1e-7)
                        bestfvs[fi] = fv
        return ers,bestfvs

    def build_tree(self,X,y,fas,kinds):
        counts = dict(Counter(y))
        result = max(counts,key=counts.get)
#         print("fas",fas,"len(counts)",len(counts))
        if len(counts) == 1 or (fas==False).all():
            return {"counts":counts,"result":result}
        ers,bestfvs = self.decide_feature(X,y,fas,kinds)
        if (ers == -1).all():
            return {"counts":counts,"result":result}
        next_ = {}
        bestfi = np.argmax(ers)
#         print(bestfi,ers)
        if kinds[bestfi] == "categorical":
            fas[bestfi] = False
            for fv in set(X[:,bestfi]):
                index = (X[:,bestfi] == fv)
#                 print("next: {} {} {}, size:{}".format(bestfi,"==",fv,len(y[index])))
                next_["{}{}".format("==",fv)] = self.build_tree(X[index],y[index],fas,kinds)
        else:
            bestfv = bestfvs[bestfi]
            index_less = X[:,bestfi] <= bestfv
            fas_less = fas.copy()
            if len(set(X[index_less][:,bestfi])) == 1:
                fas_less[bestfi] = False
#             print("next: {} {} {}, size:{}".format(bestfi,"<=",bestfv,len(y[index_less])))
            next_["{}{}".format("<=",bestfv)] = self.build_tree(X[index_less],y[index_less],fas_less,kinds)
            index_more = X[:,bestfi] > bestfv
            fas_more = fas.copy()
            if len(set(X[index_more][:,bestfi])) == 1:
                fas_more[bestfi] = False
#             print("next: {} {} {}, size:{}".format(bestfi,">=",bestfv,len(y[index_more])))
            next_["{}{}".format(">",bestfv)] = self.build_tree(X[index_more],y[index_more],fas_more,kinds)
        return {"fi":bestfi,"counts":counts,"result":None,"next":next_}
    
    def fit(self,X,y,kinds):
        fas = np.array([True]*X.shape[-1])
        self.tree = self.build_tree(X,y,fas,kinds)

    def predict_one(self,x):
        tree = self.tree
        while tree["result"] == None:
            fi = tree["fi"]
            fv = x[fi]
            flag = False
            for condition in tree["next"]:
                if eval(str(fv)+condition):
                    tree = tree["next"][condition]
                    flag = True
                    break
            if not flag:
                counts = tree["counts"]
                return max(counts,key=counts.get)
        return tree["result"]
    
    def predict(self,X):
        y_predicts = []
        for x_test in X_test:
            y_predicts.append(self.predict_one(x_test))
        return y_predicts
        
    def score(self,X_test,y_test):
        y_predicts = self.predict(X_test)
        return np.sum(y_predicts == y_test)/ len(y_test)
mytree = C45DecisionTree()
mytree.fit(X_train,y_train,kinds)
mytree.score(X_test,y_test)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16154841746087775, 'fraction_numerical': 0.005177079656430168, 'mean_word_length': 2.234398782343988, 'pattern_counts': {'":': 18, '<': 13, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '425103', 'license': 'MIT', 'n_tokens_mistral': 3070, 'n_tokens_neox': 3053, 'n_words': 499, 'path': 'DecisionTree/C4.5DecisionTree.ipynb', 'repo_name': 'QYHcrossover/ML-numpy', 'size': '8499'}