diff --git a/dingtalk.py b/dingtalk.py index bc33566..06521d1 100644 --- a/dingtalk.py +++ b/dingtalk.py @@ -6,6 +6,7 @@ import base64 import urllib.parse from loguru import logger + class DingTalkBot: """ 钉钉机器人类,通过webhook和可选的加签token向群聊发送消息提醒 @@ -26,15 +27,16 @@ class DingTalkBot: if not self.secret: return self.webhook timestamp = str(round(time.time() * 1000)) - secret_enc = self.secret.encode('utf-8') + 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() + 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): """ 发送文本消息 @@ -47,10 +49,7 @@ class DingTalkBot: data = { "msgtype": "text", "text": {"content": content}, - "at": { - "atMobiles": at_mobiles, - "isAtAll": is_at_all - } + "at": {"atMobiles": at_mobiles, "isAtAll": is_at_all}, } url = self._gen_signed_url() if self.secret else self.webhook @@ -78,14 +77,8 @@ class DingTalkBot: at_mobiles = at_mobiles or [] data = { "msgtype": "markdown", - "markdown": { - "title": title, - "text": text - }, - "at": { - "atMobiles": at_mobiles, - "isAtAll": is_at_all - } + "markdown": {"title": title, "text": text}, + "at": {"atMobiles": at_mobiles, "isAtAll": is_at_all}, } url = self._gen_signed_url() if self.secret else self.webhook @@ -105,14 +98,14 @@ class DingTalkBot: if __name__ == "__main__": webhook = "https://oapi.dingtalk.com/robot/send?access_token=fb70c1561d8beba94b4f11568f4bb15e3ae07ccbdc8ac19676434a9d1cd17546" # 填写你的webhook - secret = "SEC1ae7cd2f1a6f9da3611af37da3e7d954c1e8533fc073c6c8cc5e5af3b6e5926b" # 填写你的加签token(如果有),否则留空 + 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 = """ ## 系统通知 @@ -122,4 +115,4 @@ if __name__ == "__main__": > 详细信息请查看监控面板 """ - dingtalk.send_markdown("系统状态报告", markdown_content) \ No newline at end of file + dingtalk.send_markdown("系统状态报告", markdown_content) diff --git a/index_downloader.py b/index_downloader.py index 4fdc5d7..2b1452d 100644 --- a/index_downloader.py +++ b/index_downloader.py @@ -81,6 +81,9 @@ if __name__ == "__main__": import akshare as ak import pandas as pd - for symbol in ['NVDA', 'AAPL', 'MSFT', 'AMZN', 'TSLA', 'META', 'GOOGL']: - stock_us_daily_df = ak.stock_us_daily(symbol=symbol, adjust="qfq") - stock_us_daily_df.to_csv(f"{symbol}_stock_us_daily.csv", index=False, encoding="utf-8-sig") + + for symbol in ["NVDA", "AAPL", "MSFT", "AMZN", "TSLA", "META", "GOOGL"]: + stock_us_daily_df = ak.stock_us_daily(symbol=symbol, adjust="qfq") + stock_us_daily_df.to_csv( + f"{symbol}_stock_us_daily.csv", index=False, encoding="utf-8-sig" + ) diff --git a/signal_calc.py b/signal_calc.py index 5751539..1e3e3c7 100644 --- a/signal_calc.py +++ b/signal_calc.py @@ -10,6 +10,7 @@ import traceback from dingtalk import DingTalkBot import talib as ta from tabulate import tabulate + env = "online" # env = "on" db_config = DatabaseConfig(env=env) @@ -104,7 +105,9 @@ def main(): # dingtalk.send_markdown( # f"CCI信号", signal_df.to_markdown(tablefmt="simple", index=False) # ) - dingtalk.send_text(tabulate(signal_df, tablefmt='plain', headers="keys", showindex=False)) + dingtalk.send_text( + tabulate(signal_df, tablefmt="plain", headers="keys", showindex=False) + ) if __name__ == "__main__": diff --git a/update_data.py b/update_data.py index a9d9cda..2e9c93b 100644 --- a/update_data.py +++ b/update_data.py @@ -62,11 +62,9 @@ def get_latest_index_kline_date(): return df - - def main(): webhook = "https://oapi.dingtalk.com/robot/send?access_token=fb70c1561d8beba94b4f11568f4bb15e3ae07ccbdc8ac19676434a9d1cd17546" # 填写你的webhook - secret = "SEC1ae7cd2f1a6f9da3611af37da3e7d954c1e8533fc073c6c8cc5e5af3b6e5926b" # 填写你的加签token(如果有),否则留空 + secret = "SEC1ae7cd2f1a6f9da3611af37da3e7d954c1e8533fc073c6c8cc5e5af3b6e5926b" # 填写你的加签token(如果有),否则留空 dingtalk = DingTalkBot(webhook, secret) try: db_config = DatabaseConfig() @@ -102,4 +100,3 @@ if __name__ == "__main__": while True: schedule.run_pending() time.sleep(1) -