I. Resumen

Para estar al tanto del estado comercial del Inventor Quant Robot en transacciones del mundo real, a veces necesitamos enviar los resultados comerciales ejecutados por el robot a WeChat, correo electrónico, mensajes de texto, etc. Sin embargo, con cientos de mensajes de distintos tipos cada día, las personas ya no son sensibles a estos mensajes, lo que da como resultado la incapacidad de verificar los mensajes importantes de manera oportuna. Por lo tanto, este artículo implementa la función de envío de mensajes de robots mediante una llamada a la interfaz de grupo de DingTalk.

2. Robot de grupo DingTalk

El robot de grupo de DingTalk es una función avanzada extendida. Siempre que tenga una cuenta de DingTalk, podrá usarlo. Puede agregar información de terceros en el grupo DingTalk y realizar la sincronización automática de la información. Admite acceso personalizado al protocolo Webhook y agrega recordatorios, alarmas y otra información en el grupo DingTalk a través del Inventor Quantitative Robot. Admite tres formatos de mensajes: texto, enlace y Markdown, y cinco tipos de mensajes. El mismo mensaje también se puede enviar a varios grupos de DingTalk al mismo tiempo. Consulte el enlace oficial: https://ding-doc.dingtalk.com/doc#/serverapi2/ye8tup

3. Crea un robot

Paso 1: Crea un grupo de DingTalk Cada vez que se crea un robot personalizado en un grupo de DingTalk, se genera una dirección de Hook única, que llamamos dirección de WebHook. Al enviar mensajes a la dirección de WebHook, el grupo de DingTalk recibirá los mensajes. Tomemos como ejemplo DingTalk en PC. Primero, haz clic en el signo “+” en la esquina superior izquierda para iniciar un chat grupal. Si solo quieres recibir mensajes, puedes agregar dos personas al azar y luego expulsarlas. En el nombre del grupo: “FMZ Robot”, seleccione Normal como tipo de grupo. Solo agrupar.

Paso 2: Agregar el robot del grupo DingTalk Haga clic en su avatar, seleccione Administración de robots, luego seleccione Personalizar y haga clic en Agregar. Personaliza el nombre del robot: “FMZ” y agrégalo al grupo DingTalk que acabas de crear. El robot admite tres configuraciones de seguridad:

• Palabras clave personalizadas: la información solo se sincronizará si contiene esta palabra clave.
• Agregar firma: equivale a establecer una contraseña.
• Dirección IP: El segmento de dirección IP de información fija de terceros. Si solo se utiliza para recordatorios o alarmas, simplemente seleccione palabras clave personalizadas. La palabra clave que definimos aquí es : “, lo que significa que cuando la información enviada por el robot Inventor Quant contiene : “, esta información se enviará al grupo DingTalk. Luego haga clic en Aceptar el acuerdo para completar. Por último, copie la dirección del webhook para usarla más adelante.

4. Implementación del código

Después de obtener la dirección del Webhook, podemos iniciar una solicitud HTTP POST a esta dirección en la estrategia cuantitativa de Inventor para enviar información a este grupo de DingTalk. Se debe tener en cuenta que al iniciar una solicitud POST, la codificación del conjunto de caracteres debe establecerse en 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)  # 休眠一分钟

Al personalizar robots para sincronizar información, puede configurar los números de teléfono móvil de varios miembros del grupo. Cuando los miembros del grupo @ reciban el mensaje, se les recordará mediante el mensaje @, incluso si la sesión No molestar está configurada.

# 向钉钉群输出信息
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. Prueba del robot

En el código anterior, escribimos un caso para obtener los precios de las principales monedas digitales cada minuto y enviar esta información al grupo DingTalk: