修复同步稳定性与远程视图冲突

backend/app/database.py

@@ -102,6 +102,8 @@ class DatabaseManager:
                         local_paths.append(f"'{path}'")
                     
                     files_sql = ", ".join(local_paths)
+                    conn.execute("DROP VIEW IF EXISTS stock_daily")
+                    conn.execute("DROP TABLE IF EXISTS stock_daily")
                     conn.execute(f"CREATE OR REPLACE VIEW stock_daily AS SELECT * FROM read_parquet([{files_sql}])")
                     logger.info(f"Remote stock_daily view created with {len(parquet_files)} partitions")
                 else:

backend/scripts/sync_data.py

@@ -30,7 +30,7 @@ logger = logging.getLogger(__name__)
 
 # 配置
 YEARS_OF_DATA = 10
-MAX_WORKERS = 10
+MAX_WORKERS = 5  # 降低并发数，减少超时
 SYNC_LIMIT = -1
 USE_YAHOO_FALLBACK = 0
 
@@ -89,7 +89,7 @@ def get_stock_list() -> pd.DataFrame:
 
 def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.DataFrame]:
     """抓取单只标的数据"""
-    max_retries = 2
+    max_retries = 3  # 增加重试次数
     for attempt in range(max_retries):
         try:
             end_date = datetime.now().strftime('%Y%m%d')
@@ -102,7 +102,12 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
             elif market == 'LOF':
                 df = ak.fund_lof_hist_em(symbol=code, period="daily", start_date=fetch_start, end_date=end_date, adjust="hfq")
             elif market == '可转债':
-                df = ak.bond_zh_hs_cov_daily(symbol=code)
+                # 可转债接口通常使用 6 位数字代码，兼容 sh110xxx / sz12xxxx / bj81xxxx
+                cov_symbol = code[-6:] if len(code) > 6 else code
+                try:
+                    df = ak.bond_zh_hs_cov_daily(symbol=cov_symbol)
+                except Exception:
+                    df = ak.bond_zh_hs_cov_daily(symbol=code)
             elif market == 'REITs':
                 # REITs 接口仅支持 symbol 参数，返回全量历史后再按日期过滤
                 df = ak.reits_hist_em(symbol=code)
@@ -112,17 +117,22 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
             if df is not None and not df.empty:
                 # 标准化列名
                 rename_map = {
-                    '日期': 'trade_date', 'date': 'trade_date',
-                    '开盘': 'open', '今开': 'open',
-                    '最高': 'high',
-                    '最低': 'low',
-                    '收盘': 'close', '最新价': 'close',
-                    '成交量': 'volume',
-                    '成交额': 'amount',
+                    '日期': 'trade_date', 'date': 'trade_date', 'Date': 'trade_date',
+                    '开盘': 'open', '今开': 'open', 'Open': 'open',
+                    '最高': 'high', 'High': 'high',
+                    '最低': 'low', 'Low': 'low',
+                    '收盘': 'close', '最新价': 'close', 'Close': 'close',
+                    '成交量': 'volume', 'Volume': 'volume',
+                    '成交额': 'amount', 'Amount': 'amount',
                     '涨跌幅': 'pct_chg',
                     '换手率': 'turnover_rate', '换手': 'turnover_rate'
                 }
                 df = df.rename(columns=rename_map)
+                
+                # 如果还是没有 trade_date，尝试将索引转为列
+                if 'trade_date' not in df.columns:
+                    df = df.reset_index().rename(columns={'index': 'trade_date', 'date': 'trade_date'})
+                
                 df['trade_date'] = pd.to_datetime(df['trade_date'])
                 df = df[df['trade_date'] >= pd.to_datetime(start_date)]
                 if 'amount' not in df.columns: df['amount'] = 0
@@ -136,6 +146,62 @@ def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.Dat
             time.sleep(1)
     return None
 
+def get_last_trading_day() -> str:
+    """获取最近一个交易日（用于增量截止日）"""
+    try:
+        df = ak.stock_zh_index_daily_em(symbol="sh000300")
+        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')
+    except Exception as e:
+        logger.warning(f"Failed to get last trading day from index data: {e}")
+
+    # 回退：按工作日估算
+    d = datetime.now()
+    while d.weekday() >= 5:  # 5=周六, 6=周日
+        d -= timedelta(days=1)
+    return d.strftime('%Y-%m-%d')
+
+
+def get_index_daily(code: str) -> Optional[pd.DataFrame]:
+    """抓取指数日线（默认用于沪深300）"""
+    try:
+        symbol = f"sh{code}" if code.startswith('000') else f"sz{code}"
+        df = ak.stock_zh_index_daily_em(symbol=symbol)
+        if df is None or df.empty:
+            return None
+
+        rename_map = {
+            'date': 'trade_date', '日期': 'trade_date',
+            'open': 'open', '开盘': 'open',
+            'high': 'high', '最高': 'high',
+            'low': 'low', '最低': 'low',
+            'close': 'close', '收盘': 'close',
+            'volume': 'volume', '成交量': 'volume',
+            'amount': 'amount', '成交额': 'amount',
+            'pct_chg': 'pct_chg', '涨跌幅': 'pct_chg'
+        }
+        df = df.rename(columns=rename_map)
+        if 'trade_date' not in df.columns:
+            return None
+
+        df['trade_date'] = pd.to_datetime(df['trade_date'])
+        if 'amount' not in df.columns:
+            df['amount'] = 0
+        if 'pct_chg' not in df.columns:
+            df['pct_chg'] = df['close'].pct_change() * 100
+        if 'volume' not in df.columns:
+            df['volume'] = 0
+        df['turnover_rate'] = 0
+        df['code'] = code
+
+        return df[['code', 'trade_date', 'open', 'high', 'low', 'close', 'volume', 'amount', 'pct_chg', 'turnover_rate']]
+    except Exception as e:
+        logger.warning(f"Failed to fetch index {code}: {e}")
+        return None
+
+
 def sync_stock_daily(targets: List[Dict[str, str]], last_trade_day: str) -> int:
     """增量同步逻辑 - 兼容云端视图"""
     db = get_db()
@@ -216,12 +282,10 @@ def main():
     target_list.to_parquet(list_parquet)
 
     # 2. 行情同步
-    from scripts.sync_data import get_last_trading_day
     last_day = get_last_trading_day()
     sync_stock_daily(target_list.to_dict('records'), last_day)
     
     # 3. 指数同步
-    from scripts.sync_data import get_index_daily
     idx_df = get_index_daily('000300')
     if idx_df is not None:
         idx_path = Path("backend/data/parquet/index_000300.parquet")