get_extraData.py

import pandas as pd
import io
import requests
import os
import time

from datetime import datetime
from dateutil.relativedelta import relativedelta

from utils import gen_iter_date_by_month


def transform_date(date: str) -> str:  # 民國轉西元
    y, m, d = date.split("/")
    return str(int(y) + 1911) + "/" + m + "/" + d


def process_data(data: str) -> float | None:
    """清理原始字串並轉為 float，空值或無效值回傳 None。"""
    data = data.replace(",", "").replace("--", "").strip()
    return float(data) if data else None


def getDatas(path: str) -> dict:
    datas = {}
    for dirPath, dirNames, fileNames in os.walk(path):
        if len(fileNames) == 0:
            continue

        fileNames.sort()
        # 移除當月資料，以免資料不全
        current = f"{(datetime.now() + relativedelta(day=1)).strftime('%Y%m%d')}.csv"
        if fileNames[-1] == current:
            removed = fileNames[-1]  # 先記錄再刪除，避免 del 後索引位移
            os.remove(os.path.join(dirPath, removed))
            del fileNames[-1]
            print("remove", removed)

        for f in fileNames:
            datas[os.path.splitext(f)[0]] = True

    return datas


def save_twse_ftse_index(
    s: requests.Session, symbol: str, url_symbol: str, start: datetime
) -> None:
    savePath = os.path.join("./extraData", symbol)
    os.makedirs(savePath, exist_ok=True)
    datas = getDatas(savePath)

    end = datetime.now() + relativedelta(day=1)  # 設定為當月的 1 號

    for day in gen_iter_date_by_month(start, end):
        d = day.strftime("%Y%m%d")
        if datas.get(d, False):
            print(symbol, d, "already exists")
            continue

        url = f"https://www.twse.com.tw/rwd/zh/FTSE/{url_symbol}?response=csv&date={d}"
        print(symbol, url_symbol, d, "saving...", f"from {url}")

        c = s.get(url).content
        try:
            df = pd.read_csv(io.StringIO(c.decode("big5")), header=1)
            print("raw data")
            print(df)
        except pd.errors.EmptyDataError:
            print(symbol, d, "is empty")
            return

        print("after drop")
        df = df.dropna(axis=1)
        df = df[df[symbol.replace("指數", "報酬指數")] != "--"]
        print(df)
        df.loc[:, "Date"] = pd.to_datetime(df["日期"].apply(transform_date), format="%Y/%m/%d")
        df.loc[:, "Close"] = df[symbol].apply(process_data).astype(float)
        df.loc[:, "Adj Close"] = (
            df[symbol.replace("指數", "報酬指數")].apply(process_data).astype(float)
        )
        df.loc[:, "Dividends"] = 0
        df.loc[:, "Stock Splits"] = 0

        time.sleep(5)

        df = df[df["Adj Close"] != 0]
        df.to_csv(os.path.join(savePath, f"{d}.csv"), index=False, lineterminator="\n")


def save_TAI50I_index(s):
    save_twse_ftse_index(s, "臺灣50指數", "TAI50I", start=datetime(2002, 10, 1))


def save_TAI100I_index(s):
    save_twse_ftse_index(s, "臺灣中型100指數", "TAI100I", start=datetime(2004, 11, 1))


def save_TAIDIVIDI_index(s):
    save_twse_ftse_index(s, "臺灣高股息指數", "TAIDIVIDI", start=datetime(2007, 1, 1))


def save_TAIEX_index(s: requests.Session) -> None:
    symbol = "臺灣加權股價指數"
    savePath = os.path.join("./extraData", symbol)
    os.makedirs(savePath, exist_ok=True)
    datas = getDatas(savePath)

    start = datetime(2003, 1, 1)
    end = datetime.now() + relativedelta(day=1)  # 設定為當月的 1 號

    for day in gen_iter_date_by_month(start, end):
        d = day.strftime("%Y%m%d")
        if datas.get(d, False):
            print(symbol, d, "already exists")
            continue

        histURL = f"https://www.twse.com.tw/rwd/zh/TAIEX/MI_5MINS_HIST?response=csv&date={d}"
        totalReturnURL = f"https://www.twse.com.tw/rwd/zh/TAIEX/MFI94U?response=csv&date={d}"

        print(symbol, d, "get history...", f"from {histURL}")
        c = s.get(histURL).content
        try:
            hist = pd.read_csv(io.StringIO(c.decode("big5")), header=1)
            print("hist raw data")
            print(hist)
            print("after drop")
            hist = hist.dropna(axis=1)
            print(hist)
        except pd.errors.EmptyDataError:
            print(symbol, d, "is empty")
            return

        print(symbol, d, "get total return...", f"from {totalReturnURL}")
        c = s.get(totalReturnURL).content
        totalReturn = pd.read_csv(io.StringIO(c.decode("big5")), header=1)
        print("totalReturn raw data")
        print(totalReturn)
        print("after drop")
        totalReturn = totalReturn.dropna(axis=1)
        print(totalReturn)

        df = hist.join(totalReturn)
        if not df[df["日期"] != df["日　期"]].empty:
            raise ValueError(
                f"{symbol} {d}: hist 與 totalReturn 的日期欄位不對齊，"
                f"請確認來源資料格式是否改變"
            )

        df.loc[:, "Date"] = pd.to_datetime(df["日期"].apply(transform_date), format="%Y/%m/%d")
        df.loc[:, "Open"] = df["開盤指數"].apply(process_data).astype(float)
        df.loc[:, "High"] = df["最高指數"].apply(process_data).astype(float)
        df.loc[:, "Low"] = df["最低指數"].apply(process_data).astype(float)
        df.loc[:, "Close"] = df["收盤指數"].apply(process_data).astype(float)
        df.loc[:, "Adj Close"] = df["發行量加權股價報酬指數"].apply(process_data).astype(float)
        df.loc[:, "Dividends"] = 0
        df.loc[:, "Stock Splits"] = 0

        time.sleep(5)

        df = df[df["Adj Close"] != 0]
        df.to_csv(os.path.join(savePath, f"{d}.csv"), index=False, lineterminator="\n")


if __name__ == "__main__":
    with requests.Session() as s:
        save_TAI50I_index(s)
        save_TAI100I_index(s)
        save_TAIEX_index(s)
        save_TAIDIVIDI_index(s)