FMZ 디스크 로봇 자동 탐지 재시작 프로그램 (微信推送)
저자:에이슨04, 날짜: 2022-09-22 18:07:19태그:
FMZ 디스크 로봇 자동 탐지 재시작 프로그램 (微信推送)
여기 FMZ 플랫폼에서 신청한 API와 APIkey를 입력합니다.
변수 토큰은 WeChat 푸시를 전송하는 코드입니다.
웹 사이트로 접속하세요:https://www.pushplus.plus/자신의 WeChat로 로그인하여 대중을 추적하십시오 (광고가 아닌)
클릭 한 번
토큰을 복사하고 토큰 변수에 채우기
robotId 변수에서 자신이 감시해야 할 실제 로봇 번호를 입력합니다 (목록 형식)
실제 디스크 로봇 번호는 FMZ 웹 버전의 실제 디스크를 열고 웹 사이트에서 얻을 수 있습니다.
코드는 직접 로컬로 실행할 수 있습니다. 하지만 컴퓨터는 항상 켜져 있어야 합니다. 이 사이트는 인터넷에 접속하여 인터넷에 접속할 수 있습니다. 서버에서 실행하려면 3자 리버리를 미리 설치해야 합니다.
'''
代码可以直接放到本地运行，
不过需要一直开启电脑，也可以放到自己的服务器上运行
'''
import time
import json
import ssl
import requests
ssl._create_default_https_context = ssl._create_unverified_context
try:
import md5
import urllib2
from urllib import urlencode
except:
import hashlib as md5
import urllib.request as urllib2
from urllib.parse import urlencode
accessKey = '48xxxxxxxxxxxxxxxxxxxxxxxxxxxxde'
secretKey = '91xxxxxxxxxxxxxxxxxxxxxxxxxxxx84'
def api(method, *args):
d = {
'version': '1.0',
'access_key': accessKey,
'method': method,
'args': json.dumps(list(args)),
'nonce': int(time.time() * 1000),
}
d['sign'] = md5.md5(('%s|%s|%s|%d|%s' % (d['version'], d['method'], d['args'], d['nonce'], secretKey)).encode('utf-8')).hexdigest()
# 注意： urllib2.urlopen 函数，超时问题，可以设置超时时间，urllib2.urlopen('https://www.fmz.com/api/v1', urlencode(d).encode('utf-8'), timeout=10) 设置超时 10秒
return json.loads(urllib2.urlopen('https://www.fmz.com/api/v1', urlencode(d).encode('utf-8')).read().decode('utf-8'))
def send_wechat(msg):
token = '93xxxxxxxxxxxxxxxxxxxxxxxxxxxx57' # 前边复制到那个token
title = '【Waring】 策略信息'
content = msg
template = 'html'
url = f"https://www.pushplus.plus/send?token={token}&title={title}&content={content}&template={template}"
#print(url)
r = requests.get(url=url)
print(json.loads(r.text)['msg'])
robotId = [xxx,xxx,xxx] #需要监视的机器人代码
while True:
for j in range(len(robotId)):
detail = api('GetRobotDetail', robotId[j])
if detail['data']['result']['robot']['status'] == 1 and detail['data']['result']['robot']['wd'] == 1:
print(f"实盘{robotId[j]}状态正常 status = {detail['data']['result']['robot']['status']}，实盘监视已打开 wd = {detail['data']['result']['robot']['wd']}")
pass
elif detail['data']['result']['robot']['status'] == 1 :
print(f"实盘{robotId[j]}状态正常 status = {detail['data']['result']['robot']['status']}，实盘监视未打开 wd = {detail['data']['result']['robot']['wd']}")
pass
else:
print(f"实盘{robotId[j]}状态异常 status = {detail['data']['result']['robot']['status']}")
#尝试重启实盘 尝试次数 = 4 每5s 尝试一次
status = False
for i in range(4):
api('RestartRobot', robotId[j])
robotDetail = api('GetRobotDetail', robotId[j])
print(f"尝试重启实盘{robotId[j]}第 {i+1} 次")
if robotDetail['data']['result']['robot']['status'] == 1 :
mess = api('GetRobotLogs',robotId[j],0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
print(f"实盘{robotId[j]}重启完成 status = {api('GetRobotDetail', robotId[j])['data']['result']['robot']['status']}\n"
f"返回错误信息1：{mess['data']['result']['logs'][0]['Arr'][0][6]}\n"
f"返回错误信息2：{mess['data']['result']['logs'][0]['Arr'][1][6]}\n")
send_wechat(f"实盘{robotId[j]}重启完成 status = {api('GetRobotDetail', robotId[j])['data']['result']['robot']['status']}\n"
f"返回错误信息1：{mess['data']['result']['logs'][0]['Arr'][0][6]}\n"
f"返回错误信息2：{mess['data']['result']['logs'][0]['Arr'][1][6]}\n")
status = True
break
else:
print(f"第 {i+1} 次 重启失败!!")
time.sleep(5)
if status == False :
print(f"尝试 4 次重启实盘{robotId[j]}失败，发送警告信息！！")
send_wechat(f"尝试 4 次重启实盘{robotId[j]}失败，请及时查看！！\n尝试 4 次重启实盘{robotId[j]}失败，请及时查看！！\n尝试 4 次重启实盘{robotId[j]}失败，请及时查看！！\n")
time.sleep(60*10)
더 많은
- MACD 전략
- 트레이딩뷰 신호 실행 전략 2.1 (교육)
- SSL 채널 + STOCH RSI
- SSL 하이브리드 + STOCH RSI
- 발명가 양적 데이터베이스 실전
- Okex 거래소는 지정된 통화 보유를 취득하고 발명자를 반환합니다.
- Okex 화폐 변환, 유리 화폐 교환 화폐 수 또는 화폐 수 화폐 수
- Promise의 비동기 템플릿을 모방합니다.
- 오일 펌프의 신으로부터의 이중 EMA 평선 전략
- 비안의 소액 자산 전환
- 지표库 (python판)
- 조건 개장 마틴
- 거래 시간 V3
- RSI 추세
- BTC 봇
- 경보 (??), 경보 조건 (??) 또는 전략 경보?
- 단순 구매 판매 신호
- 부시 & 베어시 삼키기
- 구매 및 판매 전략은 SerdarYILMAZ의 AO+Stoch+RSI+ATR에 달려 있습니다.
- 이동평균
18180828122어떻게 서버를 실행시킬 수 있을까요?