添加钉钉消息通知

user_data/strategies/SimpleRSIStrategy_Fixed.py

@@ -9,6 +9,43 @@ from typing import Dict, List, Optional
 from functools import reduce
 import logging
 
+from dingtalk import DingTalkBot
+import numpy as np
+import pandas as pd
+from datetime import datetime, timedelta, timezone
+from pandas import DataFrame
+from typing import Optional, Union
+from freqtrade.strategy import (
+    IStrategy,
+    Trade,
+    Order,
+    PairLocks,
+    informative,  # @informative decorator
+    # Hyperopt Parameters
+    BooleanParameter,
+    CategoricalParameter,
+    DecimalParameter,
+    IntParameter,
+    RealParameter,
+    # timeframe helpers
+    timeframe_to_minutes,
+    timeframe_to_next_date,
+    timeframe_to_prev_date,
+    # Strategy helper functions
+    merge_informative_pair,
+    stoploss_from_absolute,
+    stoploss_from_open,
+)
+
+
+webhook = "https://oapi.dingtalk.com/robot/send?access_token=87c7abfcdd69b699c32da4e4f5981cd2ca6b0445474fc6ffb36f2ed0f6262fbb"  # 填写你的webhook
+secret = "SECf3d6b43f2f8a87ab91feffd052e71ec314fbf57a1842e483fe07af3c0a0e5aa6"  # 填写你的加签token（如果有），否则留空
+
+# CTA 群机器人
+# webhook = "https://oapi.dingtalk.com/robot/send?access_token=87c7abfcdd69b699c32da4e4f5981cd2ca6b0445474fc6ffb36f2ed0f6262fbb"
+# secret = "SECf3d6b43f2f8a87ab91feffd052e71ec314fbf57a1842e483fe07af3c0a0e5aa6"
+dingtalk = DingTalkBot(webhook, secret)
+
 logger = logging.getLogger(__name__)
 
 
@@ -57,6 +94,35 @@ class SimpleRSIStrategyFixed(IStrategy):
     ema_breakout_threshold = 0.02
     ema_separation_threshold = 0.02
 
+    def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:  
+        # 交易对  
+        trading_pair = pair  
+        
+        # 时间  
+        fill_time = order.order_filled_date or current_time  
+        
+        # 价格  
+        fill_price = order.average or order.price  
+        
+        # 买入还是卖出  
+        side = order.ft_order_side  # 'buy' 或 'sell'  
+        
+        # 仓位(当前持仓数量)  
+        position = trade.amount  
+        
+        # 或者使用日志  
+        logger.info(  
+            f"订单成交 - 交易对: {trading_pair}, "  
+            f"时间: {fill_time}, "  
+            f"价格: {fill_price}, "  
+            f"方向: {side}, "  
+            f"仓位: {position}"  
+        )  
+        dingtalk.send_text(
+            content=f"订单成交 - 交易对: {trading_pair}, 时间: {fill_time}, 价格: {fill_price}, 方向: {side}, 仓位: {position}")
+        
+        return None
+
     def populate_indicators(self, dataframe: pd.DataFrame, metadata: Dict) -> pd.DataFrame:
         """
         添加技术指标到数据框

user_data/strategies/dingtalk.py

@@ -0,0 +1,118 @@
+import requests
+import time
+import hmac
+import hashlib
+import base64
+import urllib.parse
+# from loguru import logger
+
+
+class DingTalkBot:
+    """
+    钉钉机器人类，通过webhook和可选的加签token向群聊发送消息提醒
+    """
+
+    def __init__(self, webhook: str, secret: str = None):
+        """
+        :param webhook: 钉钉自定义机器人webhook地址
+        :param secret: 加签密钥（可选）
+        """
+        self.webhook = webhook
+        self.secret = secret
+
+    def _gen_signed_url(self):
+        """
+        如果有加签token，根据钉钉接口算法拼接签名到url
+        """
+        if not self.secret:
+            return self.webhook
+        timestamp = str(round(time.time() * 1000))
+        secret_enc = self.secret.encode("utf-8")
+        string_to_sign = f"{timestamp}\n{self.secret}"
+        string_to_sign_enc = string_to_sign.encode("utf-8")
+        hmac_code = hmac.new(
+            secret_enc, string_to_sign_enc, digestmod=hashlib.sha256
+        ).digest()
+        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
+        url = f"{self.webhook}&timestamp={timestamp}&sign={sign}"
+        return url
+
+    def send_text(self, content: str, at_mobiles=None, is_at_all=False):
+        """
+        发送文本消息
+
+        :param content: 消息内容
+        :param at_mobiles: 需要@的手机号组成的列表，可选
+        :param is_at_all: 是否@所有人，默认False
+        """
+        at_mobiles = at_mobiles or []
+        data = {
+            "msgtype": "text",
+            "text": {"content": content},
+            "at": {"atMobiles": at_mobiles, "isAtAll": is_at_all},
+        }
+
+        url = self._gen_signed_url() if self.secret else self.webhook
+
+        try:
+            response = requests.post(url, json=data, timeout=5)
+            response.raise_for_status()
+            result = response.json()
+            if result.get("errcode", -1) != 0:
+                print(f"钉钉消息发送失败: {result}")
+            else:
+                print("钉钉消息发送成功")
+        except Exception as e:
+            print(f"钉钉消息发送异常: {e}")
+
+    def send_markdown(self, title: str, text: str, at_mobiles=None, is_at_all=False):
+        """
+        发送markdown消息
+
+        :param title: 消息标题
+        :param text: markdown格式的消息内容
+        :param at_mobiles: 需要@的手机号组成的列表，可选
+        :param is_at_all: 是否@所有人，默认False
+        """
+        at_mobiles = at_mobiles or []
+        data = {
+            "msgtype": "markdown",
+            "markdown": {"title": title, "text": text},
+            "at": {"atMobiles": at_mobiles, "isAtAll": is_at_all},
+        }
+
+        url = self._gen_signed_url() if self.secret else self.webhook
+
+        try:
+            response = requests.post(url, json=data, timeout=5)
+            response.raise_for_status()
+            result = response.json()
+            if result.get("errcode", -1) != 0:
+                print(f"钉钉markdown消息发送失败: {result}")
+            else:
+                print("钉钉markdown消息发送成功")
+        except Exception as e:
+            print(f"钉钉markdown消息发送异常: {e}")
+
+
+if __name__ == "__main__":
+
+    webhook = "https://oapi.dingtalk.com/robot/send?access_token=fb70c1561d8beba94b4f11568f4bb15e3ae07ccbdc8ac19676434a9d1cd17546"  # 填写你的webhook
+    secret = "SEC1ae7cd2f1a6f9da3611af37da3e7d954c1e8533fc073c6c8cc5e5af3b6e5926b"  # 填写你的加签token（如果有），否则留空
+
+    # CTA 群机器人
+    # webhook = "https://oapi.dingtalk.com/robot/send?access_token=87c7abfcdd69b699c32da4e4f5981cd2ca6b0445474fc6ffb36f2ed0f6262fbb"
+    # secret = "SECf3d6b43f2f8a87ab91feffd052e71ec314fbf57a1842e483fe07af3c0a0e5aa6"
+    dingtalk = DingTalkBot(webhook, secret)
+    dingtalk.send_text("测试消息")
+
+    # 测试markdown消息
+    markdown_content = """
+## 系统通知
+- **状态**: 正常运行
+- **CPU使用率**: 65%
+- **内存使用率**: 78%
+
+> 详细信息请查看监控面板
+    """
+    dingtalk.send_markdown("系统状态报告", markdown_content)