diff --git a/user_data/strategies/SimpleRSIStrategy_Fixed.py b/user_data/strategies/SimpleRSIStrategy_Fixed.py index 3fe0075..7ac93d8 100644 --- a/user_data/strategies/SimpleRSIStrategy_Fixed.py +++ b/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: """ 添加技术指标到数据框 diff --git a/user_data/strategies/dingtalk.py b/user_data/strategies/dingtalk.py new file mode 100644 index 0000000..a353f0b --- /dev/null +++ b/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}×tamp={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)