Sebelumnya, saya telah melakukan copywriting untuk orang lain, dan menemukan bahwa banyak teman-teman harus melakukan strategi pemantauan, dan dapat segera memberi tahu polisi jika ada situasi khusus, tetapi tidak terlalu puas dengan terus-menerus menjalankan hard disk; jadi hari ini saya akan berbagi demo peringatan harga yang mudah untuk referensi Anda. ps: bahasa yang digunakan adalah python, alarm melalui pin-pin interface, konfigurasi server tidak diperkenalkan di sini

A. Persiapan

1. Panel kontrol Di sini digunakan panel menara untuk tujuan mengunggah file ke server, tentu saja Anda juga dapat menggunakan cara lain, ingatlah jalurnya (karena server penulis adalah sistem Ubuntu, maka semua perintah di bawah ini didasarkan pada ini, perintah sistem lainnya dapat dilihat sendiri)

• Unduh

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

Setelah selesai mendapatkan alamat internet dan login ((Jika alamat internet tidak bisa dibuka maka perlu membuka port 8888)

• Mendapatkan izin Buka keamanan -shh manajemen keamanan - buka kunci shh login

• Mengunggah file

1. Screen screen adalah sebuah program manajemen multi-pakai Windows di Linux, yang bertujuan agar program tetap dapat berjalan setelah koneksi shh terminal.

• Unduh

sudo apt-get install screen

• Perintah umum Run screen, buka path run file, dan akan muncul[[name] label bisa diatur sendiri)

screen -S [name]

Keluar dari layar

ctrl+a+d

Lihat proses yang berjalan di layar di latar belakang

screen -ls

Proses berakhir (pid dapat dilihat melalui proses)

kill -9 [pid号]

Menghapus informasi proses yang sudah mati di layar

screen -wipe

1. Pengaturan Pin Ini adalah referensi dari artikel Hukybo, dimana implementasi prinsip tidak lagi dijelaskan, hanya menunjukkan kode yang mudah dibungkus, hanya untuk referensi. 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函数，输出信息

2. implementasi kode

Transaksi yang diperoleh melalui token api, menggunakan U-base kontrak fapi interface, kode berikut mudah dibungkus token api, hanya untuk referensi

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

Karena strategi hanya digunakan untuk mendapatkan harga interface, hanya demo sederhana yang dilakukan di sini.

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

Berikut ini adalah implementasi dari kode pemantauan

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

3. Beroperasi

Kode siap setelah selesai ingat jalan, buka terminal run screen

screen -S [名称]

cd [路径]

python3 [文件名]

Anda dapat keluar setelah proses konfirmasi berjalan.

Alamat kebijakan:Robot Pengingat Harga Mudah