간단한 가격에 로봇을 상기시켜

이전에는 다른 사람들에게 대사본을 썼고, 많은 친구들이 특별한 상황이 발생할 때 즉시 경보 할 수 있는 행동 모니터링 전략을 가지고 있다는 것을 발견했습니다. 그러나 항상 실제 디스크를 열고있는 것에 대해 만족스럽지 않습니다. 그래서 오늘 간단한 가격 상기 데모를 공유합니다.ps: 언어는 파이썬을 사용하고 있습니다.

一.准备

1. 컨트롤 패널 여기서는 베타 패널을 사용해서 파일을 서버에 업로드하는 것을 목적으로 하고 있습니다. 물론 다른 방법도 사용할 수 있습니다. 경로를 기억하면 됩니다. (작가 서버가 우분투 시스템이기 때문에 아래의 모든 명령은 이를 기준으로 하고, 다른 시스템 명령은 자체적으로 열 수 있습니다.)

• 다운로드

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

완료 후 URL을 얻고 로그인합니다. (이 웹 사이트가 열리지 않으면 포트 8888를 열어야합니다.)

• 권한 획득 보안 - shh 보안 관리 - shh 키 로그인

• 문서 업로드

2. 화면 스크린 (영어: screen) 은 리눅스 버전의 윈도우 멀티루터 관리자 프로그램으로, 단말기 shh 연결 후에도 프로그램을 실행할 수 있도록 하는 프로그램이다.

• 다운로드

sudo apt-get install screen

• 일반적인 명령 실행 화면, 경로를 열면 실행 파일을 실행할 수 있습니다.

screen -S [name]

스크린에서 빠져나가는 단축키

ctrl+a+d

스크린이 백그라운드에서 실행되는 프로세스를 보기

screen -ls

종료 프로세스 (pid는 프로세스를 통해 볼 수 있습니다)

kill -9 [pid号]

화면에서 죽은 프로세스 정보를 청소합니다.

screen -wipe

3. 톱니 설정 여기서는 후키보 다마의 기사를 참조하고 있습니다. 원리 구현은 더 이상 설명되지 않습니다. 간단한 패키지 코드를 보여줍니다. 참조를 위해만.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函数，输出信息

二.代码实现

거래는 위안아피를 통해 취득되며, U-비트 계약 fapi 인터페이스를 사용하며, 다음 코드는 위안아피를 간단하게 포장하여 참조하기 위해만 사용됩니다.

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

이 전략은 가격 인터페이스를 얻는 데만 사용되기 때문에, 여기서는 간단한 데모만 하고, 다른 인터페이스는 동일합니다.

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

아래는 시장 감시 코드 구현입니다.

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

三.运行

코드가 완료되면 경로를 기억하고 터미널 실행 화면을 열십시오.

screen -S [名称]

cd [路径]

python3 [文件名]

확인 절차가 실행되면 종료합니다.

이 글은 이쪽에서 작성되었습니다.로봇에 대한 간단한 가격 경고