前言

终於把资料取得都结束了，现在要制作模组化是讯号灯的部分。

本日程序码使用：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登记，就可以调用罗～