72 lines
3.1 KiB
Python
72 lines
3.1 KiB
Python
from binance.spot import Spot
|
|
import talib
|
|
import numpy as np
|
|
import pandas as pd
|
|
import datasource.crypto
|
|
import time,setting
|
|
import telegram_sender
|
|
from datasource import crypto
|
|
from datetime import datetime
|
|
import dingding
|
|
|
|
|
|
# crossover 函数：检测上穿信号
|
|
def crossover(series1, series2):
|
|
return (series1 > series2) and (series1.shift(1) <= series2.shift(1))
|
|
|
|
# crossunder 函数：检测下穿信号
|
|
def crossunder(series1, series2):
|
|
return (series1 < series2) and (series1.shift(1) >= series2.shift(1))
|
|
|
|
## 检查信号
|
|
def stratergy_run(symbol, interval, df, debug):
|
|
maxDifference = 0.005
|
|
|
|
## 计算 ema
|
|
df['ema5'] = talib.EMA(df['close'], timeperiod=5)
|
|
df['ema10'] = talib.EMA(df['close'], timeperiod=10)
|
|
df['ema30'] = talib.EMA(df['close'], timeperiod=30)
|
|
df['ema144'] = talib.EMA(df['close'], timeperiod=144)
|
|
|
|
shortDiff = abs(df['ema5'] - df['ema10']) / df['ema10']
|
|
midDiff = abs(df['ema10'] - df['ema30']) / df['ema30']
|
|
longDiff = abs(df['ema30'] - df['ema144']) / df['ema144']
|
|
|
|
# df['priceCheck'] = (longDiff <= maxDifference) and (midDiff <= maxDifference) and (shortDiff <= maxDifference)
|
|
# df['isLongArrangement'] = (df['ema5'] > df['ema10']) and (df['ema10'] > df['ema30']) and (df['ema30'] > df['ema144'])
|
|
# df['isShortArrangement'] = (df['ema5'] < df['ema10']) and (df['ema10'] < df['ema30']) and (df['ema30'] < df['ema144'])
|
|
|
|
d1 = df.iloc[-1]
|
|
d2 = df.iloc[-2]
|
|
d3 = df.iloc[-3]
|
|
|
|
d1['isLongArrangement'] = (d1['ema5'] > d1['ema10']) and (d1['ema10'] > d1['ema30']) and (d1['ema30'] > d1['ema144'])
|
|
d2['isLongArrangement'] = (d2['ema5'] > d2['ema10']) and (d2['ema10'] > d2['ema30']) and (d2['ema30'] > d2['ema144'])
|
|
|
|
d1['isShortArrangement'] = (d1['ema5'] < d1['ema10']) and (d1['ema10'] < d1['ema30']) and (d1['ema30'] < d1['ema144'])
|
|
d2['isShortArrangement'] = (d2['ema5'] < d2['ema10']) and (d2['ema10'] < d2['ema30']) and (d2['ema30'] < d2['ema144'])
|
|
|
|
|
|
isbullish = d1['isLongArrangement'] == True and d2['isLongArrangement'] == False
|
|
isBear = d1['isShortArrangement'] == True and d2['isShortArrangement'] == False
|
|
|
|
print(f"{symbol} - {interval} bullish: {isbullish} | bear : {isBear} | LongArrangement: {d1['isLongArrangement']} | ShortArrangement: {d1['isShortArrangement']}")
|
|
if(isbullish | isBear):
|
|
message = f"🦄 信 号 提 醒 🦄\r\n\r\n"
|
|
message += f"策略:【5-10-30-144均线排列】\r\n"
|
|
message += f"品种: {symbol}\r\n"
|
|
message += f"周期: {interval}\r\n"
|
|
message += f"信号: 【{'多头排列' if isbullish else '空头排列'}】\r\n"
|
|
|
|
dingding.send_message(message, isAtAll=True)
|
|
telegram_sender.send_message(setting.chat_id, message)
|
|
|
|
|
|
def run_crypto(interval, debug=False):
|
|
|
|
symbols = ['BTCUSDT','ETHUSDT','BNBUSDT','SOLUSDT','DOGEUSDT','LINKUSDT','NEARUSDT','WIFUSDT','ARBUSDT','ARUSDT','JASMYUSDT','FTMUSDT','EGLDUSDT','ENSUSDT','GALAUSDT','ORDIUSDT','TRUUSDT','LPTUSDT','JUPUSDT','WIFUSDT','AVAXUSDT','GALUSDT','BONKUSDT','UNFIUSDT','CHZUSDT']
|
|
for s in symbols:
|
|
df = crypto.get_klines(s, interval)
|
|
stratergy_run(s,interval, df, debug)
|
|
|
|
time.sleep(1) |