Nous avons créé un robot de rappel à prix abordable

J'ai déjà écrit à d'autres personnes, et j'ai découvert que beaucoup d'amis ont besoin d'une stratégie de surveillance de marché, pour alerter immédiatement en cas de situation particulière, mais ils ne sont pas très satisfaits d'avoir toujours ouvert un disque réel; alors aujourd'hui, je partage une simple démonstration de rappel de prix pour toute référence.ps: le langage est utilisé par python, par l'interface de clôture d'alarme, la configuration du serveur n'est pas présentée ici

一.准备

1. Panneau de commande Ici, le panneau de béta est utilisé pour télécharger le fichier sur le serveur, bien sûr, vous pouvez également utiliser d'autres méthodes pour mémoriser le chemin (parce que le serveur de l'auteur est un système Ubuntu, toutes les commandes ci-dessous sont basées sur cela, les autres commandes système peuvent être consultées par elles-mêmes)

• Télécharger

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

Une fois terminé, vous obtenez l'adresse Web et vous vous connectez (si l'adresse Web n'est pas ouverte, vous devez ouvrir le port 8888)

• Accéder aux droits ouvrir la sécurité - shh gérer la sécurité - ouvrir la clé de connexion shh

• Télécharger des documents

2. Écran screen est un gestionnaire de fenêtres sous linux qui permet d'exécuter des programmes après une connexion à un terminal shh.

• Télécharger

sudo apt-get install screen

• Commandements courants L'écran d'exécution ouvre le chemin d'exécution du fichier (la balise [name] peut être réglée automatiquement).

screen -S [name]

Cliquez sur le bouton de sortie

ctrl+a+d

Voir le processus en cours d'exécution en arrière-plan

screen -ls

Fin du processus (le nom de la pip peut être consulté par le processus)

kill -9 [pid号]

Nettoyer les informations de processus qui sont mortes sur l'écran

screen -wipe

3. Les clous sont fixés L'article sur le barrage de Hukybo, dont les principes de mise en œuvre ne sont pas décrits ici, montre uniquement le code de l'emballage simple, à titre de référence.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函数，输出信息

二.代码实现

La transaction est obtenue via Binance API, en utilisant l'interface U-bit contractée Fapi, le code suivant est simplement enveloppé dans Binance API, à titre de référence seulement.

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

Comme la stratégie est uniquement utilisée pour obtenir des interfaces de prix, nous n'avons fait qu'une simple démonstration ici, les autres interfaces sont similaires.

#封装获取价格接口
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"]

Ci-dessous, les codes de surveillance du secteur sont mis en œuvre:

# 监控币种&&监控价格一一对应
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()

三.运行

Lorsque le code est prêt, mémorisez le chemin et ouvrez l'écran d'exécution du terminal.

screen -S [名称]

cd [路径]

python3 [文件名]

Une fois que le processus de confirmation a été exécuté, vous pouvez vous déconnecter.

L'adresse de la stratégie:Le prix simple rappelle le robot