Вызов API DingTalk для реализации push-сообщений робота

Чтобы быть в курсе торгового статуса робота Inventor Quant в реальных транзакциях, нам иногда приходится отправлять результаты торговли, выполненной роботом, в WeChat, по электронной почте, в текстовых сообщениях и т. д. Однако, принимая сотни сообщений разного рода каждый день, люди больше не чувствительны к этим сообщениям, что приводит к невозможности своевременно проверять важные сообщения. Поэтому в этой статье реализуется отправка сообщений роботом с помощью вызова группового интерфейса DingTalk.

2. Групповой робот DingTalk

Групповой робот DingTalk — это расширенная функция. Пока у вас есть учетная запись DingTalk, вы можете ее использовать. Он может объединять стороннюю информацию в группу DingTalk и осуществлять автоматическую синхронизацию информации. Он поддерживает настраиваемый доступ к протоколу Webhook и объединяет напоминания, сигналы тревоги и другую информацию в группу DingTalk с помощью количественного робота Inventor. Поддерживает три формата сообщений: текст, ссылка и разметка, а также пять типов сообщений. Одно и то же сообщение можно отправить в несколько групп DingTalk одновременно.
Перейдите по официальной ссылке: https://ding-doc.dingtalk.com/doc#/serverapi2/ye8tup

3. Создать робота

Шаг 1: Создайте группу DingTalk

Каждый раз, когда в группе DingTalk создается пользовательский робот, генерируется уникальный адрес Hook, который мы называем адресом WebHook. Отправляя сообщения на адрес WebHook, группа DingTalk будет получать сообщения. Давайте возьмем DingTalk на ПК в качестве примера. Сначала нажмите на знак "+" в левом верхнем углу, чтобы начать групповой чат. Если вы хотите получать сообщения только сами, вы можете добавить двух случайных людей, а затем выгнать их. Заполнить в названии группы: "FMZ Robot", и выберите Normal как тип группы. Просто группа.

Шаг 2: Добавьте группового робота DingTalk
Нажмите на свой аватар, выберите «Управление роботом», затем выберите «Настроить» и нажмите «Добавить». Настройте имя робота: «FMZ» и добавьте его в только что созданную вами группу DingTalk. Робот поддерживает три настройки безопасности:

Пользовательские ключевые слова: информация будет синхронизирована только в том случае, если она содержит это ключевое слово.
Добавить подпись: эквивалентно установке пароля.
IP-адрес: сегмент IP-адреса фиксированной сторонней информации.

Если он используется только для напоминаний или будильников, просто выберите пользовательские ключевые слова. Ключевое слово, которое мы здесь определяем, — «:», что означает, что если информация, отправляемая роботом Inventor Quant, содержит «:», эта информация будет отправлена в группу DingTalk. Затем нажмите «Принять соглашение» для завершения. Наконец, скопируйте адрес Webhook для дальнейшего использования.

4. Реализация кода

Получив адрес Webhook, мы можем инициировать HTTP-запрос POST на этот адрес в количественной стратегии Inventor для отправки информации в эту группу DingTalk. Следует отметить, что при инициировании запроса POST необходимо установить кодировку набора символов 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)  # 休眠一分钟

При настройке роботов для синхронизации информации вы можете указать номера мобильных телефонов нескольких членов группы. Когда участники группы @ получат сообщение, они получат напоминание об этом с помощью сообщения @, даже если установлен сеанс «Не беспокоить».


            
            
            
# 向钉钉群输出信息
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)