로봇 푸시 메시지를 실현하기 위해 Dingding 인터페이스를 호출

로봇 푸시 메시지를 실현하기 위해 Dingding 인터페이스를 호출

1. 요약

실제 거래에서 FMZ 퀀트 로봇의 거래 상태를 시간적으로 알기 위해서는 때때로 로봇이 실행한 거래 결과를 WeChat, 이메일, SMS 등으로 전송해야 합니다. 그러나 매일 수백 가지 종류의 정보가 존재하므로 이러한 정보에 민감하지 않아 중요한 정보를 적시에 수집하는 데 실패합니다. 따라서 이 기사는 Dingding 그룹 인터페이스를 호출하여 로봇 푸시 메시지를 구현합니다.

2. 딩딩 그룹 로봇

딩딩 그룹 로봇은 고급 확장 기능입니다. 딩딩 계정이있는 한 사용할 수 있습니다. 자동 정보 동기화를 달성하기 위해 제3자 정보를 딩딩 그룹에 집계 할 수 있습니다. 웹후크 프로토콜의 사용자 정의 액세스를 지원하고 FMZ 퀀트 로봇을 통해 상기, 경고 및 기타 정보를 딩딩 그룹에 집계합니다. 텍스트, 링크 및 마크다운: 세 가지 메시지 형식 및 다섯 가지 메시지 유형이 지원됩니다. 동일한 메시지는 동시에 여러 딩딩 그룹에 전송 될 수 있습니다. 공식 링크 참조:https://ding-doc.dingtalk.com/doc#/serverapi2/ye8tup

3. 로봇 을 만들 것

단계 1: 딩딩 그룹을 만들자

딩딩 그룹에서 생성된 각 사용자 정의 로봇은 WebHook 주소라고 불리는 고유 한 주소를 생성합니다. 딩딩 그룹은 WebHook 주소로 메시지를 푸시킴으로써 메시지를 받게됩니다. 예를 들어 PC 버전의 딩딩을 보자. 먼저 그룹 채팅을 시작하기 위해 왼쪽 상단에있는 + 기호를 클릭하십시오. 메시지를 직접 받아들이고 싶다면 두 사람을 선택하고 그들을 쫓아낼 수 있습니다. 그룹 이름을 입력하십시오: FMZ Robot, 그리고 그룹 유형으로 일반 그룹을 선택하십시오.

단계 2: Dingding 그룹 로봇을 추가

아바타를 클릭하고 로봇 관리를 선택하고 사용자 정의를 선택하고 추가를 클릭합니다. 사용자 정의 로봇 이름: FMZ, 새로 만든 Dingding 그룹에 추가됩니다. 로봇은 세 가지 보안 설정을 지원합니다:

• 사용자 정의 키워드: 이 키워드를 포함할 경우에만 정보가 동기화됩니다.
• 서명: 암호를 설정하는 것과 같습니다.
• IP 주소: 고정된 제3자 정보의 IP 주소 세그먼트.

상기 또는 경고를 위해만 사용된다면 사용자 정의 키워드를 선택하십시오. 여기서 정의하는 키워드는 :입니다. 즉, FMZ 퀀트 로봇이 푸시하는 정보가 :을 포함하면 정보가 Dingding 그룹으로 푸시됩니다. 다음 동의를 클릭하여 계약을 완료합니다. 마지막으로 백업을 위해 Webhook 주소를 복사합니다.

4. 코드 구현

Webhook 주소를 얻은 후, 우리는 FMZ 퀀트 전략의 주소로 HTTP POST 요청을 전송하여 Dingding 그룹에 정보를 보낼 수 있습니다. POST 요청이 시작되면 문자 집합 인코딩이 UTF-8로 설정되어야한다는 점에 유의하십시오.

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


# Output information to Dingding group
def msg(text):
    token ="0303627a118e739e628bcde104e19cf5463f61a4a127e4f2376e6a8aa1156ef1"
    headers = {'Content-Type': 'application/json;charset=utf-8'}  # Request header
    api_url = f"https://oapi.dingtalk.com/robot/send?access_token={token}"
    json_text = {
        "msgtype": "text",  # Message type
        "text": {
            "content": text
        }
    }
    # Send and print messages
    Log(requests.post(api_url, json.dumps(json_text), headers=headers).content)

    
# Test functions
def onTick():
    arr = ['BTC', 'ETH', 'XRP', 'BCH', 'LTC']  # Mainstream digital currencies
    # Get the time of East Zone 8
    bj_dt = str(datetime.now().astimezone(timezone(timedelta(hours=8))))
    bj_dt = bj_dt.split('.')[0]  # Time of processing
    text = f'{bj_dt}\n'  # Define information content
    for i in arr:  # Loop mainstream digital currency array
        exchange.IO("currency", f"{i}_USDT")  # Switch trading pairs
        ticker = exchange.GetTicker().Last  # Get the latest price
        if i == 'LTC':
            full = ' ：'
        else:
            full = '：'
        text = text + f"{i}/USDT{full}${ticker}\n"  # Processing information content
    msg(text)  # Call msg function to output information
    

# Strategy entrance
def main():
    while True:  # Enter infinite loop 
        onTick()  # Execute onTick function
        Sleep(1000 * 60)  # Sleep for one minute

사용자 지정 로봇이 정보를 동기화 할 때, 그룹 내의 여러 구성원에게 휴대 전화 번호를 @로 설정할 수 있습니다. 그룹 구성원이 메시지를 수신 할 때 @ 메시지 상기이있을 것입니다. 방해하지 않는 세션이 설정되어 있더라도 상기이 여전히 알립니다.

# Output information to Dingding group
def msg(text):
    token = "0303627a118e739e628bcde104e19cf5463f61a4a127e4f2376e6a8aa1156ef1"
    headers = {'Content-Type': 'application/json;charset=utf-8'}  # Request header
    api_url = f"https://oapi.dingtalk.com/robot/send?access_token={token}"
    json_text = {
        "msgtype": "text",  # Message type
        "text": {
            "content": text
        },
        "at": {
            "atMobiles": [
                "16666666666",  # Phone number of the @
                "18888888888"  # Phone number of the @
            ],
            "isAtAll": False  # Not @ Everyone
        }
    }
    # Send and print messages
    Log(requests.post(api_url, json.dumps(json_text), headers=headers).content)

5. 테스트 로봇

위의 코드에서, 우리는 주류 디지털 화폐의 가격을 1분마다 얻고 이 정보를 딩딩 그룹에 보내기 위한 케이스를 작성했습니다.