"feat:integrate-baostock-and-optimize-trading-day-logic"

app/baostock_adapter.py

@@ -0,0 +1,219 @@
+"""
+BaoStock适配器 - 替代AkShare的部分接口
+"""
+
+import logging
+from typing import Optional, List, Dict, Any
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+# 尝试导入baostock
+try:
+    import baostock as bs
+    BAOSTOCK_AVAILABLE = True
+except ImportError:
+    BAOSTOCK_AVAILABLE = False
+    logger.warning("BaoStock not installed, falling back to AkShare")
+
+
+class BaoStockAdapter:
+    """BaoStock适配器类"""
+    
+    def __init__(self):
+        self._logged_in = False
+    
+    def login(self) -> bool:
+        """登录BaoStock"""
+        if not BAOSTOCK_AVAILABLE:
+            return False
+        
+        try:
+            result = bs.login()
+            if result.error_code == '0':
+                self._logged_in = True
+                logger.info("BaoStock login success")
+                return True
+            else:
+                logger.error(f"BaoStock login failed: {result.error_msg}")
+                return False
+        except Exception as e:
+            logger.error(f"BaoStock login error: {e}")
+            return False
+    
+    def logout(self):
+        """登出BaoStock"""
+        if self._logged_in and BAOSTOCK_AVAILABLE:
+            bs.logout()
+            self._logged_in = False
+    
+    def get_stock_list(self) -> Optional[pd.DataFrame]:
+        """
+        获取A股股票列表
+        替代: ak.stock_info_a_code_name()
+        
+        Returns:
+            DataFrame with columns: ['code', 'name']
+        """
+        if not BAOSTOCK_AVAILABLE:
+            return None
+        
+        if not self._logged_in:
+            if not self.login():
+                return None
+        
+        try:
+            # 获取所有A股 - 使用query_all_stock
+            # 注意：BaoStock的query_all_stock返回格式特殊
+            rs = bs.query_all_stock(day="2024-03-15")
+            if rs.error_code != '0':
+                logger.error(f"BaoStock query_all_stock failed: {rs.error_msg}")
+                return None
+            
+            data_list = []
+            codes_seen = set()
+            
+            while rs.next():
+                row = rs.get_row_data()
+                # row格式: [code, tradeStatus, code_name]
+                code = row[0]      # 如 sh.600000
+                # row[1] 是 tradeStatus (交易状态)
+                name = row[2]      # 股票名称
+                
+                # 过滤：只保留A股（code包含sh或sz）
+                # 过滤掉指数（以000/399开头的通常是指数）
+                code_short = code.replace('.', '')
+                is_index = (
+                    code_short.startswith('sh000') or 
+                    code_short.startswith('sz399') or
+                    code_short.startswith('sh88') or  # 科创板指数
+                    code_short.startswith('sh89')      # 其他指数
+                )
+                
+                if code.startswith(('sh.', 'sz.')) and not is_index and code not in codes_seen:
+                    codes_seen.add(code)
+                    data_list.append({
+                        'code': code_short,
+                        'name': name
+                    })
+            
+            df = pd.DataFrame(data_list)
+            logger.info(f"BaoStock get_stock_list: {len(df)} stocks")
+            return df
+            
+        except Exception as e:
+            logger.error(f"BaoStock get_stock_list error: {e}")
+            return None
+    
+    def get_stock_daily(self, code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]:
+        """
+        获取个股历史日K数据
+        替代: ak.stock_zh_a_hist()
+        
+        Args:
+            code: 股票代码 (如 'sh600000' 或 'sz000001')
+            start_date: 开始日期 (YYYYMMDD)
+            end_date: 结束日期 (YYYYMMDD)
+        
+        Returns:
+            DataFrame with columns compatible with existing code
+        """
+        if not BAOSTOCK_AVAILABLE:
+            return None
+        
+        if not self._logged_in:
+            if not self.login():
+                return None
+        
+        try:
+            # 转换代码格式: sh600000 -> sh.600000
+            if '.' not in code:
+                if code.startswith('sh') or code.startswith('sz'):
+                    code = code[:2] + '.' + code[2:]
+            
+            # BaoStock日期格式: YYYY-MM-DD
+            start = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:]}"
+            end = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:]}"
+            
+            # 查询数据
+            # fields参数说明: date,code,open,high,low,close,preclose,volume,amount,turn,tradestatus,pctChg,isST
+            rs = bs.query_history_k_data_plus(
+                code,
+                "date,code,open,high,low,close,preclose,volume,amount,turn,pctChg",
+                start_date=start,
+                end_date=end,
+                frequency="d",
+                adjustflag="3"  # 复权类型: 1后���权 2前复权 3不复权
+            )
+            
+            if rs.error_code != '0':
+                logger.warning(f"BaoStock query_history_k_data_plus failed for {code}: {rs.error_msg}")
+                return None
+            
+            data_list = []
+            while rs.next():
+                row = rs.get_row_data()
+                data_list.append({
+                    'trade_date': pd.to_datetime(row[0]),
+                    'code': code.replace('.', ''),
+                    'open': float(row[2]) if row[2] else None,
+                    'high': float(row[3]) if row[3] else None,
+                    'low': float(row[4]) if row[4] else None,
+                    'close': float(row[5]) if row[5] else None,
+                    'preclose': float(row[6]) if row[6] else None,
+                    'volume': float(row[7]) if row[7] else None,
+                    'amount': float(row[8]) if row[8] else None,
+                    'turnover': float(row[9]) if row[9] else None,
+                    'pct_change': float(row[10]) if row[10] else None,
+                })
+            
+            if not data_list:
+                return None
+            
+            df = pd.DataFrame(data_list)
+            logger.debug(f"BaoStock get_stock_daily for {code}: {len(df)} records")
+            return df
+            
+        except Exception as e:
+            logger.warning(f"BaoStock get_stock_daily error for {code}: {e}")
+            return None
+    
+    def get_index_daily(self, code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]:
+        """
+        获取指数历史日K数据
+        替代: ak.stock_zh_index_daily_em()
+        
+        Args:
+            code: 指数代码 (如 'sh000300' 沪深300)
+        """
+        # 指数和个股使用相同的接口
+        return self.get_stock_daily(code, start_date, end_date)
+
+
+# 全局适配器实例
+_baostock_adapter = None
+
+def get_baostock_adapter() -> BaoStockAdapter:
+    """获取BaoStock适配器实例（单例）"""
+    global _baostock_adapter
+    if _baostock_adapter is None:
+        _baostock_adapter = BaoStockAdapter()
+    return _baostock_adapter
+
+
+def get_stock_list_baostock() -> Optional[pd.DataFrame]:
+    """
+    获取股票列表（BaoStock版本）
+    如果BaoStock不可用，返回None，调用方应回退到AkShare
+    """
+    adapter = get_baostock_adapter()
+    return adapter.get_stock_list()
+
+
+def get_stock_daily_baostock(code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]:
+    """
+    获取个股日K数据（BaoStock版本）
+    如果BaoStock不可用，返回None
+    """
+    adapter = get_baostock_adapter()
+    return adapter.get_stock_daily(code, start_date, end_date)

requirements.txt

@@ -1,5 +1,6 @@
 # 同步 Space 专用依赖
 akshare>=1.12.0
+baostock>=0.8.8  # 替代AkShare部分接口
 pandas>=2.0.0
 duckdb>=0.9.0
 huggingface-hub>=0.20.0

sync_data.py

@@ -18,6 +18,9 @@ import pandas as pd
 import akshare as ak
 from huggingface_hub import hf_hub_download, upload_file, list_repo_files
 
+# BaoStock适配器（优先使用，更稳定）
+from app.baostock_adapter import get_stock_list_baostock, get_stock_daily_baostock, BAOSTOCK_AVAILABLE
+
 # 添加当前目录到路径
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
@@ -103,25 +106,41 @@ def get_stock_list() -> pd.DataFrame:
     logger.info("Fetching all-market target list...")
     all_lists = []
     
-    # A股列表获取（带重试和指数退避）
+    # A股列表获取（优先使用BaoStock，失败则回退到AkShare）
     max_retries = 5
     base_delay = 2.0
-    for attempt in range(max_retries):
+    
+    # 尝试使用BaoStock获取A股列表
+    if BAOSTOCK_AVAILABLE:
         try:
-            # 使用 stock_info_a_code_name() 替代 stock_zh_a_spot_em()，更稳定
-            df_a = ak.stock_info_a_code_name()
-            df_a.columns = ['code', 'name']
-            df_a['market'] = df_a['code'].apply(lambda x: '主板' if x.startswith(('60', '00')) else ('创业板' if x.startswith('30') else ('科创板' if x.startswith('68') else ('北交所' if x.startswith(('8', '4', '920')) else '其他'))))
-            all_lists.append(df_a)
-            logger.info(f"A-stock list fetched: {len(df_a)} stocks")
-            break  # 成功则退出重试循环
-        except Exception as e:
-            if attempt == max_retries - 1:
-                logger.error(f"Failed to fetch A-stock list after {max_retries} attempts: {e}")
+            df_a = get_stock_list_baostock()
+            if df_a is not None and len(df_a) > 0:
+                df_a['market'] = df_a['code'].apply(lambda x: '主板' if x.startswith(('60', '00')) else ('创业板' if x.startswith('30') else ('科创板' if x.startswith('68') else ('北交所' if x.startswith(('8', '4', '920')) else '其他'))))
+                all_lists.append(df_a)
+                logger.info(f"A-stock list fetched from BaoStock: {len(df_a)} stocks")
             else:
-                delay = base_delay * (2 ** attempt)
-                logger.warning(f"Attempt {attempt + 1} failed, retrying in {delay}s... Error: {e}")
-                time.sleep(delay)
+                raise Exception("BaoStock returned empty data")
+        except Exception as e:
+            logger.warning(f"BaoStock get_stock_list failed: {e}, falling back to AkShare")
+    
+    # 如果BaoStock失败或不可用，使用AkShare
+    if not all_lists:
+        for attempt in range(max_retries):
+            try:
+                # 使用 stock_info_a_code_name() 替代 stock_zh_a_spot_em()，更稳定
+                df_a = ak.stock_info_a_code_name()
+                df_a.columns = ['code', 'name']
+                df_a['market'] = df_a['code'].apply(lambda x: '主板' if x.startswith(('60', '00')) else ('创业板' if x.startswith('30') else ('科创板' if x.startswith('68') else ('北交所' if x.startswith(('8', '4', '920')) else '其他'))))
+                all_lists.append(df_a)
+                logger.info(f"A-stock list fetched from AkShare: {len(df_a)} stocks")
+                break  # 成功则退出重试循环
+            except Exception as e:
+                if attempt == max_retries - 1:
+                    logger.error(f"Failed to fetch A-stock list after {max_retries} attempts: {e}")
+                else:
+                    delay = base_delay * (2 ** attempt)
+                    logger.warning(f"Attempt {attempt + 1} failed, retrying in {delay}s... Error: {e}")
+                    time.sleep(delay)
 
     # ETF (增加容错)
     try:
@@ -292,7 +311,11 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
             fetch_start = start_date.replace('-', '')
             df = None
             if market == 'INDEX':
-                df = ak.stock_zh_index_daily_em(symbol=f"sh{code}" if code.startswith('000') else f"sz{code}")
+                # 指数：优先使用BaoStock，失败则回退AkShare
+                if BAOSTOCK_AVAILABLE:
+                    df = get_stock_daily_baostock(f"sh{code}" if code.startswith('000') else f"sz{code}", fetch_start, end_date)
+                if df is None:
+                    df = ak.stock_zh_index_daily_em(symbol=f"sh{code}" if code.startswith('000') else f"sz{code}")
             elif market == 'ETF':
                 df = ak.fund_etf_hist_em(symbol=code, period="daily", start_date=fetch_start, end_date=end_date, adjust="hfq")
             elif market == 'LOF':
@@ -306,7 +329,16 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
             elif market == 'REITs':
                 df = ak.reits_hist_em(symbol=code)
             else:
-                df = ak.stock_zh_a_hist(symbol=code, period="daily", start_date=fetch_start, end_date=end_date, adjust="hfq")
+                # A股个股：优先使用BaoStock，失败则回退AkShare
+                if BAOSTOCK_AVAILABLE:
+                    # BaoStock代码格式: sh600000, sz000001
+                    if code.startswith('6'):
+                        bs_code = f"sh{code}"
+                    else:
+                        bs_code = f"sz{code}"
+                    df = get_stock_daily_baostock(bs_code, fetch_start, end_date)
+                if df is None:
+                    df = ak.stock_zh_a_hist(symbol=code, period="daily", start_date=fetch_start, end_date=end_date, adjust="hfq")
             
             if df is not None and not df.empty:
                 # 标准化列名
@@ -340,16 +372,33 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
     return None
 
 def get_last_trading_day() -> str:
-    """获取最近一个交易日（优先使用交易日历）"""
+    """获取最近一个交易日（优先使用交易日历）
+    
+    注意：由于数据在收盘后获取，如果今天是交易日但当前时间 < 20:00，
+    则返回前一个交易日，避免获取不完整的当天数据。
+    """
+    now = get_beijing_time()
+    today = now.date()
+    current_hour = now.hour
+    
     try:
         # 获取交易日历（新浪接口，包含未来日期）
         df = ak.tool_trade_date_hist_sina()
         if df is not None and not df.empty:
-            # 转换为日期格式并过滤出 <= 今天的日期
+            # 转换为日期格式
             df['trade_date'] = pd.to_datetime(df['trade_date']).dt.date
-            today = get_beijing_time().date()
-            # 找到最后一个交易日
-            last_day = df[df['trade_date'] <= today]['trade_date'].iloc[-1]
+            
+            # 找到今天的交易日（如果今天是交易日）
+            today_is_trading_day = today in df['trade_date'].values
+            
+            # 如果今天是交易日但当前时间 < 20:00，则排除今天
+            if today_is_trading_day and current_hour < 20:
+                last_day = df[df['trade_date'] < today]['trade_date'].iloc[-1]
+                logger.info(f"Today is trading day but current time is {now.strftime('%H:%M')} (< 20:00), using previous trading day: {last_day}")
+            else:
+                # 否则使用 <= 今天的最后一个交易日
+                last_day = df[df['trade_date'] <= today]['trade_date'].iloc[-1]
+            
             return last_day.strftime('%Y-%m-%d')
     except Exception as e:
         logger.warning(f"Failed to get trading calendar: {e}")
@@ -360,11 +409,19 @@ def get_last_trading_day() -> str:
         if df is not None and not df.empty:
             date_col = 'date' if 'date' in df.columns else ('日期' if '日期' in df.columns else None)
             if date_col:
-                return pd.to_datetime(df[date_col].iloc[-1]).strftime('%Y-%m-%d')
+                last_date = pd.to_datetime(df[date_col].iloc[-1]).date()
+                # 同样检查时间条件
+                if last_date == today and current_hour < 20:
+                    # 返回前一天的数据
+                    return pd.to_datetime(df[date_col].iloc[-2]).strftime('%Y-%m-%d')
+                return last_date.strftime('%Y-%m-%d')
     except Exception: pass
 
     # 最终回退：按工作日估算
     d = get_beijing_time()
+    # 如果当前时间 < 20:00，从昨天开始查找
+    if current_hour < 20:
+        d -= timedelta(days=1)
     while d.weekday() >= 5:
         d -= timedelta(days=1)
     return d.strftime('%Y-%m-%d')