fix: Update sync script with latest AkShare interfaces and North Exchange support

backend/scripts/sync_data.py

@@ -30,7 +30,7 @@ logger = logging.getLogger(__name__)
 # 配置
 YEARS_OF_DATA = 10  # 初始获取近10年数据
 MAX_WORKERS = 5     # 并发线程数
-SYNC_LIMIT = -1     # 同步数量限制 (设置为 -1 表示全量同步)
+SYNC_LIMIT = 10     # 同步数量限制 (设置为 -1 表示全量同步)
 
 
 def get_stock_list() -> pd.DataFrame:
@@ -42,66 +42,62 @@ def get_stock_list() -> pd.DataFrame:
     
     all_lists = []
     
+    # 1. A 股列表
     try:
-        # 1. A 股列表
-        try:
-            df_a = ak.stock_zh_a_spot_em()[['代码', '名称']]
-            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 '其他')))
-            all_lists.append(df_a)
-        except Exception as e:
-            logger.warning(f"Failed to fetch A-share list from AkShare: {e}")
-
-        # 2. ETF 列表
-        try:
-            df_etf = ak.fund_etf_category_sina(symbol="ETF基金")[['代码', '名称']]
-            df_etf.columns = ['code', 'name']
-            df_etf['market'] = 'ETF'
-            all_lists.append(df_etf)
-        except Exception as e:
-            logger.warning(f"Failed to fetch ETF list from AkShare: {e}")
+        df_a = ak.stock_zh_a_spot_em()[['代码', '名称']]
+        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)
+    except Exception as e:
+        logger.warning(f"Failed to fetch A-share list: {e}")
 
-        # 3. LOF 列表
-        try:
-            df_lof = ak.fund_etf_category_sina(symbol="LOF基金")[['代码', '名称']]
-            df_lof.columns = ['code', 'name']
-            df_lof['market'] = 'LOF'
-            all_lists.append(df_lof)
-        except Exception as e:
-            logger.warning(f"Failed to fetch LOF list from AkShare: {e}")
+    # 2. ETF 列表
+    try:
+        df_etf = ak.fund_etf_category_sina(symbol="ETF基金")[['代码', '名称']]
+        df_etf.columns = ['code', 'name']
+        df_etf['market'] = 'ETF'
+        all_lists.append(df_etf)
+    except Exception as e:
+        logger.warning(f"Failed to fetch ETF list: {e}")
 
-        # 4. 可转债列表
-        try:
-            df_cb = ak.bond_zh_hs_cov_spot()[['代码', '名称']]
-            df_cb.columns = ['code', 'name']
-            df_cb['market'] = '可转债'
-            all_lists.append(df_cb)
-        except Exception as e:
-            logger.warning(f"Failed to fetch Convertible Bond list from AkShare: {e}")
+    # 3. LOF 列表
+    try:
+        df_lof = ak.fund_etf_category_sina(symbol="LOF基金")[['代码', '名称']]
+        df_lof.columns = ['code', 'name']
+        df_lof['market'] = 'LOF'
+        all_lists.append(df_lof)
+    except Exception as e:
+        logger.warning(f"Failed to fetch LOF list: {e}")
 
-        # 5. REITs 列表
-        try:
-            df_reits = ak.stock_reits_spot_em()[['代码', '名称']]
-            df_reits.columns = ['code', 'name']
-            df_reits['market'] = 'REITs'
-            all_lists.append(df_reits)
-        except Exception as e:
-            logger.warning(f"Failed to fetch REITs list from AkShare: {e}")
+    # 4. 可转债列表
+    try:
+        df_cb = ak.bond_zh_hs_cov_spot()
+        if '代码' in df_cb.columns:
+            df_cb = df_cb[['代码', '名称']]
+        elif 'symbol' in df_cb.columns:
+            df_cb = df_cb[['symbol', 'name']]
+        df_cb.columns = ['code', 'name']
+        df_cb['market'] = '可转债'
+        all_lists.append(df_cb)
+    except Exception as e:
+        logger.warning(f"Failed to fetch Convertible Bond list: {e}")
 
-    except Exception as global_e:
-        logger.error(f"Global error during list fetching: {global_e}")
+    # 5. REITs 列表
+    try:
+        df_reits = ak.reit_real_time_em()[['代码', '名称']]
+        df_reits.columns = ['code', 'name']
+        df_reits['market'] = 'REITs'
+        all_lists.append(df_reits)
+    except Exception as e:
+        logger.warning(f"Failed to fetch REITs list: {e}")
 
     if not all_lists:
-        logger.warning("AkShare list fetching failed completely. Attempting to recover from database...")
+        logger.warning("AkShare list fetching failed. Attempting to recover from database...")
         try:
             db = get_db()
             df_db = db.conn.execute("SELECT code, name, market FROM stock_list").df()
-            if not df_db.empty:
-                logger.info(f"Successfully recovered {len(df_db)} targets from database")
-                return df_db
-        except Exception as db_e:
-            logger.error(f"Failed to recover from database: {db_e}")
-        
+            if not df_db.empty: return df_db
+        except: pass
         raise Exception("Failed to fetch target list from both AkShare and Database")
         
     df = pd.concat(all_lists).drop_duplicates(subset=['code'])
@@ -145,7 +141,6 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
                     df = df.rename(columns={'日期': 'trade_date', '开盘': 'open', '最高': 'high', '最低': 'low', '收盘': 'close', '成交量': 'volume', '成交额': 'amount', '涨跌幅': 'pct_chg', '换手率': 'turnover_rate'})
                     df['trade_date'] = pd.to_datetime(df['trade_date'])
             else:
-                # 股票接口 (后复权)
                 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:
                     df = df.rename(columns={
@@ -160,8 +155,10 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
         # 2. 尝试 Yahoo Finance (作为兜底)
         if df is None or df.empty:
             # 转换代码格式
-            if code.startswith('6') or code.startswith('9') or code.startswith('5') or code.startswith('11'):
+            if code.startswith(('6', '9', '5', '11')):
                 yf_code = f"{code}.SS"
+            elif code.startswith(('8', '4', '920')):
+                yf_code = f"{code}.BJ" # 北交所
             else:
                 yf_code = f"{code}.SZ"
                 
@@ -182,7 +179,6 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
         if df is None or df.empty: return None
         
         df['code'] = code
-        # 确保列顺序一致
         cols = ['code', 'trade_date', 'open', 'high', 'low', 'close', 'volume', 'amount', 'pct_chg', 'turnover_rate']
         return df[[c for c in cols if c in df.columns]]
         
@@ -210,7 +206,6 @@ def sync_stock_daily(targets: List[Dict[str, str]]) -> int:
     db = get_db()
     success_count = 0
     
-    # 获取数据库中已有的最新日期映射
     logger.info("Checking existing data for incremental sync...")
     try:
         existing_latest = db.conn.execute("SELECT code, MAX(trade_date) FROM stock_daily GROUP BY code").fetchall()
@@ -248,7 +243,6 @@ def sync_stock_daily(targets: List[Dict[str, str]]) -> int:
 
     if all_data:
         combined_df = pd.concat(all_data, ignore_index=True)
-        # 确保列存在
         try:
             db.conn.execute("ALTER TABLE stock_daily ADD COLUMN turnover_rate DOUBLE")
         except: pass
@@ -274,30 +268,21 @@ def main():
         
         # 2. 同步日线数据
         limit = SYNC_LIMIT
-        
-        # 获取数据库中已有的标的代码 (排除指数)
         existing_codes = [row[0] for row in db.conn.execute("SELECT DISTINCT code FROM stock_daily WHERE code != '000300'").fetchall()]
         
         if limit > 0:
-            # 将总表转为字典方便查找
             all_targets_dict = {t['code']: t for t in target_list.to_dict('records')}
-            
             final_targets = []
-            # A. 先把数据库里已有的标的加进来 (保证持续更新)
             for code in existing_codes:
                 if code in all_targets_dict:
                     final_targets.append(all_targets_dict[code])
-                if len(final_targets) >= limit:
-                    break
+                if len(final_targets) >= limit: break
             
-            # B. 如果已有的不足 limit，从总表中取新的补齐
             if len(final_targets) < limit:
                 for _, row in target_list.iterrows():
                     if row['code'] not in existing_codes:
                         final_targets.append(row.to_dict())
-                    if len(final_targets) >= limit:
-                        break
-            
+                    if len(final_targets) >= limit: break
             targets = final_targets
             logger.info(f"Syncing {len(targets)} targets (Prioritized {len(existing_codes)} existing)")
         else:
@@ -316,12 +301,11 @@ def main():
         # 4. 上传
         db.upload_db()
         
-        # 5. 刷新后端缓存 (如果是在应用内运行)
+        # 5. 刷新后端缓存
         try:
             from app.core import clear_eligible_stocks_cache
             clear_eligible_stocks_cache()
-        except:
-            pass
+        except: pass
             
         logger.info("Data sync completed successfully!")