【D26】模组化#1：取股票资料

前言

在取用个资料的时候，都是一个一个执行，分散各地，为了方便需要整合；讯号灯也是如此，都放在个别的jupyter notebook中，不适合做成各自独立的讯号灯。因此这边会把他们模组化。

首先登场的是取股票资料。

本日程序码使用：stock_transaction.py

取股票资料模组化

部分资料已经把它模组化，像是取得三大法人期选交易状况（Day6)、加权指数（Day10)，但是Day4的程序还没有，

设定class和预设值

可参考Day4的程序码。这边把功能抽离出来，变成一个class：stock_transaction。并且把要取得资料的连结先设定在__init__中。

def __init__(self) -> None:
    self.title = "盘後资讯 > 个股日成交资讯"
    self.url = (
        "http://www.twse.com.tw/exchangeReport/STOCK_DAY_ALL?response=open_data"
    )
    self.df = None  # 把资料从csv转乘datframe
    self.trade_date = "2021-1-1"  # 交易日

制作所要处理的功能： _get_and_set_df_data、_create_new_header

跟之前相同，需要一个从API取得资料并放在ataframe的功能_get_and_set_df_data，以及中文栏位与资料库栏位的转换功能_create_new_header。

其中在_get_and_set_df_data特别要把日期给改好，直接从提供的档名当作日期，取代之前的「执行当日」为日期，因为今天还没有开盘或是资料仍是旧的，这个当下的日期写入资料库中，就是错误的。因此可从headers中取得档名（STOCK_DAY_ALL_20210924.csv）并且取得日期（20210924）。

csv = requests.get(self.url)
df = pandas.read_csv(StringIO(csv.text))  # 有header
# print(df)  # debug
self.df = df
# 从档名取得日期，档名：STOCK_DAY_ALL_20210924.csv
trade_date_raw = csv.headers.get("Content-Disposition")[-13:-5]
self.trade_date = (
    f"{trade_date_raw[:4]}-{trade_date_raw[4:6]}-{trade_date_raw[6:]}"
)

汇入资料库： _insert_mysql()

当然还要汇入资料库的功能_insert_mysql，原本的now参数本来是当下日期，改成从上面说明取得的日期self.trade_date。

执得注意的是，这边也是沿用之前建立的db_connect这个模组来连线MySQL。

def _insert_mysql(self) -> bool:
    try:
        new_headers = self._create_new_header(self.df.columns)
        df = self.df[1:]  # 拿掉第一行的资料
        df.columns = new_headers  # 设定资料栏位的名称
        print(df)

        counter = 0  # 记录欲新增数量
        # 建立connection物件
        my_connt_obj = db_connect.mysql_connect()
        conn = my_connt_obj.connect()
        with conn.cursor() as cursor:
            now = self.trade_date

            # 新增SQL语法
            for _, row in df.iterrows():
                try:
                    cmd = """INSERT IGNORE INTO DailyPrice 
                    (StockID, Symbol, TradeDate, OpenPrice, HighPrice,
                    LowPrice, ClosePrice, Volumn)
                    values(%s,%s,%s,%s,%s,%s,%s,%s);"""
                    cursor.execute(
                        cmd,
                        (
                            None,
                            row.stock_symbol,
                            now,
                            row.open if pandas.notnull(row.open) else 0,
                            row.high if pandas.notnull(row.high) else 0,
                            row.low if pandas.notnull(row.low) else 0,
                            row.close if pandas.notnull(row.close) else 0,
                            row.volume if pandas.notnull(row.volume) else 0,
                        ),
                    )
                    conn.commit()
                    counter += 1
                except Exception as e:
                    print(e)

给外部直接执行的function：get_and_save()

这边就很简单的呼叫使用，如果没有这个话，就要自行呼叫两个function出来

def get_and_save(self, url=None):
    """Get today transaction data and save into MySQL.

    Args:
        param1 (str): 资料的url
    Returns:
        bool: 回传结果. True 表示储存成功，False 表示没有储存至资料库

    """
    r = self._get_and_set_df_data(url)
    if r:
        r = self._insert_mysql()
    else:
        return False

    return r