1

monitors/vegas_cross.py

@@ -0,0 +1,69 @@
+import talib
+import numpy as np
+import time,setting
+import telegram_sender
+from datasource import crypto
+import dingding
+import discord_sender
+import setting
+
+
+# crossover 函数：检测上穿信号
+def crossover(series1, series2):
+    return (series1 > series2) & (series1.shift(1) <= series2.shift(1))
+
+# crossunder 函数：检测下穿信号
+def crossunder(series1, series2):
+    return (series1 < series2) & (series1.shift(1) >= series2.shift(1))
+
+## 检查信号
+def stratergy_run(symbol, interval, df, debug):
+    ## 计算 ema
+    df['ema144'] = talib.EMA(df['close'], timeperiod=144)
+    df['ema169'] = talib.EMA(df['close'], timeperiod=169)
+    df['ema576'] = talib.EMA(df['close'], timeperiod=576)
+    df['ema676'] = talib.EMA(df['close'], timeperiod=676)
+
+    max_low_line = np.maximum(df['ema144'], df['ema169'])
+    min_low_line = np.minimum(df['ema144'], df['ema169'])
+
+    max_high_line = np.maximum(df['ema576'], df['ema676'])
+    min_high_line = np.minimum(df['ema576'], df['ema676'])
+
+    df['longSignal'] = crossover(min_low_line, max_high_line)
+    df['shortSignal'] = crossunder(max_low_line, min_high_line)
+
+    latest = df.iloc[-1]
+    print(df)
+    direction = ""
+    if latest['shortSignal']==True:
+        direction = '空'
+    if latest['longSignal']==True:
+        direction = '多'
+
+    message = f"策略:【Vegas金叉死叉策略】\r\n"
+    message += f"品种: {symbol}\r\n"
+    message += f"周期: {interval}\r\n"
+    message += f"信号: 【{direction}】\r\n"
+    message += f"收盘价: {latest['close']}\r\n"
+
+    print(f"【{symbol} - {interval}】 is checked!")
+    if direction != "":
+        if debug == False:
+            url = 'https://discordapp.com/api/webhooks/1285867836454998077/JNAFUyur_ygOJIh6C4beUAVYMpm1TZf4IEeMn8Q1p0TglO1Hjyiu2LQqiU5AxVovWyiO'
+            discord_sender.send_message(url,'🦄信号提醒🦄',message)
+
+            print(f"【{symbol} - {interval}】 is singal fired!")
+        else:
+            print(message)
+    
+
+
+def run_crypto(interval, debug=False):
+    print('Vegas策略运行.')
+
+    for s in setting.symbols:
+        df = crypto.get_klines(s, interval)
+        stratergy_run(s,interval, df, debug)
+        
+        time.sleep(1)