add signals - MACD

signals/macd.py

@@ -0,0 +1,74 @@
+import talib
+import numpy as np
+import bn
+import tg
+import datetime
+import setting
+import db
+
+def check_macd(data):
+    close_prices = np.array([float(entry[4]) for entry in data])
+
+    # 自定义参数
+    fast_period = 12
+    slow_period = 26
+    signal_period = 9
+
+    print(close_prices[-1], close_prices[-2], close_prices[-3], close_prices[-4])
+    # 计算MACD指标 macd - DIF; macdsignal - DEA; macdhist- 柱状线
+    macd, macd_signal, macd_hist = talib.MACD(close_prices, fastperiod=fast_period, slowperiod=slow_period,
+                                              signalperiod=signal_period)
+
+    print("MACD:", round(macd[-1], 2))
+    print("Signal:", round(macd_signal[-1], 2))
+    print("Histogram:", round(macd_hist[-1], 2))
+    print("----------")
+    print("MACD:", round(macd[-2], 2))
+    print("Signal:", round(macd_signal[-2], 2))
+    print("Histogram:", round(macd_hist[-2], 2))
+
+    return macd, macd_signal, macd_hist
+
+
+def generate_signals(macd, macd_signal, macd_hist):
+    # 1: 多 DIF > 0 and DEA>0 and MACD>0
+    if macd[-1] > 0 and macd_signal[-1] > 0 and macd_hist[-1] > 0:
+        signal = 1
+    # 2: 强多 DIF > 0 and DEA>0 and DIF>DEA AND MACD>0
+    elif macd[-1] > 0 and macd_signal[-1] > 0 and macd[-1] > macd_signal[-1] and macd_hist[-1] > 0:
+        signal = 2
+    # 3：空
+    elif macd[-1] < 0 and macd_signal[-1] < 0 and macd[-1] < macd_signal[-1]:
+        signal = 3
+    # 4: 强空
+    elif macd[-1] < 0 and macd_signal[-1] < 0 and macd[-1] < macd_signal[-1] and macd_hist[-1] < 0:
+        signal = 2
+    else:
+        signal = 0
+    return signal
+
+
+def run(symbol, interval):
+    # 获取kline数据
+    data = bn.klines(symbol, interval)
+    macd, macd_signal, macd_hist = check_macd(data)
+    signal = generate_signals(macd, macd_signal, macd_hist)
+    print(signal)
+    text = ""
+    data = {}
+    if signal == 1 or signal == 2:
+        text = f'📶信号预警📶\r\n\r\n品种：【${symbol}】\r\n周期：{interval}\r\n信号：【多头】MACD\r\n\r\n{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
+        data = {"type": 2, "symbol": symbol, "interval": interval, "signal": signal}
+
+    if signal == 3 or signal == 4:
+        text = f'📶信号预警📶\r\n\r\n品种：【${symbol}】\r\n周期：{interval}\r\n信号：【空头】MACD\r\n\r\n{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
+        data = {"type": 2, "symbol": symbol, "interval": interval, "signal": signal}
+    if text != "":
+        signals = db.get_list('signals', f'symbol="{symbol}" and `interval`="{interval}"')
+        for s in signals:
+            print(s)
+            db.execute_sql(f'delete from signals where `id`={s["id"]}')
+
+        db.insert(data, 'signals')
+        print(text)
+        tg.send_message(setting.chat_id, text)