59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
import talib
|
|
import numpy as np
|
|
import bn
|
|
import tg
|
|
import settings
|
|
|
|
# 检查是否出现多头排列信号
|
|
def check_bullish_crossover(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)
|
|
|
|
# 判断是否出现多头排列信号
|
|
# if ema7[-1] > ema30[-1] > ema100[-1] > ema200[-1] and ema7[-2] <= ema30[-2] <= ema100[-2] <= ema200[-2]:
|
|
if ema7[-1] > ema30[-1] > ema100[-1] and ema7[-2] <= ema30[-2] <= ema100[-2]:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
# 检查是否出现空头排列信号
|
|
def check_bearish_crossover(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)
|
|
|
|
# 判断是否出现空头排列信号
|
|
if ema7[-1] < ema30[-1] < ema100[-1] < ema200[-1] >= ema30[-2] >= ema100[-2]:
|
|
# if ema7[-1] < ema30[-1] < ema100[-1] < ema200[-1] and ema7[-2] >= ema30[-2] >= ema100[-2] >= ema200[-2]:
|
|
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
|
|
def strategy_run(symbol, interval):
|
|
# 获取kline数据
|
|
data = bn.klines(symbol, interval)
|
|
|
|
if check_bullish_crossover(data):
|
|
print('多头排列信号出现！')
|
|
|
|
text = f'${symbol} - {interval}\r\n\r\n出现【多头排列】信号'
|
|
tg.send_message(settings.chat_id, text)
|
|
|
|
|
|
if check_bearish_crossover(data):
|
|
print("空头排列信号出现！")
|
|
text = f'${symbol} - {interval}\r\n\r\n出现【空头排列】信号'
|
|
tg.send_message(settings.chat_id, text) |