Tạo ra một robot nhắc nhở giá rẻ

Trước đây, tôi đã viết cho người khác và thấy nhiều người bạn muốn có chiến lược giám sát hành vi, có thể báo cảnh báo ngay lập tức khi xảy ra tình huống đặc biệt, nhưng không hài lòng với việc luôn mở máy thực; vì vậy hôm nay tôi sẽ chia sẻ một bản demo nhắc nhở giá đơn giản cho tất cả mọi người tham khảo.ps: ngôn ngữ được sử dụng là python, thông qua các giao diện móc móc báo động, cấu hình máy chủ không được giới thiệu ở đây

一.准备

1. Bảng điều khiển Ở đây, bạn sử dụng bảng điều khiển để tải tập tin lên máy chủ, và tất nhiên bạn cũng có thể sử dụng các phương thức khác, chỉ cần nhớ đường dẫn (vì máy chủ của tác giả là hệ thống Ubuntu, tất cả các lệnh dưới đây được đặt theo mục tiêu này, các lệnh hệ thống khác có thể tự truy cập)

• Tải xuống

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

Sau khi hoàn thành, bạn sẽ nhận được địa chỉ và đăng nhập (nếu bạn không thể mở địa chỉ, bạn cần mở cổng 8888)

• Nhận quyền Mở bảo mật - shh Security Manager - Mở khóa đăng nhập shh

• Tải lên tài liệu

2. Màn hình screen là một phần mềm quản lý nhiều lần sử dụng cửa sổ trên Linux, nhằm mục đích vẫn có thể chạy chương trình sau khi kết nối với thiết bị shh.

• Tải xuống

sudo apt-get install screen

• Các mệnh lệnh phổ biến Khóa chạy, mở đường dẫn sau khi chạy tập tin (([name] thẻ có thể tự cài đặt) ).

screen -S [name]

Chìa khóa thoát khỏi màn hình

ctrl+a+d

Xem màn hình đang chạy trong nền

screen -ls

Kết thúc quá trình ((pid có thể xem qua quá trình)

kill -9 [pid号]

Xóa thông tin quá trình đã chết trên màn hình

screen -wipe

3. Đặt ghim Ở đây tham chiếu đến bài viết về toà nhà Hukybo, các nguyên tắc thực hiện không được mô tả, chỉ hiển thị mã gói đơn giản, chỉ để tham khảo.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函数，输出信息

二.代码实现

Các giao dịch được lấy thông qua BinanceAPI, sử dụng giao diện giao dịch bản thân U-bit Fapi, và mã sau đây được gói gọn BinanceAPI, chỉ để tham khảo:

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

Vì chiến lược chỉ dùng để lấy giao diện giá cả nên chỉ có một bản trình bày đơn giản, các giao diện khác tương tự.

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

Dưới đây là các thực hiện mã giám sát thị trường:

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

三.运行

Khi mã đã sẵn sàng, nhớ đường dẫn, mở màn hình chạy thiết bị đầu cuối

screen -S [名称]

cd [路径]

python3 [文件名]

Sau khi xác nhận, bạn có thể thoát khỏi chương trình.

Địa chỉ chiến lược:Giá rẻ nhắc nhở robot