75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
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)
|