I. Resumo

Para nos mantermos atualizados sobre o status de negociação do Inventor Quant Robot em transações do mundo real, às vezes precisamos enviar os resultados de negociação executados pelo robô para o WeChat, e-mail, mensagens de texto, etc. No entanto, com centenas de mensagens de vários tipos todos os dias, as pessoas não são mais sensíveis a essas mensagens, resultando na incapacidade de verificar mensagens importantes em tempo hábil. Portanto, este artigo implementa o push de mensagem de robô chamando a interface de grupo DingTalk.

2. Robô do grupo DingTalk

O robô de grupo DingTalk é uma função de extensão avançada. Contanto que você tenha uma conta DingTalk, você pode usá-lo. Ele pode agregar informações de terceiros no grupo DingTalk e realizar a sincronização automática de informações. Ele suporta acesso personalizado do protocolo Webhook e agrega lembretes, alarmes e outras informações no grupo DingTalk por meio do Inventor Quantitative Robot. Suporta três formatos de mensagem: texto, link e markdown, além de cinco tipos de mensagem. A mesma mensagem também pode ser enviada para vários grupos DingTalk ao mesmo tempo. Consulte o link oficial: https://ding-doc.dingtalk.com/doc#/serverapi2/ye8tup

3. Crie um robô

Etapa 1: Crie um grupo DingTalk Toda vez que um robô personalizado é criado em um grupo DingTalk, um endereço Hook exclusivo é gerado, que chamamos de endereço WebHook. Ao enviar mensagens para o endereço WebHook, o grupo DingTalk receberá as mensagens. Vamos usar o DingTalk no PC como exemplo. Primeiro, clique no sinal “+” no canto superior esquerdo para iniciar um bate-papo em grupo. Se você só quiser receber mensagens, pode adicionar aleatoriamente duas pessoas e depois expulsá-las. Preencha no nome do grupo: “FMZ Robot”, e selecione Normal como o tipo de grupo. Apenas grupo.

Etapa 2: Adicionar robô de grupo DingTalk Clique no seu avatar, selecione Gerenciamento de robôs, depois selecione Personalizar e clique em Adicionar. Personalize o nome do robô: “FMZ” e adicione-o ao grupo DingTalk que você acabou de criar. O robô suporta três configurações de segurança:

• Palavras-chave personalizadas: as informações só serão sincronizadas se contiverem esta palavra-chave.
• Adicionar assinatura: equivalente a definir uma senha.
• Endereço IP: O segmento de endereço IP de informações fixas de terceiros. Se for usado apenas para lembretes ou alarmes, basta selecionar palavras-chave personalizadas. A palavra-chave que definimos aqui é “:”, o que significa que quando a informação enviada pelo robô Inventor Quant contiver “:”, essa informação será enviada para o grupo DingTalk. Em seguida, clique em Concordar com o acordo para concluir. Por fim, copie o endereço do Webhook para uso posterior.

4. Implementação de código

Após obter o endereço do Webhook, podemos iniciar uma solicitação HTTP POST para esse endereço na Estratégia Quantitativa do Inventor para enviar informações para esse grupo DingTalk. Vale ressaltar que ao iniciar uma solicitação POST, a codificação do conjunto de caracteres deve ser definida como UTF-8.

import requests
import json
from datetime import datetime, timedelta, timezone


# 向钉钉群输出信息
def msg(text):
    token ="0303627a118e739e628bcde104e19cf5463f61a4a127e4f2376e6a8aa1156ef1"
    headers = {'Content-Type': 'application/json;charset=utf-8'}  # 请求头
    api_url = f"https://oapi.dingtalk.com/robot/send?access_token={token}"
    json_text = {
        "msgtype": "text",  # 信息格式
        "text": {
            "content": text
        }
    }
    # 发送并打印信息
    Log(requests.post(api_url, json.dumps(json_text), headers=headers).content)

    
# 测试函数
def onTick():
    arr = ['BTC', 'ETH', 'XRP', 'BCH', 'LTC']  # 主流数字货币
    # 获取东八区时间
    bj_dt = str(datetime.now().astimezone(timezone(timedelta(hours=8))))
    bj_dt = bj_dt.split('.')[0]  # 处理时间
    text = f'{bj_dt}\n'  # 定义信息内容
    for i in arr:  # 循环主流数字货币数组
        exchange.IO("currency", f"{i}_USDT")  # 切换交易对
        ticker = exchange.GetTicker().Last  # 获取最新价格
        if i == 'LTC':
            full = ' ：'
        else:
            full = '：'
        text = text + f"{i}/USDT{full}${ticker}\n"  # 处理信息内容
    msg(text)  # 调用msg函数，输出信息
    

# 策略入口
def main():
    while True:  # 进入无线循环
        onTick()  # 执行onTick函数
        Sleep(1000 * 60)  # 休眠一分钟

Ao personalizar robôs para sincronizar informações, você pode definir os números de telefone celular de vários membros do grupo. Quando os membros do grupo @ receberem a mensagem, eles serão lembrados pela mensagem @, mesmo que a sessão Não Perturbe esteja definida.

# 向钉钉群输出信息
def msg(text):
    token = "0303627a118e739e628bcde104e19cf5463f61a4a127e4f2376e6a8aa1156ef1"
    headers = {'Content-Type': 'application/json;charset=utf-8'}  # 请求头
    api_url = f"https://oapi.dingtalk.com/robot/send?access_token={token}"
    json_text = {
        "msgtype": "text",  # 信息格式
        "text": {
            "content": text
        },
        "at": {
            "atMobiles": [
                "16666666666",  # 被@的手机号码
                "18888888888"  # 被@的手机号码
            ],
            "isAtAll": False  # 不@所有人
        }
    }
    # 发送并打印信息
    Log(requests.post(api_url, json.dumps(json_text), headers=headers).content)

5. Testando o robô

No código acima, escrevemos um caso para obter os preços das principais moedas digitais a cada minuto e enviar essas informações para o grupo DingTalk: