atmaCup #12に参加して覚えたことメモ①特徴量作成

これはなにか

データ分析コンペのatmaCup #12に参加して、他の人のコードを読んで覚えたことのメモです。

atma.connpass.com

コンペのdiscussionで公開されているコードの書き方がとても勉強になったのですが、自分のエンジニアリング力がゴミで読解に時間がかかったので解釈用にどういう処理がされてるか読解したメモです。

コンペはよく「Discussion読もうぜ！」と聞きますが、エンジニアリング力が弱いと「そこで書かれているコードが何のコードかはコメントでなんとなくわかるけど処理がよくわからんのでただコピペしてるだけ。それを利用したり加工したりはできない・・・」となってしまうのでちゃんと書いているコードの意味も紐解きながら理解する必要があるかなぁと思います。

なお、コンペ自体はクローズドなのでポリシーに違反しないようにtitanicデータを使ってますが、一部解釈用コメントを追加したりデータに合わせて処理を加工したりしていますが元のコードコメント含めてコアとなるコードはコピペです。

また、コピペ参考元のコンペリンクは貼っているものの、コンペ参加者以外は404エラーになるのでご注意ください。そのため参考コードは引用元がわかるようTwitterリンクを貼るとともに許可を得て掲載しています。

よく使う項目グループを定数化

あるデータがあって、常にgroupbyとして使うためのキーはGROUP_ID_NAME = 'id'としたり、お互い1対多結合ができるデータフレームA,BがあるときにAの特徴量をFEAT_NAMES_BASE = ['hoge','fuga','piyo']、Bの特徴量をG_FEAT_NAMES_BASE = ['address','name'] (接頭語のGはGROUPの特徴量から。つまり1対多なので結合したDFで同じGROUP(結合key)には同じ値が入る)といったように定数を置くと特徴量作成がおこないやすい。

他には、同じDF内の特徴量でも分類ができる場合それぞれ定数化すると特徴量変換がしやすい。例えばプロフィールのようなDFがあって特徴量として住所系(住んでいるところ、出身、勤務地...)、お金系(現年収、去年の年収、ボーナス...)といったものが混在している場合それぞれの定数を作成する。そうすると住所系だけ取り出したい場合はdf[住所系]のようにするといちいち住所系をすべて打ち込まなくてもよくなる。

特徴量の作成

基本的に関数化して、処理したいもとデータinputと、inputの加工結果のoutputを分けて関数の返り値とする。つまり、inputで特徴量A,B,Cがあってそれらを加工して特徴量X,Y,Zを作る場合は関数の返り値はA,B,Cを含めないX,Y,Zのみのデータとする。
要するに、関数は新たな特徴量のみが返ってくる仕様とすることで他の関数に依存しない独立した関数とすることで再現性を担保している。

例えばtitanicでNameを加工する処理は以下のようにまとめる*1。

なおデータはtitanicのtrain,teatデータをそれぞれDataFrameとしてtrain testに格納している。

def create_name_features(input_df: pd.DataFrame) -> pd.DataFrame:
  out_df = pd.DataFrame()

  # 処理自体は以下から拝借
  # https://qiita.com/jun40vn/items/d8a1f71fae680589e05c

  # 名字
  out_df['Surname'] = input_df['Name'].map(lambda name:name.split(',')[0].strip())

  # Nameから敬称(Title)を抽出し、グルーピング
  out_df['Title'] = input_df['Name'].map(lambda x: x.split(', ')[1].split('. ')[0])
  out_df['Title'].replace(['Capt', 'Col', 'Major', 'Dr', 'Rev'], 'Officer', inplace=True)
  out_df['Title'].replace(['Don', 'Sir',  'the Countess', 'Lady', 'Dona'], 'Royalty', inplace=True)
  out_df['Title'].replace(['Mme', 'Ms'], 'Mrs', inplace=True)
  out_df['Title'].replace(['Mlle'], 'Miss', inplace=True)
  out_df['Title'].replace(['Jonkheer'], 'Master', inplace=True)

  return out_df

create_name_features(train) # SurnameとTitleのDFが返ってくる

def create_surname_agg_feature(input_df: pd.DataFrame) -> pd.DataFrame:
  surname_df = create_name_features(input_df)
  df = pd.concat([input_df, surname_df], axis=1)
  
  surname_aggregations = [
        # サンプル用なのであまり意味のない集計もしているので注意

        # 平均
        df.groupby('Surname')[["Age", "SibSp", "Parch"]].mean().add_prefix("mean_"),

        # 最小値
        df.groupby('Surname')[["Age", "SibSp", "Parch"]].min().add_prefix("min_"),

        # 家族の数
        df.groupby('Surname').size().rename("n_surname"),
        ]
    
  agg_df = pd.concat(surname_aggregations, axis=1) # 全結果を結合
  out_df = pd.merge(df['Surname'], agg_df, on='Surname', how="left").drop(columns=['Surname'])

  return out_df

surname_aggregations(train) # => Surnameで集計した系のDFが返ってくる

また、この関数が元データの行数を増やすような処理をおこなってないか都度テストするとロバストなので関数処理とセットでおこなうとベター

func = create_name_features
for df in [train, test]:
    assert func(df).equals(func(df))

こちらのテスト方法は@nyker_gotoさんのdiscussionを参考にしました。

特徴量の処理の仕方

人によって何パターンかあったので2つ書く。

まとめて一気に関数処理

こちらのコードは@nyker_gotoさんのdiscussionを参考にしました。

まずは、前述のように作成した特徴量作成関数のうち、使用する関数をリストアップ

# 使用する関数
feature_functions = [
    create_name_features,
    create_surname_agg_feature
]

その後、引数で渡された上記関数listを1つずつ取り出し、処理をおこないつつ連結して返す関数を設定することで特徴量の処理をシンプルにおこなうことができます。

from typing import List

def build_feature(input_df: pd.DataFrame, feature_functions: List) -> pd.DataFrame:
    # 出力するデータフレームを空で用意して
    out_df = pd.DataFrame()

    print("start build features...")

    # 各特徴生成関数ごとで
    for func in feature_functions:

        # 特徴量を作成し
        _df = func(input_df)
        print(f"\t- {func.__name__}:\tn={len(_df.T)}")

        # 横方向 (axis=1) にがっちゃんこ (concat) する
        out_df = pd.concat([out_df, _df], axis=1)

    return out_df

# 実行
feat_train_df = build_feature(input_df=train, feature_functions=feature_functions)
feat_test_df = build_feature(input_df=test, feature_functions=feature_functions)

X, y = feat_train_df.values, train["Survived"].values

'''
start build features...
  - create_name_features: n=2
  - create_surname_agg_feature:   n=7
start build features...
  - create_name_features: n=2
  - create_surname_agg_feature:   n=7
'''

この流れのメリットは

• 関数作成パートと実際に作成するパートがはっきり分かれているため可読性が高い
• 関数listに使用したい関数を記載するだけなので、関数の取捨選択を気軽にできる
• trainとtestで漏れなく同じ関数処理ができる

一方デメリットとして、build_featureの引数がinput_dateのみであるように別の関数結果を使った処理の際に、create_surname_agg_featureのように一度別の関数を関数内では知らせる必要があり計算の無駄が発生する*2。

なお、関数処理をもうちょっと発展させてClassで処理を管理する処理の場合は過去記事の以下(作成者は同じく@nyker_gotoさん)

knknkn.hatenablog.com

ひとつずつ処理

こちらは作成した関数listをもとに一気にまとめて処理するのでなく、都度処理をおこなう。

これは@tawatawaraさんのDiscussionを参考にさせてもらっている。

まずはtrain,testに関数処理をした結果の値を格納するlistおよび、作成した特徴量名を格納するlistを作成する

train_feat_list = []
test_feat_list = []
feat_names = []

次にまずはNameベースの加工をする関数を書く。これは先程と同じ処理

def create_name_features(input_df: pd.DataFrame) -> pd.DataFrame:
  output = pd.DataFrame()

  # 処理自体は以下から拝借
  # https://qiita.com/jun40vn/items/d8a1f71fae680589e05c

  # 名字
  output['Surname'] = input_df['Name'].map(lambda name:name.split(',')[0].strip())

  # Nameから敬称(Title)を抽出し、グルーピング
  output['Title'] = input_df['Name'].map(lambda x: x.split(', ')[1].split('. ')[0])
  output['Title'].replace(['Capt', 'Col', 'Major', 'Dr', 'Rev'], 'Officer', inplace=True)
  output['Title'].replace(['Don', 'Sir',  'the Countess', 'Lady', 'Dona'], 'Royalty', inplace=True)
  output['Title'].replace(['Mme', 'Ms'], 'Mrs', inplace=True)
  output['Title'].replace(['Mlle'], 'Miss', inplace=True)
  output['Title'].replace(['Jonkheer'], 'Master', inplace=True)

  return output

そしてこの関数処理をおこなってtrain(test)_feat_listに格納する。

# trainに対する処理
with Timer(prefix="[train] create name features:"):
    train_feat_list.append(create_name_features(train))

# testに対して同様の処理
with Timer(prefix="[test ] create name features:"):
    test_feat_list.append(create_name_features(test))

print("num of created features:", train_feat_list[-1].shape[-1]) # 直前に格納された位置=今回の処理結果を参照
feat_names.extend(train_feat_list[-1].columns) # 直前に格納された位置=今回の処理結果を参照

'''
[train] create name features: 0.008[s]
[test ] create name features: 0.006[s]
num of created features: 2
'''

今回Timer というものが処理に出ているがこれは処理の際に時間を測るClassとなっていて事前に実行するなりimportするなりしておく必要があります。これは詳細は割愛しますが、withで展開しつつその中に行いたい処理を書くことでその処理の時間を出力してくれるClassで便利なので合わせて使用しています。
これは先程同様@nyker_gotoさんのdiscussionコードにあるものです。

from time import time
class Timer:
    def __init__(self, logger=None, format_str="{:.3f}[s]", prefix=None, suffix=None, sep=" "):

        if prefix: format_str = str(prefix) + sep + format_str
        if suffix: format_str = format_str + sep + str(suffix)
        self.format_str = format_str
        self.logger = logger
        self.start = None
        self.end = None

    @property
    def duration(self):
        if self.end is None:
            return 0
        return self.end - self.start

    def __enter__(self):
        self.start = time()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.end = time()
        out_str = self.format_str.format(self.duration)
        if self.logger:
            self.logger.info(out_str)
        else:
            print(out_str)

次に、前述のcreate_surname_agg_featureと同様の結果を返すが引数で上述のcreate_name_featuresの処理結果を直接参照できるように少し変える。

def create_surname_agg_feature2(input_df: pd.DataFrame, surname_df: pd.DataFrame) -> pd.DataFrame:
  
  #surname_df = create_name_features(input_df) # 引数で参照しているので必要なし
  df = pd.concat([input_df, surname_df], axis=1)
  
  surname_aggregations = [
        # サンプル用なのであまり意味のない集計もしているので注意

        # 平均
        df.groupby('Surname')[["Age", "SibSp", "Parch"]].mean().add_prefix("mean_"),

        # 最小値
        df.groupby('Surname')[["Age", "SibSp", "Parch"]].min().add_prefix("min_"),

        # 家族の数
        df.groupby('Surname').size().rename("n_surname"),
        ]
    
  agg_df = pd.concat(surname_aggregations, axis=1) # 全結果を結合
  out_df = pd.merge(df['Surname'], agg_df, on='Surname', how="left").drop(columns=['Surname'])

  return out_df

先程の違いとしては、引数としてcreate_name_featuresの結果を渡すことで再計算を発生させてない。
また、引数に指定しているcreate_name_featuresの結果は、train_feat_listの0番目に格納されているのでtrain_feat_list[0]で取り出すことができる。

with Timer(prefix="[train] create surname_agg features:"):
  train_surname_df = train_feat_list[0] # create_name_featuresの結果
  train_feat_list.append(create_surname_agg_feature2(train, train_surname_df))

with Timer(prefix="[test ] create surname_agg features:"):
  test_surname_df = test_feat_list[0]
  test_feat_list.append(create_surname_agg_feature2(test, test_surname_df))

print("num of created features:", train_feat_list[-1].shape[-1])
feat_names.extend(train_feat_list[-1].columns)

'''
[train] create surname_agg features: 0.016[s]
[test ] create surname_agg features: 0.013[s]
num of created features: 7
'''

なお、この結果はtrain_feat_list[1]に格納される。

この流れのメリットは処理の再利用が可能なことがあげられる。

一方で、デメリットとしては結果を都度train_feat_listに入れているので途中で処理のミスに気づいた場合差し替えあるいはtrain_feat_listの作成時点(空list)から流し直す必要がある。また、過去の処理がtrain_feat_listの何番目にあるかが直感的に出せないのでやや面倒。ただ、そのあたりは一度作りきったら問題ないので利便性がまさる。

なお、@tawatawaraさんにお話を伺ったところ、「関数定義後その場で処理を確認したかった」とのことです。実際こちらの方が確認しながら処理をおこないやすいので好みだと思います。

また、

train_feat_listの何番目にあるかが直感的に出せないのでやや面倒

に関しては「OrderedDictとかで持ったほうがわかりやすいかも」とのこと。

実際に、単純にOrderedDictに書き換えてみると以下のような感じかなと。

# OrderedDict版
from collections import OrderedDict

feat_names = []
train_feat_dict = OrderedDict()
test_feat_dict = OrderedDict()

with Timer(prefix="[train] create name features:"):
    train_feat_dict['create_name_features'] = create_name_features(train)

with Timer(prefix="[test ] create name features:"):
    test_feat_dict['create_name_features'] = create_name_features(test)

print("num of created features:", train_feat_dict['create_name_features'].shape[-1])
feat_names.extend(train_feat_dict['create_name_features'].columns)

# => train_feat_dict['create_name_features']でcreate_name_featuresのDFを取り出す

ただ、この場合は関数名を何箇所も書かないといけなくてミスに繋がりそうなので、以下のように関数名を変数で格納してそれをもとに書くとよさそう

# dict版
from collections import OrderedDict

feat_names = []
train_feat_dict = OrderedDict()
test_feat_dict = OrderedDict()

f = create_name_features # ここのみ変える
f_name = f.__name__

with Timer(prefix=f"[train] {f_name}:"):
    train_feat_dict[f_name] = f(train)

with Timer(prefix=f"[test] {f_name}:"):
    test_feat_dict[f_name] = f(test)

print("num of created features:", train_feat_dict[f_name].shape[-1])
feat_names.extend(train_feat_dict[f_name].columns)

集約関数のエレガントな処理

前述で集約関数系をおこなったが、量が多くなると面倒なので、集約対象と集約関数のセットを渡すと集約対象名_集約関数名という列を作成するような関数を作成する。

これも@tawatawaraさんのDiscussionを参考にさせてもらっている。

import typing as tp
from itertools import product

def create_aggregated_feature(
    df               : pd.DataFrame,
    group_ids        : tp.Sequence,
    feature_names    : tp.List[str],
    aggregation_names: tp.List[str],
) -> pd.DataFrame:
    """
    pandas.core.groupby.DataFrameGroupBy.aggregate で集約特徴を作成する関数
    
    Args:
        df (pd.DataFrame)            : 特徴量の集約元となるデータ
        group_ids (Sequence)         : 集約する group を示す id のシーケンス
        feature_names (List[str])    : 集約の対象となるカラムのリスト
        aggregation_names (List[str]): 集約の操作のリスト
    
    Returns:
        agg_feat (pd.DataFrame): 集約した特徴. index は group_id となっている.
    """
    # # pandas.core.groupby.DataFrameGroupBy.aggregate に渡す辞書を作成
    agg_dict = {}
    for f_name, a_name in product(feature_names, aggregation_names):
        agg_dict[f"{f_name}_{a_name}"] = pd.NamedAgg(column=f_name, aggfunc=a_name) # NamedAggはpandas 0.25以降
        #agg_dict[f"{f_name}_{a_name}"] = (f_name, a_name) # 左記のように省略してpd.NamedAggを明示しなくても可能
        
    # # group_id で集約. agg_dict は unpack して渡す
    # よくある使い方は agg(x_min=pd.NamedAgg('x', 'min'),...)だが、
    # forで取り出した{f_name}_{a_name}という組み合わせ列を {f_name}_{a_name}:pd.NamedAgg(column=f_name, aggfunc=a_name) という辞書を作成し、
    agg_feat = df.groupby(group_ids).agg(**agg_dict)
    # この辞書をaggにアンパックで渡すと、辞書のkeyが列名、valueが処理として作成できる！！！(辞書のアンパックなので複数展開される)
    
    return agg_feat

あとはこれをもとに前述の特徴量作成方法で作成

def create_surname_agg_feature3(
    df: pd.DataFrame, surname_df: pd.DataFrame
) -> pd.DataFrame:

  # Age, SibSpのmean,medianの組み合わせの4列(2*2)を作成
  age_sibsp_agg = create_aggregated_feature(
      df, surname_df["Surname"],
      ["Age", "SibSp"],
      ["mean", "median"])

  # Parchのmin,maxの2列(1*2)を作成
  parch_agg = create_aggregated_feature(
      df, surname_df["Surname"],
      ["Parch"],
      ["min", "max"])

  # # 作成した特徴量を concat.
  # # Surname が index なのでindex毎にくっつけてくれる
  all_agg = pd.concat([age_sibsp_agg, parch_agg], axis=1)
  
  # # Surnameごとの特徴量として merge
  agg_feat = pd.merge(surname_df[["Surname"]], all_agg, left_on="Surname", right_index=True)
  
  # # group_id を drop. mergeのときにindexがバラバラになるので元の並びに戻す
  agg_feat = agg_feat.drop("Surname", axis=1).sort_index()

  return agg_feat

実行

with Timer(prefix="[train] create surname_agg features:"):
  train_surname_df = train_feat_list[0]
  train_feat_list.append(create_surname_agg_feature3(train, train_surname_df))

with Timer(prefix="[test ] create surname_agg features:"):
  test_surname_df = test_feat_list[0]
  test_feat_list.append(create_surname_agg_feature3(test, test_surname_df))

print("num of created features:", train_feat_list[-1].shape[-1])
feat_names.extend(train_feat_list[-1].columns)

'''
[train] create surname_agg features: 0.034[s]
[test ] create surname_agg features: 0.022[s]
num of created features: 6
'''

*1:加工自体のコードは https://qiita.com/jun40vn/items/d8a1f71fae680589e05c を参考にした

*2:少し改変したら可能ではある