34 lines
1.2 KiB
Python
34 lines
1.2 KiB
Python
# 突然放量策略
|
|
import numpy as np
|
|
import datasource,time, dingding
|
|
|
|
def stratergy_run(symbol, interval, df, debug):
|
|
# 计算过去 30 根 K 线的平均交易量
|
|
average_volume = df["volume"].iloc[-31:-1].mean() # 倒数第 31 到倒数第 2 根的平均
|
|
|
|
# 获取当前（最后一根，即第 500 根）交易量
|
|
current_volume = df["volume"].iloc[-1]
|
|
|
|
timestamp = df['timestamp'].iloc[-1]
|
|
|
|
rate = current_volume / average_volume
|
|
print(f"【{timestamp}】【{symbol}】当前交易量: {current_volume} , 平均交易量: {average_volume} , 倍数: {rate}")
|
|
|
|
if rate > 3:
|
|
message = "📢信号提醒📢\r\n\r\n"
|
|
message += f"{symbol} 在 {interval} 出现放量！\r\n\r\n"
|
|
message += timestamp
|
|
|
|
if debug == False:
|
|
dingding.send_message(message, True)
|
|
print(f"【{symbol} - {interval}】 is singal fired!")
|
|
else:
|
|
print(message)
|
|
|
|
def run_crypto(interval, debug=False):
|
|
symbols = datasource.crypto.get_future_symbols()
|
|
for s in symbols:
|
|
df = datasource.crypto.get_klines(s, interval, True,limit=50)
|
|
stratergy_run(s,interval, df, debug)
|
|
|
|
time.sleep(1) |