import pandas as pd
import matplotlib.pyplot as plt
from scorecardbundle.feature_discretization import ChiMerge as cm # ChiMerge特征离散
from scorecardbundle.feature_encoding import WOE as woe # WOE编码实现
from scorecardbundle.model_training import LogisticRegressionScoreCard as lrsc # 模型训练-逻辑回归
from scorecardbundle.model_evaluation import ModelEvaluation as me # 模型评估
# 01读取数据
def read_csv():
bd_data = pd.read_csv(r'20200326.csv', encoding='utf_8', low_memory=False)
bd_data = bd_data.set_index('bd_code') # 设置bd_code索引
# 将object转化为float
col = list(bd_data.columns)
bd_data[col] = bd_data[col].apply(pd.to_numeric, errors='coerce').fillna(0.0)
# 获取关键字表
bd_data = bd_data[bd_data['con_num'] > 5] # 合同数小于0的BD不参与评分
bd_data = bd_data[['amount_char_rate', 'loss_num_rate', 'loss_rate']]
# 归一化
bd_data = normalized(bd_data, 'amount_char_rate') # 归一化
bd_data = normalized(bd_data, 'loss_num_rate') # 归一化
bd_data = normalized(bd_data, 'loss_rate') # 归一化
bd_data.to_csv('01归一化后的样本集.csv', header=True, index=True)
return bd_data
# 归一化
def normalized(X, feature_name):
max_x = X[feature_name].max()
min_x = X[feature_name].min()
X[feature_name] = X[feature_name].apply(lambda x: (x - min_x) / (max_x - min_x))
return X
def mark_score(train_data, column, flag):
train_data[column + '_num'] = train_data[column].rank(ascending=flag, method='dense')
max_num = max(train_data[column + '_num'])
train_data[column + '_num'] = train_data[column + '_num'] / max_num * 100
return train_data
# 03 样本标注
def feature_goal(dataset):
dataset['score_num'] = dataset['amount_char_rate'] * 0.5 + dataset[
'loss_num_rate'] * 0.25 + dataset['loss_rate'] * 0.25
q95 = dataset.score_num.quantile(0.95)
q05 = dataset.score_num.quantile(0.05)
# 截尾，避免离群值对数据造成影响
dataset = dataset.loc[lambda x: x['score_num'] > q05]
dataset = dataset.loc[lambda x: x['score_num'] < q95]
# 平均值
truncated_average = dataset.score_num.quantile(0.5)
dataset.loc[dataset['score_num'] >= truncated_average, 'score_num'] = 1
dataset.loc[dataset['score_num'] < truncated_average, 'score_num'] = 0
dataset.rename(columns={'score_num': 'tag'}, inplace=True)
dataset.to_csv('02标注后的样本集.csv', header=True, index=True)
# 获取训练集
train_data = dataset.sample(frac=0.75, random_state=0)
# 获取测试集
test_data = dataset[~dataset.index.isin(train_data.index)]
train_data.to_csv('03训练集.csv', header=True, index=True)
test_data.to_csv('04测试集.csv', header=True, index=True)
# 拆分特征和标签
train_X, train_y = train_data[['amount_char_rate', 'loss_num_rate', 'loss_rate']], train_data['tag']
test_X, test_y = test_data[['amount_char_rate', 'loss_num_rate', 'loss_rate']], test_data['tag']
X, y = dataset[['amount_char_rate', 'loss_num_rate', 'loss_rate']], dataset['tag']
return train_X, train_y, test_X, test_y, X, y
# 04特征离散化（基于ChiMerge）分箱
def ChiMerge(train_X, train_y):
trans_cm = cm.ChiMerge(max_intervals=6, min_intervals=5, output_dataframe=True)
result_cm = trans_cm.fit_transform(train_X, train_y)
return result_cm
# 05特征编码(基于证据权重WOE)
def woe_fun(result_cm, train_y):
trans_woe = woe.WOE_Encoder(output_dataframe=True)
result_woe = trans_woe.fit_transform(result_cm, train_y) # WOE运行很快，此任务仅需1秒
return trans_woe, result_woe
# 06模型训练
def model_train(trans_woe, result_woe, train_X, train_y):
model = lrsc.LogisticRegressionScoreCard(trans_woe, PDO=-5, basePoints=60, verbose=True)
model.fit(result_woe, train_y)
model.woe_df_.to_csv(r'05模型详情.csv', header=True, index=False)
return model
def predict_result(model, X):
result = model.predict(X) # 得出训练集的结果分数
result.index = X.index # 使结果对应BD号
result.to_csv(r'06预测结果.csv', header=True, index=True)
return result
# 08模型评估
def model_evaluation(y, result):
evaluation = me.BinaryTargets(y, result['TotalScore'])
print("模型评估结果：")
print(evaluation.ks_stat())
print(evaluation.plot_all())
# 09分数校正
def correction_score(result_score):
min_score = min(result_score['TotalScore'])
max_score = max(result_score['TotalScore'])
print("#####模型分数概况：######")
print('最小值:' + str(min_score))
print('最大值:' + str(max_score))
print('平均值:' + str(result_score['TotalScore'].mean()))
print('中位数:' + str(result_score['TotalScore'].median()))
q5 = result_score.TotalScore.quantile(0.5)
q7 = result_score.TotalScore.quantile(0.7)
q9 = result_score.TotalScore.quantile(0.9)
# D:70以下 C:70-80 B:80-90 A:90-100
result_score['level'] = result_score['TotalScore'].apply(lambda x: get_level(x, q5, q7, q9))
result_score.to_csv(r'07划分等级后的结果.csv', header=True, index=True)
# 等级划分函数
def get_level(score, q5, q7, q9):
if score > q9:
return 'A'
elif score > q7:
return 'B'
elif score > q5:
return 'C'
else:
return 'D'
# 数据结果分布展示
def display(data_df):
data_df.TotalScore.hist(bins=50)
# 构建图像
plt.ylabel('BD数量')
plt.xlabel('BD信用分')
plt.show()
# 主程序入口
if __name__ == '__main__':
# 读取csv,数据处理
bd_data = read_csv()
# 样本标注 划分训练集和测试集
train_X, train_y, test_X, test_y, X, y = feature_goal(bd_data)
# 离散化处理
result_cm = ChiMerge(train_X, train_y)
# 计算woe
trans_woe, result_woe = woe_fun(result_cm, train_y)
# 训练模型
model = model_train(trans_woe, result_woe, train_X, train_y)
# 预测训练集
train_result = predict_result(model, train_X)
# 训练集评估
model_evaluation(train_y, train_result)
# 预测测试集
test_result = predict_result(model, test_X)
# 测试集评估
model_evaluation(test_y, test_result)
# 预测总体
X_result = predict_result(model, X)
# 分数简单统计 等级划分
correction_score(X_result)