CHAIDモデルのルール識別子毎の分岐条件出力方法について part1

目的

SPSS Modelerで作成したCHAIDモデルについて、それぞれのレコードの分岐条件とスコアを出力する方法をご紹介します。
SPSS Modeler18.5から追加されたネイティブPythonの機能を活用してXMLファイルの読み込み、読み込んだ情報とナゲットの結合方法についてまとめています。
part1ではXMLファイルを読み込んでcsvファイルを作成するところまでをご紹介します。

対象読者

・日常業務でSPSS Modeler 18.5を使用しているデータ分析初級者～中級者の方
・今後SPSS Modeler18.5の導入を検討されている方
・作成したCHAIDモデルの現場への活用方法を検討されている方

目次

• 1章 今回の記事の趣旨について
• 2章 使用データについて
• 3章 PMMLファイルの出力方法について
• 4章 PMMLファイルの読み取りについて
• 5章 SPSS Modelerでの操作について

1章 今回の記事の趣旨について

SPSS Modelerでは作成したCHAIDモデルについて設定タブの「ルール識別子」にチェックを入れると各レコードの分岐条件のルールID($RI-flag)を出力できます。
一方で、ルールIDだけでは分岐条件の中身は分かりません。

ルールIDはCHAIDモデルで作成された決定木の各ノード番号と対応しています。

今回は作成したCHAIDモデルのPMMLファイルから決定木モデルの情報を読み取り、その情報を先ほどのルールIDと結合することで表の中で各レコードの分岐条件を見られるようにしてみたいと思います。
＊PMMLファイルは、決定木モデルを含む機械学習モデルをXML形式で保存するための標準フォーマットです。

2章 使用データについて

今回使用するデータを準備していきたいと思います。
使用するデータは問いませんので、ご自身でCHAIDモデルを作成したいデータがある方は
そちらを使用いただき、3章に進んでください。
記事内と同じデータを使用する方は引き続き2章の作業を進めてください。

ここではUCI Machine Learning Repositoryにて提供されている、アメリカの国勢調査データに基づき、個人の属性から年収が50Kドルを超えるかどうかを予測するためのオープンデータソースを使用したいと思います。
下記からデータをダウンロードしてください。

ダウンロードしたデータをSPSS Modelerからアクセス可能な位置に配置して、「adult.data」を読み込んでください。
ここでは「D:\adult\adult.data」として配置しています。
今回のデータセットはフィールド名がデフォルトではないため、ファイルからフィールド名を取得のチェックを外してください。

フィルタータブに移動してフィールド名を変更します。
ダウンロードページ内の「Variables Table」と同じフィールド名にしています。

置換フィールドを配置してadult.dataへリンクします。
カテゴリ型フィールドについて" ?"の値を空文字に変換します。

adult.dataからデータ検査ノードにリンクし、実行してください。
次に、欠損値検査タブに移動して、生成タブの「欠損値選択ノード」を選択し、有効ボタンにチェックを入れてOKを選択してください。
この作業により欠損値を含むレコードを除外するノードを作成出来ます。

先ほど生成したノードに置換ノードからリンクし、更にフィールド作成ノードを配置してリンクします。
ここで分析対象フィールドとなるincome列の値である" >50K"と" <=50K"を1/0に変換した列を作成します。
今回使用するpythonスクリプトの仕様上1/0に変換する必要があります。

ここまでがデータの準備になります。次の章ではCHAIDモデルを作成して、PMMLファイルを出力するところまで進みます。

3章 PMMLファイルの出力方法について

データ型ノードを配置してリンクさせ、値の読み込みを行ってください。
ここで2章にてご紹介したデータをお使いの方は「income」列のロールをなしにして、「flag」列のロールを対象にしてください。他のフィールドについては入力のままで問題ありません。

＊ご自身のデータセットをお使いの方は対象の列の値が1/0になっていることを確認してください。
今回作成するPythonスクリプトの仕様上対象フィールドの値を1/0に変換する必要がありますので、置換フィールドやフィールド作成ノードを使用して変換をお願いいたします。

モデル作成パレットからCHAIDモデルノードを選択し、データ型ノードからリンクさせて実行してください。

作成したナゲットをダブルクリックしてください。
ファイルタブの「PMMLをエクスポート」を選択し、SPSS Modelerからアクセス可能なフォルダへ保存してください。

ここでは「adult.data」と同じフォルダにPMMLファイルを保存しています。

4章 PMMLファイルの読み取りについて

PMMLファイルの中には決定木モデル内の各ノードの分岐条件が含まれていますので、それぞれを抽出するスクリプトを作成していきたいと思います。

今回はXMLファイルを操作するためのモジュールのxml.etree.ElementTreeとデータ操作を行うためのpandasをインポートします。

import xml.etree.ElementTree as ET
import pandas as pd

xmlファイルを読み込んで各要素の情報を取得する関数を作成します。
下記の部分では要素をパースし、重要度に関する情報を取得しています。

# XMLファイルを読み込む
def parse_decision_tree(xml_file_path):
    # XMLファイルを解析
    tree = ET.parse(xml_file_path)
    root = tree.getroot()

    # 名前空間を処理
    ns = {'pmml': 'http://www.dmg.org/PMML-4_3'}

    # 重要度情報を取得
    importance_dict = {}
    mining_fields = root.findall('.//pmml:MiningField', ns)
    for field in mining_fields:
        name = field.get('name')
        importance = field.get('importance')
        if importance:
            importance_dict[name] = float(importance)

各ノードの分岐条件を取得します。
分岐条件の中にはandやorを含むものもあるため、それらを1行でまとめて取得するための処理を行っています。

    # 述語（条件）をテキストに変換する関数
    def get_predicate_text(predicate):
        if predicate is None:
            return "条件なし"

        # 複合条件（CompoundPredicate）の処理
        if predicate.tag.endswith('CompoundPredicate'):
            # まずは直下のCompoundPredicateのbooleanOperatorを取得
            boolean_operator = predicate.get('booleanOperator', 'and')
        
            # surrogateの場合、内側のCompoundPredicateのbooleanOperatorを優先
            if boolean_operator == 'surrogate':
                inner_predicates = predicate.findall('.//pmml:CompoundPredicate', ns)
                if inner_predicates:
                   # 最初に見つかった内側のCompoundPredicateのbooleanOperatorを使用
                   boolean_operator = inner_predicates[0].get('booleanOperator', 'and')
        
            # 条件を格納するリスト
            conditions = []
        
            # 内部のSimplePredicateを取得
            for simple_predicate in predicate.findall('.//pmml:SimplePredicate', ns):
                conditions.append(get_simple_predicate_text(simple_predicate))
        
            return f" {boolean_operator} ".join(conditions)

        # 単一条件（SimplePredicate）の処理
        else:
            return get_simple_predicate_text(predicate)

ここでは単一条件の場合の処理方法と分岐条件の変換方法をまとめています。
PMMLファイルで演算子を"eaual"や"lessorEqual"として表現していため、これらを記号に変換するためのマップを作成します。
演算子の情報はoperatorタブに含まれているため、このタブにマッピングを適用して記号に変換します。

    # 単純な述語をテキストに変換
    def get_simple_predicate_text(predicate):
        field = predicate.get('field')
        operator = predicate.get('operator')
        value = predicate.get('value')

        # Shift-JIS対応の演算子にマッピング
        op_map = {
            'equal': '=',
            'lessOrEqual': '≦',  # ≤ → ≦
            'greaterThan': '>',
            'greaterOrEqual': '≧',  # ≥ → ≧
            'lessThan': '<',
            'notEqual': '≠'
        }

        op_symbol = op_map.get(operator, operator)
        return f"{field} {op_symbol} {value}"

この部分では決定木の各ノードから詳細情報を抽出して、構造化されたデータに変換しています。
再帰的に木構造をたどって、階層構造を保ったままノード情報を取得しています。
ここで、スコア分布を取得する際に、1と予測されたノードのスコアを取得する形のスクリプトにしているため、3章冒頭にて対象フィールドの値は1/0に指定しました。

    # ノード情報を取得するための関数
    def get_node_info(node, parent_id=None, level=1):
        node_id = node.get('id')
        score = node.get('score')
        record_count = node.get('recordCount')

        # 分岐条件を取得
        predicates = node.findall('.//pmml:CompoundPredicate', ns) + node.findall('.//pmml:SimplePredicate', ns)
        condition = "ルートノード" if parent_id is None else get_predicate_text(predicates[0] if predicates else None)

        # スコア分布（確率）を取得
        score_distributions = node.findall('.//pmml:ScoreDistribution', ns)
        probability = next((sd.get('probability') for sd in score_distributions if sd.get('value') == '1'), "0")

        return {
            'ノードID': node_id,
            '階層': level,
            '親ノード': parent_id if parent_id is not None else '-',
            '分岐条件': condition,
            'スコア (成約確率)': f"{score} ({float(probability) * 100:.1f}%)"
        }

    # 再帰的にノードを処理してデータフレームを作成
    def process_nodes(node, parent_id=None, level=1):
        result = [get_node_info(node, parent_id, level)]

        # 子ノードを処理
        for child in node.findall('./pmml:Node', ns):
            result.extend(process_nodes(child, node.get('id'), level + 1))

        return result

    # ルートノードを取得
    root_node = root.find('.//pmml:Node', ns)

    # 全ノードの情報を取得
    node_data = process_nodes(root_node)

    # 重要度ランキングのデータフレームを作成
    importance_data = [{'変数名': k, '重要度': v} for k, v in importance_dict.items()]
    importance_df = pd.DataFrame(importance_data).sort_values('重要度', ascending=False)

    return pd.DataFrame(node_data), importance_df

ここまでに整理した各ノードの分岐条件と重要度の情報についての出力方式で決める関数を作成します。
xml_file_pathには読み取るPMMLファイルのパスを指定します。ここでは3章で保存したPMMLファイルのパスを指定してください。
output_csv_pathには各ノードの分岐条件と重要度の情報をまとめたcsvファイルを出力するファイル名を指定してください。

# 出力形式を選択してファイルを作成
def create_decision_tree_tables(xml_file_path, output_path, format='csv'):
    nodes_df, importance_df = parse_decision_tree(xml_file_path)

    if format == 'csv':
        # CSVとして保存（Shift-JIS）
        nodes_csv_path = output_path.replace('.csv', '_nodes.csv')
        importance_csv_path = output_path.replace('.csv', '_importance.csv')
        nodes_df.to_csv(nodes_csv_path, index=False, encoding='cp932')
        importance_df.to_csv(importance_csv_path, index=False, encoding='cp932')
        print(f"CSVファイルを出力しました: {nodes_csv_path}, {importance_csv_path}")

# 実行例
xml_file_path = r'D:\adult\flag.xml'
output_csv_path = r'D:\adult\flag.csv'

# CSV形式で出力
create_decision_tree_tables(xml_file_path, output_csv_path, format='csv')

5章 SPSS Modelerでの操作について

作成したスクリプトを実行して、PMMLファイルの内容をcsvファイルに出力してみます。
入力パレットのユーザー入力ノードをキャンバスに配置して、ダミーの入力ノードを作成します。

出力パレットの拡張の出力ノードをキャンバスに配置して、ユーザー入力ノードからリンクします。
pythonを選択し、4章で作成したスクリプトを貼り付けて実行します。
xml_file_pathには3章で保存したPMMLファイルのパスを、output_csv_pathには作成するcsvファイルの出力先とファイル名を指定してください。
(作成したスクリプトは下記です。)
付録（スクリプト）

実行すると指定したフォルダに「_importance.csv」と「_nodes.csv」が作成されます。

「_importance.csv」についてはCHAIDモデルの中で重要度の高い変数を昇順で出力しています。

「_nodes.csv」については決定木の各ノードの分岐条件を出力しています。

次回のpart2ではこの「_nodes.csv」を使用して、各分岐を1列にまとめ直して、下記のように複数の分岐条件とスコアを一列で出力してみたいと思います。

付録(スクリプト)

import xml.etree.ElementTree as ET
import pandas as pd

# XMLファイルを読み込む
def parse_decision_tree(xml_file_path):
   # XMLファイルを解析
   tree = ET.parse(xml_file_path)
   root = tree.getroot()

   # 名前空間を処理
   ns = {'pmml': 'http://www.dmg.org/PMML-4_3'}

   # 重要度情報を取得
   importance_dict = {}
   mining_fields = root.findall('.//pmml:MiningField', ns)
   for field in mining_fields:
       name = field.get('name')
       importance = field.get('importance')
       if importance:
           importance_dict[name] = float(importance)

   # 述語（条件）をテキストに変換する関数
   def get_predicate_text(predicate):
       if predicate is None:
           return "条件なし"

       # 複合条件（CompoundPredicate）の処理
       if predicate.tag.endswith('CompoundPredicate'):
           # まずは直下のCompoundPredicateのbooleanOperatorを取得
           boolean_operator = predicate.get('booleanOperator', 'and')
       
           # surrogateの場合、内側のCompoundPredicateのbooleanOperatorを優先
           if boolean_operator == 'surrogate':
               inner_predicates = predicate.findall('.//pmml:CompoundPredicate', ns)
               if inner_predicates:
                  # 最初に見つかった内側のCompoundPredicateのbooleanOperatorを使用
                  boolean_operator = inner_predicates[0].get('booleanOperator', 'and')
       
           # 条件を格納するリスト
           conditions = []
       
           # 内部のSimplePredicateを取得
           for simple_predicate in predicate.findall('.//pmml:SimplePredicate', ns):
               conditions.append(get_simple_predicate_text(simple_predicate))
       
           return f" {boolean_operator} ".join(conditions)

       # 単一条件（SimplePredicate）の処理
       else:
           return get_simple_predicate_text(predicate)

   # 単純な述語をテキストに変換
   def get_simple_predicate_text(predicate):
       field = predicate.get('field')
       operator = predicate.get('operator')
       value = predicate.get('value')

       # Shift-JIS対応の演算子にマッピング
       op_map = {
           'equal': '=',
           'lessOrEqual': '≦',  # ≤ → ≦
           'greaterThan': '>',
           'greaterOrEqual': '≧',  # ≥ → ≧
           'lessThan': '<',
           'notEqual': '≠'
       }

       op_symbol = op_map.get(operator, operator)
       return f"{field} {op_symbol} {value}"

   # ノード情報を取得するための関数
   def get_node_info(node, parent_id=None, level=1):
       node_id = node.get('id')
       score = node.get('score')
       record_count = node.get('recordCount')

       # 分岐条件を取得
       predicates = node.findall('.//pmml:CompoundPredicate', ns) + node.findall('.//pmml:SimplePredicate', ns)
       condition = "ルートノード" if parent_id is None else get_predicate_text(predicates[0] if predicates else None)

       # スコア分布（確率）を取得
       score_distributions = node.findall('.//pmml:ScoreDistribution', ns)
       probability = next((sd.get('probability') for sd in score_distributions if sd.get('value') == '1'), "0")

       return {
           'ノードID': node_id,
           '階層': level,
           '親ノード': parent_id if parent_id is not None else '-',
           '分岐条件': condition,
           'スコア (成約確率)': f"{score} ({float(probability) * 100:.1f}%)"
       }

   # 再帰的にノードを処理してデータフレームを作成
   def process_nodes(node, parent_id=None, level=1):
       result = [get_node_info(node, parent_id, level)]

       # 子ノードを処理
       for child in node.findall('./pmml:Node', ns):
           result.extend(process_nodes(child, node.get('id'), level + 1))

       return result

   # ルートノードを取得
   root_node = root.find('.//pmml:Node', ns)

   # 全ノードの情報を取得
   node_data = process_nodes(root_node)

   # 重要度ランキングのデータフレームを作成
   importance_data = [{'変数名': k, '重要度': v} for k, v in importance_dict.items()]
   importance_df = pd.DataFrame(importance_data).sort_values('重要度', ascending=False)

   return pd.DataFrame(node_data), importance_df

# 出力形式を選択してファイルを作成
def create_decision_tree_tables(xml_file_path, output_path, format='csv'):
   nodes_df, importance_df = parse_decision_tree(xml_file_path)

   if format == 'csv':
       # CSVとして保存（Shift-JIS）
       nodes_csv_path = output_path.replace('.csv', '_nodes.csv')
       importance_csv_path = output_path.replace('.csv', '_importance.csv')
       nodes_df.to_csv(nodes_csv_path, index=False, encoding='cp932')
       importance_df.to_csv(importance_csv_path, index=False, encoding='cp932')
       print(f"CSVファイルを出力しました: {nodes_csv_path}, {importance_csv_path}")

# 実行例
xml_file_path = r'D:\adult\flag.xml'
output_csv_path = r'D:\adult\flag.csv'

# CSV形式で出力
create_decision_tree_tables(xml_file_path, output_csv_path, format='csv')