Ein einfaches Preis-Reminder-Roboter

Ich habe es schon früher für andere geschrieben und habe festgestellt, dass viele Freunde eine Strategie zur Überwachung der Märkte haben, um bei besonderen Situationen sofort zu melden, aber nicht sehr zufrieden sind mit der ständigen Einrichtung eines echten Ladens; Deshalb teile ich heute eine einfache Preis-Erinnerungsdemo für alle.ps: Die Sprache wird mit Python verwendet, um durch die Nail Interface zu alarmieren, die Konfiguration des Servers wird hier nicht beschrieben.

一.准备

1. Kontrollpanel Hier wird das Bauplatz verwendet, um die Dateien auf den Server hochzuladen, aber natürlich kann man auch andere Methoden verwenden, um den Pfad zu merken. (Da der Autorserver ein Ubuntu-System ist, sind alle folgenden Befehle darauf ausgerichtet, andere Systembefehle können sich selbst durchsuchen.)

• Herunterladen

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

Wenn Sie fertig sind, erhalten Sie die Webseite und melden Sie sich an (wenn die Webseite nicht geöffnet wird, müssen Sie den Port 8888 öffnen)

• Zugriff auf Berechtigungen Öffnen Sie die Sicherheits-shh-Sicherheitsverwaltung-Öffnen Sie die Schlüssel-shh-Anmeldung

• Dokumente hochladen

2. Bildschirm screen ist ein Windows-Multi-Use-Manager unter Linux, mit dem ein Programm ausgeführt werden kann, wenn ein Terminal mit einer SHH-Verbindung verbunden ist.

• Herunterladen

sudo apt-get install screen

• Häufige Befehle Der Startbildschirm kann nach dem Öffnen des Pfades ausgeführt werden.

screen -S [name]

Schaltfläche zum Verlassen des Bildschirms

ctrl+a+d

Prozesse im Hintergrund des Bildschirms sehen

screen -ls

Beenden des Prozesses (Pid kann durch den Prozess angezeigt werden)

kill -9 [pid号]

Löschen Sie abgestorbene Prozessinformationen aus dem Bildschirm

screen -wipe

3. Nägel eingestellt Hier wird auf den Artikel über den Hukybo-Damm verwiesen. Die Implementierung der Grundlagen wird nicht beschrieben, sondern nur der einfache Code für die Verpackung gezeigt.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函数，输出信息

二.代码实现

Die Transaktionen werden über BitAnapi abgerufen, wobei die U-Bit-Kontrakt-Fapi-Schnittstelle verwendet wird.

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

Da die Strategie nur für die Beschaffung von Preisinterfaces verwendet wird, wird hier nur eine einfache Demonstration durchgeführt, die andere Schnittstellen sind gleich.

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

Hier sind die Codeimplementierungen für die Marktüberwachung:

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

三.运行

Wenn der Code fertig ist, erinnern Sie sich an den Pfad und öffnen Sie den Terminallaufbildschirm

screen -S [名称]

cd [路径]

python3 [文件名]

Sie können die Bestätigung abbrechen, wenn sie ausgeführt wurde.

Die Strategie ist unter:Ein einfacher Preis erinnert an Roboter