前言

終於把資料取得都結束了，現在要製作模組化是訊號燈的部分。

本日程式碼使用：fin_signal.py

將會改寫：

• d20_singalTaiexAmount.ipynb
• d21_singalLegal.ipynb
• d23_singalLegal_2.ipynb
• d24_stock_ma.ipynb

製作module: fin_signal

• 設定初始值：init
  製作一個class，命名為fin_signal，並且設定好__init__，其中預設值是把三個dataframe資料集設為None。

class fin_signal:
    def __init__(self):
        self.df_taiex = None
        self.df_legal = None
        self.df_stock = None

• 取得資料：_get_db_data

取得近一年的加權指數、三大法人期選交易狀況、股價等等資料，並且放在self底下：

def _get_db_data(self):
    # 建立與MySQL的連線
    my_connt_obj = db_connect.mysql_connect()
    conn = my_connt_obj.connect()

    # 半年前的日期
    date_year = datetime.today() - relativedelta(year=1)
    date_year = f"{date_year.year}-{date_year.month}-{date_year.date}"
    # 加權指數資料
    sql_script = (
        f"SELECT * FROM StockTransactionInfo WHERE TradeDate > '{date_year}'"
    )
    self.df_taiex = pandas.read_sql(sql_script, con=conn)

    # 取得三大法人期選的資料
    sql_script = (
        f"SELECT * FROM LegalDailyFutureOption WHERE TradeDate > '{date_year}'"
    )
    self.df_legal = pandas.read_sql(sql_script, con=conn)

    # 取得股價資訊
    sql_script = f"SELECT * FROM DailyPrice WHERE TradeDate > '{date_year}'"
    self.df_stock = pandas.read_sql(sql_script, con=conn)

    # 關閉連線
    conn.close()

製作訊號燈

因訊號燈的程式碼眾多，直接去Github上面觀看。

Day20: 加權指數資料

建立以下兩個訊號燈：

• 最近交易日，是近期六天最大的交易金額: _get_signal_1
• 顯示交易是否熱絡：_get_signal_2

Day21: 三大法人

外資部分為：

• 期貨留倉是否空單還是多單：_get_signal_3_1
• 期貨留倉數量是否增加：_get_signal_4_1
• 期貨留倉數量變化多少：_get_signal_5_1
• 期貨留倉數變動比例：_get_signal_6_1

自營商部分為：

• 期貨留倉是否空單還是多單：_get_signal_3_2
• 期貨留倉數量是否增加：_get_signal_4_2
• 期貨留倉數量變化多少：_get_signal_5_2
• 期貨留倉數變動比例：_get_signal_6_2

投信部分為：

• 期貨留倉是否空單還是多單：_get_signal_3_3
• 期貨留倉數量是否增加：_get_signal_4_3
• 期貨留倉數量變化多少：_get_signal_5_3
• 期貨留倉數變動比例：_get_signal_6_3

Day24: MA線

移動平均線的穿過，也沒有很難寫，就直接把之前的帶入，基本上就完成了。

def _get_signal_7(self) -> int:
    """signal_7: 5MA > 20MA
    近期5日平均比20日還要高

    Retrun:
        bool: True:高；False：低
    """

    # 取得台積電資料，代號：2330
    df_2330 = self.df_stock.loc[self.df_stock["Symbol"] == "2330"]

    # 製作5日移動平均數
    close_price_5 = df_2330["ClosePrice"].rolling(5, min_periods=1).mean()

    # 製作20日移動平均數
    close_price_20 = df_2330["ClosePrice"].rolling(20, min_periods=1).mean()

    r = True if close_price_5.iloc[-1] > close_price_20.iloc[-1] else False

    return r

取得所有訊號燈：get_signal

在這邊使用get_signal，裡面把所有的訊號燈function加入，變成一個tuple，之後根據所要呈現的方式拆分此tuple即可。這邊就先不多著墨如何呈現。

def get_signal(self) -> tuple:
    r = (
        self._get_signal_1(),
        self._get_signal_2(),
        self._get_signal_3_1(),
        self._get_signal_3_2(),
        self._get_signal_3_3(),
        self._get_signal_4_1(),
        self._get_signal_4_2(),
        self._get_signal_4_3(),
        self._get_signal_5_1(),
        self._get_signal_5_2(),
        self._get_signal_5_3(),
        self._get_signal_6_1(),
        self._get_signal_6_2(),
        self._get_signal_6_3(),
        self._get_signal_7(),
    )

    return r

測試

執行下列的程式碼：

signal = fin_signal()
signal._get_db_data()
signal.show_data()
r = signal.get_signal()

print(r)

可以看得到結果是：

(True, True, False, True, False, False, True, False, -11118, 2341, -2205, 0.04248570817156308, 0.07184728232513887, 1.3602714373843308, True)

後記

訊號燈看起來不怎麼好讀，這就需要之後的優化，但我們已經取得所有的訊號燈。當以後需要別的訊號燈的話，直接新增判別訊號燈的功能並且在get_signal登記，就可以調用囉～