I. 요약

실제 거래에서 Inventor Quant Robot의 거래 상태를 파악하기 위해 때때로 로봇이 실행한 거래 결과를 WeChat, 이메일, 문자 메시지 등으로 보내야 합니다. 하지만 매일 수백 개의 다양한 종류의 메시지가 오면서 사람들은 더 이상 이러한 메시지에 민감하지 않아 중요한 메시지를 적시에 확인할 수 없게 되었습니다. 따라서 이 문서에서는 DingTalk 그룹 인터페이스를 호출하여 로봇 메시지 푸시를 구현합니다.

2. DingTalk 그룹 로봇

DingTalk 그룹 로봇은 고급 확장 기능입니다. DingTalk 계정만 있으면 사용할 수 있습니다. 타사 정보를 DingTalk 그룹에 집계하고 정보의 자동 동기화를 실현할 수 있습니다. Webhook 프로토콜의 사용자 정의 액세스를 지원하고 Inventor Quantitative Robot을 통해 DingTalk 그룹에 알림, 알람 및 기타 정보를 집계합니다. 텍스트, 링크, 마크다운의 세 가지 메시지 형식과 다섯 가지 메시지 유형을 지원합니다. 동일한 메시지를 여러 DingTalk 그룹에 동시에 보낼 수도 있습니다. 공식 링크를 참조하세요: https://ding-doc.dingtalk.com/doc#/serverapi2/ye8tup

3. 로봇 만들기

1단계: DingTalk 그룹 만들기 DingTalk 그룹에서 사용자 정의 로봇이 생성될 때마다 고유한 Hook 주소가 생성되는데, 이를 WebHook 주소라고 합니다. WebHook 주소로 메시지를 푸시하면 DingTalk 그룹이 메시지를 받게 됩니다. PC에서 DingTalk를 예로 들어보겠습니다. 먼저 왼쪽 상단 모서리에 있는 “+” 기호를 클릭하여 그룹 채팅을 시작합니다. 본인만 메시지를 받고 싶다면 무작위로 두 명을 추가한 다음 쫓아낼 수 있습니다. 채우기 그룹 이름에 “FMZ Robot”을 입력하고 그룹 유형으로 Normal을 선택합니다. 그냥 그룹입니다.

2단계: DingTalk 그룹 로봇 추가 아바타를 클릭하고 로봇 관리를 선택한 다음 사용자 지정을 선택하고 추가를 클릭합니다. 로봇 이름을 “FMZ”로 지정하고 방금 만든 DingTalk 그룹에 추가하세요. 로봇은 세 가지 보안 설정을 지원합니다.

• 사용자 지정 키워드: 정보는 이 키워드가 포함된 경우에만 동기화됩니다.
• 서명 추가: 비밀번호 설정과 동일합니다.
• IP 주소: 고정된 제3자 정보의 IP 주소 세그먼트입니다. 미리알림이나 알람으로만 사용하는 경우 사용자 지정 키워드를 선택하기만 하면 됩니다. 여기서 정의하는 키워드는 “:“입니다. 즉, Inventor Quant 로봇이 푸시한 정보에 “:“가 포함되어 있으면 이 정보가 DingTalk 그룹으로 푸시된다는 의미입니다. 그런 다음 계약에 동의를 클릭하여 완료하세요. 마지막으로 나중에 사용하기 위해 Webhook 주소를 복사하세요.

4. 코드 구현

Webhook 주소를 얻은 후 Inventor Quantitative Strategy에서 이 주소로 HTTP POST 요청을 시작하여 이 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)

5. 로봇 테스트

위 코드에서 우리는 매분 주요 디지털 통화의 가격을 얻고 이 정보를 DingTalk 그룹에 푸시하는 사례를 작성했습니다.