Hacer un robot de alerta de precio fácil

Antes de escribir para otros, encontré que muchos amigos quieren una estrategia de monitoreo de comportamiento, para alertar de inmediato en caso de situaciones especiales, pero no están muy satisfechos con el hecho de que siempre estén abiertos; Por lo tanto, hoy comparto una simple demo de precio para recordar a todos.ps: el lenguaje utiliza python, alerta por interfaz de clavijas, la configuración del servidor no se presenta aquí

一.准备

1. Panel de control Aquí se utiliza el panel de torres para subir archivos al servidor, pero también se pueden usar otros métodos para recordar el camino (ya que el servidor del autor es el sistema Ubuntu, todos los siguientes comandos se basan en él y otros comandos del sistema pueden consultarse por sí mismos)

• Descargar

wget -O install.sh http://download.bt.cn/install/install-ubuntu_6.0.sh && sudo bash install.sh

Cuando termine, obtenga el sitio web y inicie sesión (si no puede abrir el sitio web, debe abrir el puerto 8888)

• Acceso a los permisos Abrir seguridad-shh Administrador de seguridad-abrir la clave de acceso-shh

• Carga de archivos

2. La pantalla screen es un administrador de Windows multiuso bajo Linux, cuyo objetivo es que los programas puedan seguir ejecutándose después de una conexión con un terminal shh.

• Descargar

sudo apt-get install screen

• Las órdenes comunes La pantalla de ejecución, que se abre después de que el camino se ejecuta el archivo (([name] etiqueta se puede configurar automáticamente)).

screen -S [name]

Escoge el teclado de acceso rápido de la pantalla

ctrl+a+d

Ver el proceso en el fondo de la pantalla

screen -ls

Terminar el proceso (pID puede ser visto por el proceso)

kill -9 [pid号]

Limpiar información de proceso de pantalla muerta

screen -wipe

3. Configuración de los clavos Aquí se hace referencia al artículo de Hukybo Dam, el principio de la implementación no se describe en detalle, solo se muestra el código de envase sencillo, solo para referencia.https://www.fmz.com/digest-topic/5840

# 钉钉输出接口
class DING:
    # 向钉钉群输出信息
    def msg(self,text):
        token ="***"
        headers = {'Content-Type': 'application/json;charset=utf-8'}  # 请求头
        api_url = f"***={token}"
        json_text = {
            "msgtype": "text",  # 信息格式
            "text": {
                "content": text
            }
        }
        # 发送并打印信息
        requests.post(api_url, json.dumps(json_text), headers=headers).content
    
    #拼接输出信息
    def dd(self,logging):
        bj_dt = str(datetime.datetime.now())
        bj_dt = bj_dt.split('.')[0]  # 处理时间
        text = f'{bj_dt}\n'  # 定义信息内容
        text = text + logging # 处理信息内容
        log.msg(text)  # 调用msg函数，输出信息

二.代码实现

Las transacciones se obtienen a través de Binanceapi, utilizando una interfaz de contrato de U-bit con Binanceapi, el siguiente código se envuelve fácilmente en Binanceapi, solo para referencia.

import requests,json,time,hmac,hashlib,datetime

# APIKEY填写位置
apikey = '***'
Secret_KEY = '***'

#币安接口
class bian:
    #拼接请求参数
    def param2string(self,param):
        s = ''
        for k in param.keys():
            s += k
            s += '='
            s += str(param[k])
            s += '&'
        return s[:-1]
    
    # 参数为get，post请求方式，路径，body
    def IO(self,method,request_path,body):
        header = {
        'X-MBX-APIKEY': apikey,
        }
        #选择请求方式
        if body != '':
            #签名
            body['signature'] = hmac.new(Secret_KEY.encode('utf-8'), self.param2string(body).encode('utf-8'), hashlib.sha256).hexdigest()
            if method == 'GET':
                body = self.param2string(body)
                tell = 'https://fapi.binance.com{0}?{1}'.format(request_path,body)
                response = requests.get(url=tell, headers=header).json()
                return response
            elif method == 'POST':
                response = requests.post(url='https://fapi.binance.com'+str(request_path), headers=header, data=body).json()
                return response
        else:
            response = requests.get(url='https://fapi.binance.com'+str(request_path), headers=header).json()
            return response

Dado que la estrategia solo se utiliza para obtener interfaces de precios, aquí solo se hace una simple demostración, las otras interfaces son similares.

#封装获取价格接口
def price(self,Name):
    body = {"symbol":str(Name)}
    info = self.IO("GET","/fapi/v1/ticker/price",body)
    for i in info:
        if i == "code":
            #设计一个接口错误容错功能
            time.sleep(0.5)
            letgo = '调用price函数接口返回错误,再次尝试 返回错误代码:{0}'.format(str(info))
            log.dd(str(letgo))
            exchange.price(Name)
    return info["price"]

A continuación se muestra la implementación del código de vigilancia de mercado.

# 监控币种&&监控价格一一对应
ccy = ["BTCUSDT","ETHUSDT","LTCUSDT"]
PriceTIME = ["100000;28000","500000000;1200","500;100"]

#行情监控逻辑
def pricewarm():
    #轮询获取当前价格
    for i in range(len(PriceTIME)):
        info = exchange.price(str(PriceTIME[i]))
        priceindex = PriceTIME[i].find(";") #提取价格区间
        #价格上限
        priceup = PriceTIME[i][:priceindex]
        #价格下限
        pricedown = PriceTIME[i][priceindex+1:]
        if float(info) >= float(priceup): #钉钉接口输出
            letgo = f'当前价格{info}USDT大于所设定上限{priceup}USDT'
            log.dd(letgo)
        elif float(info) <= float(pricedown):
            letgo = f'当前价格{info}USDT小于等于设定下限{pricedown}USDT'
            log.dd(letgo)
        time.sleep(0.2)

# 主函数
def main():
    global exchange,log
    log = DING
    exchange = bian
    while True:
        try:
            pricewarm()
            time.sleep(1)
        except:
            time.sleep(1)

if __name__ == "__main__":
    main()

三.运行

Cuando el código está listo, recuerde el camino y abra la pantalla de ejecución del terminal.

screen -S [名称]

cd [路径]

python3 [文件名]

El proceso de confirmación se ejecuta y se puede salir.

La dirección de la estrategia:El precio de los robots es sencillo