More than 3 years have passed since last update.

ディープインパクト産駒の早熟性を検証(スレイピング含む&テスト大盛)してみる。

Last updated at 2021-07-20Posted at 2021-07-18

はじめに

先週に「仕事ではじめる機械学習第2版」を読みました。「帯」の「機械学習でいい感じにしてくれ」と上司に言われたらのイメージとはかなり異なる(あくまで私の感想)ような気もしますが、機械学習を組み込んだシステムの設計、実装、運用の勘所が丁度よい深さで説明されていて、とても勉強になりました。
一番印象に残っているのは、「データサイエンティスト VS ソフトウェアエンジニア」と題された箇所です。

機械学習のワークフローでは、データサイエンティストなどと呼ばれる人々とソフトウェアエンジニアなどと呼ばれる人々との間で、共同作業が必要となります。しかし、ここで大きな問題が1つあります。それは、データサイエンティストと呼ばれる人々の多くがアナリストや研究者出身であり、多くの場合、ソフトウェア開発に関する知識や経験は比較的少ないという点です。

確かにそうですよね、経験豊富なソフトウェアエンジニアは、様々なシステムの設計、実装、テスト、保守を経験しており、どう実装すれば変更に強く、技術的負債がたまりにくいシステムになるかとの問いに関する答えを持っているはずです。データサイエンティストの歴史はまだ浅いし、得意分野も求められる役割も大きく異なるわけで、どうしても対立構造になりがちですよね。まあこれは顧客とソフトウェアエンジニアや営業とエンジニアの関係にも言えることですが、お互いに違いがあることを認識した上で、協力していくことが重要であると心得ることがプロジェクト成功の鍵ということだと思います。

Python入門(pirnt少なめ、assert多め)では、「Jupyter Notebook」上でassertを書いてテストを行うとの内容を投稿させていただきましたが、今回はロジックを複数のpyファイルに分割し、対応するpyファイルのテストを実装するとの内容となります。

本投稿の具体的な内容ですが、「netkeiba」からデータを収集(スクレイピング)し、ディープインパクト産駒の賞金獲得率(年単位、年月単位)を算出し、その値を分析(といっても折れ線グラフ作るだけですが・・・)することで、ディープインパクト産駒の早熟性を検証してみます。といっても、分析の色合いはあまり強くなく、スレイピングと収集したデータの集計処理とそのテスト(たぶん大盛)がメインとなっておりますので、pythonのユニットテストに関する投稿と生暖かく見ていただければ幸いです。

全てのソースはgithubに登録しております。

本投稿の作成理由(競馬ファン視点)

競馬ファン以外でも「ディープインパクト」をご存じの方は多いと思います。種牡馬としても優秀で、国内での産駒重賞勝利数は263勝、産駒G1級競走勝利は65勝とすごい数値となっています。国外でもディープインパクト産駒のスノーフォールが現地時間7月17日に愛オークスを8.1/2馬身差で圧勝しています。スノーフォールは前走の英オークスでも16馬身差で圧勝しており、今年の凱旋門賞の制覇に期待が高まっています。

そんな優秀なディープインパクトなのですが、牡馬産駒は「ディープタイマー」搭載と揶揄される事も多いのが事実です。
(本投稿で記載されている数値や各種データは2021年7月現在の値となります。)

「ディープタイマー」ですが、ディープインパクトの牡馬は、成長して体重増えると筋肉が固くなって能力が発揮できにくくなる。牝馬は筋肉が柔らかいから古馬になっても能力が発揮できるとの意味で利用されます。

日本ダービーを優勝した産駒は、「ディープブリランテ」、「キズナ」、「マカヒキ」、「ワグネリアン」、「ロジャーバローズ」、「コントレイル」、「シャフリヤール」の7頭です。ダービーは満年齢で3歳時に実施され、この7頭の中で4歳以降で勝ち星を挙げている産駒は「キズナ」だけで、1勝のみとなっております。「コントレイル」、「シャフリヤール」はまだ現役&若いので一括りにしてしまうのは微妙ですし、「ディープブリランテ」と「ロジャーバローズ」は3歳で引退しておりますので、誰もが納得できるデータではないとも言えます。

2010年以降のダービー馬の12頭が古馬になってG1を勝利したのは3頭のみですが、12頭中の7頭がディープインパクト産駒で、ディープインパクト産駒以外で4歳以降にG1勝利がないのは2頭のみであることは事実ですので、ディープインパクト産駒の早熟さは疑いのない事実です。商業的には早熟な産駒のほうが需要が高いのも事実です。サラブレッドを競走馬としてどこかに預託したら毎月結構な額の資金が必要ですので、早期に投資を回収できるのは大きな魅力です。

ディープインパクト産駒で4歳以降にG1を勝利した牡馬は、「フィエールマン」、「ミッキーアイル」、「アルアイン」、「ダノンキングリー」、「ダノンシャーク」、「ワールドプレミア」、「サトノアラジン」、「グローリーヴェイズ」、「トーセンラー」、「スピルバーグ」の10頭です。
牝馬は、「ジェンティルドンナ」、「グランアレグリア」、「ショウナンパンドラ」、「ヴィルシーナ」、「マリアライト」、「ラヴズオンリーユー」、「ラキシス」、「レイパパレ」、「ヴィブロス」の9頭です。

4歳以降にG1を複数回勝利した産駒は、「ジェンティルドンナ」、「グランアレグリア」、「フィエールマン」、「ヴィルシーナ」、「マリアライト」となります。「フィエールマン」だけが牡馬なので大物産駒に関しては「ディープタイマー」が当てはまると言えます。全産駒の早熟性に関しては前述の事実だけで証明されているとは言い難いですので検証したいと思います。比較する種牡馬は「キングカメハメハ」と「ハーツクライ」となります。

ファイル構成の概説

複数のpyファイルで処理を構成し、テスタブル、実装効率、変更への強さを意識した構成となっております。

各種クラスや処理の概説

SeleniumのWebDriverのファサードクラス
テスタブルでDRY(Don't repeat yourself)なコードの構成にするために、SeleniumのWebDriverのファサードクラスを作成します。
関連するファイルのパスの処理用のユーティリティクラス
ファイルパスを各種処理で意識する必要がないようにとの意図のクラスとなります。
CSVファイル処理用のユーティリティクラス
CSVのオープン(io.TextIOWrapper)、行データの書き込み、CSVのクローズ処理を含みます。
対象の種牡馬の産駒一覧を収集しCSVを出力する処理
「netkeiba」の競走馬検索画面 https://db.netkeiba.com/?pid=horse_list で種牡馬名(父名)を指定して検索行い、結果をCSV(horse_base.csv)に出力します。
ディープインパクトが処理対象の場合のパスは、プロジェクトのルートパス/datas/ディープインパクト/horse_base.csvとなります。
対象の種牡馬の産駒一覧の各競走馬の競争成績を収集しCSVを出力する処理
horse_base.csvの各行のhorse_idに対応する競走馬の競争成績を取得し、horse_id.csvを出力します。
ディープインパクトの産駒であるジェンティルドンナ(horse_idは2009106253)が処理対象の場合のパスは、プロジェクトのルートパス/datas/ディープインパクト/detail/2009106253.csvとなります。
各競走馬の競争成績から賞金獲得率(年単位、年月単位)を算出しCSVを出力する処理
各競走馬の競争成績のCSVファイルを読み込んで、賞金獲得率(年単位、年月単位)のCSVファイルを出力します。
各種処理を呼び出して結果のグラフを表示する処理

ファイル構成

sire_nameは種牡馬名、getは産駒を意味する単語として利用しています。
実施の動作確認はWindows上で実施しておりますので、プロジェクトルート直下にchromedriver.exeを配置する必要があります。

プロジェクトルート
├─datas
├─getmodule
│  │─analyze.py
│  │─csv_util.py
│  │─file_path_util.py
│  │─get_collector.py
│  │─get_collector_executer.py
│  │─get_detail_collector.py
│  │─get_list_analyzer.py
│  │─get_row_analyzer.py
│  │─horse_search.py
│  │─race_analyze_executer.py
│  │─race_analyzer.py
│  │─scraping.py
│  └─web_driver_facade.py
├─tests
│  │─datas
│  │  └─ナリタブライアン
│  │  　  └─growth_data.csv
│  │─__init__.py
│  │─test_csv_util.py
│  │─test_file_path_util.py
│  │─test_get_collector.py
│  │─test_get_collector_executer.py
│  │─test_get_detail_collector.py
│  │─test_get_list_analyzer.py
│  │─test_get_row_analyzer.py
│  │─test_horse_search.py
│  │─test_race_analyze_executer.py
│  │─test_race_analyzer.py
│  │─test_web_driver_facade.py
├─chromedriver.exe

SeleniumのWebDriverのファサードクラス

seleniumのwebdriverをそれぞれの処理から呼び出すのは、様々な点から問題がありますので、webdriverをラップしたファサードクラスを実装します。

WebDriverFacadeクラス

web_driver_facade.py

from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait

class WebDriverFacade:
    driver: webdriver = None

    @staticmethod
    def init_driver() -> None:
        if WebDriverFacade.driver == None:
            WebDriverFacade.driver = webdriver.Chrome()

    @staticmethod
    def quit() -> None:
        WebDriverFacade.driver.quit()
        WebDriverFacade.driver = None

    # 引数で指定されたurlに対してgetリクエストを送信します。
    @staticmethod
    def get(url:str) -> None:
        WebDriverFacade.init_driver()
        WebDriverFacade.driver.get(url)

    # 引数で指定されたcss_selectorに対応するエレメントのtext属性の値を返却します。
    @staticmethod
    def get_text(css_selector:str) -> str:
        text = WebDriverFacade.driver.find_element_by_css_selector(css_selector)
        return text.get_attribute("textContent")

    # 引数で指定されたcss_selectorに対応するテキストボックスにテキストを入力します。
    @staticmethod
    def send_keys(css_selector:str, input_value:str) -> None:
        input_box = WebDriverFacade.driver.find_element_by_css_selector(css_selector)
        input_box.clear()
        input_box.send_keys(input_value)

    # 指定されたcss_selectorに対応するエレメントをクリックします。
    @staticmethod
    def click(css_selector:str) -> None:
        target = WebDriverFacade.driver.find_element_by_css_selector(css_selector)
        target.click()

    # 指定されたcss_selectorに対応するエレメントをJavaScriptでクリックします。
    @staticmethod
    def click_button_js(css_selector:str) -> None:
        botton = WebDriverFacade.driver.find_element_by_css_selector(css_selector)
        WebDriverFacade.driver.execute_script('arguments[0].click();', botton)

    # 引数で指定されたテキストのリンクをクリックします。
    @staticmethod
    def click_by_link_text(link_text:str) -> None:
        link = WebDriverFacade.driver.find_element_by_link_text(link_text)
        link.click()

    # 引数で指定されたcss_selectorに対応するエレメントが表示されまでウエイトします。
    @staticmethod
    def wait_until(css_selector:str, timeout:int = 10) -> None:
        WebDriverWait(WebDriverFacade.driver, timeout).until(lambda x: x.find_element_by_css_selector(css_selector))

    # 現在開いているページのソースの文字列を返却します。
    @staticmethod
    def get_page_source() -> str:
        return WebDriverFacade.driver.page_source.encode("utf-8")

WebDriverFacadeクラスのテスト

test_web_driver_facade.py

import pytest
from getmodule.web_driver_facade import WebDriverFacade
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException, TimeoutException

def test_init_driver(mocker):
    webdriver_mock = mocker.Mock()
    init_driver_mock = mocker.patch.object(webdriver, 'Chrome')
    init_driver_mock.return_value = webdriver_mock

    # テスト対象の処理 WebDriverFacade.init_driver()を呼び出し
    WebDriverFacade.init_driver()

    # WebDriverFacade.driverにwebdriver_mockがセットされていること
    assert WebDriverFacade.driver == webdriver_mock

    # init_driver_mockが一回だけ呼び出されていること
    assert init_driver_mock.call_count == 1

def test_quit(mocker):
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    quit_mock = mocker.patch.object(webdriver_mock, 'quit')
   
    # テスト対象の処理 WebDriverFacade.quit()を呼び出し
    WebDriverFacade.quit()
    
    # quit_mockが一回だけ呼び出されていること
    assert quit_mock.call_count == 1

def test_get(mocker):
    # WebDriverFacade.driverにモックをセット    
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    get_mock = mocker.patch.object(webdriver_mock, 'get')
    get_mock.return_value = True
   
    dummy_url = 'dummy_url'
    
    # テスト対象の処理 WebDriverFacade.get(dummy_url)を呼び出し
    WebDriverFacade.get(dummy_url)

    # get_mockが引数：dummy_urlで一回だけ呼び出されていること
    get_mock.assert_called_once_with(dummy_url)

def test_get_text(mocker):
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    text_mock = mocker.Mock()
    
    # WebDriverFacade.driver.find_element_by_css_selectorの結果にtext_mockをセット
    find_element_by_css_selector_mock = mocker.patch.object(webdriver_mock, 'find_element_by_css_selector')
    find_element_by_css_selector_mock.return_value = text_mock

    dummy_css_selector = 'dummy_css_selector'
    dummy_text = 'dummy_text'

    # text_mock.get_attributeをモック化
    get_attribute_mock = mocker.patch.object(text_mock, 'get_attribute')
    get_attribute_mock.return_value = dummy_text

    # テスト対象の処理 WebDriverFacade# テスト対象の処理 WebDriverFacade.get_text(dummy_css_selector)を呼び出し結果がdummy_textと一致すること
    assert WebDriverFacade.get_text(dummy_css_selector) == dummy_text

    # find_element_by_css_selector_mockがdummy_css_selectorを引数として一回呼び出されていることを検証
    find_element_by_css_selector_mock.assert_called_once_with(dummy_css_selector)

    # get_attribute_mockが'textContent'を引数として一回呼び出されていることを検証
    get_attribute_mock.assert_called_once_with('textContent')

def test_send_keys(mocker):
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    input_box = mocker.Mock()

    find_element_by_css_selector_mock = mocker.patch.object(webdriver_mock, 'find_element_by_css_selector')
    find_element_by_css_selector_mock.return_value = input_box
   
    clear_mock = mocker.patch.object(input_box, 'clear')
    send_keys_mock = mocker.patch.object(input_box, 'send_keys')

    dummy_css_selector = 'dummy_css_selector'
    dummy_input_value = 'dummy_input_value'

    # テスト対象の処理 WebDriverFacade.send_keysを呼び出し
    WebDriverFacade.send_keys(dummy_css_selector, dummy_input_value)

    find_element_by_css_selector_mock.assert_called_once_with(dummy_css_selector)
    clear_mock.assert_called_once_with()
    send_keys_mock.assert_called_once_with('dummy_input_value')

def test_click(mocker):
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    target = mocker.Mock()

    find_element_by_css_selector_mock = mocker.patch.object(webdriver_mock, 'find_element_by_css_selector')
    find_element_by_css_selector_mock.return_value = target
   
    click_mock = mocker.patch.object(target, 'click')

    dummy_css_selector = 'dummy_css_selector'
    
    WebDriverFacade.click(dummy_css_selector)

    find_element_by_css_selector_mock.assert_called_once_with(dummy_css_selector)
    click_mock.assert_called_once()
    
def test_click_button_js(mocker):
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    button = mocker.Mock()

    find_element_by_css_selector_mock = mocker.patch.object(webdriver_mock, 'find_element_by_css_selector')
    find_element_by_css_selector_mock.return_value = button
   
    execute_script_mock = mocker.patch.object(webdriver_mock, 'execute_script')
    
    dummy_css_selector = 'dummy_css_selector'
    
    WebDriverFacade.click_button_js(dummy_css_selector)

    find_element_by_css_selector_mock.assert_called_once_with(dummy_css_selector)
    execute_script_mock.assert_called_once_with('arguments[0].click();', button)

def test_click_by_link_text(mocker):
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    link = mocker.Mock()

    dumy_link_text = 'dumy_link_text'
    
    find_element_by_link_text_mock = mocker.patch.object(webdriver_mock, 'find_element_by_link_text')
    find_element_by_link_text_mock.return_value = link
   
    click_mock = mocker.patch.object(link, 'click')
    
    WebDriverFacade.click_by_link_text(dumy_link_text)

    find_element_by_link_text_mock.assert_called_once_with(dumy_link_text)
    assert click_mock.call_count == 1

def test_wait_until(mocker):
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    dummy_element = mocker.Mock()

    dummy_css_selector = 'dummy_css_selector'
    
    # 初回はNoSuchElementExceptionがスロー、2回目dummy_elementが返却されるように振る舞いをセット
    find_element_by_css_selector_mock = mocker.patch.object(webdriver_mock, 'find_element_by_css_selector', side_effect=[NoSuchElementException,dummy_element])
    find_element_by_css_selector_mock.return_value = dummy_element
    
    WebDriverFacade.wait_until(dummy_css_selector)

    # 呼び出し回数は2であること   
    assert find_element_by_css_selector_mock.call_count == 2

def test_wait_until_fail(mocker):
    webdriver_mock = mocker.Mock()
    WebDriverFacade.driver = webdriver_mock

    dummy_css_selector = 'dummy_css_selector'
    
    # NoSuchElementExceptionがスローされるように振る舞いをセット
    find_element_by_css_selector_mock = mocker.patch.object(webdriver_mock, 'find_element_by_css_selector', side_effect=NoSuchElementException)
    
    timeout = 3
    POLL_FREQUENCY = 0.5

    # TimeoutExceptionが発生すること
    with pytest.raises(TimeoutException):
        WebDriverFacade.wait_until(dummy_css_selector, timeout)

    # 呼び出し回数はtimeout / POLL_FREQUENCY = 6であること    
    assert find_element_by_css_selector_mock.call_count == timeout / POLL_FREQUENCY

#関連するファイルのパスの処理用のユーティリティクラス
関連するファイルのパスを生成するクラスとなります。当然なのですが、複数の処理で共通に利用するファイルのパスを各処理が意識する必要がある構成では変更に弱いです。
各ファイルの利用目的はのちほど説明させていただきます。

FilePathUtil

csv_util.py

import os

class FilePathUtil:
    CSV_FILE_PATH_TEMPLATE = '{}/{}.csv'

    @staticmethod
    def get_sire_dir(sire_name:str) -> str:
        # 種牡馬のディレクトリを返却
        return 'datas/{}'.format(sire_name)

    @staticmethod
    def create_sire_dir(sire_name:str) -> str:
        sire_dir = FilePathUtil.get_sire_dir(sire_name)
        # 種牡馬のディレクトリを作成
        os.makedirs(sire_dir, exist_ok=True)

    @staticmethod
    def get_horse_base_csv_path(sire_name:str) -> str:
        # datas/種牡馬名/horse_base.csvの文字列を返却
        return FilePathUtil.CSV_FILE_PATH_TEMPLATE.format(FilePathUtil.get_sire_dir(sire_name), 'horse_base')

    @staticmethod
    def get_growth_data_csv_path(sire_name:str) -> str:
        # datas/種牡馬名/growth_data.csvの文字列を返却
        return FilePathUtil.CSV_FILE_PATH_TEMPLATE.format(FilePathUtil.get_sire_dir(sire_name), 'growth_data')

    @staticmethod
    def get_growth_montly_data_csv_path(sire_name:str, accumulate_flag:bool, gender:str) -> str:
        # 年月単位の集計結果のCSVファイル名を生成して返却
        file_name = 'growth_data_montly'
        if accumulate_flag:
            if not gender:
                file_name = file_name + '_accumulate'
            else:
                file_name = file_name + '_accumulate_' + gender
        else:
            if gender:
                file_name = file_name + '_' + gender
                
        return FilePathUtil.CSV_FILE_PATH_TEMPLATE.format(FilePathUtil.get_sire_dir(sire_name), file_name)

    @staticmethod
    def get_sire_detail_dir(sire_name:str) -> str:
        # 種牡馬毎の詳細ディレクトリを返却
        # datas/種牡馬名/detail
        return 'datas/{}/detail'.format(sire_name)

    @staticmethod
    def create_sire_detail_dir(sire_name:str) -> str:
        # 種牡馬毎の詳細ディレクトリを作成
        sire_detail_dir = FilePathUtil.get_sire_detail_dir(sire_name)
        os.makedirs(sire_detail_dir, exist_ok=True)
        return sire_detail_dir

    @staticmethod
    def get_horse_detail_csv_path(sire_name:str, horse_id:str) -> str:
        # horse_idに対応する競走馬の詳細(競争)データのcsvのファイルパスを返却
        csv_path = FilePathUtil.CSV_FILE_PATH_TEMPLATE.format(FilePathUtil.get_sire_detail_dir(sire_name), horse_id)
        return csv_path

FilePathUtilのテスト

test_file_path_util.py

import os
import pytest
from getmodule.file_path_util import FilePathUtil

SIRE_NAME = 'sire_name'
SIRE_DETAIL_DIR = 'datas/sire_name/detail'

def test_get_sire_dir():
    assert FilePathUtil.get_sire_dir(SIRE_NAME) == 'datas/sire_name'

def test_create_sire_dir(mocker):
    # os.makedirsをモック化
    makedirs_mock = mocker.patch.object(os, 'makedirs')

    # テスト対象の処理を呼び出し
    FilePathUtil.create_sire_dir(SIRE_NAME)
    
    # os.makedirsが引数'datas/sire_name', exist_ok=Trueで一回だけ呼び出されていることを検証
    makedirs_mock.assert_called_once_with('datas/sire_name', exist_ok=True)

def test_get_horse_base_csv_path():
    assert FilePathUtil.get_horse_base_csv_path(
        SIRE_NAME) == 'datas/sire_name/horse_base.csv'

def test_get_growth_data_csv_path():
    assert FilePathUtil.get_growth_data_csv_path(
        SIRE_NAME) == 'datas/sire_name/growth_data.csv'

@pytest.mark.parametrize(('accumulate_flag', 'gender', 'expected_file_name'),
                         [(True, '牡', 'growth_data_montly_accumulate_牡'),
                          (True, None, 'growth_data_montly_accumulate'),
                          (False, '牝', 'growth_data_montly_牝'),
                          (False, None, 'growth_data_montly')])
def test_get_growth_montly_data_csv_path(mocker, accumulate_flag, gender, expected_file_name):
    # テストのパラメタを引数と期待値としてFilePathUtil.get_growth_montly_data_csv_pathの結果を検証
    assert FilePathUtil.get_growth_montly_data_csv_path(
        SIRE_NAME, accumulate_flag, gender) == 'datas/sire_name/{}.csv'.format(expected_file_name)

def test_get_sire_detail_dir():
    assert FilePathUtil.get_sire_detail_dir(
        SIRE_NAME) == SIRE_DETAIL_DIR

def test_create_sire_detail_dir(mocker):
    # os.makedirsをモック化
    makedirs_mock = mocker.patch.object(os, 'makedirs')

    assert FilePathUtil.create_sire_detail_dir(
        SIRE_NAME) == SIRE_DETAIL_DIR

    makedirs_mock.assert_called_once_with(SIRE_DETAIL_DIR, exist_ok=True)

def test_get_horse_detail_csv_path():
    assert FilePathUtil.get_horse_detail_csv_path(
        SIRE_NAME, 'horse_id_dummy') == 'datas/sire_name/detail/horse_id_dummy.csv'

#CSVファイル処理用のユーティリティクラス
対象ファイルをオープンし、データの書き込みを行うユーティリティクラスです。

CsvWriterクラス

csv_util.py

import csv
import pandas as pd
from typing import List
from pandas.core.frame import DataFrame
from io import TextIOWrapper

class CsvWriter:
    def __init__(self, file_name:str):
        self.file_name = file_name
        self.csv_file_io:TextIOWrapper = None
        self.csv_writer = None

    def open_file(self, mode:str) -> None:
        # ファイル(file_name)をオープン
        self.csv_file_io = open(self.file_name, mode=mode, encoding='utf-8', newline='')
        # writerを生成
        self.csv_writer = csv.writer(self.csv_file_io, delimiter=',', quotechar='"')

    def close_file(self) -> None:
        if self.csv_file_io != None:
            # ファイル(file_name)をクローズ
            self.csv_file_io.close()

    def writerow(self, csv_row_data:List[str]) -> None:
        self.csv_writer.writerow(csv_row_data)

class CsvReader:
    @staticmethod
    def read_horse_base(csv_path:str) -> DataFrame:
        return pd.read_csv(csv_path, index_col=0, quoting=csv.QUOTE_ALL)

CsvWriterクラスのテスト

import csv
import pandas as pd
import pytest
from typing import List
from pandas.core.frame import DataFrame
from getmodule.csv_util import CsvWriter, CsvReader
from unittest.mock import mock_open, patch
from pandas.util.testing import assert_frame_equal

def test_open_file(mocker):
    # builtins.openをモック化するためのmock_ioを宣言
    mock_io = mock_open()

    # builtins.openをモック化
    with patch('builtins.open', mock_io):
        file_name = 'dummy_file_name'

        # csv.writerをモック化
        csv_writer_mock = mocker.Mock()
        writer_mock = mocker.patch.object(csv, 'writer')
        writer_mock.return_value = csv_writer_mock

        # テスト対象の処理を呼び出し
        csv_writer = CsvWriter(file_name)
        csv_writer.open_file('w')       

        # mock_ioの呼び出し確認
        mock_io.assert_called_once_with(file_name, mode='w', encoding='utf-8', newline='')
        # writer_mockの呼び出し確認
        writer_mock.assert_called_once_with(csv_writer.csv_file_io, delimiter=',', quotechar='"')

        # csv_writer.csv_writerにcsv_writer_mockがセットされているか確認
        assert csv_writer.csv_writer == csv_writer_mock

@pytest.mark.parametrize(('open_flag'), [(True), (False)])
def test_close_file(mocker, open_flag:bool):
    """
        Parameters
        ----------
        open_flag : bool
            Falseの場合はcsv_writer.csv_file_ioがNoneの状態でテストが実行される
    """
    
    file_name = 'dummy_file_name'
    
    # csv_file_io.close()のモック
    csv_file_io_mock = mocker.Mock()
    close_mock = mocker.patch.object(csv_file_io_mock, 'close')

    csv_writer = CsvWriter(file_name)

    # open_flagがTrueの場合はcsv_file_ioにcsv_file_io_mockをセット
    if open_flag:
        csv_writer.csv_file_io = csv_file_io_mock

    # テスト対象の処理を呼び出し
    csv_writer.close_file()

    # close_mockの呼び出し回数を検証
    assert close_mock.call_count == (1 if open_flag else 0)
        
def test_writerow(mocker):
    file_name = 'dummy_file_name'
    
    csv_writer_mock = mocker.Mock()
    writerow_mock = mocker.patch.object(csv_writer_mock, 'writerow')

    csv_writer = CsvWriter(file_name)
    csv_writer.csv_writer = csv_writer_mock

    csv_row_data = ['dummy1', 'dummy2']
    csv_writer.writerow(csv_row_data)

    writerow_mock.assert_called_once_with(csv_row_data)

def test_read_horse_base(mocker):
    csv_path = 'dummy_file_name'
    read_csv_mock = mocker.patch.object(pd, 'read_csv')
    df:DataFrame = pd.DataFrame()
    read_csv_mock.return_value = df

    result = CsvReader.read_horse_base(csv_path)
    assert_frame_equal(result, df)  

    read_csv_mock.assert_called_once_with(csv_path, index_col=0, quoting=csv.QUOTE_ALL)

#対象の種牡馬の産駒一覧を収集しCSVを出力する処理
「netkeiba」の競走馬検索画面 https://db.netkeiba.com/?pid=horse_list で種牡馬名(父名)を指定して検索行い、結果をCSV(horse_base.csv)に出力します。

競走馬の検索項目のフォームは以下のようになっています。

競走馬検索画面の結果一覧は以下のようになっています。

20件毎にページングされる構成となっていますので、ページを移動しながら結果をhorse_base.csvに出力していきます。
画像の「1,791件中21～40件目」の親要素のdivのCSSセレクタが#contents_liquid > div > div > div.pagerとなります。

対象の種牡馬の名前にディープインパクトを指定して処理を実行すると、プロジェクトの配下のdatas/ディープインパクト/horse_base.csvが生成されます。

horse_base.csvのイメージ

horse_id,馬名,性,生年,馬主
2009106253,ジェンティルドンナ,牝,2009,サンデーレーシング
2016104532,グランアレグリア,牝,2016,サンデーレーシング
2013106101,サトノダイヤモンド,牡,2013,サトミホースカンパニー
2017101835,コントレイル,牡,2017,前田晋二
2015105075,フィエールマン,牡,2015,サンデーレーシング

##GetCollectorExecuterクラス

GetCollectorExecuterクラスがdatas/種牡馬の名前/horse_base.csvのファイルのオープンや書き込みを制御します。GetCollectorExecuterのイニシャライザでsire_name(対象の種牡馬の名前)を指定します。
horse_base.csvに書き込むデータは、GetCollectorクラスがnetkeibaから取得します。

get_collector_executer.py

import os
import time
from getmodule.get_collector import GetCollector
from getmodule.csv_util import CsvWriter
from getmodule.file_path_util import FilePathUtil

class GetCollectorExecuter:
    HEADARE = ['horse_id', '馬名', '性', '生年', '馬主']
    
    def __init__(self, sire_name):
        self.sire_name = sire_name

    # datas/種牡馬の名前/horse_base.csvのファイルを作成
    def get_get_dict(self) -> None:
        # 種牡馬のサブディレクトリの作成
        sire_dir = FilePathUtil.get_sire_dir(self.sire_name)
        os.makedirs(sire_dir, exist_ok=True)

        # 産駒のCSVのパスを生成
        horse_base_csv_path = FilePathUtil.get_horse_base_csv_path(self.sire_name)
  
        # horse_base_csv_pathに対するCsvWriterを宣言
        csv_writer = CsvWriter(horse_base_csv_path)
        # ファイルをwモードでオープン
        csv_writer.open_file('w')
        # ヘッダーを出力
        csv_writer.writerow(GetCollectorExecuter.HEADARE)

        # GetCollectorを作成し処理を実施
        get_collector = GetCollector(self.sire_name)

        try:
            while True:
                # 現在の処理対象ページの産駒情報を取得
                get_dict_rsult = get_collector.get_get_dict()
                
                # 全ての処理対象ページの処理が完了した場合はbreak
                if len(get_dict_rsult) == 0:
                    break

                # 現在の処理対象ページの産駒情報をCSVに出力
                for horse_id, horse_datas in get_dict_rsult.items():
                    csv_writer.writerow([horse_id] + horse_datas)

                #「1回URL叩いたら1秒Sleepしましょう」なので1秒スリープ
                time.sleep(1)
        finally:
            # CsvWriterをクローズ
            csv_writer.close_file()

##GetCollectorクラス

GetCollectorクラスでは、対象の種牡馬の名前、対象の種牡馬の検索結果(産駒一覧)の総ページ数、現在の処理対象ページを保持しており、get_get_dictが呼び出されると現在の処理対象ページの情報であるDict[str, List[str]]が返却されます。Dictのキーはhorse_idで、値は[horse_name, horse_gender, birth_year, horse_owner]の文字列のリストとなります。

現在の処理対象ページの情報は、HorseSearchService.get_horse_idsメソッド内でnetkeibaから取得しています。

get_collector.py

from typing import Dict, List
from getmodule.horse_search import HorseSearchService

class GetCollector:
    def __init__(self, sire_name):
        # 対象の種牡馬名
        self.sire_name = sire_name
        # 対象の種牡馬の産駒一覧の総ページ数
        self.total_page = -1
        # 現在の処理対象ページ
        self.current_page = 0

    # 現在の処理対象ページの産駒情報を返却
    def get_get_dict(self) -> Dict[str, List[str]]:
        # total_pageがセットされていない場合はセット
        if self.total_page == -1:
            self.total_page = HorseSearchService.get_total_page_by_sire_name(
                self.sire_name)

        # 全てのページを処理した場合は空の辞書を返却
        if self.total_page == self.current_page:
            return {}
        else:
            try:
                # 現在の処理対象ページの産駒情報を実際に取得しリターン
                return HorseSearchService.get_horse_ids(self.current_page)
            finally:
                self.current_page += 1
            
        print('GetCollector get_get_dict end')

##HorseSearchServiceクラス

horse_search.py

import time
from typing import Counter, Dict, Tuple, List
from getmodule.web_driver_facade import WebDriverFacade
from getmodule.get_list_analyzer import GetListAnalyzer

# pagerのCSSセレクタ
PAGER_CSS_SELECTOR = '#contents_liquid > div > div > div.pager'

class HorseSearchService:
    # 競走馬検索画面で父名指定で検索を実施、検索結果から総ページ数を取得
    @staticmethod
    def get_total_page_by_sire_name(sire_name:str) -> int:      
        # 競走馬検索画面で父名指定で検索
        HorseSearchService.search(sire_name)
        
        # 検索結果からpagerのテキスト(533件中1～20件目の形式)を取得
        pager_text = WebDriverFacade.get_text(PAGER_CSS_SELECTOR)

        # pagerのテキストから総ページ数を取得し返却
        return GetListAnalyzer.get_total_page_no(pager_text)

    # 指定ページの競走馬検索結果の辞書を返却
    @staticmethod
    def get_horse_ids(page:int) -> Dict[str, List[str]]:
        # 現在表示しているページ番号を取得
        pager_text = WebDriverFacade.get_text(PAGER_CSS_SELECTOR)
        current_page_no = GetListAnalyzer.get_page_no(pager_text)
        
        # 現在表示しているページ番号が引数のpageより大きい場合はValueErrorをスロー
        if current_page_no > page:
            raise ValueError('current_page_no={} page={}'.format(current_page_no, page))

        # 現在表示しているページ番号が引数のpageより小さい場合は「次」のリンクをクリックしページを移動
        while(current_page_no < page):
            time.sleep(1)

            print('current_page_no={} page={}'.format(current_page_no, page))
              
            # 「次」のリンクをクリック
            WebDriverFacade.click_by_link_text("次")

            # pagerのテキストを取得し現在表示しているページ番号に反映
            pager_text = WebDriverFacade.get_text(PAGER_CSS_SELECTOR)
            current_page_no = GetListAnalyzer.get_page_no(pager_text)
        
            print('set current_page_no={}'.format(current_page_no))

        # 表示しているページのソースを取得
        html = WebDriverFacade.get_page_source()

        # ページのソースからDict[str, List[str]]を生成して返却
        return GetListAnalyzer.analyze_get_list(html)

    # 競走馬検索画面で父名指定で検索を実施
    @staticmethod
    def search(sire_name:str) -> None: 
        WebDriverFacade.get('https://db.netkeiba.com/?pid=horse_list')
        # 父名を入力
        WebDriverFacade.send_keys('#db_search_detail_form > form > table > tbody > tr:nth-child(2) > td > input', sire_name)
        # 検索ボタンをクリック(JavaScript利用)
        WebDriverFacade.click_button_js('#db_search_detail_form > form > div > input:nth-child(1)')
        # pagerのCSSセレクタに対応する要素が出現するまでウエイト
        WebDriverFacade.wait_until(PAGER_CSS_SELECTOR)

##GetListAnalyzerクラス
対象の種牡馬の検索結果(産駒一覧)の表示内容のHTMLの文字列から総ページ数、現在のページ、産駒一覧を辞書に変換したデータを取得する処理を行います。

get_list_analyzer.py

import re
from typing import Counter, Dict, Tuple, List
from bs4 import BeautifulSoup
from getmodule.get_row_analyzer import GetRowAnalyzer

class GetListAnalyzer:

    # 競走馬検索画面の検索結果のhtmlから産駒一覧の辞書を生成
    @staticmethod
    def analyze_get_list(html:str) -> Dict[str, List[str]]:
        result:Dict[str, List[str]] = {}

        # BeautifulSoupで引数のhtmlをhtmlとして解析するインスタンスを生成
        soup = BeautifulSoup(html, "html.parser")

        # tableタグでclass=nk_tb_common race_table_01の要素を取得
        table = soup.find('table', {'class':'nk_tb_common race_table_01'}).tbody
        
        # trの一覧を取得
        rows = table.find_all('tr')

        # rowsの各要素(競走馬の基本情報)に対する処理
        for row in rows:
            # 現在のrowから競走馬の基本情報を取得
            row_result = GetRowAnalyzer.analyze_row(row)
            
            if len(row_result) > 1:
                # row_result[0]はhorse_id
                # result辞書にキー:horse_id、値:[horse_name, horse_gender, birth_year, horse_owner]をセット
                horse_id = row_result[0]
                horse_data= row_result[1]            
                result[horse_id] = horse_data

        return result

    # 533件中1～20件目 形式の文字列から現在のページを取得
    @staticmethod
    def get_page_no(pager:str) -> int:
        # 件中の後の数値はフォーマットされない
        search_result = re.search(r'\d{1,8}～', pager)
        result = search_result.group(0).replace('～', '')
        
        try:
            return int(int(result) / 20)
        except ValueError:
            return -1

    # 533件中1～20件目 形式の文字列から総ページ数を取得
    @staticmethod
    def get_total_page_no(pager:str) -> int:
        pager = pager.replace('\n', '')
        
        search_result = re.search(r'^\d{1,3}(,\d{1,3})*', pager)
        result = search_result.group(0).replace(',', '')
        
        try:
            if int(result) == 0:
                return 0
            else:
                return int(int(result) / 20) + 1
        except ValueError:
            return 0

##GetRowAnalyzerクラス
競走馬検索画面の検索結果のhtmlの任意の競走馬の行データから、(horse_id, [horse_name, horse_gender, birth_year, horse_owner])のtupleを返却する処理が含まれています。これぐらいだとGetListAnalyzer内で定義した方が良いかもしれません・・・

具体的には、現在のrowに対応する競走馬がラストインパクトの場合は、以下の画像の赤枠部分のrow:Tagがanalyze_rowメソッドに渡されます。

メソッドの戻り値は
('2010104439', ['ラストインパクト', '牡', '2010', 'シルクレーシング'])
とのタプルとなります。

get_row_analyzer.py

from bs4.element import Tag

class GetRowAnalyzer:
    # 引数のrowをtuple (horse_id, [horse_name, horse_gender, birth_year, horse_owner])に変換して返却
    @staticmethod
    def analyze_row(row:Tag) -> tuple:
        # rowの配下のtdを取得
        tds = row.find_all('td')
        
        if(len(tds) >= 10):
            horse_name_td = tds[1]
            horse_gender_td = tds[2]
            birth_year_td = tds[3]
            horse_owner_td = tds[9]

            # horse_name_tdのaタグを取得
            horse_name_link = horse_name_td.find('a')
            # title属性をhorse_nameにセット
            horse_name = horse_name_link['title']
            # href属性からhorse_idを取得
            horse_id = horse_name_link['href'].replace('/horse/', '')
            horse_id = horse_id.replace('/', '')

            horse_gender = horse_gender_td.text
            birth_year = birth_year_td.text
            horse_owner = horse_owner_td.text
            
            # horse_ownerの不要な文字を除去
            horse_owner = horse_owner.replace('\n', '')
            horse_owner = horse_owner.replace('\t', '')

            return (horse_id, [horse_name, horse_gender, birth_year, horse_owner])

        return ()

##GetCollectorExecuterクラスのテスト

test_get_collector_executer.py

import os
from getmodule.get_collector_executer import GetCollectorExecuter
from getmodule.get_collector import GetCollector
from getmodule.get_list_analyzer import GetListAnalyzer
from getmodule.get_detail_collector import GetDetailCollector

from getmodule.csv_util import CsvWriter
from getmodule.file_path_util import FilePathUtil
from unittest.mock import call

def test_get_get_dict(mocker):
    sire_name = "dummy_sire_name"
    sire_dir = "dummy_sire_dir"
    get_horse_base_csv_path = "dummy_get_horse_base_csv_path"

    makedirs_mock = mocker.patch.object(os, 'makedirs')

    get_sire_dir_mock = mocker.patch.object(FilePathUtil, 'get_sire_dir')
    get_sire_dir_mock.return_value = sire_dir

    get_horse_base_csv_path_mock = mocker.patch.object(
        FilePathUtil, 'get_horse_base_csv_path')
    get_horse_base_csv_path_mock.return_value = get_horse_base_csv_path

    open_file_mock = mocker.patch.object(CsvWriter, 'open_file')
    writerow_mock = mocker.patch.object(CsvWriter, 'writerow')

    horse_1_data = ["horse_1_data_1"]
    horse_2_data = ["horse_2_data_1"]

    get_get_dict_mock_result = [
        {'horse_id_1': horse_1_data}, {'horse_id_2': horse_2_data}, {}]

    get_get_dict_mock = mocker.patch.object(
        GetCollector, 'get_get_dict', side_effect=get_get_dict_mock_result)

    close_file_mock = mocker.patch.object(CsvWriter, 'close_file')

    executer = GetCollectorExecuter(sire_name)
    executer.get_get_dict()

    get_sire_dir_mock.assert_called_once_with(sire_name)
    makedirs_mock.assert_called_once_with(sire_dir, exist_ok=True)
    get_horse_base_csv_path_mock.assert_called_once_with(sire_name)
    open_file_mock.assert_called_once_with('w')

    assert writerow_mock.call_args_list == [
        call(GetCollectorExecuter.HEADARE), 
        call(['horse_id_1'] + horse_1_data), 
        call(['horse_id_2'] + horse_2_data)
    ]

    assert get_get_dict_mock.call_count == 3
    assert close_file_mock.call_count == 1

##GetCollectorExecuterクラスのテスト

test_get_collector_executer.py

import pytest
from typing import Counter, Dict, Tuple, List
from getmodule.get_collector import GetCollector
from getmodule.horse_search import HorseSearchService

test_get_get_dict_datas = [
    (1, -1, 0, 1, {"dummy1":["dummy"]}),
    (1, 1, 0, 1, {"dummy2":["dummy"]}),
    (1, 1, 1, 1, {}),
    (3, 3, 1, 2, {"dummy3":["dummy"]})
]

@pytest.mark.parametrize("total_page_mock_result, current_total_page, current_page, expect_current_page, expect", test_get_get_dict_datas)
def test_get_get_dict(mocker, total_page_mock_result:int, current_total_page:int, current_page:int, expect_current_page:int, expect:Dict[str, List[str]]):
    sire_name = 'ナリタブライアン'
    get_collector = GetCollector(sire_name)
    get_collector.total_page = current_total_page
    get_collector.current_page = current_page

    get_total_page_by_sire_name_mock = mocker.patch.object(HorseSearchService, 'get_total_page_by_sire_name')
    get_total_page_by_sire_name_mock.return_value = total_page_mock_result
    
    get_horse_ids_mock = mocker.patch.object(HorseSearchService, 'get_horse_ids')
    get_horse_ids_mock.return_value = expect
    
    assert get_collector.get_get_dict() == expect
    assert get_collector.total_page == total_page_mock_result
    assert get_collector.current_page == expect_current_page

    if current_total_page == -1:
        get_total_page_by_sire_name_mock.assert_called_once_with(sire_name)
    else:
        assert get_total_page_by_sire_name_mock.call_count == 0

    if current_total_page != current_page:
        get_horse_ids_mock.assert_called_once_with(current_page)
    else:
        assert get_horse_ids_mock.call_count == 0

##HorseSearchServiceクラスのテスト

test_horse_search.py

import pytest
from unittest.mock import call
from getmodule.horse_search import HorseSearchService, PAGER_CSS_SELECTOR
from getmodule.web_driver_facade import WebDriverFacade
from getmodule.get_list_analyzer import GetListAnalyzer

def test_get_total_page_by_sire_name(mocker):
    sire_name = 'sire_name'
    get_text_mock_result = 'dummy_pager'

    search_mock = mocker.patch.object(HorseSearchService, 'search')
    get_text_mock = mocker.patch.object(WebDriverFacade, 'get_text')
    get_text_mock.return_value = get_text_mock_result

    get_total_page_no_mock = mocker.patch.object(GetListAnalyzer, 'get_total_page_no')
    get_total_page_no_mock.return_value = 10

    assert HorseSearchService.get_total_page_by_sire_name(sire_name) == 10

    search_mock.assert_called_once_with(sire_name)
    get_text_mock.assert_called_once_with(PAGER_CSS_SELECTOR)
    get_total_page_no_mock.assert_called_once_with(get_text_mock_result)

def test_get_horse_ids_by_sire_name_value_error(mocker):
    get_text_mock = mocker.patch.object(WebDriverFacade, 'get_text')
    get_text_mock.return_value = 'dummy_pager_text'

    get_page_no_mock = mocker.patch.object(GetListAnalyzer, 'get_page_no', side_effect=ValueError)
   
    with pytest.raises(ValueError):
        HorseSearchService.get_horse_ids(1)

    get_text_mock.assert_called_once_with(PAGER_CSS_SELECTOR)
    get_page_no_mock.assert_called_once_with('dummy_pager_text')
    
@pytest.mark.parametrize(('page', 'get_text_result', 'get_page_no_result'), [
    (1, ['dummy_pager_text1'], [1]),
    (2, ['dummy_pager_text1', 'dummy_pager_text2'], [1,2])
])
def test_get_horse_ids_by_sire_name_value(mocker, page, get_text_result, get_page_no_result):
    get_text_mock = mocker.patch.object(WebDriverFacade, 'get_text', side_effect=get_text_result)

    get_page_no_mock = mocker.patch.object(GetListAnalyzer, 'get_page_no', side_effect=get_page_no_result)
   
    click_by_link_text_mock = mocker.patch.object(WebDriverFacade, 'click_by_link_text')

    dummy_html = 'dummy_html'
    get_page_source_mock = mocker.patch.object(WebDriverFacade, 'get_page_source')
    get_page_source_mock.return_value = dummy_html

    analyze_get_list_result = {'dummy_key':['dummy1', 'dummy1']}
    analyze_get_list_mock = mocker.patch.object(GetListAnalyzer, 'analyze_get_list')
    analyze_get_list_mock.return_value = analyze_get_list_result

    assert HorseSearchService.get_horse_ids(page) == analyze_get_list_result

    if len(get_text_result) == 1:
        click_by_link_text_mock.call_count = 0
    else:
        click_by_link_text_mock.assert_called_once_with('次')
    
    assert get_text_mock.call_args_list == [call(PAGER_CSS_SELECTOR) for i in range(len(get_text_result))]
    assert get_page_no_mock.call_args_list == [call(get_text_result[i]) for i in range(len(get_text_result))]


def test_search(mocker):
    sire_name = 'sire_name'

    get_mock = mocker.patch.object(WebDriverFacade, 'get')
    send_keys_mock = mocker.patch.object(WebDriverFacade, 'send_keys')
    click_button_js_mock = mocker.patch.object(WebDriverFacade, 'click_button_js')
    wait_until_mock = mocker.patch.object(WebDriverFacade, 'wait_until')
    
    HorseSearchService.search(sire_name)

    get_mock.assert_called_once_with('https://db.netkeiba.com/?pid=horse_list')
    send_keys_mock.assert_called_once_with('#db_search_detail_form > form > table > tbody > tr:nth-child(2) > td > input', sire_name)
    click_button_js_mock.assert_called_once_with('#db_search_detail_form > form > div > input:nth-child(1)')
    wait_until_mock.assert_called_once_with(PAGER_CSS_SELECTOR)

##GetListAnalyzerクラスのテスト

test_get_list_analyzer.py

import pytest
import re
from typing import Counter, Dict, Tuple, List
from bs4 import BeautifulSoup
from getmodule.get_list_analyzer import GetListAnalyzer
from getmodule.get_row_analyzer import GetRowAnalyzer
from unittest.mock import MagicMock, Mock, PropertyMock
from unittest.mock import call

def test_analyze_get_list(mocker):
    beautifulsoup_mock = mocker.Mock(spec=BeautifulSoup)
    mocker.patch('getmodule.get_list_analyzer.BeautifulSoup', return_value=beautifulsoup_mock)

    table_element = mocker.Mock()
    find_mock = mocker.patch.object(beautifulsoup_mock, 'find')
    find_mock.return_value = table_element

    table_mock = mocker.Mock()
    table_element.tbody = table_mock

    rows_mock_value_0 = mocker.Mock()
    rows_mock_value_1 = mocker.Mock()
    rows_mock_value_2 = mocker.Mock()
   
    find_all_mock = mocker.patch.object(table_mock, 'find_all')
    find_all_mock.return_value = [rows_mock_value_0, rows_mock_value_1, rows_mock_value_2]
    
    row_result_0 = MagicMock()
    row_result_1 = MagicMock()
    row_result_2 = MagicMock()
    row_result = [row_result_0, row_result_1, row_result_2]
    
    analyze_row_mock = mocker.patch.object(GetRowAnalyzer, 'analyze_row', side_effect=row_result)
   
    horse_data_1 = ('horse_id_0', ['horse_id_0_value'])
    row_result_0.__len__.return_value = 2
    row_result_0.__getitem__.side_effect = horse_data_1

   
    horse_data_2 = ('horse_id_1', ['horse_id_1_value'])
    row_result_1.__len__.return_value = 3
    row_result_1.__getitem__.side_effect = horse_data_2
    
    row_result_2.__len__.return_value = 1

    result = GetListAnalyzer.analyze_get_list('dummy_html')
    assert result == {horse_data_1[0]:horse_data_1[1], horse_data_2[0]:horse_data_2[1], }

    find_mock.assert_called_once_with('table', {'class':'nk_tb_common race_table_01'})
    find_all_mock.assert_called_once_with('tr')
    
    assert analyze_row_mock.call_args_list == [
        call(rows_mock_value_0),
        call(rows_mock_value_1),
        call(rows_mock_value_2)
    ]

@pytest.mark.parametrize("pager, expect",  [
    ("10件中1～20件目", 1),
    ("1,001件中21～20件目", 51)
])
def test_get_total_page_no(pager: str, expect: int):
    assert GetListAnalyzer.get_total_page_no(pager) == expect

@pytest.mark.parametrize("pager, expect", [
    ("1件中1～20件目", 0),
    ("1,785件中21～20件目", 1),
    ("1,785件中1000～1020件目", 50)
])
def test_get_page_no(pager: str, expect: int):
    assert GetListAnalyzer.get_page_no(pager) == expect

##GetRowAnalyzerクラスのテスト

test_get_row_analyzer.py

from getmodule.get_row_analyzer import GetRowAnalyzer
from bs4.element import Tag
from bs4.element import PageElement
from unittest.mock import MagicMock, PropertyMock

def test_analyze_row(mocker):
    page_element = PageElement()
    row = Tag(page_element, name='tr')
    
    tds_mock = MagicMock()

    # row.find_allをモック化
    find_all_mock = mocker.patch.object(
        row, 'find_all')

    # row.find_all('td')の結果にtds_mockをセット
    find_all_mock.return_value = tds_mock

    horse_name_td = mocker.Mock()
    horse_gender_td = mocker.Mock()
    birth_year_td = mocker.Mock()
    horse_owner_td = mocker.Mock()

    # tds[1]の結果をhorse_name_td
    # tds[2]の結果をhorse_gender_td
    # tds[3]の結果をbirth_year_td
    # tds[9]の結果をhorse_owner_td
    # とするために、tds_mock.__getitem__.side_effectを指定
    tds_mock.__getitem__.side_effect = [horse_name_td, horse_gender_td, birth_year_td, horse_owner_td]
    # len(tds)を10にセット
    tds_mock.__len__.return_value = 10
    
    horse_name = 'dummy_horse_name'
    horse_id = 'dummy_horse_id'
    horse_href = '/horse/{}/'.format(horse_id)

    horse_name_link = MagicMock()

    # orse_name_td.find('a')をモック化
    find_horse_name_link_mock = mocker.patch.object(
        horse_name_td, 'find')
    # orse_name_td.find('a')の結果にhorse_name_linkをセット
    find_horse_name_link_mock.return_value = horse_name_link

    # horse_name_link['title']の結果にhorse_name
    # horse_name_link['href']horse_hrefをそれぞれセットし
    # analyze_rowの結果のタプルに含まれるhorse_nameとhorse_idを固定化
    horse_name_link.__getitem__.side_effect = [horse_name, horse_href]

    horse_gender = 'dummy_horse_gender'

    # horse_gender_tdのプロパティをモックするためにPropertyMockを使用
    horse_gender_td_text = PropertyMock()
    horse_gender_td_text.return_value = horse_gender
    # horse_gender_td.textの振る舞いを変更するので、type(horse_gender_td).textの値にhorse_gender_td_textをセットする必要がある
    type(horse_gender_td).text = horse_gender_td_text

    # horse_gender_tdと同様にbirth_year_tdも処理
    birth_year = 'dummy_birth_year'
    birth_year_text = PropertyMock()
    birth_year_text.return_value = birth_year
    type(birth_year_td).text = birth_year_text
    
    # horse_gender_tdと同様にhorse_ownerも処理
    horse_owner = 'dummy_horse_owner'
    horse_owner_text = PropertyMock()
    horse_owner_text.return_value = horse_owner
    type(horse_owner_td).text = horse_owner_text

    # テスト対象のGetRowAnalyzer.analyze_rowを呼び出し
    result = GetRowAnalyzer.analyze_row(row)

    assert result[0] == horse_id
    assert result[1] == [horse_name, horse_gender, birth_year, horse_owner]

    find_horse_name_link_mock.assert_called_once_with('a')

対象の種牡馬の産駒一覧の各競走馬の競争成績を収集しCSVを出力する処理

datas/種牡馬の名前/horse_base.csvを読み込み、各行のhourse_idに対応する競走馬の競争成績をnetkeibaから取得しhorse_idに対応するCSVファイルに出力します。

horse_id=2009106253はジェンティルドンナの値となっていますので、ジェンティルドンナの競争成績を取得する場合
https://db.netkeiba.com/horse/2009106253
にアクセスし、表示される競争成績のテーブルをdatas/ディープインパクト/detail/2009106253.csvに出力します。

競争成績のテーブルは以下のようになります。

対象の種牡馬の名前にディープインパクトを指定した場合、以下のようなcsvが出力されます。

datas
├─ディープインパクト/detail
│  ├─2009106253.csv
│  ├─2016104532.csv
│  ├─2013106101.csv
 　　　　　・
 　　　　　・
 　　　　　・

GetDetailCollectorクラス

get_detail_collector.py

import csv
import time
from collections import namedtuple
import pandas as pd
from getmodule.file_path_util import FilePathUtil
from getmodule.csv_util import CsvReader

class GetDetailCollector:
    def __init__(self, sire_name):
        # 対象の種牡馬名
        self.sire_name = sire_name

    def get_get_detail(self) -> None:
        # 対象種牡馬の産駒一覧(horse_base.csv)のパスを取得
        base_csv_path = FilePathUtil.get_horse_base_csv_path(self.sire_name)
        # horse_base.csvを読み込んでDataFrameを生成
        horse_base = CsvReader.read_horse_base(base_csv_path)

        # datas\対象種牡馬名\detailフォルダを作成
        FilePathUtil.create_sire_detail_dir(self.sire_name)

        for horse_id, item in horse_base.iterrows():
            if(not horse_id):
                break
                
            print('get_get_detail horse_id={}'.format(horse_id))

            try:
                time.sleep(1)

                # horse_idに対応する競走馬の競走戦績のtableデータを取得
                race_results = pd.read_html('https://db.netkeiba.com/horse/{}'.format(
                    horse_id), attrs={'class': 'db_h_race_results nk_tb_common'})[0]
                
                # 競走戦績を出力するCSVのパスを取得
                detail_csv_path = FilePathUtil.get_horse_detail_csv_path(
                    self.sire_name, horse_id)
                
                # 競走戦績をCSVに出力
                race_results.to_csv(
                    detail_csv_path, index=False, quoting=csv.QUOTE_ALL)
                    
            except ValueError as e:
                print(e)
            # 競走成績が存在しない
            except ImportError as e:
                print(e)

GetDetailCollectorクラスのテスト

test_get_detail_collector.py

import csv
from typing import Counter, Dict, Tuple, List
from getmodule.horse_search import HorseSearchService
from collections import namedtuple
import pandas as pd
from getmodule.file_path_util import FilePathUtil
from getmodule.csv_util import CsvReader
from getmodule.get_detail_collector import GetDetailCollector
from unittest.mock import call

def test_get_get_dict(mocker):
    sire_name = 'dummy_sire_name'
    base_csv_path = "dummy_get_horse_base_csv_path"

    get_horse_base_csv_path_mock = mocker.patch.object(
        FilePathUtil, 'get_horse_base_csv_path')
    get_horse_base_csv_path_mock.return_value = base_csv_path

    horse_base = pd.DataFrame([["horse_name_{}".format(id)] for id in range(3)],
                              index=["horse_id_{}".format(
                                  id) for id in range(2)] + [''],
                              columns=["馬名"])

    read_horse_base_mock = mocker.patch.object(
        CsvReader, 'read_horse_base')
    read_horse_base_mock.return_value = horse_base

    create_sire_detail_dir_mock = mocker.patch.object(FilePathUtil, 'create_sire_detail_dir')

    race_results_1 = mocker.Mock()
    race_results_list_1 = [race_results_1]
    race_results_2 = mocker.Mock()
    race_results_list_2 = [race_results_2]

    read_html_mock = mocker.patch.object(
        pd, 'read_html', side_effect=[race_results_list_1, race_results_list_2])

    get_horse_detail_csv_path_mock = mocker.patch.object(
        FilePathUtil, 'get_horse_detail_csv_path', side_effect=["detail_csv_path_{}".format(id) for id in range(2)])

    to_csv_mock1 = mocker.patch.object(
        race_results_1, 'to_csv')

    to_csv_mock2 = mocker.patch.object(
        race_results_2, 'to_csv')

    executer = GetDetailCollector(sire_name)
    executer.get_get_detail()

    get_horse_base_csv_path_mock.assert_called_once_with(sire_name)
    read_horse_base_mock.assert_called_once_with(base_csv_path)
    create_sire_detail_dir_mock.assert_called_once_with(sire_name)

    assert read_html_mock.call_args_list == [
        call('https://db.netkeiba.com/horse/horse_id_0', attrs={'class': 'db_h_race_results nk_tb_common'}),
        call('https://db.netkeiba.com/horse/horse_id_1', attrs={'class': 'db_h_race_results nk_tb_common'})
        ]

    assert get_horse_detail_csv_path_mock.call_args_list == [
        call(sire_name, 'horse_id_0'),
        call(sire_name, 'horse_id_1')
    ]

    to_csv_mock1.assert_called_once_with('detail_csv_path_0', index=False, quoting=csv.QUOTE_ALL)
    to_csv_mock2.assert_called_once_with('detail_csv_path_1', index=False, quoting=csv.QUOTE_ALL)

各競走馬の競争成績から賞金獲得率(年単位、年月単位)を算出しCSVを出力する処理

収集した競争成績から、対象種牡馬の産駒の賞金獲得率を算出します。集計単位として、年単位、年月単位、年月単位(累計)、性別指定の年月単位、性別指定の年月単位(累計)が選択可能となります。

|ファイル名|集計単位|
|---|---|---|
|growth_data.csv |年|
|growth_data_montly.csv|年月|
|growth_data_montly_牡.csv|年月(牡馬のみ)|
|growth_data_montly_牝.csv|年月(牝馬のみ)|
|growth_data_montly_accumulate|年月(累計)|
|growth_data_montly_accumulate_牡.csv|年月(牡馬のみ、累計)|
|growth_data_montly_accumulate_牝.csv|年月(牝馬のみ、累計)|

対象の種牡馬の名前にディープインパクトを指定した場合、以下のようなcsvが出力されます。

datas
├─ディープインパクト
│  ├─growth_data.csv
│  ├─growth_data_montly.csv
│  ├─growth_data_montly_牡.csv
│  ├─growth_data_montly_牝.csv
│  ├─growth_data_montly_accumulate.csv
│  ├─growth_data_montly_accumulate_牡.csv
│  ├─growth_data_montly_accumulate_牝csv

賞金獲得率のCSVの例

growth_data.csv

horse_idと年齢(2歳から9歳)に獲得した賞金の全獲得賞金に対する百分率の数値となります。

horse_id,2,3,4,5,6,7,8,9
2009106253,0.5881421494333476,51.40498111158866,25.728956952518832,22.277919786459158,0.0,0.0,0.0,0.0
2016104532,6.480398363878483,23.129564724705595,50.24631634269851,20.143720568717445,0.0,0.0,0.0,0.0
2013106101,1.6413832005585327,76.37795275590553,12.278239882375244,9.702424161160712,0.0,0.0,0.0,0.0

growth_data_montly.csv

horse_idと年齢_月に獲得した賞金の全獲得賞金に対する百分率の数値となります。

horse_id,2_01,2_02,2_03,2_04,2_05,2_06,2_07,2_08,2_09,2_10,2_11,2_12,3_01,3_02,3_03,3_04,3_05,3_06,3_07,3_08,3_09,3_10,3_11,3_12,4_01,4_02,4_03,4_04,4_05,4_06,4_07,4_08,4_09,4_10,4_11,4_12,5_01,5_02,5_03,5_04,5_05,5_06,5_07,5_08,5_09,5_10,5_11,5_12,6_01,6_02,6_03,6_04,6_05,6_06,6_07,6_08,6_09,6_10,6_11,6_12,7_01,7_02,7_03,7_04,7_05,7_06,7_07,7_08,7_09,7_10,7_11,7_12,8_01,8_02,8_03,8_04,8_05,8_06,8_07,8_08,8_09,8_10,8_11,8_12,9_01,9_02,9_03,9_04,9_05,9_06,9_07,9_08,9_09,9_10,9_11,9_12
2009106253,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.21112795107863763,0.37701419835471006,2.8352975773067612,0.0,0.3845544823218042,8.769350253730556,9.468636188838872,0.0,0.0,0.0,3.8313690893599053,6.995875464669999,19.119898055360764,0.0,0.0,0.0,0.0,0.0,0.0,2.513629063270523,0.0,0.0,0.0,4.079595237556647,19.13573265169166,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,6.943998310976392,15.333921475482766,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0
2016104532,0.0,0.0,0.0,0.0,0.0,0.776148727836796,0.0,0.0,0.0,3.691585106371184,0.0,2.012664529670503,0.0,0.0,0.0,14.395230455189056,1.1642230917551941,0.0,0.0,0.0,0.0,0.0,0.0,7.570111177761345,0.0,0.0,5.8907470875019,0.0,0.0,14.75414380261873,0.0,0.0,0.0,14.791398941554895,14.81002651102298,0.0,0.0,0.0,0.0,2.217567793819417,12.066008122950832,5.860144651947192,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0
2013106101,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.8091325636556147,0.832250636902918,0.0,4.430463147479437,0.0,3.350271174999191,10.288698498712325,0.0,0.0,0.0,6.366486191574849,16.87168544624817,0.0,35.070348296891545,0.0,0.0,7.82546779421216,4.452772088163084,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,1.860773715675441,0.0,0.0,0.0,0.0,0.0,0.0,7.841650445485272,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0

growth_data_montly_accumulate.csv

horse_idと年齢_月までに獲得した賞金の全獲得賞金に対する百分率の数値となります。

horse_id,2_01,2_02,2_03,2_04,2_05,2_06,2_07,2_08,2_09,2_10,2_11,2_12,3_01,3_02,3_03,3_04,3_05,3_06,3_07,3_08,3_09,3_10,3_11,3_12,4_01,4_02,4_03,4_04,4_05,4_06,4_07,4_08,4_09,4_10,4_11,4_12,5_01,5_02,5_03,5_04,5_05,5_06,5_07,5_08,5_09,5_10,5_11,5_12,6_01,6_02,6_03,6_04,6_05,6_06,6_07,6_08,6_09,6_10,6_11,6_12,7_01,7_02,7_03,7_04,7_05,7_06,7_07,7_08,7_09,7_10,7_11,7_12,8_01,8_02,8_03,8_04,8_05,8_06,8_07,8_08,8_09,8_10,8_11,8_12,9_01,9_02,9_03,9_04,9_05,9_06,9_07,9_08,9_09,9_10,9_11,9_12
2009106253,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.21112795107863763,0.5881421494333476,3.423439726740109,3.423439726740109,3.8079942090619134,12.57734446279247,22.04598065163134,22.04598065163134,22.04598065163134,22.04598065163134,25.877349740991246,32.87322520566124,51.993123261022006,51.993123261022006,51.993123261022006,51.993123261022006,51.993123261022006,51.993123261022006,51.993123261022006,54.50675232429253,54.50675232429253,54.50675232429253,54.50675232429253,58.58634756184917,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,77.72208021354083,84.66607852451723,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0,100.0
2016104532,0.0,0.0,0.0,0.0,0.0,0.776148727836796,0.776148727836796,0.776148727836796,0.776148727836796,4.46773383420798,4.46773383420798,6.480398363878483,6.480398363878483,6.480398363878483,6.480398363878483,20.87562881906754,22.039851910822733,22.039851910822733,22.039851910822733,22.039851910822733,22.039851910822733,22.039851910822733,22.039851910822733,29.609963088584077,29.609963088584077,29.609963088584077,35.500710176085974,35.500710176085974,35.500710176085974,50.2548539787047,50.2548539787047,50.2548539787047,50.2548539787047,65.0462529202596,79.85627943128259,79.85627943128259,79.85627943128259,79.85627943128259,79.85627943128259,82.073847225102,94.13985534805283,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003,100.00000000000003
2013106101,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.8091325636556147,1.6413832005585327,1.6413832005585327,6.07184634803797,6.07184634803797,9.42211752303716,19.710816021749487,19.710816021749487,19.710816021749487,19.710816021749487,26.077302213324337,42.94898765957251,42.94898765957251,78.01933595646406,78.01933595646406,78.01933595646406,85.84480375067622,90.29757583883931,90.29757583883931,90.29757583883931,90.29757583883931,90.29757583883931,90.29757583883931,90.29757583883931,90.29757583883931,90.29757583883931,90.29757583883931,90.29757583883931,92.15834955451474,92.15834955451474,92.15834955451474,92.15834955451474,92.15834955451474,92.15834955451474,92.15834955451474,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000001,100.00000000000000

RaceAnalyzeExecuterクラス

競争成績が少ない競走馬の早熟性を判断することは不可能ですので、2016年以前に生まれた競走馬のみを処理対象としています。

race_analyze_executer.py

from typing import List
from getmodule.get_collector import GetCollector
from getmodule.csv_util import CsvWriter
from getmodule.race_analyzer import RaceAnalyzer
from getmodule.file_path_util import FilePathUtil
from getmodule.csv_util import CsvReader


class RaceAnalyzeExecuter:
    ANALYZE_START_YEAR = 2016
    
    def __init__(self, sire_name):
        self.sire_name = sire_name

    def execute(self, monthly:bool, accumulate_flag:bool = False, gender:str = None) -> None:
        """
        対象の種牡馬の産駒(複数)の成長曲線(賞金ベース)を描くためのデータを集計しCSVに出力

        Parameters
        ----------
        monthly : bool
            集計単位が月の場合Trueを指定
        accumulate_flag : bool
            累積のデータを作成する場合はTrueを指定
        gender : str
            性別を指定して集計したい場合に利用 牡 or 牝を指定可能
        """

        # datas/種牡馬名/detailのフォルダを作成
        FilePathUtil.create_sire_detail_dir(self.sire_name)

        # 集計結果のCSVを書き出すクラスのインスタンスを生成
        if monthly:
            csv_writer = CsvWriter(FilePathUtil.get_growth_montly_data_csv_path(self.sire_name, accumulate_flag, gender))
        else:
            csv_writer = CsvWriter(FilePathUtil.get_growth_data_csv_path(self.sire_name))

        # 集計結果のCSVをwモードでオープン
        csv_writer.open_file('w')
        
        try:     
            # 集計結果のCSVのヘッダーの生成とCSVへの書き込み
            index_list :List[str] = RaceAnalyzer.create_index(monthly)
            csv_writer.writerow(index_list)

            # datas/種牡馬名/horse_base.csvをDataFrameに読み込む
            base_csv_path =FilePathUtil.get_horse_base_csv_path(self.sire_name)
            horse_base = CsvReader.read_horse_base(base_csv_path)

            for index,item in horse_base.iterrows():
                birth_year = int(item['生年'])
                horse_gender = item['性']

                # 引数のgenderが指定されている AND 現在の処理対象の競走馬の性と一致しない場合は処理対象外
                if gender != None and horse_gender != gender:
                    continue  
               
                # 競争成績が少ない競走馬は集計対象外とするif文
                if(birth_year <= RaceAnalyzeExecuter.ANALYZE_START_YEAR):
                    try:
                        # 実際の集計
                        if monthly:
                            current_horse_result = RaceAnalyzer.analyze_race_montly_data(self.sire_name, index, birth_year, accumulate_flag)
                        else:
                            current_horse_result = RaceAnalyzer.analyze_race_data(self.sire_name, index, birth_year)
                        
                        # 集計結果をCSVに出力
                        values = current_horse_result.values.tolist()
                        csv_writer.writerow([index] + values)
                    except OSError as e:
                        print(e)
                    except KeyError as e:
                        print(e)
                    except ValueError as e:
                        print(e)                   
        finally:
            #CSVファイルをクローズ
            csv_writer.close_file()

RaceAnalyzerクラス

race_analyzer.py

import os
from typing import List
from selenium.webdriver.support.ui import WebDriverWait
from getmodule.web_driver_facade import WebDriverFacade
from getmodule.file_path_util import FilePathUtil
import pandas as pd
from pandas.core.frame import Series


class RaceAnalyzer:
    START_AGE: int = 2
    END_AGE: int = 10

    @staticmethod
    def analyze_race_data(sire_name: str, horse_id: str, birth_year: int) -> Series:
        """
        対象の種牡馬の対象の産駒の成長曲線(年毎の賞金ベース)を描くためのデータを集計

        Parameters
        ----------
        sire_name : str
            種牡馬名
        horse_id : str
            対象の競走馬を識別するためのhorse_id
        birth_year : int
            対象の競走馬の生年
        """

        # horse_idに対応する競争データのCSVであるdatas/種牡馬名/detail/horse_id.csvのパスを取得
        horse_detail_csv = FilePathUtil.get_horse_detail_csv_path(
            sire_name, horse_id)

        if not os.path.isfile(horse_detail_csv):
            raise OSError('file not exist {}'.format(horse_detail_csv))

        # horse_detail_csvの読み込み
        race_data = pd.read_csv(horse_detail_csv, thousands=',')

        # race_dataにrace_day:datetimeを追加
        race_data['race_day'] = pd.to_datetime(race_data['日付'])

        # 賞金列の欠損値を0に置換しfloatに変換
        race_data.fillna(value={'賞金': '0'}, inplace=True)
        race_data['賞金'] = race_data['賞金'].astype(float)

        # 総賞金を産出
        total_prize_money = race_data['賞金'].sum()

        # 総賞金が0の競走馬は処理対象外なのでValueErrorをスロー
        if(total_prize_money == 0):
            raise ValueError("total_prize_money is zero")

        # 年毎の賞金の合計を算出
        prize_money_year = race_data.groupby(
            race_data['race_day'].dt.strftime('%Y'))['賞金'].sum()
        # prize_money_year.indexをintに変換
        prize_money_year.index = prize_money_year.index.astype(int)

        target_age = 2

        # 総賞金に対する年毎の賞金の割合を格納する辞書
        prize_money_percentage = {}

        # 2歳から9歳までの処理ループ
        for year in range(birth_year + RaceAnalyzer.START_AGE, birth_year + RaceAnalyzer.END_AGE):
            try:
                # target_ageの賞金の割合を計算して格納
                prize_money_percentage[target_age] = prize_money_year[year] * \
                    100 / total_prize_money
            except KeyError:
                # target_ageに対応する年の賞金が存在しない場合は0をセット
                prize_money_percentage[target_age] = float(0)
            finally:
                target_age += 1

        # 結果のSeriesを生成してリターン
        return pd.Series(data=prize_money_percentage, name=horse_id)

    @staticmethod
    def analyze_race_montly_data(sire_name: str, horse_id: str, birth_year: str, accumulate_flag: bool) -> Series:
        """
        対象の種牡馬の対象の産駒の成長曲線(年月毎の賞金ベース)を描くためのデータを集計

        Parameters
        ----------
        sire_name : str
            種牡馬名
        horse_id : str
            対象の競走馬を識別するためのhorse_id
        birth_year : int
            対象の競走馬の生年
        accumulate_flag : bool
            累積値で集計する場合にTrueを指定
        """

        horse_detail_csv = FilePathUtil.get_horse_detail_csv_path(
            sire_name, horse_id)

        if not os.path.isfile(horse_detail_csv):
            raise OSError('file not exist {}'.format(horse_detail_csv))

        birth_year_int: int = int(birth_year)

        race_data = pd.read_csv(horse_detail_csv, thousands=',')

        race_data.head()

        race_data['race_day'] = pd.to_datetime(race_data['日付'])

        race_data.fillna(value={'賞金': '0'}, inplace=True)
        race_data['賞金'] = race_data['賞金'].astype(float)

        total_prize_money = race_data['賞金'].sum()

        if(total_prize_money == 0):
            raise ValueError("total_prize_money is zero")

        prize_money_year_month = race_data.groupby(
            race_data['race_day'].dt.strftime('%Y/%m'))['賞金'].sum()

        sum_value: float = 0

        target_age = 2
        prize_money_percentage = {}

        for year in range(birth_year_int + RaceAnalyzer.START_AGE, birth_year_int + RaceAnalyzer.END_AGE):
            for month in range(1, 13):
                target_age_month_key_1 = "{}_{:02}".format(target_age, month)
                target_age_month_key_2 = "{}/{:02}".format(year, month)
                try:
                    current_value = prize_money_year_month[target_age_month_key_2] * \
                        100 / total_prize_money

                    prize_money_percentage[target_age_month_key_1] = sum_value + \
                        current_value

                    if accumulate_flag:
                        sum_value = sum_value + current_value
                except KeyError:
                    prize_money_percentage[target_age_month_key_1] = sum_value
 
            target_age += 1
        result = pd.Series(data=prize_money_percentage, name=horse_id)

        return result

    @staticmethod
    def create_index(monthly: bool) -> List[str]:
        target_age = 1
        index_list: List[str] = []

        if monthly:
            for year in range(RaceAnalyzer.START_AGE, RaceAnalyzer.END_AGE):
                target_age += 1
                for month in range(1, 13):
                    target_age_month_key_1 = "{}_{:02}".format(
                        target_age, month)
                    index_list.append(target_age_month_key_1)

            return ['horse_id'] + index_list
        else:
            return ['horse_id'] + [str(i) for i in range(RaceAnalyzer.START_AGE, RaceAnalyzer.END_AGE)]

RaceAnalyzeExecuterクラスのテスト

test_race_analyze_executer.py

from typing import Counter, Dict, Tuple, List
from getmodule.get_collector import GetCollector
from getmodule.race_analyze_executer import RaceAnalyzeExecuter
from getmodule.file_path_util import FilePathUtil
from pandas.testing import assert_series_equal
import pandas as pd
from getmodule.csv_util import CsvWriter
from getmodule.csv_util import CsvReader
from pandas.core.frame import DataFrame
from pandas.core.frame import Series
from unittest.mock import call
from getmodule.race_analyzer import RaceAnalyzer

HORSE_ID_TEMPLATE = 'horse_id_{}'

def test_analyze_race_data_year(mocker):
    create_sire_detail_dir_mock = mocker.patch.object(FilePathUtil, 'create_sire_detail_dir')
    
    get_growth_data_csv_path_result = 'get_growth_data_csv_path_result'

    get_growth_data_csv_path_mock = mocker.patch.object(FilePathUtil, 'get_growth_data_csv_path')
    get_growth_data_csv_path_mock.return_value = get_growth_data_csv_path_result

    csv_writer_mock = mocker.Mock(spec=CsvWriter)
    mocker.patch('getmodule.race_analyze_executer.CsvWriter', return_value=csv_writer_mock)

    open_file_mock = mocker.patch.object(csv_writer_mock, 'open_file')
    
    create_index_mock = mocker.patch.object(RaceAnalyzer, 'create_index')
    create_index_reslt = ['create_index_reslt']
    create_index_mock.return_value = create_index_reslt

    writerow_mock = mocker.patch.object(csv_writer_mock, 'writerow')

    base_csv_path = "dummy_base_csv_path"
    get_horse_base_csv_path_mock = mocker.patch.object(FilePathUtil, 'get_horse_base_csv_path')
    get_horse_base_csv_path_mock.return_value = base_csv_path

    horse_base_num = 4

    horse_base:DataFrame = pd.DataFrame([
        [RaceAnalyzeExecuter.ANALYZE_START_YEAR+1-id, '牡' if id % 2 == 0 else '牝'] for id in range(horse_base_num)],
                              index=[HORSE_ID_TEMPLATE.format(
                                  id) for id in range(horse_base_num)],
                              columns=['生年', '性'])

    read_horse_base_mock = mocker.patch.object(CsvReader, 'read_horse_base')
    read_horse_base_mock.return_value = horse_base

    current_horse_result_list = [pd.Series(data={'2':id}, name='data{}'.format(id)) for id in range(1, horse_base_num)]

    analyze_race_data_mock = mocker.patch.object(RaceAnalyzer, 'analyze_race_data', side_effect=current_horse_result_list)

    close_file_mock = mocker.patch.object(csv_writer_mock, 'close_file')

    sire_name = "dummy_sire_name"
    race_analyze_executer = RaceAnalyzeExecuter(sire_name)
    race_analyze_executer.execute(monthly=False)

    create_sire_detail_dir_mock.assert_called_once_with(sire_name)
    open_file_mock.assert_called_once_with('w')
  
    get_horse_base_csv_path_mock.assert_called_once_with(sire_name)
    read_horse_base_mock.assert_called_once_with(base_csv_path)

    assert analyze_race_data_mock.call_args_list == [
        call(sire_name, HORSE_ID_TEMPLATE.format(id), RaceAnalyzeExecuter.ANALYZE_START_YEAR+1-id) for id in range(1, horse_base_num)
    ]

    close_file_mock.assert_called_once_with()
   
    assert writerow_mock.call_args_list == [
        call(create_index_reslt),
        call(['horse_id_1'] + current_horse_result_list[0].values.tolist()),
        call(['horse_id_2'] +current_horse_result_list[1].values.tolist()),
        call(['horse_id_3'] +current_horse_result_list[2].values.tolist())
    ]
    
def test_analyze_race_data_monthly(mocker):
    create_sire_detail_dir_mock = mocker.patch.object(FilePathUtil, 'create_sire_detail_dir')
    
    get_growth_montly_data_csv_path_result = 'get_growth_montly_data_csv_path_result'

    get_growth_montly_data_csv_path_mock = mocker.patch.object(FilePathUtil, 'get_growth_data_csv_path')
    get_growth_montly_data_csv_path_mock.return_value = get_growth_montly_data_csv_path_result

    csv_writer_mock = mocker.Mock(spec=CsvWriter)
    mocker.patch('getmodule.race_analyze_executer.CsvWriter', return_value=csv_writer_mock)

    open_file_mock = mocker.patch.object(csv_writer_mock, 'open_file')

    create_index_mock = mocker.patch.object(RaceAnalyzer, 'create_index')
    create_index_reslt = ['create_index_reslt']
    create_index_mock.return_value = create_index_reslt

    writerow_mock = mocker.patch.object(csv_writer_mock, 'writerow')

    base_csv_path = "dummy_base_csv_path"
    get_horse_base_csv_path_mock = mocker.patch.object(FilePathUtil, 'get_horse_base_csv_path')
    get_horse_base_csv_path_mock.return_value = base_csv_path

    horse_base_num = 4

    horse_base:DataFrame = pd.DataFrame([
        [RaceAnalyzeExecuter.ANALYZE_START_YEAR+1-id, '牝' if id % 2 == 0 else '牡'] for id in range(horse_base_num)],
                              index=[HORSE_ID_TEMPLATE.format(
                                  id) for id in range(horse_base_num)],
                              columns=['生年', '性'])

    read_horse_base_mock = mocker.patch.object(CsvReader, 'read_horse_base')
    read_horse_base_mock.return_value = horse_base

    current_horse_result_list = [pd.Series(data={'2':id}, name='data{}'.format(id)) for id in range(1, horse_base_num)]

    analyze_race_montly_data_mock = mocker.patch.object(RaceAnalyzer, 'analyze_race_montly_data', side_effect=current_horse_result_list)

    close_file_mock = mocker.patch.object(csv_writer_mock, 'close_file')

    sire_name = "dummy_sire_name"
    race_analyze_executer = RaceAnalyzeExecuter(sire_name)
    race_analyze_executer.execute(monthly=True, accumulate_flag=True, gender='牡')

    create_sire_detail_dir_mock.assert_called_once_with(sire_name)
    open_file_mock.assert_called_once_with('w')
  
    get_horse_base_csv_path_mock.assert_called_once_with(sire_name)
    read_horse_base_mock.assert_called_once_with(base_csv_path)

    assert analyze_race_montly_data_mock.call_args_list == [
        call(sire_name, "horse_id_1", 2016, True),
        call(sire_name, "horse_id_3", 2014, True)
    ]

    close_file_mock.assert_called_once_with()
   
    assert writerow_mock.call_args_list == [
        call(create_index_reslt), 
        call(['horse_id_1'] + current_horse_result_list[0].values.tolist()),
        call(['horse_id_3'] +current_horse_result_list[1].values.tolist())
    ]

RaceAnalyzerクラスのテスト

test_race_analyzer.py

import pytest
from typing import Counter, Dict, Tuple, List
from getmodule.get_collector import GetCollector
from getmodule.race_analyzer import RaceAnalyzer
from getmodule.file_path_util import FilePathUtil
import time
from pandas.testing import assert_series_equal
import pandas as pd
from pandas.core.frame import Series

# 年単位の集計のテスト
def test_analyze_race_data(mocker):
    horse_id  = '1999104283'

    # 'testdata/detail/1999104283.csv'が対象CSVになるようにモックをセット
    get_horse_detail_csv_path_mock = mocker.patch.object(
        FilePathUtil, 'get_horse_detail_csv_path')
    get_horse_detail_csv_path_mock.return_value = 'testdata/detail/{}.csv'.format(horse_id)

    sire_name = "dummy_sire_name"

    # テスト対象の処理の呼び出し
    result = RaceAnalyzer.analyze_race_data(sire_name, horse_id, 1999)

    # 期待値の値を準備
    expected = pd.Series([0.0, 0.0, 26.887965282174093, 32.18371822043618, 22.208765700549577, 18.719550796840164, 0.0, 0.0], index=[2, 3, 4, 5, 6, 7, 8, 9], name=horse_id)
    
    # 期待値と一致することを検証
    assert_series_equal(result, expected)

# 年月単位の集計のテスト
def test_analyze_race_montly_data(mocker):
    horse_id  = '1999104641'

    get_horse_detail_csv_path_mock = mocker.patch.object(
        FilePathUtil, 'get_horse_detail_csv_path')
    get_horse_detail_csv_path_mock.return_value = 'testdata/detail/{}.csv'.format(horse_id)
    birth_year_int = 1999

    sire_name = "dummy_sire_name"

    result = RaceAnalyzer.analyze_race_montly_data(sire_name, horse_id, birth_year_int, False)


    index_list :List[str] = RaceAnalyzer.create_index(True)
    index_list = index_list[1:]

    expected = pd.Series([0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 74.97656982193065, 0.0, 0.0, 0.0, 0.0, 18.775382692908465, 6.248047485160887, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], index=index_list, name=horse_id)
    assert_series_equal(result, expected)

    result = RaceAnalyzer.analyze_race_montly_data(sire_name, horse_id, birth_year_int, True)
   
    expected = pd.Series([0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 74.97656982193065, 74.97656982193065, 74.97656982193065, 74.97656982193065, 74.97656982193065, 93.75195251483912, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0], index=index_list, name=horse_id)
    assert_series_equal(result, expected)

各種処理を呼び出して結果のグラフを表示する処理

scraping.py

GetCollectorExecuterとGetDetailCollectorを呼び出す処理となります。
プロジェクトルートに移動し、python -m getmodule.scrapingで実行します。

scraping.py

import traceback
from getmodule.get_collector_executer import GetCollectorExecuter
from getmodule.get_detail_collector import GetDetailCollector
from getmodule.web_driver_facade import WebDriverFacade

def main():
    target_sires = ['ディープインパクト', 'キングカメハメハ', 'ステイゴールド', 'ハーツクライ', 'クロフネ', 'スクリーンヒーロー']
    
    # datas/種牡馬名/horse_base.csvの作成
    for sire_name in target_sires:
        try:
            get_collector_executer = GetCollectorExecuter(sire_name)
            get_collector_executer.get_get_dict()
            WebDriverFacade.quit()
        except Exception :
            print(traceback.format_exc())
    
    # datas/種牡馬名/horse_base.csvの各行の競走馬のデータ(datas/種牡馬名/detail/horse_id.csv)を取得
    for sire_name in target_sires:
        try:
            get_detail_collector = GetDetailCollector(sire_name)
            get_detail_collector.get_get_detail()
        except Exception :
            print(traceback.format_exc())
            
if __name__ == "__main__":
    main()

analyze.py

RaceAnalyzeExecuterを呼び出す処理となります。
プロジェクトルートに移動し、python -m getmodule.analyzeで実行します。

analyze.py

from getmodule.race_analyze_executer import RaceAnalyzeExecuter

def main():
    target_sires = ['ディープインパクト', 'キングカメハメハ', 'ステイゴールド', 'ハーツクライ', 'クロフネ', 'スクリーンヒーロー']
    
    for sire_name in target_sires:
        race_analyze_executer = RaceAnalyzeExecuter(sire_name)
        # 年データ(非累積)の集計
        race_analyze_executer.execute(False, False)
        
        # 年月データ(非累積)の集計
        race_analyze_executer.execute(True, False)
        race_analyze_executer.execute(True, False, '牡')
        race_analyze_executer.execute(True, False, '牝')
        
        # 年月データ(累積)の集計
        race_analyze_executer.execute(True, True)
        race_analyze_executer.execute(True, True, '牡')
        race_analyze_executer.execute(True, True, '牝')
            
if __name__ == "__main__":
    main()

集計結果の検証

Jupyter Notebookで集計結果のCSVを読み込んで結果を検証してみます。全然DRYじゃないのですが・・・

年単位の集計

ディープインパクト、キングカメハメハ、ハーツクライの比較

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# ディープインパクトの年単位の集計結果CSVを読み込み
growth_data_deep_impact = pd.read_csv('datas/ディープインパクト/growth_data.csv')

# 折れ線グラフを描画するためにhorse_idカラムを削除
growth_data_deep_impact.drop(['horse_id'], axis=1, inplace=True)

# 平均値を取得
growth_data_mean_deep_impact = growth_data_deep_impact.mean()

# キングカメハメハも同様に処理
growth_data_king_kamehameha = pd.read_csv('datas/キングカメハメハ/growth_data.csv')
growth_data_king_kamehameha.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_king_kamehameha = growth_data_king_kamehameha.mean()

# ハーツクライも同様に処理
growth_data_hart_cry = pd.read_csv('datas/ハーツクライ/growth_data.csv')
growth_data_hart_cry.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_hart_cry = growth_data_hart_cry.mean()

result = pd.DataFrame({
    'ディープインパクト': growth_data_mean_deep_impact,
    'キングカメハメハ': growth_data_mean_king_kamehameha,
    'ハーツクライ': growth_data_mean_hart_cry
})

result.plot().grid()

結果は以下のようになりました。

差異が小さく分かりにくいのですね、それほどディープインパクト産駒の早熟性を決定付ける結果ではないです・・・、ディープインパクト産駒でも半分近くがJRA未勝利馬で、その大半が3歳で引退してしまう or 地方に移籍して4歳以降の獲得賞金なしとの状態ですので、全体だとこんな感じになるのも納得はいきます。

ディープインパクト、キングカメハメハ、ハーツクライの上位100頭の比較

4歳以降も現役である可能性が高い賞金獲得額の上位100頭で比較してみます。

growth_data_deep_impact = pd.read_csv('datas/ディープインパクト/growth_data.csv')
growth_data_deep_impact.drop(['horse_id'], axis=1, inplace=True)
growth_data_deep_impact = growth_data_deep_impact.drop(growth_data_deep_impact.index[range(100, len(growth_data_deep_impact))])
growth_data_mean_deep_impact = growth_data_deep_impact.mean()

growth_data_king_kamehameha = pd.read_csv('datas/キングカメハメハ/growth_data.csv')
growth_data_king_kamehameha.drop(['horse_id'], axis=1, inplace=True)
growth_data_king_kamehameha = growth_data_king_kamehameha.drop(growth_data_king_kamehameha.index[range(100, len(growth_data_king_kamehameha))])
growth_data_mean_king_kamehameha = growth_data_king_kamehameha.mean()

growth_data_hart_cry = pd.read_csv('datas/ハーツクライ/growth_data.csv')
growth_data_hart_cry.drop(['horse_id'], axis=1, inplace=True)
growth_data_hart_cry = growth_data_hart_cry.drop(growth_data_hart_cry.index[range(100, len(growth_data_hart_cry))])
growth_data_mean_hart_cry = growth_data_hart_cry.mean()

result = pd.DataFrame({
    'ディープインパクト_上位100頭': growth_data_mean_deep_impact,
    'キングカメハメハ_上位100頭': growth_data_mean_king_kamehameha,
    'ハーツクライ_上位100頭': growth_data_mean_hart_cry
})

result.plot().grid()

結果は以下のようになりました。

ディープインパクト産駒の早熟性が見て取れるグラフになっていると言えます。

年月単位の集計

ディープインパクト、キングカメハメハ、ハーツクライの比較

growth_data_deep_impact = pd.read_csv('datas/ディープインパクト/growth_data_montly.csv')
growth_data_deep_impact.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_deep_impact = growth_data_deep_impact.mean()

growth_data_king_kamehameha = pd.read_csv('datas/キングカメハメハ/growth_data_montly.csv')
growth_data_king_kamehameha.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_king_kamehameha = growth_data_king_kamehameha.mean()

growth_data_hart_cry = pd.read_csv('datas/ハーツクライ/growth_data_montly.csv')
growth_data_hart_cry.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_hart_cry = growth_data_hart_cry.mean()

result = pd.DataFrame({
    'ディープインパクト': growth_data_mean_deep_impact,
    'キングカメハメハ': growth_data_mean_king_kamehameha,
    'ハーツクライ': growth_data_mean_hart_cry
})
result.plot().grid()

結果は以下のようになりました。

3歳の前半に大きな山があるのは事実ですが。年単位の結果とほぼ同じで、全体的に早熟かどうかは微妙な結果となります。

ディープインパクト、キングカメハメハ、ハーツクライの上位100頭の比較

growth_data_deep_impact = pd.read_csv('datas/ディープインパクト/growth_data_montly.csv')
growth_data_deep_impact.drop(['horse_id'], axis=1, inplace=True)
growth_data_deep_impact = growth_data_deep_impact.drop(growth_data_deep_impact.index[range(100, len(growth_data_deep_impact))])
growth_data_mean_deep_impact = growth_data_deep_impact.mean()

growth_data_king_kamehameha = pd.read_csv('datas/キングカメハメハ/growth_data_montly.csv')
growth_data_king_kamehameha.drop(['horse_id'], axis=1, inplace=True)
growth_data_king_kamehameha = growth_data_king_kamehameha.drop(growth_data_king_kamehameha.index[range(100, len(growth_data_king_kamehameha))])
growth_data_mean_king_kamehameha = growth_data_king_kamehameha.mean()

growth_data_hart_cry = pd.read_csv('datas/ハーツクライ/growth_data_montly.csv')
growth_data_hart_cry.drop(['horse_id'], axis=1, inplace=True)
growth_data_hart_cry = growth_data_hart_cry.drop(growth_data_hart_cry.index[range(100, len(growth_data_hart_cry))])
growth_data_mean_hart_cry = growth_data_hart_cry.mean()

result = pd.DataFrame({
    'ディープインパクト_上位100頭': growth_data_mean_deep_impact,
    'キングカメハメハ_上位100頭': growth_data_mean_king_kamehameha,
    'ハーツクライ_上位100頭': growth_data_mean_hart_cry
})
result.plot().grid()

結果は以下のようになりました。

ディープインパクト産駒の早熟性が見て取れるグラフになっていると言えます。年単位のグラフよりもインパクト大きいでね。
とはいえ、5_05の前後でも、それほど大きくないですが山があることも事実です。

性別の比較

「ディープタイマー」の存在の確認と社台系のクラブ（社台RH、サンデーレーシング、キャロット、シルク）の牝馬は「６歳３月」までしか現役を続けられないとの事実から、
牡馬と牝馬の比較も行ってみます。まあ「６歳３月」以降もバリバリ賞金稼げる馬なんてほぼいないので、クラブのくだりは不要かもしれませんが・・・

growth_data_deep_impact_1 = pd.read_csv('datas/ディープインパクト/growth_data_montly_牡.csv')
growth_data_deep_impact_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_deep_impact_1 = growth_data_deep_impact_1.mean()

growth_data_deep_impact_2 = pd.read_csv('datas/ディープインパクト/growth_data_montly_牝.csv')
growth_data_deep_impact_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_deep_impact_2 = growth_data_deep_impact_2.mean()

growth_data_king_kamehameha_1 = pd.read_csv('datas/キングカメハメハ/growth_data_montly_牡.csv')
growth_data_king_kamehameha_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_king_kamehameha_1 = growth_data_king_kamehameha_1.mean()

growth_data_king_kamehameha_2 = pd.read_csv('datas/キングカメハメハ/growth_data_montly_牝.csv')
growth_data_king_kamehameha_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_king_kamehameha_2 = growth_data_king_kamehameha_2.mean()

growth_data_hart_cry_1 = pd.read_csv('datas/ハーツクライ/growth_data_montly_牡.csv')
growth_data_hart_cry_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_hart_cry_1 = growth_data_hart_cry_1.mean()

growth_data_hart_cry_2 = pd.read_csv('datas/ハーツクライ/growth_data_montly_牝.csv')
growth_data_hart_cry_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_hart_cry_2 = growth_data_hart_cry_2.mean()

_, axes = plt.subplots(2, 3,figsize=(20,15)) # 論理サイズ縦20 x 横15、縦2 x 横3枠の領域を確保

result_1 = pd.DataFrame({
    'ディープインパクト_牡': growth_data_mean_deep_impact_1,
    'ディープインパクト_牝': growth_data_mean_deep_impact_2
})
result_1.plot.line(ax=axes[0, 0]).grid()

result_2 = pd.DataFrame({
    'キングカメハメハ_牡': growth_data_mean_king_kamehameha_1,
    'キングカメハメハ_牝': growth_data_mean_king_kamehameha_2
})
result_2.plot.line(ax=axes[0, 1]).grid()

result_3 = pd.DataFrame({
    'ハーツクライ_牡': growth_data_mean_hart_cry_1,
    'ハーツクライ_牝': growth_data_mean_hart_cry_2
})
result_3.plot.line(ax=axes[0, 2]).grid()

result_4 = pd.DataFrame({
    'ディープインパクト_牡': growth_data_mean_deep_impact_1,
    'キングカメハメハ_牡': growth_data_mean_king_kamehameha_1,
    'ハーツクライ_牡': growth_data_mean_hart_cry_1
})
result_4.plot.line(ax=axes[1, 0]).grid()

result_5 = pd.DataFrame({
    'ディープインパクト_牝': growth_data_mean_deep_impact_2,
    'キングカメハメハ_牝': growth_data_mean_king_kamehameha_2,
    'ハーツクライ_牝': growth_data_mean_hart_cry_2
})
result_5.plot.line(ax=axes[1, 1]).grid()

plt.show()

結果は以下のようになりました。

ハーツクライ以外は牝馬の方が早熟なことが分かります。筋肉が固くなるのくだりの説得力なくなりますね・・・

性別(上位100頭)の比較

growth_data_deep_impact_1 = pd.read_csv('datas/ディープインパクト/growth_data_montly_牡.csv')
growth_data_deep_impact_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_deep_impact_1 = growth_data_deep_impact_1.drop(growth_data_deep_impact_1.index[range(100, len(growth_data_deep_impact_1))])
growth_data_mean_deep_impact_1 = growth_data_deep_impact_1.mean()

growth_data_deep_impact_2 = pd.read_csv('datas/ディープインパクト/growth_data_montly_牝.csv')
growth_data_deep_impact_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_deep_impact_2 = growth_data_deep_impact_2.drop(growth_data_deep_impact_2.index[range(100, len(growth_data_deep_impact_2))])
growth_data_mean_deep_impact_2 = growth_data_deep_impact_2.mean()

growth_data_king_kamehameha_1 = pd.read_csv('datas/キングカメハメハ/growth_data_montly_牡.csv')
growth_data_king_kamehameha_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_king_kamehameha_1 = growth_data_king_kamehameha_1.drop(growth_data_king_kamehameha_1.index[range(100, len(growth_data_king_kamehameha_1))])
growth_data_mean_king_kamehameha_1 = growth_data_king_kamehameha_1.mean()

growth_data_king_kamehameha_2 = pd.read_csv('datas/キングカメハメハ/growth_data_montly_牝.csv')
growth_data_king_kamehameha_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_king_kamehameha_2 = growth_data_king_kamehameha_2.drop(growth_data_king_kamehameha_2.index[range(100, len(growth_data_king_kamehameha_2))])
growth_data_mean_king_kamehameha_2 = growth_data_king_kamehameha_2.mean()

growth_data_hart_cry_1 = pd.read_csv('datas/ハーツクライ/growth_data_montly_牡.csv')
growth_data_hart_cry_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_hart_cry_1 = growth_data_hart_cry_1.drop(growth_data_hart_cry_1.index[range(100, len(growth_data_hart_cry_1))])
growth_data_mean_hart_cry_1 = growth_data_hart_cry_1.mean()

growth_data_hart_cry_2 = pd.read_csv('datas/ハーツクライ/growth_data_montly_牝.csv')
growth_data_hart_cry_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_hart_cry_2 = growth_data_hart_cry_2.drop(growth_data_hart_cry_2.index[range(100, len(growth_data_hart_cry_2))])
growth_data_mean_hart_cry_2 = growth_data_hart_cry_2.mean()

_, axes = plt.subplots(2, 3,figsize=(20,15)) # 論理サイズ縦20 x 横15、縦2 x 横3枠の領域を確保

result_1 = pd.DataFrame({
    'ディープインパクト_牡_上位100頭': growth_data_mean_deep_impact_1,
    'ディープインパクト_牝_上位100頭': growth_data_mean_deep_impact_2
})
result_1.plot.line(ax=axes[0, 0]).grid()

result_2 = pd.DataFrame({
    'キングカメハメハ_牡_上位100頭': growth_data_mean_king_kamehameha_1,
    'キングカメハメハ_牝_上位100頭': growth_data_mean_king_kamehameha_2
})
result_2.plot.line(ax=axes[0, 1]).grid()

result_3 = pd.DataFrame({
    'ハーツクライ_牡_上位100頭': growth_data_mean_hart_cry_1,
    'ハーツクライ_牝_上位100頭': growth_data_mean_hart_cry_2
})
result_3.plot.line(ax=axes[0, 2]).grid()

result_4 = pd.DataFrame({
    'ディープインパクト_牡_上位100頭': growth_data_mean_deep_impact_1,
    'キングカメハメハ_牡_上位100頭': growth_data_mean_king_kamehameha_1,
    'ハーツクライ_牡_上位100頭': growth_data_mean_hart_cry_1
})
result_4.plot.line(ax=axes[1, 0]).grid()

result_5 = pd.DataFrame({
    'ディープインパクト_牝_上位100頭': growth_data_mean_deep_impact_2,
    'キングカメハメハ_牝_上位100頭': growth_data_mean_king_kamehameha_2,
    'ハーツクライ_牝_上位100頭': growth_data_mean_hart_cry_2
})
result_5.plot.line(ax=axes[1, 1]).grid()

plt.show()

結果は以下のようになりました。

左下の牡馬のグラフからディープインパクト産駒が3歳春のクラシックに強い事が理解できる結果となっています。
下の段の真ん中のグラフからもディープインパクト産駒の牝馬が3歳春のクラシックに強い事が理解できる結果となっています。

筋肉が固くなるのくだりの説得力はやはりないですね、左上のグラフを見る限り、むしろ牝馬の方が早熟であるとの結果となっていますが、社台系のクラブ（社台RH、サンデーレーシング、キャロット、シルク）の牝馬は「６歳３月」までしか現役を続けられないことを考慮するれば、まだ分析が足りないと言わざるおえません。

左上のグラフの横軸を3歳時のみで表示してみます。

# ディープインパクトの牡馬の3歳時データ
growth_data_mean_deep_impact_1_3 = growth_data_mean_deep_impact_1[12:24]
# ディープインパクトの牝馬の3歳時データ
growth_data_mean_deep_impact_2_3 = growth_data_mean_deep_impact_2[12:24]

result = pd.DataFrame({
    'ディープインパクト_牡_上位100頭_3歳時': growth_data_mean_deep_impact_1_3,
    'ディープインパクト_牝_上位100頭_3歳時': growth_data_mean_deep_impact_2_3
})
result.plot().grid()

明らかに3歳のG1の開催時期のところに山があることが確認できます。

年月単位(累計)の集計

ディープインパクト、キングカメハメハ、ハーツクライの比較

growth_data_deep_impact_1 = pd.read_csv('datas/ディープインパクト/growth_data_montly_accumulate_牡.csv')
growth_data_deep_impact_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_deep_impact_1 = growth_data_deep_impact_1.mean()

growth_data_deep_impact_2 = pd.read_csv('datas/ディープインパクト/growth_data_montly_accumulate_牝.csv')
growth_data_deep_impact_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_deep_impact_2 = growth_data_deep_impact_2.mean()

growth_data_king_kamehameha_1 = pd.read_csv('datas/キングカメハメハ/growth_data_montly_accumulate_牡.csv')
growth_data_king_kamehameha_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_king_kamehameha_1 = growth_data_king_kamehameha_1.mean()

growth_data_king_kamehameha_2 = pd.read_csv('datas/キングカメハメハ/growth_data_montly_accumulate_牝.csv')
growth_data_king_kamehameha_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_king_kamehameha_2 = growth_data_king_kamehameha_2.mean()

growth_data_hart_cry_1 = pd.read_csv('datas/ハーツクライ/growth_data_montly_accumulate_牡.csv')
growth_data_hart_cry_1.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_hart_cry_1 = growth_data_hart_cry_1.mean()

growth_data_hart_cry_2 = pd.read_csv('datas/ハーツクライ/growth_data_montly_accumulate_牝.csv')
growth_data_hart_cry_2.drop(['horse_id'], axis=1, inplace=True)
growth_data_mean_hart_cry_2 = growth_data_hart_cry_2.mean()

growth_data_deep_impact_1 = growth_data_deep_impact_1.drop(growth_data_deep_impact_1.index[range(100, len(growth_data_deep_impact_1))])
growth_data_mean_deep_impact_1 = growth_data_deep_impact_1.mean()

growth_data_deep_impact_2 = growth_data_deep_impact_2.drop(growth_data_deep_impact_2.index[range(100, len(growth_data_deep_impact_2))])
growth_data_mean_deep_impact_2 = growth_data_deep_impact_2.mean()

growth_data_king_kamehameha_1 = growth_data_king_kamehameha_1.drop(growth_data_king_kamehameha_1.index[range(100, len(growth_data_king_kamehameha_1))])
growth_data_mean_king_kamehameha_1 = growth_data_king_kamehameha_1.mean()

growth_data_king_kamehameha_2 = growth_data_king_kamehameha_2.drop(growth_data_king_kamehameha_2.index[range(100, len(growth_data_king_kamehameha_2))])
growth_data_mean_king_kamehameha_2 = growth_data_king_kamehameha_2.mean()

growth_data_hart_cry_1 = growth_data_hart_cry_1.drop(growth_data_hart_cry_1.index[range(100, len(growth_data_hart_cry_1))])
growth_data_mean_hart_cry_1 = growth_data_hart_cry_1.mean()

growth_data_hart_cry_2 = growth_data_hart_cry_2.drop(growth_data_hart_cry_2.index[range(100, len(growth_data_hart_cry_2))])
growth_data_mean_hart_cry_2 = growth_data_hart_cry_2.mean()

_, axes = plt.subplots(2, 1,figsize=(20,15)) 

result_1 = pd.DataFrame({
    'ディープインパクト_牡_上位100頭': growth_data_mean_deep_impact_1,
    'キングカメハメハ_牡_上位100頭': growth_data_mean_king_kamehameha_1,
    'ハーツクライ_牡_上位100頭': growth_data_mean_hart_cry_1
})
result_1.plot.line(ax=axes[0]).grid()

result_2 = pd.DataFrame({
    'ディープインパクト_牝_上位100頭': growth_data_mean_deep_impact_2,
    'キングカメハメハ_牝_上位100頭': growth_data_mean_king_kamehameha_2,
    'ハーツクライ_牝_上位100頭': growth_data_mean_hart_cry_2
})
result_2.plot.line(ax=axes[1]).grid()


plt.show()

結果は以下のようになりました。累計だからこそ分かることがありますね。どの種牡馬の牡馬でも5歳時点で60%弱で同じになる点が面白いですね。
牝馬、牡馬ともにキングカメハメハが一番晩成なことが明確に分かります。

全体を見ての考察(感想)ですが、ディープインパクト産駒には明らかに早熟傾向があり、3歳時の賞金獲得率は目を見張るものがあります。「ディープタイマー」に関しては今回の結果からは存在の有無は確認できませんでしたので、社台系のクラブの牝馬の分析を追加で行う必要があると感じます。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up