Я уже писал для других, и я обнаружил, что многие мои друзья должны следить за ситуацией, чтобы в случае возникновения каких-либо особых ситуаций они могли немедленно вызвать полицию, но я не был очень доволен тем, что постоянно работал на дискете; поэтому сегодня я хочу поделиться простым напоминанием о цене для вашего сведения.
ps: язык используется в Python, через интерфейс, на который указывают, что сервер настроен не для этого.
1. Подготовка
- Панель управления
Здесь используется панель Botox, предназначенная для загрузки файлов на сервер, но, конечно, можно использовать и другие способы, помните, что путь доступен ((поскольку сервер автора является системой Ubuntu, все следующие команды используются в качестве ориентира, другие системные команды могут просматриваться самостоятельно))
- скачать
wget -O install.sh http://download.bt.cn/install/install-ubuntu_6.0.sh && sudo bash install.sh
По завершении получите URL и войдите в систему ((если URL не открывается, необходимо открыть порт 8888)
-
Получение разрешения
Открыть безопасность - shh безопасность управления - открыть ключ shh вход
-
Загрузка файлов
- Screen
screen - многоразовый менеджер окон под Linux, предназначенный для работы с терминалом shh.
- скачать
sudo apt-get install screen
- Обычные команды
Запустите экран, затем откройте путь и запустите файл.[[name] (название может быть изменено самостоятельно) <unk>
screen -S [name]
Выйти из экранной кнопки
ctrl+a+d
Просмотр процесса, который происходит на экране в фоновом режиме
screen -ls
Завершение процесса (пид можно просмотреть через процесс)
kill -9 [pid号]
Очистить информацию о мертвых процессах на экране
screen -wipe
- Настройка гвоздей
Здесь ссылаемся на статью 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函数，输出信息
Код реализации
Операция была получена через API, используя интерфейс fapi с U-образным контрактом.
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()
3. Работа
Код готов, запомните путь, откройте экран запуска терминала
screen -S [名称]
cd [路径]
python3 [文件名]
Вы можете выйти после завершения процедуры.
Адрес программы:Простые напоминания о ценах
