এর আগে আমি অন্যদের জন্য কপি করেছি, এবং অনেক বন্ধুকে খুঁজে পেয়েছি যে তাদের পরিস্থিতি পর্যবেক্ষণের কৌশল অনুসরণ করতে হবে, বিশেষ পরিস্থিতিতে অবিলম্বে সতর্কতা অবলম্বন করতে হবে, কিন্তু আমি সবসময় রিয়েল-ডিস্কে কাজ করার জন্য খুব সন্তুষ্ট নই; তাই আজ আমি আপনাদের সাথে একটি সহজ দামের ডেমো শেয়ার করছি। পিএসঃ ভাষাটি পাইথন ব্যবহার করা হয়, এটি একটি পিক-ইন-পিক ইন্টারফেসের মাধ্যমে সতর্ক করে, সার্ভারের কনফিগারেশন এখানে বর্ণনা করা হয়নি

১. প্রস্তুতি

1. কন্ট্রোল প্যানেল এখানে পট্টা প্যানেল ব্যবহার করা হয়েছে, যার উদ্দেশ্য ফাইলগুলি সার্ভারে আপলোড করা, অবশ্যই অন্যান্য পদ্ধতিও ব্যবহার করা যেতে পারে, পথটি মনে রাখবেন ((কারণ লেখক সার্ভারটি উবুন্টু সিস্টেম, তাই নীচের সমস্ত কমান্ড এটির জন্য, অন্যান্য সিস্টেম কমান্ডগুলি স্বয়ংক্রিয়ভাবে অনুসন্ধান করতে পারে))

• ডাউনলোড

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

এটি করার পরে, আপনি একটি URL পাবেন এবং লগইন করবেন (যদি আপনি এটি খুলতে না পারেন তবে আপনাকে পোর্ট 8888 খুলতে হবে)

• অনুমতি পেতে সিকিউরিটি-এসএইচএস সিকিউরিটি ম্যানেজমেন্ট-এসএইচএস কী লগইন চালু করুন

• ফাইল আপলোড করুন

1. Screen screen হল লিনাক্সের একটি উইন্ডো মাল্টি ইউজ ম্যানেজমেন্ট প্রোগ্রাম, যার উদ্দেশ্য হল টার্মিনাল shh সংযোগের পরেও প্রোগ্রামটি চালানো যায়।

• ডাউনলোড

sudo apt-get install screen

• সাধারণ আদেশ রান স্ক্রিন, পাথ রান ফাইল খোলার পরই রান করা যাবে।[[name] ট্যাগটি নিজে নিজে সেট করতে পারেন)

screen -S [name]

স্ক্রিন শর্টকাট থেকে প্রস্থান করুন

ctrl+a+d

স্ক্রিনটি ব্যাকগ্রাউন্ডে চলমান দেখুন

screen -ls

প্রক্রিয়া সমাপ্তি

kill -9 [pid号]

স্ক্রিনের মৃত প্রক্রিয়ার তথ্য সাফ করুন

screen -wipe

1. পিন সেটিং এখানে Hukybo-এর লেখা থেকে উদ্ধৃতি দেওয়া হয়েছে, কিন্তু বাস্তবায়নের মূলনীতি নিয়ে আর কিছু বলা হয়নি, শুধু সহজেই বানানো কোড দেখানো হয়েছে, শুধুমাত্র রেফারেন্সের জন্য 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函数，输出信息

২. কোড বাস্তবায়ন

প্রক্রিয়াকরণটি Binance API এর মাধ্যমে প্রাপ্ত করা হয়েছে, যা U- স্থানীয় চুক্তি fapi ইন্টারফেস ব্যবহার করে, নিম্নলিখিত কোডটি কেবলমাত্র রেফারেন্সের জন্য Binance API এর সাথে সংযুক্ত করা হয়েছে

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 [文件名]

নিশ্চিতকরণ প্রক্রিয়া চালানোর পর আপনি প্রস্থান করতে পারবেন।

নীতিমালার ঠিকানাঃরোবটকে সহজেই দাম জানাতে হবে