代码格式化

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')
+        string_to_sign_enc = string_to_sign.encode("utf-8")
-        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
+        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):
         """
         发送文本消息
@@ -47,10 +49,7 @@ class DingTalkBot:
         data = {
             "msgtype": "text",
             "text": {"content": content},
-            "at": {
+            "at": {"atMobiles": at_mobiles, "isAtAll": is_at_all},
-                "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": {
+            "markdown": {"title": title, "text": text},
-                "title": title,
+            "at": {"atMobiles": at_mobiles, "isAtAll": is_at_all},
-                "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)
+    dingtalk.send_markdown("系统状态报告", markdown_content)

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")  
+    for symbol in ["NVDA", "AAPL", "MSFT", "AMZN", "TSLA", "META", "GOOGL"]:
-        stock_us_daily_df.to_csv(f"{symbol}_stock_us_daily.csv", index=False, encoding="utf-8-sig")
+        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"
+        )

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__":

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)
-