1 更新自己的东方财富爬虫函数 2 测试突破买入 3 测试保存yaml及读取功能

EmailTest.py

@@ -101,7 +101,8 @@ def parse_return_email(to_email, send_email_time):
         encoded_email = to_email.encode('utf-8')
         
         # 构建搜索条件：HEADER FROM "<发件人邮箱>, 时间大于 <send_email_time>"
-        search_criteria = f'FROM "{encoded_email.decode("utf-8")}" SINCE "{send_email_time}"'
+        search_criteria = ['FROM', encoded_email.decode('utf-8'), 'SINCE', send_email_time]
+
         # search_criteria = f'FROM "{encoded_email.decode("utf-8")}"'
         
         # 执行搜索操作
@@ -148,9 +149,9 @@ def parse_return_email(to_email, send_email_time):
         # 根据“实际买入价格-买入份额-手续费”格式解析body中的内容
         parse_states = True
         body.split("-")
-        price = body.split("-")[0]
-        share = body.split("-")[1]
-        fee = body.split("-")[2]
+        price = float(body.split("-")[0])
+        share = float(body.split("-")[1])
+        fee = float(body.split("-")[2])
         return parse_states, price, share, fee
     
 def check_email(to_email, send_email_time):

TurtleOnTime.py

@@ -11,12 +11,21 @@ from dataclasses import dataclass
 import time
 import threading
 import yaml  # 添加YAML支持
+import logging
+from stock_em import stock_zh_a_spot_em
+from etf_em import fund_etf_spot_em
+import random
+import urllib.parse
+import requests
 
 '''
 todo
 
-1 运行过程框架调整，支持多个turtle同时监测
-2 增加运行状态写入yaml文件，读取文件恢复状态
+1 运行过程框架调整，支持多个turtle同时监测 done
+2 增加运行状态写入yaml文件，读取文件恢复状态 done
+3 测试每分钟获取实时信息，是否稳定。 done
+4 etf实时数据使用异步方式，获取速度加快了 done
+4 获取数据，调用clash代理 done
 '''
 
 @dataclass
@@ -90,10 +99,10 @@ class TurtleTrading(object):
         self.Capital = Capital
         self.cash = cash
         self.TrigerTime = 0
-        self.BuyStates = list[BuyState]
+        self.BuyStates = []
         
-        self.tradeslog = list[TradeLog]
-        self.BreakOutLog = list[BreakOutLog]
+        self.tradeslog = []
+        self.BreakOutLog = []
         self.PriceNow = 0.0
         self.Donchian_20_up = 0.0
         self.Donchian_10_down = 0.0
@@ -295,7 +304,7 @@ class TurtleTrading(object):
         
     def system1EnterSafe(self, PriceNow, TempDonchian55Upper):
         
-        if PriceNow > TempDonchian55Upper[-1]: # 保底的55日突破
+        if PriceNow > TempDonchian55Upper: # 保底的55日突破
             return True
         else:
             return False
@@ -345,6 +354,34 @@ class TurtleTrading_OnTime(object):
     3、实时监测主流程
     '''
     
+    def __init__(self, turtles: list[TurtleTrading], user_email):
+        self.turtles = turtles  # List of TurtleTrading instances
+        self.user_email = user_email
+        self.email_events = {}  # Track email response events for each turtle
+        logging.basicConfig(level=logging.INFO)
+        # Load previous state from YAML if exists
+        self.load_previous_state()
+        
+        self.clash_api = "http://127.0.0.1:9090"
+        self.node_group_name_encoded = urllib.parse.quote("🚀代理线路", safe='')  # 根据 Clash 中策略组的名字填写
+        self.clash_secret = "6Gp-fdt-veS-ugv"
+        self.node_names = [
+                "R1-0|香港-NF|深|负载均衡",
+                "R1-1|香港-NF|粤|负载均衡",
+                "R1-2|香港-NF|沪|负载均衡",
+                "R1-3|香港-NF|湘|负载均衡",
+                "R2-1|香港-NF|HKT家宽",
+                "R2-2|香港-NF|HKT家宽",
+                "R2-3|香港-NF|HKT家宽",
+                "R2-5|香港-NF|HKT家宽",
+                "R3-1|香港-NF|BGP静态",
+                "R4-1|台湾-NF|家宽|原生",
+                "R5-1|日本-NF|GMO|原生IP",
+                "R5-4|日本-NF|IIJ|精品",
+                "R6-1|美国-NF|BGP",
+                "R9-1|狮城-NF|FDC",
+            ]
+    
     def load_previous_state(self):
         """Load previous state from YAML file if exists"""
         state_dir = "state"
@@ -355,40 +392,48 @@ class TurtleTrading_OnTime(object):
         if os.path.exists(filename):
             with open(filename, 'r') as f:
                 state_data = yaml.safe_load(f)
+                main_state = state_data.get('main_state', {})
                 
             # Restore state
-            for turtle_data in state_data.get('turtles', []):
-                # Find or create TurtleTrading instance
-                turtle = next((t for t in self.turtles if t.TradeCode == turtle_data['TradeCode']), None)
-                if not turtle:
-                    # Create new instance if not found (should not happen)
-                    turtle = TurtleTrading(**turtle_data)
-                    self.turtles.append(turtle)
-                
-                # Restore attributes
-                turtle.TradeCode = turtle_data['TradeCode']
-                turtle.type = turtle_data['type']
-                turtle.riskcoe = turtle_data['riskcoe']
-                turtle.Capital = turtle_data['Capital']
-                turtle.cash = turtle_data['cash']
-                turtle.TrigerTime = turtle_data['TrigerTime']
-                turtle.BuyStates = [BuyState(**bs) for bs in turtle_data['BuyStates']]
-                turtle.tradeslog = [TradeLog(**tl) for tl in turtle_data['tradeslog']]
-                turtle.BreakOutLog = [BreakOutLog(**bol) for bol in turtle_data['BreakOutLog']]
-                turtle.PriceNow = turtle_data['PriceNow']
-                turtle.Donchian_20_up = turtle_data['Donchian_20_up']
-                turtle.Donchian_10_down = turtle_data['Donchian_10_down']
-                turtle.Donchian_50_up = turtle_data['Donchian_50_up']
-                turtle.is_gap_up = turtle_data['is_gap_up']
-                turtle.prev_heigh = turtle_data['prev_heigh']
+            try:
+                for turtle_data in main_state.get('turtles', []):
+                    # Find or create TurtleTrading instance
+                    turtle = next((t for t in self.turtles if t.TradeCode == turtle_data['TradeCode']), None)
+                    if not turtle:
+                        # Create new instance if not found (should not happen)
+                        turtle = TurtleTrading(**turtle_data)
+                        self.turtles.append(turtle)
+                    
+                    # Restore attributes
+                    turtle.TradeCode = turtle_data['TradeCode']
+                    turtle.type = turtle_data['type']
+                    turtle.riskcoe = turtle_data['riskcoe']
+                    turtle.Capital = turtle_data['Capital']
+                    turtle.cash = turtle_data['cash']
+                    turtle.TrigerTime = turtle_data['TrigerTime']
+                    turtle.BuyStates = [BuyState(**bs) for bs in turtle_data['BuyStates']]
+                    turtle.tradeslog = [TradeLog(**tl) for tl in turtle_data['tradeslog']]
+                    turtle.BreakOutLog = [BreakOutLog(**bol) for bol in turtle_data['BreakOutLog']]
+    
+            except Exception as e:
+                logging.error(f"Error loading previous state: {e}")
+    def switch_random_node(self):
+        '''随机切换clash节点
+        '''
+        selected_node = random.choice(self.node_names)
+        url = f"{self.clash_api}/proxies/{self.node_group_name_encoded}"
+        headers = {
+        "Authorization": f"Bearer {self.clash_secret}"
+        }
+        try:
+            res = requests.put(url, headers=headers, json={"name": selected_node})
+            if res.status_code == 204:
+                print(f"[✓] 成功切换节点为：{selected_node}")
+            else:
+                print(f"[✗] 切换失败：{res.status_code} - {res.text}")
+        except Exception as e:
+            print(f"[!] 请求失败: {e}")
 
-    def __init__(self, turtles: list[TurtleTrading], user_email):
-        self.turtles = turtles  # List of TurtleTrading instances
-        self.user_email = user_email
-        self.email_events = {}  # Track email response events for each turtle
-        
-        # Load previous state from YAML if exists
-        self.load_previous_state()
     
     def day_end(self):
         """Save current state to YAML file at the end of the day"""
@@ -428,20 +473,21 @@ class TurtleTrading_OnTime(object):
     def get_stocks_data(self):
         """获取实时股票、基金数据，不保存
         """
-        stock_data = ak.stock_zh_a_spot_em()
-        stock_data = stock_data.dropna(subset=['最新价'])
-        # # print(stock_zh_a_spot_df)
-        # # stock_zh_a_spot_df第一列加上时间，精确到分钟
-        # stock_zh_a_spot_df['时间'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        # mysql_database.insert_db(stock_zh_a_spot_df, "stock_price", True, "代码")
+        try:
+            self.switch_random_node()
+            stock_data = stock_zh_a_spot_em()
+            stock_data = stock_data.dropna(subset=['最新价'])
+            
+            etf_data = fund_etf_spot_em()
+            etf_data = etf_data.dropna(subset=['最新价'])
+            
+            # 成功调用，使用logging记录日志，加上时间
+            logging.info(f"Successfully fetched stock and ETF data at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+            
+        except Exception as e:
+            logging.error(f"Error occurred while getting stock data: {e}")
+            return None, None
         
-        etf_data = ak.fund_etf_spot_em()
-        # etf_data = ak.fund_etf_spot_ths()
-        # etf_data = etf_data.dropna(subset=['当前-单位净值'])
-        etf_data = etf_data.dropna(subset=['最新价'])
-        
-        # etf_data['时间'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        # mysql_database.insert_db(etf_data, "etf_price", True, "代码")
         return stock_data, etf_data
 
     def Buy_stock(self, turtle: TurtleTrading, price_now):
@@ -453,8 +499,8 @@ class TurtleTrading_OnTime(object):
         body = f"{turtle.TradeCode}，价格{price_now}，份额{turtle.IntPositionSize} \n "
         body += "回复：实际买入价格-买入份额-手续费"
         send_email(subject, body, self.user_email)
-        send_email_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        
+        # send_email_time = datetime.strftime(datetime.now(),"%Y%m%d %H:%M:%S").date()
+        send_email_time = datetime.now().date()
         #每隔1分钟检测回信，解析邮件。
         
         parsed_email_flag = False
@@ -519,7 +565,7 @@ class TurtleTrading_OnTime(object):
         body = f"{turtle.TradeCode}，价格{price_now}，份额{turtle.IntPositionSize} \n "
         body += "回复：实际买入价格-买入份额-手续费"
         send_email(subject, body, self.user_email)
-        send_email_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        send_email_time = datetime.now().date()
 
         #每隔1分钟检测回信，解析邮件。
 
@@ -619,7 +665,7 @@ class TurtleTrading_OnTime(object):
         body = f"{turtle.TradeCode}，价格{price_now}，份额{turtle.IntPositionSize * sale_shares} \n "
         body += "回复：实际卖出价格-卖出份额-手续费"
         send_email(subject, body, self.user_email)
-        send_email_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        send_email_time = datetime.now().date()
         
         # 每隔1分钟检测回信，解析邮件。
 
@@ -669,7 +715,7 @@ class TurtleTrading_OnTime(object):
         body = f"{turtle.TradeCode}，价格{price_now}，份额{turtle.IntPositionSize} \n "
         body += "回复：实际卖出价格-卖出份额-手续费"
         send_email(subject, body, self.user_email)
-        send_email_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        send_email_time = datetime.now().date()
         
         # 每隔1分钟检测回信，解析邮件。
 
@@ -703,88 +749,6 @@ class TurtleTrading_OnTime(object):
                                     Net_return=abs(turtle.Capital - available_cash))
         turtle.tradeslog.append(sale_this_time)
 
-    # def run_short_trading_loop(self, stock_data, etf_data):
-
-    #     now = datetime.now().time()
-    #     # 根据类型获取当前价格
-    #     if self.turtle.type == "stock":
-    #         self.turtle.PriceNow = float(stock_data.loc[etf_data['代码'] == self.turtle.TradeCode, '最新价'].values[0])
-
-    #     elif self.turtle.type == "etf":
-    #         # self.turtle.PriceNow = float(etf_data.loc[etf_data['基金代码'] == self.turtle.TradeCode, '当前-单位净值'].values[0])
-    #         self.turtle.PriceNow = float(etf_data.loc[etf_data['代码'] == self.turtle.TradeCode, '最新价'].values[0])
-
-    #     # # 9点30 判断是否跳空高开
-    #     if now.hour == 9 and now.minute == 30 and self.turtle.PriceNow > self.turtle.prev_heigh:
-    #         self.turtle.is_gap_up = True
-            
-            
-    #     # 判断当前仓位状态并执行相应操作
-    #     if self.turtle.TrigerTime == 0:
-    #         # 空仓状态
-    #         if self.turtle.system1EnterNormal(
-    #             self.turtle.PriceNow,
-    #             self.turtle.Donchian_20_up,
-    #             self.turtle.BreakOutLog
-    #         ):
-    #             self.Buy_stock(self.turtle.PriceNow)
-                
-    #             # 突破 记录self.turtle.breakoutlog
-    #             today = datetime.now().strftime("%Y-%m-%d")
-    #             breakout_this_time = BreakOutLog(today,
-    #                                             self.turtle.Donchian_20_up, 
-    #                                             self.turtle.Donchian_20_up - 2 * self.turtle.N,
-    #                                             'valid',
-    #                                             None)
-    #             self.turtle.BreakOutLog.append(breakout_this_time)
-
-    #         elif self.turtle.system1EnterSafe(
-    #             self.turtle.PriceNow,
-    #             self.turtle.Donchian_50_up
-    #         ):
-    #             self.Buy_stock(self.turtle.PriceNow)
-
-    #     elif 1 <= self.turtle.TrigerTime <= 3:
-    #         # # 突破状态
-    #         # if self.turtle.system1EnterNormal(
-    #         #     self.turtle.PriceNow,
-    #         #     self.turtle.Donchian_20_up,
-    #         #     self.turtle.BreakOutLog
-    #         # ):
-    #         #     self.Buy_stock(self.turtle.PriceNow)
-    #         # elif self.turtle.system1EnterSafe(
-    #         #     self.turtle.PriceNow,
-    #         #     self.turtle.Donchian_50_up
-    #         # ):
-    #         #     self.Buy_stock(self.turtle.PriceNow)
-            
-    #         # 加仓状态
-    #         if self.turtle.add(self.turtle.PriceNow):
-    #             self.add_stock(self.turtle.PriceNow)
-                
-    #         # 止损状态
-    #         elif self.turtle.system_1_stop(self.turtle.PriceNow):
-    #             self.stop_sale_stock(self.turtle.PriceNow)
-            
-    #         # 止盈
-    #         elif self.turtle.system_1_Out(
-    #             self.turtle.PriceNow,
-    #             self.turtle.Donchian_10_down
-    #         ):
-    #             self.out_sale_stock(self.turtle.PriceNow)
-
-    #     elif self.turtle.TrigerTime == 4:
-    #         # 满仓  止损 止盈
-    #         if self.turtle.system_1_stop(self.turtle.PriceNow):
-    #             self.stop_sale_stock(self.turtle.PriceNow)
-    #         elif self.turtle.system_1_Out(
-    #             self.turtle.PriceNow,
-    #             self.turtle.Donchian_10_down
-    #         ):
-    #             self.out_sale_stock(self.turtle.PriceNow)
-
-    #     # 等待 1 分钟后下一次循环
-    #     time.sleep(60)
         
     def Start_short_system(self):
         """启动short系统
@@ -797,7 +761,7 @@ class TurtleTrading_OnTime(object):
             # 准备数据
             turtle.get_ready(100)
             turtle.N = float(turtle.CurrentData['ATR'].iloc[-1])
-            turtle.prev_heigh = float(turtle.CurrentData['最高价'].iloc[-1])
+            turtle.prev_heigh = float(turtle.CurrentData['最高'].iloc[-1])
             turtle.Donchian_20_up = float(turtle.CurrentData['Donchian_20_upper'].iloc[-1])
             turtle.Donchian_50_up = float(turtle.CurrentData['Donchian_50_upper'].iloc[-1])
             turtle.Donchian_10_down = float(turtle.CurrentData['Donchian_10_lower'].iloc[-1])
@@ -823,7 +787,7 @@ class TurtleTrading_OnTime(object):
                 (now.hour == 14 and 0 <= now.minute <= 59) or
                 (now.hour == 15 and now.minute <= 0)
             )
-
+            # is_trading_time = True
 
                 
             if not is_trading_time:
@@ -831,14 +795,15 @@ class TurtleTrading_OnTime(object):
                 time.sleep(60)
                 continue
             
-            is_stop_time = (now.hour > 15 and now.minute > 0) #收盘时间
+            is_stop_time = (now.hour >= 15 and now.minute > 0) #收盘时间
+            # is_stop_time = (now.hour >= 18 and now.minute > 32) #收盘时间   
             if is_stop_time:
                 break
 
             # 获取股票和ETF数据
             self.monitor_all_turtles()
             # 等待一段时间后再次检查
-            time.sleep(60)  # 每分钟检查一次
+            time.sleep(random.randint(60, 65))# 每分钟检查一次
         # ------------------结束阶段--------------------
         # 数据库更新当天数据，增加ATR、donchian数据
         # 直接做个新表
@@ -856,9 +821,19 @@ class TurtleTrading_OnTime(object):
         stock_data, etf_data = self.get_stocks_data()
         
         # 遍历所有turtle进行监控
+        # 为每个 Turtle 启动一个线程
+        threads = []
         for turtle in self.turtles:
-            self.monitor_single_turtle(turtle, stock_data, etf_data)
-        
+            thread = threading.Thread(
+                target=self.monitor_single_turtle,
+                args=(turtle, stock_data, etf_data)
+            )
+            thread.start()
+            threads.append(thread)
+
+        # 可选：等待所有线程完成（如果需要）
+        for thread in threads:
+            thread.join()
 
     
     def monitor_single_turtle(self, turtle: TurtleTrading, stock_data, etf_data):
@@ -866,15 +841,17 @@ class TurtleTrading_OnTime(object):
 
         now = datetime.now().time()
         if turtle.type == "stock":
-            turtle.PriceNow = float(stock_data.loc[etf_data['代码'] == self.turtle.TradeCode, '最新价'].values[0])
+            turtle.PriceNow = float(stock_data.loc[stock_data['代码'] == turtle.TradeCode, '最新价'].values[0])
 
         elif turtle.type == "etf":
             # self.turtle.PriceNow = float(etf_data.loc[etf_data['基金代码'] == self.turtle.TradeCode, '当前-单位净值'].values[0])
-            turtle.PriceNow = float(etf_data.loc[etf_data['代码'] == self.turtle.TradeCode, '最新价'].values[0])
+            turtle.PriceNow = float(etf_data.loc[etf_data['代码'] == turtle.TradeCode, '最新价'].values[0])
         
-        if now.hour == 9 and now.minute == 30 and self.turtle.PriceNow > self.turtle.prev_heigh:
+        if now.hour == 9 and now.minute == 30 and turtle.PriceNow > turtle.prev_heigh:
             turtle.is_gap_up = True
-            
+        
+        fake_price = 1.492
+        turtle.PriceNow = fake_price
         # 判断当前仓位状态并执行相应操作
         if turtle.TrigerTime == 0:
             if turtle.system1EnterNormal(
@@ -929,37 +906,39 @@ class TurtleTrading_OnTime(object):
 
     def start_email_thread(self, turtle:TurtleTrading, action, price_now):
         """启动邮件处理线程"""
-        event = threading.Event()
-        self.email_events[turtle.TradeCode] = event
-        
-        thread = threading.Thread(
-            target=self.handle_email_response,
-            args=(turtle, action, price_now, event)
-        )
-        thread.start()
+
+        self.handle_email_response(turtle, action, price_now)
+
     
-    def handle_email_response(self, turtle:TurtleTrading, action, price_now, event):
+    def handle_email_response(self, turtle:TurtleTrading, action, price_now):
         """处理邮件响应的线程"""
-        # 发送邮件
-        if action == "买入":
-            self.Buy_stock(turtle, price_now)
-        elif action == "加仓":
-            self.add_stock(turtle, price_now)
-        elif action == "止损":
-            self.stop_sale_stock(turtle, price_now)
-        elif action == "止盈":
-            self.out_sale_stock(turtle, price_now)
         
-        # 等待邮件响应
-        event.wait()
+        try:
+            logging.info("handle_email_response is called with action: {}".format(action))
+            # 发送邮件
+            if action == "买入":
+                self.Buy_stock(turtle, price_now)
+            elif action == "加仓":
+                self.add_stock(turtle, price_now)
+            elif action == "止损":
+                self.stop_sale_stock(turtle, price_now)
+            elif action == "止盈":
+                self.out_sale_stock(turtle, price_now)
+            else:
+                logging.warning(f"Unknown action: {action} for TradeCode: {turtle.TradeCode}")
+
+
+        except Exception as e:
+            logging.error(f"Error in handle_email_response for TradeCode: {turtle.TradeCode}, Error: {e}")
 
 if __name__ == '__main__':
     
     user_email = "guoyize2209@163.com"
-    t = TurtleTrading('513870', "etf", 0.0025, 100000, 200000)
+    nsdk = TurtleTrading('513870', "etf", 0.0025, 100000, 200000)
+    cjdl = TurtleTrading('600900', "stock", 0.0025, 100000, 200000)
     # t.get_ready(100)
     
-    a = TurtleTrading_OnTime(t, user_email)
+    a = TurtleTrading_OnTime([nsdk, cjdl], user_email)
     a.Start_short_system()
 
     # # 全是股票

etf_em.py

@@ -0,0 +1,231 @@
+import asyncio
+import httpx
+import math
+import pandas as pd
+from datetime import datetime, timezone, timedelta
+import logging
+
+logging.getLogger("httpx").setLevel(logging.WARNING)
+
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/122.0.0.0 Safari/537.36"
+    ),
+    "Referer": "https://quote.eastmoney.com/center/gridlist.html#fund_etf",  # 合理来源页
+    "Accept": "application/json, text/plain, */*",
+    "Accept-Language": "zh-CN,zh;q=0.9",
+    "Connection": "keep-alive",
+}
+
+
+async def fetch_page(client: httpx.AsyncClient, url: str, params: dict) -> pd.DataFrame:
+    try:
+        resp = await client.get(url, params=params, timeout=10)
+        data_json = resp.json()
+        if "data" in data_json and data_json["data"].get("diff"):
+            return pd.DataFrame(data_json["data"]["diff"])
+    except Exception as e:
+        print(f"Error on page {params.get('pn')}: {e}")
+    return pd.DataFrame()
+
+
+async def fund_etf_spot_em_async() -> pd.DataFrame:
+    url = "https://88.push2.eastmoney.com/api/qt/clist/get"
+    params = {
+        "pn": "1",
+        "pz": "100",
+        "po": "1",
+        "np": "1",
+        "ut": "bd1d9ddb04089700cf9c27f6f7426281",
+        "fltt": "2",
+        "invt": "2",
+        "wbp2u": "|0|0|0|web",
+        "fid": "f12",
+        "fs": "b:MK0021,b:MK0022,b:MK0023,b:MK0024,b:MK0827",
+        "fields": (
+            "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,"
+            "f12,f13,f14,f15,f16,f17,f18,f20,f21,"
+            "f23,f24,f25,f22,f11,f30,f31,f32,f33,"
+            "f34,f35,f38,f62,f63,f64,f65,f66,f69,"
+            "f72,f75,f78,f81,f84,f87,f115,f124,f128,"
+            "f136,f152,f184,f297,f402,f441"
+        ),
+    }
+
+    async with httpx.AsyncClient(headers=HEADERS, http2=True) as client:
+        # 获取首页数据，确认总页数
+        resp = await client.get(url, params=params)
+        data_json = resp.json()
+        if "data" not in data_json or not data_json["data"].get("diff"):
+            return pd.DataFrame()
+
+        total = data_json["data"]["total"]
+        per_page = len(data_json["data"]["diff"])
+        total_pages = math.ceil(total / per_page)
+
+        tasks = []
+        for page in range(1, total_pages + 1):
+            new_params = params.copy()
+            new_params["pn"] = str(page)
+            tasks.append(fetch_page(client, url, new_params))
+
+        dfs = await asyncio.gather(*tasks)
+        temp_df = pd.concat([df for df in dfs if not df.empty], ignore_index=True)
+
+        temp_df.rename(
+            columns={
+                "f12": "代码",
+                "f14": "名称",
+                "f2": "最新价",
+                "f4": "涨跌额",
+                "f3": "涨跌幅",
+                "f5": "成交量",
+                "f6": "成交额",
+                "f7": "振幅",
+                "f17": "开盘价",
+                "f15": "最高价",
+                "f16": "最低价",
+                "f18": "昨收",
+                "f8": "换手率",
+                "f10": "量比",
+                "f30": "现手",
+                "f31": "买一",
+                "f32": "卖一",
+                "f33": "委比",
+                "f34": "外盘",
+                "f35": "内盘",
+                "f62": "主力净流入-净额",
+                "f184": "主力净流入-净占比",
+                "f66": "超大单净流入-净额",
+                "f69": "超大单净流入-净占比",
+                "f72": "大单净流入-净额",
+                "f75": "大单净流入-净占比",
+                "f78": "中单净流入-净额",
+                "f81": "中单净流入-净占比",
+                "f84": "小单净流入-净额",
+                "f87": "小单净流入-净占比",
+                "f38": "最新份额",
+                "f21": "流通市值",
+                "f20": "总市值",
+                "f402": "基金折价率",
+                "f441": "IOPV实时估值",
+                "f297": "数据日期",
+                "f124": "更新时间",
+            },
+            inplace=True,
+        )
+        temp_df = temp_df[
+            [
+                "代码",
+                "名称",
+                "最新价",
+                "IOPV实时估值",
+                "基金折价率",
+                "涨跌额",
+                "涨跌幅",
+                "成交量",
+                "成交额",
+                "开盘价",
+                "最高价",
+                "最低价",
+                "昨收",
+                "振幅",
+                "换手率",
+                "量比",
+                "委比",
+                "外盘",
+                "内盘",
+                "主力净流入-净额",
+                "主力净流入-净占比",
+                "超大单净流入-净额",
+                "超大单净流入-净占比",
+                "大单净流入-净额",
+                "大单净流入-净占比",
+                "中单净流入-净额",
+                "中单净流入-净占比",
+                "小单净流入-净额",
+                "小单净流入-净占比",
+                "现手",
+                "买一",
+                "卖一",
+                "最新份额",
+                "流通市值",
+                "总市值",
+                "数据日期",
+                "更新时间",
+            ]
+        ].copy()
+        temp_df["最新价"] = pd.to_numeric(temp_df["最新价"], errors="coerce")
+        temp_df["涨跌额"] = pd.to_numeric(temp_df["涨跌额"], errors="coerce")
+        temp_df["涨跌幅"] = pd.to_numeric(temp_df["涨跌幅"], errors="coerce")
+        temp_df["成交量"] = pd.to_numeric(temp_df["成交量"], errors="coerce")
+        temp_df["成交额"] = pd.to_numeric(temp_df["成交额"], errors="coerce")
+        temp_df["开盘价"] = pd.to_numeric(temp_df["开盘价"], errors="coerce")
+        temp_df["最高价"] = pd.to_numeric(temp_df["最高价"], errors="coerce")
+        temp_df["最低价"] = pd.to_numeric(temp_df["最低价"], errors="coerce")
+        temp_df["昨收"] = pd.to_numeric(temp_df["昨收"], errors="coerce")
+        temp_df["换手率"] = pd.to_numeric(temp_df["换手率"], errors="coerce")
+        temp_df["量比"] = pd.to_numeric(temp_df["量比"], errors="coerce")
+        temp_df["委比"] = pd.to_numeric(temp_df["委比"], errors="coerce")
+        temp_df["外盘"] = pd.to_numeric(temp_df["外盘"], errors="coerce")
+        temp_df["内盘"] = pd.to_numeric(temp_df["内盘"], errors="coerce")
+        temp_df["流通市值"] = pd.to_numeric(temp_df["流通市值"], errors="coerce")
+        temp_df["总市值"] = pd.to_numeric(temp_df["总市值"], errors="coerce")
+        temp_df["振幅"] = pd.to_numeric(temp_df["振幅"], errors="coerce")
+        temp_df["现手"] = pd.to_numeric(temp_df["现手"], errors="coerce")
+        temp_df["买一"] = pd.to_numeric(temp_df["买一"], errors="coerce")
+        temp_df["卖一"] = pd.to_numeric(temp_df["卖一"], errors="coerce")
+        temp_df["最新份额"] = pd.to_numeric(temp_df["最新份额"], errors="coerce")
+        temp_df["IOPV实时估值"] = pd.to_numeric(temp_df["IOPV实时估值"], errors="coerce")
+        temp_df["基金折价率"] = pd.to_numeric(temp_df["基金折价率"], errors="coerce")
+        temp_df["主力净流入-净额"] = pd.to_numeric(
+            temp_df["主力净流入-净额"], errors="coerce"
+        )
+        temp_df["主力净流入-净占比"] = pd.to_numeric(
+            temp_df["主力净流入-净占比"], errors="coerce"
+        )
+        temp_df["超大单净流入-净额"] = pd.to_numeric(
+            temp_df["超大单净流入-净额"], errors="coerce"
+        )
+        temp_df["超大单净流入-净占比"] = pd.to_numeric(
+            temp_df["超大单净流入-净占比"], errors="coerce"
+        )
+        temp_df["大单净流入-净额"] = pd.to_numeric(
+            temp_df["大单净流入-净额"], errors="coerce"
+        )
+        temp_df["大单净流入-净占比"] = pd.to_numeric(
+            temp_df["大单净流入-净占比"], errors="coerce"
+        )
+        temp_df["中单净流入-净额"] = pd.to_numeric(
+            temp_df["中单净流入-净额"], errors="coerce"
+        )
+        temp_df["中单净流入-净占比"] = pd.to_numeric(
+            temp_df["中单净流入-净占比"], errors="coerce"
+        )
+        temp_df["小单净流入-净额"] = pd.to_numeric(
+            temp_df["小单净流入-净额"], errors="coerce"
+        )
+        temp_df["小单净流入-净占比"] = pd.to_numeric(
+            temp_df["小单净流入-净占比"], errors="coerce"
+        )
+        temp_df["数据日期"] = pd.to_datetime(
+            temp_df["数据日期"], format="%Y%m%d", errors="coerce"
+        )
+        temp_df["更新时间"] = (
+            pd.to_datetime(temp_df["更新时间"], unit="s", errors="coerce")
+            .dt.tz_localize("UTC")
+            .dt.tz_convert("Asia/Shanghai")
+        )
+    return temp_df
+
+
+# 示例同步调用封装（如需在同步代码中使用）
+def fund_etf_spot_em() -> pd.DataFrame:
+    return asyncio.run(fund_etf_spot_em_async())
+
+
+if __name__ == "__main__":
+    df = fund_etf_spot_em()
+    print(df.head())

state/2025-05-30.yaml

@@ -0,0 +1,46 @@
+main_state:
+  email_events: {}
+  turtles:
+  - BreakOutLog:
+    - breakout_price: 1.491
+      data: '2025-05-31'
+      lose_price: 1.459
+      valid_or_not: valid
+      win_or_lose: null
+    BuyStates:
+    - add_price: 1.5
+      atr: 0.016
+      available_cash: 76724.9
+      buy_price: 1.492
+      is_gap_up: false
+      shares: 15600.0
+      stop_price: 1.46
+      trigger_time: 1
+    Capital: 100000
+    TradeCode: '513870'
+    TrigerTime: 1
+    cash: 200000
+    riskcoe: 0.0025
+    tradeslog:
+    - Net_return: 0
+      Net_value: 23275.2
+      all_cost: 23275.100000000002
+      all_shares: 15600.0
+      atr: 0.016
+      available_cash: 76724.9
+      buy_price: 1.492
+      cost: 23275.100000000002
+      data: '2025-05-31'
+      shares: 15600.0
+      type: "\u4E70\u5165"
+    type: etf
+  - BreakOutLog: []
+    BuyStates: []
+    Capital: 100000
+    TradeCode: '600900'
+    TrigerTime: 0
+    cash: 200000
+    riskcoe: 0.0025
+    tradeslog: []
+    type: stock
+  user_email: guoyize2209@163.com

stock_em.py

@@ -0,0 +1,165 @@
+import asyncio
+from typing import Dict, List
+
+import aiohttp
+import pandas as pd
+
+
+# 增加了 User-Agent 头
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/122.0.0.0 Safari/537.36"
+    ),
+    "Referer": "https://quote.eastmoney.com/center/gridlist.html#hs_a_board",  # 合理来源页
+    "Accept": "application/json, text/plain, */*",
+    "Accept-Language": "zh-CN,zh;q=0.9",
+    "Connection": "keep-alive",
+}
+
+
+async def fetch_single_page(
+    session: aiohttp.ClientSession, url: str, params: Dict
+) -> Dict:
+    """异步获取单页数据"""
+    async with session.get(url, params=params, headers=HEADERS, ssl=False) as response:
+        return await response.json()
+
+
+async def fetch_all_pages_async(url: str, base_params: Dict) -> List[Dict]:
+    """异步获取所有页面数据"""
+    first_page_params = base_params.copy()
+    first_page_params["pn"] = "1"
+
+    async with aiohttp.ClientSession() as session:
+        first_page_data = await fetch_single_page(session, url, first_page_params)
+
+        if first_page_data.get("rc") != 0 or not first_page_data.get("data"):
+            return [first_page_data]
+
+        total = first_page_data["data"]["total"]
+        page_size = int(base_params["pz"])
+        total_pages = (total + page_size - 1) // page_size
+        total_pages = min(total_pages, 100)
+
+        tasks = []
+        for page in range(1, total_pages + 1):
+            page_params = base_params.copy()
+            page_params["pn"] = str(page)
+            tasks.append(fetch_single_page(session, url, page_params))
+
+        results = await asyncio.gather(*tasks)
+        return results
+
+
+def process_data(page_results: List[Dict]) -> pd.DataFrame:
+    """处理数据为 DataFrame"""
+    all_data = []
+    page_number = 1
+    items_per_page = 100
+
+    for result in page_results:
+        if result.get("rc") == 0 and result.get("data") and result["data"].get("diff"):
+            page_data = result["data"]["diff"]
+            for item in page_data:
+                item["page_number"] = page_number
+                item["page_index"] = page_data.index(item)
+            all_data.extend(page_data)
+            page_number += 1
+
+    if not all_data:
+        return pd.DataFrame()
+
+    df = pd.DataFrame(all_data)
+    df["序号"] = df.apply(
+        lambda row: (row["page_number"] - 1) * items_per_page + row["page_index"] + 1,
+        axis=1,
+    )
+    df.drop(columns=["page_number", "page_index"], inplace=True, errors="ignore")
+
+    column_map = {
+        "f1": "原序号",
+        "f2": "最新价",
+        "f3": "涨跌幅",
+        "f4": "涨跌额",
+        "f5": "成交量",
+        "f6": "成交额",
+        "f7": "振幅",
+        "f8": "换手率",
+        "f9": "市盈率-动态",
+        "f10": "量比",
+        "f11": "5分钟涨跌",
+        "f12": "代码",
+        "f13": "_",
+        "f14": "名称",
+        "f15": "最高",
+        "f16": "最低",
+        "f17": "今开",
+        "f18": "昨收",
+        "f20": "总市值",
+        "f21": "流通市值",
+        "f22": "涨速",
+        "f23": "市净率",
+        "f24": "60日涨跌幅",
+        "f25": "年初至今涨跌幅",
+    }
+
+    df.rename(columns=column_map, inplace=True)
+
+    desired_columns = [
+        "序号", "代码", "名称", "最新价", "涨跌幅", "涨跌额", "成交量", "成交额", "振幅",
+        "最高", "最低", "今开", "昨收", "量比", "换手率", "市盈率-动态", "市净率",
+        "总市值", "流通市值", "涨速", "5分钟涨跌", "60日涨跌幅", "年初至今涨跌幅"
+    ]
+    available_columns = [col for col in desired_columns if col in df.columns]
+    df = df[available_columns]
+
+    for col in [
+        "最新价", "涨跌幅", "涨跌额", "成交量", "成交额", "振幅", "最高", "最低", "今开",
+        "昨收", "量比", "换手率", "市盈率-动态", "市净率", "总市值", "流通市值", "涨速",
+        "5分钟涨跌", "60日涨跌幅", "年初至今涨跌幅"
+    ]:
+        if col in df.columns:
+            df[col] = pd.to_numeric(df[col], errors="coerce")
+
+    df.sort_values(by="涨跌幅", ascending=False, inplace=True)
+    df.reset_index(drop=True, inplace=True)
+    df["序号"] = df.index + 1
+    return df
+
+
+async def stock_zh_a_spot_em_async() -> pd.DataFrame:
+    url = "https://82.push2.eastmoney.com/api/qt/clist/get"
+    params = {
+        "pn": "1",
+        "pz": "100",
+        "po": "1",
+        "np": "1",
+        "ut": "bd1d9ddb04089700cf9c27f6f7426281",
+        "fltt": "2",
+        "invt": "2",
+        "fid": "f12",
+        "fs": "m:0 t:6,m:0 t:80,m:1 t:2,m:1 t:23,m:0 t:81 s:2048",
+        "fields": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f12,f13,f14,f15,f16,f17,f18,"
+                  "f20,f21,f23,f24,f25,f22,f11,f62,f128,f136,f115,f152",
+    }
+    results = await fetch_all_pages_async(url, params)
+    return process_data(results)
+
+
+def stock_zh_a_spot_em() -> pd.DataFrame:    
+    """
+    东方财富网-沪深京 A 股-实时行情 (同步接口)
+    https://quote.eastmoney.com/center/gridlist.html#hs_a_board
+    :return: 实时行情
+    :rtype: pandas.DataFrame
+    """
+    import nest_asyncio
+    nest_asyncio.apply()
+    return asyncio.run(stock_zh_a_spot_em_async())
+
+
+if __name__ == "__main__":
+    df = stock_zh_a_spot_em()
+    print(df)