DingTalk APIを呼び出してロボットプッシュメッセージを実装する

実際の取引におけるInventor Quant Robotの取引状況を把握するために、ロボットが実行した取引結果をWeChat、電子メール、テキストメッセージなどに送信する必要がある場合があります。しかし、毎日何百ものさまざまな種類のメッセージを受信すると、人々はこれらのメッセージに敏感ではなくなり、重要なメッセージをタイムリーに確認できなくなります。そこで、この記事では、DingTalk グループインターフェースを呼び出してロボットメッセージプッシュを実装します。

2. DingTalkグループロボット

DingTalkグループロボットは高度な拡張機能です。DingTalkアカウントをお持ちであればご利用いただけます。サードパーティの情報をDingTalkグループに集約し、情報の自動同期を実現します。 Webhook プロトコルのカスタムアクセスをサポートし、Inventor Quantitative Robot を通じてリマインダー、アラーム、その他の情報を DingTalk グループに集約します。テキスト、リンク、マークダウンの 3 つのメッセージ形式と 5 つのメッセージタイプをサポートします。同じメッセージを複数のDingTalkグループに同時に送信することもできます。
公式リンクを参照してください: https://ding-doc.dingtalk.com/doc#/serverapi2/ye8tup

3. ロボットを作る

ステップ1: DingTalkグループを作成する

DingTalk グループでカスタムロボットが作成されるたびに、WebHook アドレスと呼ばれる一意のフックアドレスが生成されます。WebHook アドレスにメッセージをプッシュすると、DingTalk グループがメッセージを受信します。 PC版のDingTalkを例に挙げてみましょう。まず、左上隅の「+」記号をクリックしてグループチャットを開始します。自分だけがメッセージを受け取りたい場合は、ランダムに2人を追加して追い出すことができます。グループ名に「FMZ Robot」と入力し、グループの種類として「通常」を選択します。グループ化するだけです。

ステップ2: DingTalkグループロボットを追加する
アバターをクリックし、「ロボット管理」を選択してから、「カスタマイズ」を選択し、「追加」をクリックします。ロボット名「FMZ」をカスタマイズし、作成した DingTalk グループに追加します。ロボットは次の 3 つのセキュリティ設定をサポートしています。

カスタムキーワード: このキーワードが含まれている場合にのみ情報が同期されます。
署名を追加: パスワードの設定と同等です。
IP アドレス: 固定サードパーティ情報の 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)