【はじめての Databricks】金融取引データから異常検知 #5 リモデル/モデル比較

1# Train データセットをラベルごとに抽出
2dfn = train.filter(train.label == 0)
3dfy = train.filter(train.label == 1)
4
5# 要素数
6N = train.count()
7y = dfy.count()
8p = y/N
9
10# 通常取引dfの一部を抜粋、不正取引の df に union
11train_b = dfn.sample(False, p, seed = 92285).union(dfy)
12
13# データ全体の要素数、不正取引要素数、構成日比率、train データセットの件数表示
14print("Total count: %s \nFraud cases count: %s \nProportion of fraud cases: %s" % (N, y, p))
15print("Balanced training dataset count: %s" % train_b.count())

1display(train_b.groupBy("label").count())

1# 作成したパイプラインに修正を加えて定義
2crossval_b = CrossValidator(estimator = dt,
3                          estimatorParamMaps = paramGrid,
4                          evaluator = evaluatorAUC,
5                          numFolds = 3)  
6pipelineCV_b = Pipeline(stages=[indexer, va, crossval_b])
7
8# 新たに作成したデータセットで学習
9cvModel_b = pipelineCV_b.fit(train_b)

1dt_model_b = pipeline.fit(train_b)
2display(dt_model_b.stages[-1])

1ExtractFeatureImp(dt_model_b.stages[-1].featureImportances, train_pred_b, "features").head(10)

1# パイプラインの中で一番いいモデルを構築
2train_pred_b = cvModel_b.transform(train_b)
3test_pred_b = cvModel_b.transform(test)
4
5# Training データで評価指標を算出
6pr_train_b = evaluatorPR.evaluate(train_pred_b)
7auc_train_b = evaluatorAUC.evaluate(train_pred_b)
8
9# Test データで評価指標算出
10pr_test_b = evaluatorPR.evaluate(test_pred_b)
11auc_test_b = evaluatorAUC.evaluate(test_pred_b)
12
13# 評価指標の出力
14print("PR train:", pr_train_b)
15print("AUC train:", auc_train_b)
16print("PR test:", pr_test_b)
17print("AUC test:", auc_test_b)

1# 予測結果の一時ビュー作成
2test_pred_b.createOrReplaceTempView("test_pred_b")
3
4# テスト結果の混合行列 spark df の作成
5test_pred_b_cmdf = spark.sql("select a.label, a.prediction, coalesce(b.count, 0) as count from cmt a left outer join (select label, prediction, count(1) as count from test_pred_b group by label, prediction) b on b.label = a.label and b.prediction = a.prediction order by a.label desc, a.prediction desc")
6
7# Pandas df に変換
8cm_b_pdf = test_pred_b_cmdf.toPandas()
9
10# 混合行列の配列生成
11cm_b_1d = cm_b_pdf.iloc[:, 2]
12TP, FN, FP, TN = cm_b_1d
13
14# 2次元にする
15cm_b = np.array([[TP, FN], [FP, TN]])
16
17# 混合行列描画 
18plot_confusion_matrix(cm_b, "Confusion Matrix (balanced Test)")

1with mlflow.start_run(experiment_id = mlflow_experiment_id) as run:
2
3  mlflow.log_param("balanced", "yes")
4  mlflow.log_metric("AUC train", auc_train_b)
5  mlflow.log_metric("AUC test", auc_test_b)
6  mlflow.log_metric("PR train", pr_train_b)
7  mlflow.log_metric("PR test", pr_test_b)
8
9# モデルのロギング
10  mlflow.spark.log_model(dt_model_b, "model")
11
12# 混合行列もログとして残す
13  mlflow.log_artifact("confusion-matrix.png")

1print("---model #1---")
2print("PR train:", pr_train)
3print("AUC train:", auc_train)
4print("PR test:", pr_test)
5print("AUC test:", auc_test)
6
7print("---model #2---")
8print("PR train:", pr_train_b)
9print("AUC train:", auc_train_b)
10print("PR test:", pr_test_b)
11print("AUC test:", auc_test_b)