【Python】機械学習を用いた競馬予想【前処理・特徴量生成編】

.
|-- Dockerfile
|-- docker-compose.yml
|-- entrypoint.sh
`-- workspace
    `-- keiba
        |-- data
        |   |-- html
        |   |   |-- horse
        |   |   |-- ped
        |   |   `-- race
        |   |-- master
        |   `-- raw
        |       |-- results.pkl
        |       |-- race_info.pkl
        |       |-- payback.pkl
        |       |-- horse_results.pkl
        |       `-- ped_results.pkl
        |-- models
        |-- modules
        |   |-- __init__.py # [変更]モジュールロード用
        |   |-- Constants.py # [変更]定数用
        |   |-- Collection.py
        |   |-- Preprocess.py # [追加]前処理用
        |   |-- DataManager.py # [追加]マージ用
        |   `-- FeatureEngineering.py # [追加]特徴量生成用
        |-- scrape.ipynb
        `-- preprocess_learning.ipynb # [追加]前処理・特徴量生成用（学習部分は次回）

%load_ext autoreload
%reload_ext autoreload
%autoreload 2

import pandas as pd
from modules import Preprocess
from UserParams import Params
from datetime import datetime
pd.set_option('display.max_columns', 100)
pd.set_option('display.max_rows', 30)

# ==================
# 前処理＆データ結合
# ==================
# 集計開始年月日
_year = 2007
# インスタンス生成
preprocess = Preprocess()
# 各テーブルの前処理
manager = preprocess.preprocess_rawdata(since_=datetime(_year, 1, 1))
columns_info = manager.get_columns()

# データ内容の確認（Jupyter用）
#_results = manager.__dict__['_DataManager__results']
#_raceinfo = manager.__dict__['_DataManager__raceinfo']
#_horseresults = manager.__dict__['_DataManager__horseresults']
#_paybacks = manager.__dict__['_DataManager__paybacks']
#_peds = manager.__dict__['_DataManager__peds']
#_results.head()
#_raceinfo.head()
#_paybacks.head(11)
#_horseresults.head()
#_peds.head()
#columns_info.horseresults.T.set_axis(['対応するcolumn名の意味'], axis='columns')

# データの結合
# 「馬の過去成績」を対象としたデータ結合
horseresults = columns_info.horseresults
target_columns = [horseresults[key][0] for key in Params.TARGET_COLUMNS]
group_columns =  [horseresults[key][0] for key in Params.GROUP_COLUMNS]
merged_data = manager.merge(target_columns, group_columns)

# データ内容の確認（Jupyter用）
#merged_data.df.head()
#merged_data.df.columns
#merged_data.df.describe()

# ==========
# 特徴量生成
# ==========
table_df = merged_data.table

# ラベルエンコーディング＆ダミー変数化の処理による特徴量生成
functions = {
    'dumminize':        {'columns': []},
    'label_encoding':   {'columns': [table_df[key][0] for key in Params.CATEGORY_COLUMNS]},
    'encode_jockey_id': {},
    'encode_horse_id':  {},
    'drop':             {'columns': table_df['DATE'].to_list()},
}
inputs = manager.get_features(merged_data.df, functions)

# データ内容の確認（Jupyter用）
#inputs.features
#inputs.features.columns

from dataclasses import dataclass

@dataclass(frozen=True)
class Params:
    # 馬の過去成績を取得する際の対象
    TARGET_COLUMNS = ('LAST_TIME', 'RANK', 'RANK_DIFF', 'PRIZE', 'first_corner', 'final_corner', 'mean_corner')
    # 馬の過去成績を取得する際のグループ化して取得する対象
    GROUP_COLUMNS = ('race_type', 'course_len', 'PLACE')
    # カテゴリー変数化する対象
    CATEGORY_COLUMNS = ('PLACE', 'WEATHER', 'GROUND_STATE', 'sex', 'race_type', 'around', 'race_class')

import numpy as np
import pandas as pd
import re
from .Constants import LocalPaths, HorseResultsCols, ResultsCols, Master
from .DataManager import DataManager

class _Extractor:
    def __init__(self, name, pattern, callback=None):
        # 詳細は以降で解説
    def extract(self, series):
        # 詳細は以降で解説

class Preprocess:
    def __init__(self):
        """
        初期化処理
        """
        # クラスメソッド用の変数
        self.__class_name = self.__class__.__name__
        self.__date_format = '%Y-%m-%d'
        self.__course_len_scaler = 100.0  # course_lenの正規化
        self.__max_relatives = 2          # 対象とする親等の最大値（例：2親等 -> 2）

    def __read_pickle(self, filepath, err_msg='function'):
        """
        pickleファイルの読み込み
        
        Parameters
        ----------
        filepath : str
            ファイルパス

        Returns
        -------
        df : pd.DataFrame
            読み込んだDataFrame        
        """
        try:
            df = pd.read_pickle(filepath)
        except Exception as e:
            raise Exception(f'Error({self.__class_name}::{err_msg}) {e}')
        
        return df

    def preprocess_rawdata(self, 
                           results_filepath=LocalPaths.RAW_RESULTS_PATH,
                           raceinfo_filepath=LocalPaths.RAW_RACEINFO_PATH,
                           payback_filepath=LocalPaths.RAW_PAYBACK_PATH,
                           horseresults_filepath=LocalPaths.RAW_HORSERESULTS_PATH,
                           peds_filepath=LocalPaths.RAW_PEDS_PATH,
                           since_=None):
        """
        生データの前処理
        
        Parameters
        ----------
        results_filepath : str
            レース結果テーブルのファイルパス
        raceinfo_filepath : str
            レース情報テーブルのファイルパス
        payback_filepath : str
            払い戻し結果テーブルのファイルパス
        horseresults_filepath : str
            馬の過去成績テーブルのファイルパス
        peds_filepath : str
            血統情報テーブルのファイルパス
        since_ : datetime or None
            集計開始年月日
            
        Returns
        -------
        manager : _DataManager
            データ管理用インスタンス
        """
        # 前処理実行
        results = self.__preprocess_results(results_filepath)
        raceinfo = self.__preprocess_raceinfo(raceinfo_filepath)
        paybacks = self.__preprocess_payback(payback_filepath)
        horseresults = self.__preprocess_horseresults(horseresults_filepath)
        peds = self.__preprocess_peds(peds_filepath)
        # 集計期間の更新
        if since_ is not None:
            # 集計開始年月日以降に開催されたレースを抽出
            valid_indices = raceinfo[raceinfo[HorseResultsCols.DATE] >= since_].index
            # フィルタリング
            _func = lambda _df: _df[_df.index.isin(valid_indices)]
            results = _func(results)
            raceinfo = _func(raceinfo)
            paybacks = _func(paybacks)
        
        # データ管理用のインスタンス作成
        manager = DataManager(results, raceinfo, paybacks, horseresults, peds)
        
        return manager

    def __preprocess_results(self, filepath):
        # 詳細は以降で解説
    def __preprocess_raceinfo(self, filepath):
        # 詳細は以降で解説
    def __preprocess_payback(self, filepath):
        # 詳細は以降で解説
    def __preprocess_horseresults(self, filepath):
        # 詳細は以降で解説
    def __preprocess_peds(self, filepath):
        # 詳細は以降で解説

import numpy as np
import pandas as pd
import re
from dataclasses import dataclass
from typing import Any
from .Constants import HorseResultsCols, ResultsCols
from .FeatureEngineering import FeatureEngineering

@dataclass
class _ValidColumns:
    """
    DataManagerが持っている各データcolumn一覧
    """
    results: pd.DataFrame
    raceinfo: pd.DataFrame
    horseresults: pd.DataFrame

@dataclass
class _MergedData:
    """
    結合後のデータとcolumn一覧
    """
    df: pd.DataFrame
    table: pd.DataFrame

class DataManager:
    """
    DataManager : データ管理用クラス
    
    Attributes
    ----------
    __results : pd.DataFrame
        レース結果テーブル
    __raceinfo : pd.DataFrame
        レース情報テーブル
    __paybacks : pd.DataFrame
        払い戻し結果テーブル
    __horseresults : pd.DataFrame
        過去成績テーブル
    __peds : pd.DataFrame
        血統情報テーブル
    """
    def __init__(self, results, raceinfo, paybacks, horseresults, peds):
        """
        初期化処理
        """
        self.__results = results
        self.__raceinfo = raceinfo
        self.__paybacks = paybacks
        self.__horseresults = horseresults
        self.__peds = peds
        
    def __create_column_df(self, _cls, columns, description):
        """
        定数定義されたcolumn名とPythonでの変数定義のペアを生成する
        
        Parameters
        ----------
        _cls : object
            集計対象のクラス
        columns : list-like object
            該当するcolumn
        description : str
            概要説明
        """
        pattern = re.compile(r'^__')
        pairs = dict(map(lambda key: (_cls.__dict__[key], key),
                         filter(lambda val: not bool(pattern.match(val)), dir(_cls))))
        data = {'description': description}
        data.update({pairs.get(name, name): [name] for name in columns})
        df = pd.DataFrame(data)
        
        return df

    def get_columns(self):
        """
        有効なcolumn一覧の取得
        
        Returns
        -------
        instance : _ValidColumns
            _ValidColumnsのインスタンス
        """
        columns = {
            'results': self.__create_column_df(ResultsCols, self.__results.columns, 'レース結果テーブル'),
            'raceinfo': self.__create_column_df(HorseResultsCols, self.__raceinfo.columns, 'レース情報テーブル'),
            'horseresults': self.__create_column_df(HorseResultsCols, self.__horseresults.columns, '馬の過去成績テーブル'),
        }
        instance = _ValidColumns(**columns)
        
        return instance

    @property
    def paybacks(self):
        return self.__paybacks

    def __custom_aggregation(self, _df, target_columns, group_columns):
        # 詳細は以降で解説
    def merge(self, target_columns, group_columns):
        # 詳細は以降で解説
    def get_features(self, df, functions=None):
        """
        特徴量の取得
        
        Parameters
        ----------
        df : pd.DataFrame
            処理対象のDataFrame
        functions : dict or None
            呼び出す関数一覧
                key:   関数名（str）
                value: 関数の引数（dict）
            
        Returns
        -------
        instance : FeatureEngineering
            FeatureEngineeringのインスタンス
        """
        if functions is None:
            functions = {}

        # 初期化
        instance = FeatureEngineering(df)
        # 指定された関数を順に呼び出す
        for name, kwargs in functions.items():
            _func = getattr(instance, name, None)
            # 関数が存在するかつ呼び出し可能
            if hasattr(instance, name) and callable(_func):
                _func(**kwargs)

        return instance

import pandas as pd
from sklearn.preprocessing import LabelEncoder
from .Constants import SystemPaths

class FeatureEngineering:
    """
    FeatureEngineering : 特徴量エンジニアリング用クラス
    
    Attributes
    ----------
    features : pd.DataFrame
        特徴量
    """
    def __init__(self, df):
        """
        初期化
        
        Parameters
        ----------
        df : pd.DataFrame
            入力データ
        """
        self.features = df

    def drop(self, columns):
        # 詳細は以降で解説
    def dumminize(self, columns):
        # 詳細は以降で解説
    def encode_jockey_id(self):
        # 詳細は以降で解説
    def encode_horse_id(self):
        # 詳細は以降で解説
    def label_encoding(self, columns):
        # 詳細は以降で解説

# 体重を馬体重と体重変化に分割
# ResultsCols.HORSE_WEIGHTは、Constants.pyで「'馬体重'」として定義されている
weight_diff_df = df[ResultsCols.HORSE_WEIGHT].str.extract('(?P\d+)\((?P[-+\d]+)\)')

# 頭数の設定
# resultsにはrace_idをIndexとするレース結果が格納されている前提
index_counts = results.index.value_counts(sort=False)
results['n_horse'] = 0
results.loc[index_counts.index, 'n_horse'] = index_counts

    def __preprocess_results(self, filepath):
        """
        レース結果テーブルに対する前処理を行う
        
        Parameters
        ----------
        filepath : str
            ファイルパス
        Returns
        -------
        results : pd.DataFrame
            前処理後のレース結果テーブル
        """
        # データ読み込み
        df = self.__read_pickle(filepath, err_msg='preprocess_results')
        # columnのスペースを削除
        df = df.set_axis(df.columns.map(lambda x: x.replace(' ', '')), axis='columns', copy=False)
        # RANKが数字以外のものは削除・降着判定されたものも正しく扱う
        df = df[df[ResultsCols.RANK].str.contains('^\d+', na=True)].dropna(subset=[ResultsCols.RANK], axis=0)
        df[ResultsCols.RANK] = df[ResultsCols.RANK].replace('\D+', '', regex=True).map(int)
        # 抽出する列を指定
        same_cols = [
            ResultsCols.FRAME_NUMBER,
            ResultsCols.HORSE_NUMBER,
            ResultsCols.WEIGHT,
            'horse_id',
            'jockey_id',
        ]
        # 性齢を性別と年齢に分割
        sex_age_df = df[ResultsCols.SEX_AGE].str.extract('(?P\D+)(?P\d+)')
        sex_age_df['age'] = sex_age_df['age'].map(int)
        # 体重を馬体重と体重変化に分割
        weight_diff_df = df[ResultsCols.HORSE_WEIGHT].str.extract(
            '(?P\d+)\((?P[-+\d]+)\)'
        )
        weight_diff_df = weight_diff_df.fillna(weight_diff_df.mode().iloc[0]).applymap(float)
        # データ結合
        results = pd.concat([df[same_cols].copy(), sex_age_df, weight_diff_df], axis=1)
        # 頭数の設定
        index_counts = results.index.value_counts(sort=False)
        results['n_horse'] = 0
        results.loc[index_counts.index, 'n_horse'] = index_counts
        # 着順処理
        results['rank'] = df[ResultsCols.RANK]
        return results

        # columnのスペースを削除
        df = df.set_axis(df.columns.map(lambda x: x.replace(' ', '')), axis='columns', copy=False)
        # RANKが数字以外のものは削除・降着判定されたものも正しく扱う
-       df = df[df[ResultsCols.RANK].str.contains('^\d+', na=False)]
-       df[ResultsCols.RANK] = df[ResultsCols.RANK].str.replace('\D+', '', regex=True).map(int)
+       df = df[df[ResultsCols.RANK].str.contains('^\d+', na=True)].dropna(subset=[ResultsCols.RANK], axis=0)
+       df[ResultsCols.RANK] = df[ResultsCols.RANK].replace('\D+', '', regex=True).map(int)
        # 抽出する列を指定
        same_cols = [
            ResultsCols.FRAME_NUMBER,

class _Extractor:
    """
    _Extractor : データ抽出器
    
    Attributes
    ----------
    __name : str
        column name of DataFrame
    __pattern : re.compile
        正規表現パターン
    __callback : callable or None
        callback function
    """
    def __init__(self, name, pattern, callback=None):
        """
        初期化処理
        
        Parameters
        ----------
        name : str
            column name of DataFrame
        pattern : re.compile
            正規表現パターン
        callback : callable or None
            callback function
        """
        self.__name = name
        self.__pattern = pattern
        self.__callback = callback
    def extract(self, series):
        """
        抽出器
        
        Parameters
        ----------
        series : pd.Series
            変換対象
                    
        Returns
        -------
        result : pd.DataFrame
            変換結果
        """
        result = series.str.extract(f'(?P<{self.__name}>{self.__pattern})')
        if self.__callback is not None:
            result = result[self.__name].map(self.__callback)
        return result

'ダ右1200m-天候-曇-ダート-稍重-発走-10-25-2020年4月12日-3回中山6日目-3歳未勝利-馬齢'

    def __preprocess_raceinfo(self, filepath):
        """
        レース情報テーブルに対する前処理を行う
        
        Parameters
        ----------
        filepath : str
            ファイルパス

        Returns
        -------
        results : pd.DataFrame
            前処理後のレース情報テーブル
        """
        df = self.__read_pickle(filepath, err_msg='preprocess_raceinfo')
        
        extractors = [
            # PLACEの抽出
            _Extractor(HorseResultsCols.PLACE, '|'.join(Master.PLACES.keys()).replace('(', '\(').replace(')', '\)')),
            # WEATHERの抽出
            _Extractor(HorseResultsCols.WEATHER, '|'.join(Master.WEATHER)),
            # GROUND_STATEの抽出
            _Extractor(HorseResultsCols.GROUND_STATE, '|'.join(Master.GROUND_STATES.values())),
            # dateの抽出
            _Extractor(HorseResultsCols.DATE, '\d+年\d+月\d+', 
                       callback=lambda x: re.sub(r'年|月', '-', x)),
            # RACE_TYPEの抽出
            _Extractor('race_type', '|'.join(Master.RACE_TYPES.keys()), 
                       callback=lambda x: Master.RACE_TYPES[x]),
            # AROUNDの抽出
            _Extractor('around', '|'.join(Master.AROUND)),
            # RACE_CLASSの抽出
            _Extractor('race_class', '|'.join(Master.RACE_CLASSES)),
            # course_lenの抽出
            _Extractor('course_len', '\d+m', 
                       callback=lambda x: float(x[:-1]) / self.__course_len_scaler),
        ]
        texts = df['texts']
        results = pd.concat([instance.extract(texts) for instance in extractors], axis=1)
        results.fillna({'race_class': 'unknown'}, inplace=True)
        results[HorseResultsCols.DATE] = pd.to_datetime(results[HorseResultsCols.DATE], format=self.__date_format)

        return results

    def __preprocess_payback(self, filepath):
        """
        払い戻しテーブルに対する前処理を行う
        
        Parameters
        ----------
        filepath : str
            ファイルパス

        Returns
        -------
        results : pd.DataFrame
            前処理後の払い戻しテーブル
        """
        df = self.__read_pickle(filepath, err_msg='preprocess_payback')
        # column名変更
        df = df.set_axis(['type', 'matched', 'price', 'popular'], axis='columns', copy=False)
        # popularの列を削除
        df.drop(columns='popular', inplace=True)
        # ワイドと複勝を分割
        judge = df['type'].isin(['ワイド', '複勝'])
        filtered_df = df.loc[~judge, :].apply(lambda x: x.str.replace(' ', ''))
        matched = df.loc[judge, 'matched'].replace(r'\s-\s', '-', regex=True).str.split()
        price = df.loc[judge, ['type', 'price']].apply(lambda x: x.str.split() if x.name != 'type' else x)
        modified_df = pd.concat([matched.explode(), price.explode('price')], axis=1)
        # バラバラにしたデータを結合・整形
        results = pd.concat([filtered_df, modified_df]).sort_index().fillna(0)
        results['price'] = results['price'].map(lambda x: int(str(x).replace(',', '')))
        
        return results

        # ワイドと複勝を分割
        judge = df['type'].isin(['ワイド', '複勝'])
        filtered_df = df.loc[~judge, :].apply(lambda x: x.str.replace(' ', ''))
-       matched = df.loc[judge, 'matched'].str.replace(r'\s-\s', '-', regex=True).str.split()
+       matched = df.loc[judge, 'matched'].replace(r'\s-\s', '-', regex=True).str.split()
        price = df.loc[judge, ['type', 'price']].apply(lambda x: x.str.split() if x.name != 'type' else x)
        modified_df = pd.concat([matched.explode(), price.explode('price')], axis=1)
        # バラバラにしたデータを結合・整形

    def __preprocess_horseresults(self, filepath):
        """
        馬の過去成績テーブルに対する前処理を行う
        
        Parameters
        ----------
        filepath : str
            ファイルパス
        Returns
        -------
        results : pd.DataFrame
            前処理後の馬の過去成績テーブル
        """
        df = self.__read_pickle(filepath, err_msg='preprocess_horseresults')
        # columnのスペースを削除
        df = df.set_axis(df.columns.map(lambda x: x.replace(' ', '')), axis='columns', copy=False)
        # RANKが数字以外のものは削除・降着判定されたものも正しく扱う
        df = df[df[HorseResultsCols.RANK].str.contains('^\d+', na=True)].dropna(subset=[
            HorseResultsCols.RANK, HorseResultsCols.R
        ], how='any', axis=0)
        df[HorseResultsCols.RANK] = df[HorseResultsCols.RANK].replace('\D+', '', regex=True).map(int)
        # GROUND_STATEがNaNとなっているものを削除
        df = df.dropna(subset=[HorseResultsCols.GROUND_STATE], how='any', axis=0)
        # PRIZEの値の補完
        df.fillna({HorseResultsCols.PRIZE: 0.0}, inplace=True)
        # 実数を整数に変換
        targets = [
            HorseResultsCols.R, 
            HorseResultsCols.N_HORSES,
            HorseResultsCols.FRAME_NUMBER,
        ]
        df = df[df[targets].isna().sum(axis=1).map(lambda x: not bool(x))]
        df[targets] = df[targets].applymap(int)
        # 日付のフォーマット変更
        df[HorseResultsCols.DATE] = pd.to_datetime(df[HorseResultsCols.DATE], format=self.__date_format)
        # GROUND_STATEのフォーマット変更
        df[HorseResultsCols.GROUND_STATE] = df[HorseResultsCols.GROUND_STATE].map(
            lambda x: Master.GROUND_STATES[x]
        )
        # PLACEのフォーマット変更
        df[HorseResultsCols.PLACE] = df[HorseResultsCols.PLACE].replace('\d', '', regex=True)
        # 抽出する列を指定
        same_cols = [
            HorseResultsCols.DATE,
            HorseResultsCols.PLACE,
            HorseResultsCols.WEATHER,
            HorseResultsCols.R,
            HorseResultsCols.N_HORSES,
            HorseResultsCols.FRAME_NUMBER,
            HorseResultsCols.HORSE_NUMBER,
            HorseResultsCols.WEIGHT,
            HorseResultsCols.GROUND_STATE,
            HorseResultsCols.RANK,
            HorseResultsCols.RANK_DIFF,
            HorseResultsCols.LAST_TIME,
            HorseResultsCols.PRIZE,
        ]
        # 体重を馬体重と体重変化に分割
        weight_diff_df = df[HorseResultsCols.HORSE_WEIGHT].str.extract(
            '(?P\d+)\((?P[-+\d]+)\)'
        )
        weight_diff_df = weight_diff_df.fillna(weight_diff_df.mode().iloc[0]).applymap(float)
        # 距離をレースタイプと全長に分割
        race_type_distance_df = df[HorseResultsCols.COURSE_LEN].str.extract(
            '(?P\w)(?P\d+)'
        )
        race_type_distance_df['race_type'] = race_type_distance_df['race_type'].map(
            lambda x: Master.RACE_TYPES[x]
        )
        # コーナー算出
        corners = df[HorseResultsCols.CORNER].str.split('-').fillna('0')
        several_corner_df = pd.DataFrame({
            'first_corner': corners.map(lambda xs: int(xs[0])),
            'final_corner': corners.map(lambda xs: int(xs[-1])),
            'mean_corner':  corners.map(lambda xs: np.mean(list(map(int, xs))) if isinstance(xs, list) else float(xs)),
        })
        # データ結合
        results = pd.concat([
            df[same_cols].copy(), weight_diff_df, race_type_distance_df, several_corner_df
        ], axis=1)
        results['course_len'] = results['course_len'].map(lambda x: float(x) / self.__course_len_scaler)
        return results

        # columnのスペースを削除
        df = df.set_axis(df.columns.map(lambda x: x.replace(' ', '')), axis='columns', copy=False)
        # RANKが数字以外のものは削除・降着判定されたものも正しく扱う
-       df = df[df[HorseResultsCols.RANK].str.contains('^\d+', na=False)]
-       df[HorseResultsCols.RANK] = df[HorseResultsCols.RANK].str.replace('\D+', '', regex=True).map(int)
-       # HORSE_WEIGHTが不適切なものは削除
-       df = df[df[HorseResultsCols.HORSE_WEIGHT].str.contains('^\d+', na=False)]
+       df = df[df[HorseResultsCols.RANK].str.contains('^\d+', na=True)].dropna(subset=[
+           HorseResultsCols.RANK, HorseResultsCols.R
+       ], how='any', axis=0)
+       df[HorseResultsCols.RANK] = df[HorseResultsCols.RANK].replace('\D+', '', regex=True).map(int)
        # GROUND_STATEがNaNとなっているものを削除
        df = df.dropna(subset=[HorseResultsCols.GROUND_STATE], how='any', axis=0)
        # PRIZEの値の補完

            lambda x: Master.GROUND_STATES[x]
        )
        # PLACEのフォーマット変更
-       df[HorseResultsCols.PLACE] = df[HorseResultsCols.PLACE].str.replace('\d', '', regex=True)
+       df[HorseResultsCols.PLACE] = df[HorseResultsCols.PLACE].replace('\d', '', regex=True)
        # 抽出する列を指定
        same_cols = [
            HorseResultsCols.DATE,

    def __preprocess_peds(self, filepath):
        """
        血統テーブルに対する前処理を行う
        
        Parameters
        ----------
        filepath : str
            ファイルパス

        Returns
        -------
        results : pd.DataFrame
            前処理後の血統テーブル
        """
        df = self.__read_pickle(filepath, err_msg='preprocess_peds')
        # 対象とする「n親等」までの情報を抽出
        judge = lambda xs: np.nan if (isinstance(xs, float) and np.isnan(xs)) or xs[1] > self.__max_relatives else xs[0]
        results = df.applymap(judge).dropna(axis=1)
        
        return results

    def __custom_aggregation(self, _df, target_columns, group_columns):
        """
        対象のhorse_idとdateに対する集計処理
        Parameters
        ----------
        _df : pd.Series
            集計時の評価データが含まれるDataFrame
        target_columns : list
            結合時のcolumn
        group_columns : list
            グループ化して集計する際に対象とするcolumn
        Returns
        -------
        results : pd.DataFrame
            集計結果のDataFrame
        """
        def _calc_mean_mode(base_series, target_df, columns):
            """
            平均値と最頻値を算出
            Parameters
            ----------
            base_series : pd.Series or None
                集計対象のSeries
            target_df : pd.DataFrame
                集計対象のDataFrame
            columns : list
                集計対象の列名のリスト
            Returns
            -------
            output : pd.Series
                出力結果
            """
            @dataclass
            class _CondParams:
                name: str
                condition: Any
            if base_series is None:
                # 集計時の制約条件なしの場合
                _params = _CondParams(
                    name='',
                    condition=map(bool, np.ones(len(target_df), dtype=np.int64)),
                )
            else:
                # 集計時の制約条件ありの場合
                _params = _CondParams(
                    name=base_series.name,
                    condition=target_df[base_series.name].isin(base_series.values),
                )
            # 集計方法の決定（True: 何もしない, False: mean）
            # pd.to_numericで無効値がない（すべて数値である）場合、平均値を算出
            agg_table = target_df[columns].apply(
                lambda _series: pd.to_numeric(_series, errors='coerce').isna().sum() > 0
            ).to_dict()
            # 集計対象の分類
            mean_info = {key: f'{_params.name}{key}_mean' for key, value in agg_table.items() if not value}
            mean_cols = mean_info.keys()
            # 集計処理
            ret_mean = target_df.loc[_params.condition, mean_cols].agg('mean').rename(mean_info)
            # 出力結果の生成
            output = ret_mean.dropna()
            if output.empty:
                output = pd.Series(dtype=object)
            return output
        # 集計対象の情報を取得
        name = HorseResultsCols.DATE
        date = _df[name].values[0]
        horse_id = _df['horse_id'].values
        # 過去成績テーブルのフィルタリング
        target_df = self.__horseresults[
            self.__horseresults.index.isin(horse_id) & (self.__horseresults[name] < date)
        ]
        # ========
        # 集計処理
        # ========
        if target_df.empty:
            results = pd.DataFrame({'dummy': [0]})
        else:
            outputs = {'dummy': 0}
            _kwargs = {'target_df': target_df, 'columns': target_columns}
            # 制約条件なしの場合の処理
            outputs.update(_calc_mean_mode(None, **_kwargs).to_dict())
            # group_columnsの各要素に対する処理
            _agg_series = _df[group_columns].agg(lambda _series: _calc_mean_mode(_series, **_kwargs))
            outputs.update(_agg_series.stack().reset_index(level=1, drop=True).to_dict())
            # 結果の集計
            results = pd.DataFrame({str(key): [value] for key, value in outputs.items()})
        return results.set_index('dummy')
    def merge(self, target_columns, group_columns):
        """
        データ結合
        Parameters
        ----------
        target_columns : list
            結合時のcolumn
        group_columns : list
            グループ化して集計する際に対象とするcolumn
        Returns
        -------
        result : _MergedData
            _MergedDataのインスタンス
                df: 結合後のデータ
                table: 列名とPythonコードの変数名との対応関係
        """
        # レース結果テーブルとレース情報テーブルの結合
        df = self.__results.merge(self.__raceinfo, left_index=True, right_index=True).reset_index()
        # 馬IDを基準に血統情報テーブルを結合（欠損値は最頻値で置換）
        df = df.merge(self.__peds.reset_index(drop=False), on='horse_id').fillna('mode')
        # 集計対象に存在する馬IDのみ抽出
        valid_horse_ids = self.__horseresults.index.unique()
        filtered_df = df[df['horse_id'].isin(valid_horse_ids)]
        # 集計
        _groupby_target = ['horse_id', HorseResultsCols.DATE]
        _extraction_cols = _groupby_target + group_columns
        aggregated_df = filtered_df.groupby(by=_groupby_target, group_keys=True)[_extraction_cols].apply(
            self.__custom_aggregation, target_columns=target_columns, group_columns=group_columns
        ).reset_index().drop(columns=['dummy'])
        # 集計結果を結合
        merged_df = df.merge(aggregated_df, on=_groupby_target, how='outer')
        filled_agg_df = pd.concat([
            merged_df[df.columns],
            merged_df.loc[:, merged_df.columns.str.match('.*mean$')].fillna(0),
        ], axis=1).set_index('race_id')
        # インスタンス生成
        result = _MergedData(**{
            'df': filled_agg_df,
            'table': self.__create_column_df(HorseResultsCols, filled_agg_df.columns, '結合後テーブル'),
        })
        return result

    def drop(self, columns):
        """
        指定されたcolumnを削除

        Parameters
        ----------
        columns : list
            削除対象のcolumn

        Returns
        -------
        self : FeatureEngineering
            FeatureEngineeringのインスタンス
        """
        self.features.drop(columns=columns, inplace=True)

        return self

    def dumminize(self, columns):
        """
        指定されたcolumnをダミー変数化

        Parameters
        ----------
        columns : list
            ダミー変数化の対象となるcolumn

        Returns
        -------
        self : FeatureEngineering
            FeatureEngineeringのインスタンス
        """
        if len(columns) > 0:
            dummy_df = pd.get_dummies(self.features[columns])
            extracted_df = self.features.loc[:, ~self.features.columns.isin(columns)].copy()
            self.features = pd.concat([extracted_df, dummy_df], axis=1)

        return self

    def encode_jockey_id(self):
        """
        jockey id をラベルエンコーディング

        Returns
        -------
        self : FeatureEngineering
            FeatureEngineeringのインスタンス
        """
        # jockey idを変換
        name = 'jockey_id'
        encoder = LabelEncoder()
        original = self.features[name].values
        transformed = encoder.fit_transform(original)
        df = pd.DataFrame({
            name: original,
            'encoded_id': transformed,
        })
        # 対応関係を保存
        df.to_pickle(SystemPaths.ENCODED_JOCKEYID_PATH)
        # 変換結果を格納
        self.features[name] = transformed
        self.features[name] = self.features[name].astype('category')

        return self

    def encode_horse_id(self):
        """
        horse id をラベルエンコーディング

        Returns
        -------
        self : FeatureEngineering
            FeatureEngineeringのインスタンス
        """
        # 集計対象のcolumnの定義
        columns = self.features.columns
        _peds = columns[columns.str.match('^peds_.*')].to_list()
        horse_ids = ['horse_id'] + _peds
        # 対象データの取得
        data = self.features[horse_ids].stack().reset_index(drop=True).unique()
        # ラベルエンコーディングの実施
        encoder = LabelEncoder()
        transformed = encoder.fit_transform(data)
        # 対応関係を保存
        df = pd.DataFrame({
            'horse_id': data,
            'encoded_id': transformed,
        })
        df.to_pickle(SystemPaths.ENCODED_HORSEID_PATH)
        # 変換結果を格納
        self.features[horse_ids] = self.features[horse_ids].apply(lambda series: encoder.transform(series))
        self.features[horse_ids] = self.features[horse_ids].astype('category')

        return self

    def label_encoding(self, columns):
        """
        指定されたcolumnをラベルエンコーディング

        Parameters
        ----------
        columns : list
            ラベルエンコーディングの対象となるcolumn

        Returns
        -------
        self : FeatureEngineering
            FeatureEngineeringのインスタンス
        """
        encoder = LabelEncoder()

        for name in columns:
            self.features[name] = encoder.fit_transform(self.features[name])
            self.features[name] = self.features[name].astype('category')

        return self

# インスタンス生成
preprocess = Preprocess()
# 各テーブルの前処理（整形処理）
manager = preprocess.preprocess_rawdata(since_=datetime(_year, 1, 1))
columns_info = manager.get_columns()
# 「馬の過去成績」を対象としたデータ結合
horseresults = columns_info.horseresults
target_columns = [horseresults[key][0] for key in Params.TARGET_COLUMNS]
group_columns =  [horseresults[key][0] for key in Params.GROUP_COLUMNS]
merged_data = manager.merge(target_columns, group_columns)

# ===========
# 特徴量生成
# ===========
table_df = merged_data.table
# ラベルエンコーディング＆ダミー変数化の処理による特徴量生成
functions = {
    'dumminize':        {'columns': []},
    'label_encoding':   {'columns': [table_df[key][0] for key in Params.CATEGORY_COLUMNS]},
    'encode_jockey_id': {},
    'encode_horse_id':  {},
    'drop':             {'columns': table_df['DATE'].to_list()},
}
inputs = manager.get_features(merged_data.df, functions)

# functionsのフォーマット
functions = {
    function_name: {argument_name: argument_data},
}
# function_name: 呼び出すメソッドの名称
#   例： 'drop'
# argument_name: メソッドの引数名
#   例： 'columns'
# argument_data: 引数の内容
#   例： table_df['DATE'].to_list()