Article détaillé:https://www.fmz.com/bbs-topic/5969
Comme le serveur HTTPS a ses propres failles, vous pouvez envisager d'utiliser le serveur ThreadingHTTPS à la place. Pour référence:https://docs.python.org/3.7/library/http.server.htmlVous avez besoin de Python 3.7.
Le problème du serveur HTTPS:https://www.zybuluo.com/JunQiu/note/1350528
'''
请求格式：http://x.x.x.x:xxxx/data?access_key=xxx&secret_key=yyy&type=buy&amount=0.001
策略机器人参数：
- 类型：加密字符串，AccessKey , SecretKey ，可以用FMZ平台的低权限的API KEY，或者自己生成KEY也可以。
- 类型：字符串，合约ID，ContractType
- 类型：数值，端口号，Port
'''
import re
import _thread
import json
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse
def url2Dict(url):
query = urlparse(url).query
params = parse_qs(query)
result = {key: params[key][0] for key in params}
return result
class Executor(BaseHTTPRequestHandler):
def do_GET(self):
try:
dictParam = url2Dict(self.path)
Log("测试", dictParam)
except Exception as e:
Log("Provider do_GET error, e:", e)
def do_POST(self):
try:
self.send_response(200)
self.send_header("Content-type", "application/json")
self.end_headers()
dictParam = url2Dict(self.path)
# 测试POST请求Body信息
data = self.rfile.read(200) # 指定了读取长度
Log("data:", data) # 打印POST请求的数据，可以根据请求中的数据具体再让机器人执行对应的操作
# 校验
if len(dictParam) == 4 and dictParam["access_key"] == AccessKey and dictParam["secret_key"] == SecretKey:
del dictParam["access_key"]
del dictParam["secret_key"]
Log("接收到请求", "参数：", dictParam, "#FF0000")
'''
map[access_key:xxx amount:0.001 secret_key:yyy type:buy]
'''
isSpot = True
if exchange.GetName().find("Futures") != -1:
if ContractType != "":
exchange.SetContractType(ContractType)
isSpot = False
else :
raise "未设置期货合约"
q = None
if exchange.GetName() == "Futures_CTP" and UseMarketOrderForCTP == False:
q = ext.NewTaskQueue()
if isSpot and dictParam["type"] == "buy":
exchange.Buy(-1, float(dictParam["amount"]))
Log(exchange.GetAccount())
elif isSpot and dictParam["type"] == "sell":
exchange.Sell(-1, float(dictParam["amount"]))
Log(exchange.GetAccount())
elif not isSpot and dictParam["type"] == "long":
exchange.SetDirection("buy")
if not q:
exchange.Buy(-1, float(dictParam["amount"]))
else :
q.pushTask(exchange, ContractType, "buy", float(dictParam["amount"]), lambda task, ret: Log(task["desc"], ret, "#FF0000"))
Log("持仓：", exchange.GetPosition())
elif not isSpot and dictParam["type"] == "short":
exchange.SetDirection("sell")
if not q:
exchange.Sell(-1, float(dictParam["amount"]))
else :
q.pushTask(exchange, ContractType, "sell", float(dictParam["amount"]), lambda task, ret: Log(task["desc"], ret, "#FF0000"))
Log("持仓：", exchange.GetPosition())
elif not isSpot and dictParam["type"] == "cover_long":
exchange.SetDirection("closebuy")
if not q:
exchange.Sell(-1, float(dictParam["amount"]))
else :
q.pushTask(exchange, ContractType, "closebuy", float(dictParam["amount"]), lambda task, ret: Log(task["desc"], ret, "#FF0000"))
Log("持仓：", exchange.GetPosition())
elif not isSpot and dictParam["type"] == "cover_short":
exchange.SetDirection("closesell")
if not q:
exchange.Buy(-1, float(dictParam["amount"]))
else :
q.pushTask(exchange, ContractType, "closesell", float(dictParam["amount"]), lambda task, ret: Log(task["desc"], ret, "#FF0000"))
Log("持仓：", exchange.GetPosition())
if q is not None:
while q.size() > 0:
q.poll()
Sleep(500)
# 处理body数据
if isDealBodyMsg:
if exchange.GetName().find("Futures") != -1:
Log("data:", data.decode('utf-8')) # 测试
if re.search(r'buy', data.decode('utf-8')):
Log("触发buy")
exchange.SetContractType(ct)
exchange.SetDirection("buy")
exchange.Buy(-1, amount)
elif re.search(r'sell', data.decode('utf-8')):
Log("触发sell")
exchange.SetContractType(ct)
exchange.SetDirection("sell")
exchange.Sell(-1, amount)
# 写入数据应答
self.wfile.write(json.dumps({"state": "ok"}).encode())
except Exception as e:
Log("Provider do_POST error, e:", e)
def createServer(host):
try:
server = ThreadingHTTPServer(host, Executor)
Log("Starting server, listen at: %s:%s" % host)
server.serve_forever()
except Exception as e:
Log("createServer error, e:", e)
raise Exception("stop")
def main():
# 开启一个线程
try:
_thread.start_new_thread(createServer, (("0.0.0.0", Port), )) # VPS服务器上测试
except Exception as e:
Log("错误信息：", e)
raise Exception("stop")
if exchange.GetName().find("Futures") != -1:
exchange.SetContractType(ContractType)
Log("账户资产信息：", _C(exchange.GetAccount))
while True:
if exchange.GetName() == "Futures_CTP":
if exchange.IO("status"):
LogStatus(_D(), "CTP连接")
else:
LogStatus(_D(), "CTP未连接")
else:
LogStatus(_D())
Sleep(2000)
et8888/upload/asset/2457669d9c32568732cb7.jpg ouvrir le disque dur indique cette erreur, où est-ce que c'est mal
et8888/upload/asset/2457669d9c32568732cb7.jpg ouvrir le disque dur indique cette erreur, où est-ce que c'est mal
Les épées de HuashanLe professeur: Pourriez-vous ajouter un message?
Les épées de Huashan2021/11/21 11:53:33 CMD 199866 buy=0.1, l'administrateur a reçu ce message et aucun mouvement n'a été effectué par le robot.
sur surJe n'ai pas compris ce que vous avez écrit, mais je peux vous aider à le traduire.
Je vous en prie.Pourriez-vous écrire un exemple d'une liste complète de signaux de télévision, comment obtenir des informations dans le corps; voir les gens dans le brouillard.
DietQuantProfesseur, comment faire pour augmenter la paire de devises sur un seul disque?
Les rêves coûtent huit chiffresMon Dieu, est-ce que tu peux me faire sortir un disque simulé OKv5?
Je suis désolé.La question est la suivante: la version 3.7 est-elle la version de l'administrateur de déploiement? Comment le mettre à jour?
2021-08-11 11:08:55 erreur Traceback (most recent call last): Fichier "
Le numéro de téléphone:Je n'ai pas compris ce que tu as écrit, comment peux-tu me contacter?
Je ne sais pas.Bonjour, professeur, comment est l'adresse du webhook pour la stratégie d'alerte directe sur la télévision?
lanbnBon, merci professeur, je vais essayer!
lanbnProfesseur, bonjour, pourriez-vous me donner un peu plus de détails sur le code écrit dans le contrat, le programmeur est vraiment incroyable, merci!
Le petit rêveLa documentation de l'API FMZ contient finalement la fonction de commutation de disque d'analogie OKEX.
wbe3- petit déjeuner de crêpesComment faire pour changer le disque d'analogie?
Le petit rêveL'erreur est due au fait que Python n'a pas de module associé ThreadingHTTPServer
clé986Vous n'avez pas résolu ce problème?
et8888C'est clair.
Le petit rêveIl n'est pas recommandé de créer une demande de service de surveillance télévisée comme celle-ci, mais utilisez l'API étendue de FMZ, il y a des articles dans la bibliothèque.
Le petit rêveCette stratégie est basée sur l'exemple des contrats à terme sur les produits, qui nécessitent un code de contrat si vous voulez exécuter une crypto-monnaie.
Le petit rêveL'exemple le plus direct de cette tactique est que le robot tactique crée un service pour écouter les demandes de signaux de télévision. Je ne sais pas. # Testez la requête POST pour l'information du corps data = self.rfile.read ((200) # spécifie la longueur de lecture Log (("data:", data) # Imprime les données de la requête POST et demande au robot d'exécuter les opérations correspondantes en fonction des données de la requête Je ne sais pas. Il y a déjà une note dans le code, c'est l'information du corps pour traiter les demandes de télévision.
Le petit rêveIl a également écrit sur Twitter: "Les gens ne sont pas prêts de changer de stratégie.
Le petit rêveLa configuration de l'API KEY de l'analogue OKEX sur FMZ est effectuée en ajoutant une phrase ``exchange.IO (simuler, vrai) ``exchange dans le code.
Le petit rêveJe ne comprends pas, je suppose que c'est une erreur, vérifiez-la.
Je ne sais pas.Merci, si mon adresse webhook indique que la direction de transaction est d'acheter, mais que la stratégie de télévision indique que la stratégie est de vendre?
Le petit rêveL'article et la stratégie de https://www.fmz.com/bbs-topic/5969 contiennent des exemples.
Le petit rêveVous pouvez consulter la description de la fonction. https://www.fmz.com/api#exchange.setcontracttype...
PY008Quel est le code du contrat?
Le petit rêveJe suis désolée.
Le petit rêveSi c'est pour un contrat d'échange, écrivez l'adresse du webhook sur votre téléviseur: Il y a un autre contrat: Je ne sais pas. Il est également possible de télécharger des fichiers sur les réseaux sociaux. Je ne sais pas. Il n'y a qu'un seul contrat: Je ne sais pas. Il est également possible de télécharger des fichiers sur les réseaux sociaux, y compris les réseaux sociaux. Je ne sais pas. N'oubliez pas de définir le code du contrat dans les paramètres de stratégie.