Making a simple price to remind the robot

I have written to others before, and found that many friends want to monitor the strategy of the market, and immediately alert the police in case of special situations, but not very satisfied with the fact that they have been open all the time; so today I share a simple price reminder demo for everyone's reference.ps: the language is used by python, it alerts through the pins interface, the configuration of the server is not introduced here

一.准备

1. Control panel The purpose of this is to upload files to the server, but of course you can also use other methods to remember the path. (As the author's server is the Ubuntu system, all the following commands are based on this, and other system commands can be read by themselves)

• Downloaded

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

Once you're done, get the URL and log in. (If the URL is not open, you need to open port 8888)

• Access to permissions Open security - shh security management - open shh key log in

• Uploading of documents

2. Screen screen is a Windows reusable window manager under Linux that is designed to allow programs to run even after a terminal shh connection has been made.

• Downloaded

sudo apt-get install screen

• Common commands Run the screen, open the path and run the file ([name] tag can be set by yourself).

screen -S [name]

Exit screen shortcut

ctrl+a+d

View the screen running in the background

screen -ls

End of process (pid can be viewed by the process)

kill -9 [pid号]

Clearing process information that is dead on the screen

screen -wipe

3. Set the nail This is a reference to the article on the Hukybo Dam, where the principles of implementation are not described, only the simple packaging code is shown, for reference only.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函数，输出信息

二.代码实现

The transaction is accessed via Binance API, using the U-bit contract fapi interface, the following code is simply wrapped in Binance API, for reference only

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

Since the strategy is only used to get the price interface, this is just a simple demonstration, other interfaces are the same.

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

Below is the implementation of the industry monitoring code:

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

三.运行

When the code is ready, remember the path, open the terminal run screen

screen -S [名称]

cd [路径]

python3 [文件名]

You can log out after the confirmation process has run.

The policy address:A simple price reminds the robot