53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
import talib
|
|
import numpy as np
|
|
import bn
|
|
import tg
|
|
import datetime
|
|
import setting
|
|
import db
|
|
|
|
flags = {}
|
|
|
|
def check_ma_arrange(data):
|
|
# 提取收盘价
|
|
close_prices = np.array([float(entry[4]) for entry in data])
|
|
|
|
# 计算移动平均线
|
|
ema7 = talib.EMA(close_prices, timeperiod=7)
|
|
ema30 = talib.EMA(close_prices, timeperiod=30)
|
|
ema100 = talib.EMA(close_prices, timeperiod=100)
|
|
ema200 = talib.EMA(close_prices, timeperiod=200)
|
|
|
|
bullish = ema7[-1] > ema30[-1] > ema100[-1] > ema200[-1]
|
|
bearish = ema7[-1] < ema30[-1] < ema100[-1] < ema200[-1]
|
|
|
|
return bullish, bearish
|
|
|
|
|
|
def run(symbol, interval):
|
|
# 获取kline数据
|
|
data = bn.klines(symbol, interval)
|
|
|
|
bullish, bearish = check_ma_arrange(data)
|
|
|
|
flag_name = symbol + '_' + interval
|
|
text = ""
|
|
data = {}
|
|
if bullish and (flag_name not in flags or flags[flag_name] == False):
|
|
flags[flag_name] = True
|
|
text = f'📶信号预警📶\r\n\r\n品种：【${symbol}】\r\n周期：{interval}\r\n信号：【多头】排列\r\n\r\n{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
|
|
data = {"type" : 1, "symbol": symbol, "interval" : interval, "signal": 1}
|
|
|
|
if bearish and (flag_name not in flags or flags[flag_name] == True):
|
|
flags[flag_name] = False
|
|
text = f'📶信号预警📶\r\n\r\n品种：【${symbol}】\r\n周期：{interval}\r\n信号：【空头】排列\r\n\r\n{datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
|
|
data = {"type" : 1, "symbol": symbol, "interval" : interval, "signal": 3}
|
|
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) |