37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
# 突然放量策略
|
||
import numpy as np
|
||
from datasource import crypto
|
||
import 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
|
||
|
||
rate_limit = 10
|
||
if rate > rate_limit:
|
||
message = "📢Signals Notificaiton📢\r\n\r\n"
|
||
message += f"{symbol} 在 {interval} 出现放量 {round(rate, 2)}倍 ！\r\n\r\n"
|
||
message += f"{timestamp}"
|
||
|
||
if debug == False:
|
||
dingding.send_message(message)
|
||
print(f"【{symbol} - {interval}】 is singal fired!")
|
||
else:
|
||
print(message)
|
||
|
||
def run_crypto(interval, debug=False):
|
||
print(f"Vols-up strategy {interval} is running...")
|
||
|
||
symbols = crypto.get_future_symbols()
|
||
for s in symbols:
|
||
df = crypto.get_klines(s, interval, True,limit=50)
|
||
stratergy_run(s,interval, df, debug)
|
||
|
||
time.sleep(1) |