楊育晟(Peter Yang)

嗨, 我叫育晟, 部落格文章主題包含了程式設計、財務金融及投資...等等,內容多是記錄一些學習的過程和心得,任何想法都歡迎留言一起討論。



Email: ycy.tai@gmail.com
LinkedIn: Peter Yang
Github: ycytai

用 Python 蒐集 ETF 股利分配歷史資料-附程式碼(Use Python to collect ETF dividend data)

前言

ETF 是近年來廣受投資大眾喜愛的金融商品,其投資一籃子標的的特性解決了許多投資人在選股上的問題,而高配息 ETF 更是受到青睞,今天本篇文章將介紹如何使用 Python 蒐集證交所公佈的 ETF 股利資料,之後就不需要再手動到網站上蒐集了。

關於 ETF 的介紹可以參考這篇文章認識ETF(Things you should know about ETF),裡頭詳述了 ETF 的基礎知識。

另外常常和 ETF 一同出現的名詞就是定期定額 ,有興趣也能夠參考ETF定期定額績效回測(Backtesting for dollar-averaging strategy) ,裡頭也用了 Python 手把手地完成定期定額的投資策略回測。

如果想爬股票股價的話,可以看看證交所股票價格爬蟲實作教學-附程式碼(Use Python to collect stock price),裡頭介紹了蒐集股價及爬蟲的基本概念。

用Python爬取ETF股利資料

此次蒐集任務的資料來源為ETF 分配收益

點進後可以看到裡頭可以選擇不同 ETF 標的以及期間,查詢後出現的結果就能得到 ETF 在各年的收益分配資料。

在標的下拉選單如果是請選擇,則會選取期間所有的 ETF 商品配息紀錄,目前網頁上並沒有提供直接下載的功能,也就是說如果投資人要做研究,需要自己上網搜詢後記錄起來,或是用滑鼠選取複製。

爬蟲解析

直接查看網頁呼叫的 api

https://www.twse.com.tw/rwd/zh/ETF/etfDiv?stkNo=&startDate=20050101&endDate=20231028&response=json&_=1698477064915

可以清楚地看到幾個參數 stkNo, startDate, endDate, response 以及 _,最後一個_ 看起來不像後端需要接收的參數,因此忽略他,其他四個參數可以迅速地從字面理解意思,就是股票代號、起始日、終止日以及回傳的資料格式,日期都是用西元。

知道來源和參數後,就可以寫 code 囉。

爬蟲程式碼

from urllib.parse import urlencode, urljoin
from datetime import datetime
import requests


def get_etf_dividend_history(
        symbol: str = "", 
        start_date: str = '20050101', 
        end_date: str = datetime.now().strftime('%Y%m%d'),
    ) -> dict:

    base_url = 'https://www.twse.com.tw/rwd/zh/ETF/etfDiv'
    params = {
        'stkNo': symbol,
        'startDate': start_date,
        'endDate': end_date,
        'response': 'json',
    }
    query_string = f'?{urlencode(params)}'
    res = requests.get(urljoin(base_url, query_string))

    if res.status_code == 200:
        return res.json()
    else:
        raise Exception(f'Get data fail. {res.text}')

上方將爬取的過程包成一個函式,再用 urlencode() 把參數轉換成 query string 放在 url 送入後端。

接著執行後便能順利取得資料。

>>> data = get_etf_dividend_history()

資料整理

抓到資料後會是 dict 類型,接著來把它轉成平常習慣用的表格,用 pandas 整理成 DataFrame

資料整理程式碼

下方是資料整理的程式碼,對到各個欄位之後整理成表格

import pandas as pd
from pandas.core.frame import DataFrame

def map_fields_and_values(source: dict) -> DataFrame:
    fields = source.get('fields')
    data = source.get('data')
    output = {}
    for idx, row in enumerate(data):
        output.setdefault(
            idx,
            {
                k: v
                for k, v in zip(fields, row)
            }
        )
    df = pd.DataFrame(output).T
    return df

執行結果如下,完成!

>>> df = map_fields_and_values(data)
>>> df.head()
     證券代號       證券簡稱       除息交易日     收益分配基準日     收益分配發放日 收益分配金額 (每1受益權益單位)                                  收益分配標準 (102年度起啟用) 公告年度
0   00929   復華台灣科技優息  112年10月24日  112年10月30日  112年11月17日              0.11  1.本基金投資中華民國境內所得之下列各款收益,做為本基金之可分配收益:(1)除息交易日前(不...  112
1   00904  新光臺灣半導體30  112年10月23日  112年10月29日  112年11月17日              0.13  本基金可分配收益,除應符合下列規定外,並應經金管會核准辦理公開發行公司之簽證會計師查核簽證後...  112
2   00904  新光臺灣半導體30  112年10月23日  112年10月29日  112年11月17日              0.13  本基金可分配收益,除應符合下列規定外,並應經金管會核准辦理公開發行公司之簽證會計師查核簽證後...  112
3    0056      元大高股息  112年10月19日  112年10月25日  112年11月14日               1.2  本基金可分配收益,除應符合下列規定外,並應經金管會核准辦理公開發行公司之簽證會計師查核出具收...  112
4  006204     永豐臺灣加權  112年10月19日  112年10月27日  112年11月17日             3.959  經理公司應於收益評價日(即每年九月三十日,如該日為非營業日則延至次一營業日)檢視自成立日起至...  112

客製化表格

如此一來就得到了網頁上的表格,大家可以再依自己的需求整理成想要的樣子,這邊提供我客製化的結果以及整理的程式碼。

對我來說,我只需要 證券代號除息交易日收益分配發放日 以及 收益分配金額 (每1受益權益單位) 這四個欄位,另外我也希望把日期整理成西元,然後欄位名稱改為英文。

def handle_date(v: str) -> datetime:
    return datetime.strptime(
        f'{int(v[0:3])+1911}-{v[4:6]}-{v[7:9]}', 
        '%Y-%m-%d',
    )

def customize_dividend_info_table(df: DataFrame) -> DataFrame:
    required_fields = [
        '證券代號',
        '除息交易日',
        '收益分配發放日',
        '收益分配金額 (每1受益權益單位)',
    ]
    dividend_df = df[required_fields].copy()

    dividend_df['除息交易日'] = dividend_df['除息交易日'].apply(handle_date)
    dividend_df['收益分配發放日'] = dividend_df['收益分配發放日'].apply(handle_date)

    fields_chi_to_en = {
        '證券代號': 'symbol',
        '證券簡稱': 'short_name',
        '除息交易日': 'ex_dividend_date',
        '收益分配基準日': 'dividend_based_date',
        '收益分配發放日': 'dividend_receive_date',
        '收益分配金額 (每1受益權益單位)': 'dividend_amount',
        '收益分配標準 (102年度起啟用)': 'dividend_detail',
        '公告年度': 'year'
    }
    dividend_df.rename(columns=fields_chi_to_en, inplace=True)
    return dividend_df

執行結果如下,看起來清爽很多囉!

>>> dividend_df = customize_dividend_info_table(df)
     symbol ex_dividend_date dividend_receive_date dividend_amount
0     00929       2023-10-24            2023-11-17            0.11
1     00904       2023-10-23            2023-11-17            0.13
2     00904       2023-10-23            2023-11-17            0.13
3      0056       2023-10-19            2023-11-14             1.2
4    006204       2023-10-19            2023-11-17           3.959
..      ...              ...                   ...             ...
587    0051       2007-11-23            2007-12-28            0.77
588    0054       2007-11-23            2007-12-28            0.35
589    0050       2007-10-24            2007-12-03             2.5
590    0050       2006-10-26            2006-12-01             4.0
591    0050       2005-05-19            2005-06-13            1.85

最後輸出成 csv 檔就完成囉! 是不是很輕鬆寫意 XD

>>> dividend_df.to_csv('etf_dividend.csv', index=False)

完整程式碼

from urllib.parse import urlencode, urljoin
from datetime import datetime
from pandas.core.frame import DataFrame
import requests
import pandas as pd


def get_etf_dividend_history(
        symbol: str = "", 
        start_date: str = '20050101', 
        end_date: str = datetime.now().strftime('%Y%m%d'),
    ) -> dict:

    base_url = 'https://www.twse.com.tw/rwd/zh/ETF/etfDiv'
    params = {
        'stkNo': symbol,
        'startDate': start_date,
        'endDate': end_date,
        'response': 'json',
    }
    query_string = f'?{urlencode(params)}'
    res = requests.get(urljoin(base_url, query_string))

    if res.status_code == 200:
        return res.json()
    else:
        raise Exception(f'Get data fail. {res.text}')


def map_fields_and_values(source: dict) -> DataFrame:
    fields = source.get('fields')
    data = source.get('data')
    output = {}
    for idx, row in enumerate(data):
        output.setdefault(
            idx,
            {
                k: v
                for k, v in zip(fields, row)
            }
        )
    df = pd.DataFrame(output).T
    return df


def handle_date(v: str) -> datetime:
    return datetime.strptime(
        f'{int(v[0:3])+1911}-{v[4:6]}-{v[7:9]}', 
        '%Y-%m-%d',
    )


def create_customize_dividend_table(df: DataFrame) -> DataFrame:
    required_fields = [
        '證券代號',
        '除息交易日',
        '收益分配發放日',
        '收益分配金額 (每1受益權益單位)',
    ]
    dividend_df = df[required_fields].copy()

    dividend_df['除息交易日'] = dividend_df['除息交易日'].apply(handle_date)
    dividend_df['收益分配發放日'] = dividend_df['收益分配發放日'].apply(handle_date)

    fields_chi_to_en = {
        '證券代號': 'symbol',
        '證券簡稱': 'short_name',
        '除息交易日': 'ex_dividend_date',
        '收益分配基準日': 'dividend_based_date',
        '收益分配發放日': 'dividend_receive_date',
        '收益分配金額 (每1受益權益單位)': 'dividend_amount',
        '收益分配標準 (102年度起啟用)': 'dividend_detail',
        '公告年度': 'year'
    }
    dividend_df.rename(columns=fields_chi_to_en, inplace=True)
    return dividend_df


data = get_etf_dividend_history()
df = map_fields_and_values(data)
dividend_df = create_customize_dividend_table(df)
dividend_df.to_csv('etf_dividend.csv', index=False)
Tags:
# python
# etf
# 爬蟲
# finance