sync_data.py

"""
数据同步脚本 - Sync Space 专用版
从 AkShare 抓取数据并同步到 Hugging Face Dataset
"""

import os
import sys
import logging
import time
import threading
import gc
from datetime import datetime, timedelta
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Optional, Dict, Any
from pathlib import Path

import pandas as pd
import akshare as ak
from huggingface_hub import hf_hub_download, upload_file, list_repo_files

# Tushare 适配器（优先使用）
from app.tushare_adapter import (
    get_stock_list_tushare,
    get_stock_daily_tushare,
    get_dividend_tushare,
    TUSHARE_AVAILABLE
)

# 混合数据适配器（yfinance + efinance，作为回退）
from app.hybrid_adapter import (
    get_stock_daily_yfinance, 
    get_index_daily_yfinance,
    get_fund_flow_efinance,
    YFINANCE_AVAILABLE
)

# 添加当前目录到路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from app.database import get_db
from app.database_user import get_beijing_time
from app.sync_status import get_sync_status
from app.stock_list_cache import get_cached_stock_list, save_stock_list_cache

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 配置
YEARS_OF_DATA = 10

def _safe_int_env(var_name: str, default: int) -> int:
    """安全地读取环境变量并转换为整数"""
    try:
        value = os.getenv(var_name)
        if value is None:
            return default
        return int(value)
    except (ValueError, TypeError):
        logger.warning(f"Invalid value for {var_name}, using default: {default}")
        return default

# 动态线程数配置（延迟计算，避免导入时触发 multiprocessing）
def get_thread_config():
    """获取线程池配置（延迟计算）"""
    import multiprocessing
    cpu_count = multiprocessing.cpu_count()
    
    # 分层并发策略（降低并发，避免触发服务端限流）
    config = {
        'daily': _safe_int_env("MAX_WORKERS_DAILY", min(4, cpu_count)),
        'fund': _safe_int_env("MAX_WORKERS_FUND", min(3, cpu_count)),
        'valuation': _safe_int_env("MAX_WORKERS_VALUATION", min(3, cpu_count)),
        'margin': _safe_int_env("MAX_WORKERS_MARGIN", min(3, cpu_count)),
        'financial': _safe_int_env("MAX_WORKERS_FINANCIAL", min(2, cpu_count)),
        'dividend': _safe_int_env("MAX_WORKERS_DIVIDEND", min(2, cpu_count)),
    }
    
    # 向后兼容
    legacy = _safe_int_env("MAX_WORKERS", 0)
    if legacy > 0:
        config = {k: legacy for k in config}
    
    return cpu_count, config

# 延迟初始化线程配置（在 main 中调用）
_CPU_COUNT = None
_THREAD_CONFIG = None
_thread_config_lock = threading.Lock()

def get_thread_config_safe():
    """安全获取线程配置（自动初始化，线程安全）"""
    global _CPU_COUNT, _THREAD_CONFIG
    if _THREAD_CONFIG is None:
        with _thread_config_lock:
            # 双重检查锁定模式
            if _THREAD_CONFIG is None:
                _CPU_COUNT, _THREAD_CONFIG = get_thread_config()
                logger.info(f"Thread pool config: CPU={_CPU_COUNT}, "
                            f"Daily={_THREAD_CONFIG['daily']}, Fund={_THREAD_CONFIG['fund']}, "
                            f"Valuation={_THREAD_CONFIG['valuation']}, Margin={_THREAD_CONFIG['margin']}, "
                            f"Financial={_THREAD_CONFIG['financial']}, Dividend={_THREAD_CONFIG['dividend']}")
    return _THREAD_CONFIG

def init_thread_config():
    """初始化线程配置（在 main 中调用）"""
    get_thread_config_safe()

def get_stock_list() -> pd.DataFrame:
    """获取全市场标的列表（带缓存机制）"""
    # 1. 尝试使用缓存
    cached_df = get_cached_stock_list()
    if cached_df is not None:
        return cached_df
    
    # 2. 缓存无效，重新获取
    logger.info("Fetching all-market target list...")
    all_lists = []
    
    # A股列表获取（优先使用Tushare，失败则回退到AkShare）
    max_retries = 5
    base_delay = 2.0
    
    # 尝试使用Tushare获取A股列表（更稳定，字段完整）
    if TUSHARE_AVAILABLE:
        try:
            df_a = get_stock_list_tushare()
            if df_a is not None and len(df_a) > 0:
                all_lists.append(df_a)
                logger.info(f"A-stock list fetched from Tushare: {len(df_a)} stocks")
            else:
                raise Exception("Tushare returned empty data")
        except Exception as e:
            logger.warning(f"Tushare get_stock_list failed: {e}, falling back to AkShare")
    
    # 如果Tushare失败或不可用，使用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:
        df_etf = ak.fund_etf_spot_em()
        # 检查实际列名
        code_cols = ['代码', 'code', 'fund_code', 'ETF代码']
        name_cols = ['名称', 'name', 'fund_name', 'ETF名称']
        
        c_code = None
        c_name = None
        
        for col in code_cols:
            if col in df_etf.columns:
                c_code = col
                break
        
        for col in name_cols:
            if col in df_etf.columns:
                c_name = col
                break
        
        if c_code and c_name:
            df_etf = df_etf[[c_code, c_name]]
            df_etf.columns = ['code', 'name']
            df_etf['market'] = 'ETF'
            all_lists.append(df_etf)
            logger.info(f"ETF list fetched: {len(df_etf)} funds")
        else:
            logger.warning(f"Could not find code/name columns in ETF data. Available columns: {df_etf.columns.tolist()}")
    except Exception as e:
        logger.warning(f"ETF list fetch failed: {e}")

    # LOF
    try:
        df_lof = ak.fund_lof_spot_em()
        # 检查实际列名
        code_cols = ['代码', 'code', 'fund_code', 'LOF代码']
        name_cols = ['名称', 'name', 'fund_name', 'LOF名称']
        
        c_code = None
        c_name = None
        
        for col in code_cols:
            if col in df_lof.columns:
                c_code = col
                break
        
        for col in name_cols:
            if col in df_lof.columns:
                c_name = col
                break
        
        if c_code and c_name:
            df_lof = df_lof[[c_code, c_name]]
            df_lof.columns = ['code', 'name']
            df_lof['market'] = 'LOF'
            all_lists.append(df_lof)
            logger.info(f"LOF list fetched: {len(df_lof)} funds")
        else:
            logger.warning(f"Could not find code/name columns in LOF data. Available columns: {df_lof.columns.tolist()}")
    except Exception as e:
        logger.warning(f"LOF list fetch failed: {e}")

    # REITs
    try:
        df_reits = ak.reits_realtime_em()
        # 检查实际列名
        code_cols = ['代码', 'code', 'REITs代码']
        name_cols = ['名称', 'name', 'REITs名称']
        
        c_code = None
        c_name = None
        
        for col in code_cols:
            if col in df_reits.columns:
                c_code = col
                break
        
        for col in name_cols:
            if col in df_reits.columns:
                c_name = col
                break
        
        if c_code and c_name:
            df_reits = df_reits[[c_code, c_name]]
            df_reits.columns = ['code', 'name']
            df_reits['market'] = 'REITs'
            all_lists.append(df_reits)
            logger.info(f"REITs list fetched: {len(df_reits)} products")
        else:
            logger.warning(f"Could not find code/name columns in REITs data. Available columns: {df_reits.columns.tolist()}")
    except Exception as e:
        logger.warning(f"REITs list fetch failed: {e}")
    
    # 可转债
    try:
        df_cb = ak.bond_zh_hs_cov_spot()
        # 检查实际列名
        logger.info(f"Convertible bond columns: {df_cb.columns.tolist()}")
        
        # 尝试找到正确的代码列名
        code_cols = ['代码', 'symbol', 'bond_code', '代码', '转债代码', '代码']
        name_cols = ['名称', 'name', 'bond_name', '转债名称', '名称']
        
        c_code = None
        c_name = None
        
        for col in code_cols:
            if col in df_cb.columns:
                c_code = col
                break
        
        for col in name_cols:
            if col in df_cb.columns:
                c_name = col
                break
        
        if c_code and c_name:
            # 过滤掉未上市可转债（成交额=0 或 最新价=0）
            # 检查是否有成交额列
            amount_col = None
            for col in ['amount', '成交额', 'Amount']:
                if col in df_cb.columns:
                    amount_col = col
                    break
            
            if amount_col:
                before_count = len(df_cb)
                df_cb = df_cb[df_cb[amount_col] > 0]
                filtered_count = before_count - len(df_cb)
                if filtered_count > 0:
                    logger.info(f"Filtered {filtered_count} unlisted convertible bonds (amount=0)")
            
            df_cb = df_cb[[c_code, c_name]]
            df_cb.columns = ['code', 'name']
            df_cb['market'] = '可转债'
            all_lists.append(df_cb)
            logger.info(f"Convertible bond list fetched: {len(df_cb)} bonds")
        else:
            logger.warning(f"Could not find code/name columns in convertible bond data. Available columns: {df_cb.columns.tolist()}")
    except Exception as e:
        logger.warning(f"Convertible bond list fetch failed: {e}")

    if not all_lists:
        db = get_db()
        return db.conn.execute("SELECT code, name, market FROM stock_list").df()
        
    df = pd.concat(all_lists).drop_duplicates(subset=['code'])
    df['list_date'] = None
    
    # 3. 保存到缓存
    save_stock_list_cache(df)
    
    return df

def get_target_daily(code: str, start_date: str, market: str) -> Optional[pd.DataFrame]:
    """抓取单只标的数据"""
    max_retries = 5
    base_delay = 1.0  # 寂础延迟秒
    
    for attempt in range(max_retries):
        try:
            # 指数退避：每次重试增加延迟
            if attempt > 0:
                delay = base_delay * (2 ** attempt)  # 1s, 2s, 4s, 8s
                time.sleep(delay)
            end_date = get_beijing_time().strftime('%Y%m%d')
            fetch_start = start_date.replace('-', '')
            df = None
            if market == 'INDEX':
                # 指数：优先使用 yfinance，失败则回退 AkShare
                if YFINANCE_AVAILABLE:
                    df = get_index_daily_yfinance(code, fetch_start, end_date)
                    if df is not None:
                        logger.debug(f"Got index data from yfinance for {code}")
                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':
                df = ak.fund_lof_hist_em(symbol=code, period="daily", start_date=fetch_start, end_date=end_date, adjust="hfq")
            elif market == '可转债':
                # bond_zh_hs_cov_daily 需要带交易所前缀的代码（如 sh110048, sz123015）
                # 北交所可转债（bj开头）不支持
                if code.startswith('bj'):
                    return None  # 北交所可转债 API 不支持
                df = ak.bond_zh_hs_cov_daily(symbol=code)
            elif market == 'REITs':
                df = ak.reits_hist_em(symbol=code)
            else:
                # A股个股：优先使用 Tushare，失败则回退 yfinance/AkShare
                if TUSHARE_AVAILABLE:
                    df = get_stock_daily_tushare(code, fetch_start, end_date, adj='qfq')
                    if df is not None:
                        logger.debug(f"Got data from Tushare for {code}")
                if df is None and YFINANCE_AVAILABLE:
                    df = get_stock_daily_yfinance(code, fetch_start, end_date)
                    if df is not None:
                        logger.debug(f"Got data from yfinance for {code}")
                if df is None:
                    logger.debug(f"Tushare/yfinance failed, falling back to AkShare for {code}")
                    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:
                # 标准化列名
                rename_map = {
                    '日期': '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)
                
                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
                if 'pct_chg' not in df.columns: df['pct_chg'] = df['close'].pct_change() * 100
                if 'turnover_rate' not in df.columns: 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:
            if attempt == max_retries - 1:
                logger.warning(f"Failed to fetch {code} ({market}): {str(e)}")
            time.sleep(1)
    return None

def check_today_data_available() -> bool:
    """
    探测当日数据是否已更新可用
    通过尝试获取一只活跃股票（平安银行 000001）的当日数据来判断
    
    Returns:
        True: 当日数据已可用
        False: 当日数据尚未更新
    """
    try:
        today = get_beijing_time().strftime('%Y%m%d')
        
        # 优先使用 Tushare 探测
        if TUSHARE_AVAILABLE:
            try:
                import tushare as ts
                ts.set_token(TUSHARE_TOKENS[0])
                pro = ts.pro_api()
                df = pro.daily(ts_code='000001.SZ', trade_date=today)
                if df is not None and not df.empty:
                    logger.debug(f"Tushare check: today data available")
                    return True
            except Exception:
                pass
        
        # 回退到 AkShare 探测
        try:
            df = ak.stock_zh_a_spot_em()
            if df is not None and not df.empty:
                # 检查是否有今日数据
                logger.debug(f"AkShare check: today data available")
                return True
        except Exception:
            pass
            
    except Exception as e:
        logger.debug(f"Check today data failed: {e}")
    
    return False


def get_last_trading_day() -> str:
    """获取最近一个交易日（通过探测数据可用性）
    
    逻辑：
    1. 先尝试获取当日数据（探测 000001.SZ）
    2. 如果能获取到，说明当日数据已更新，返回今日
    3. 如果获取不到，说明当日数据未更新，返回上一个交易日
    """
    now = get_beijing_time()
    today = now.date()
    today_str = today.strftime('%Y-%m-%d')
    
    # 探测当日数据是否可用
    logger.info(f"Checking if today ({today_str}) data is available...")
    
    if check_today_data_available():
        logger.info(f"Today data is available, using today: {today_str}")
        return today_str
    else:
        # 获取上一个交易日
        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
                prev_trading_days = df[df['trade_date'] < today]['trade_date']
                if not prev_trading_days.empty:
                    last_day = prev_trading_days.iloc[-1]
                    logger.info(f"Today data not available yet, using previous trading day: {last_day}")
                    return last_day.strftime('%Y-%m-%d')
        except Exception as e:
            logger.warning(f"Failed to get previous trading day: {e}")
        
        # 回退：昨天
        yesterday = (today - timedelta(days=1)).strftime('%Y-%m-%d')
        logger.info(f"Using yesterday as fallback: {yesterday}")
        return yesterday

    # 备用：使用指数行情获取
    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:
                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')


def get_index_daily(code: str) -> Optional[pd.DataFrame]:
    """抓取指数日线"""
    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) -> Dict[str, Any]:
    """增量同步逻辑，返回详细结果（采用全局水位线机制）"""
    logger.info("Syncing daily data...")
    
    # 1. 扫描本地 parquet 文件获取全局最新日期（类似其他指标）
    parquet_dir = Path("/tmp/data/parquet")
    parquet_dir.mkdir(parents=True, exist_ok=True)
    
    global_latest_date = "2000-01-01"
    existing_codes = set()
    
    for f in parquet_dir.glob("*.parquet"):
        if f.name.startswith('index_'):  # 跳过指数文件
            continue
        try:
            df = pd.read_parquet(f)
            if not df.empty and 'trade_date' in df.columns:
                max_date = df['trade_date'].max()
                if isinstance(max_date, pd.Timestamp):
                    max_date = max_date.strftime('%Y-%m-%d')
                if max_date > global_latest_date:
                    global_latest_date = max_date
                existing_codes.update(df['code'].unique())
        except Exception:
            pass
    
    # 2. 如果本地没有数据，尝试从云端下载最近3个月作为基准
    if global_latest_date == "2000-01-01":
        repo_id = os.getenv("DATASET_REPO_ID")
        if repo_id:
            try:
                files = list_repo_files(repo_id=repo_id, repo_type="dataset")
                parquet_files = sorted([f for f in files if f.startswith("data/parquet/") and f.endswith(".parquet")])
                
                # 下载最近3个月的数据作为基准
                for pf in parquet_files[-3:]:
                    try:
                        local_file = hf_hub_download(repo_id=repo_id, filename=pf, repo_type="dataset")
                        df = pd.read_parquet(local_file)
                        if not df.empty and 'trade_date' in df.columns:
                            max_date = df['trade_date'].max()
                            if isinstance(max_date, pd.Timestamp):
                                max_date = max_date.strftime('%Y-%m-%d')
                            if max_date > global_latest_date:
                                global_latest_date = max_date
                            existing_codes.update(df['code'].unique())
                    except Exception:
                        pass
                logger.info(f"Downloaded daily data from cloud, latest date: {global_latest_date}")
            except Exception as e:
                logger.info(f"No existing daily data in cloud: {e}")
    
    # 3. 区分新股和存量股票
    new_codes = [t for t in targets if t['code'] not in existing_codes]
    
    # 4. 全局水位线拦截
    if global_latest_date >= last_trade_day and not new_codes:
        logger.info(f"Daily data is already up to date ({global_latest_date}) and no new stocks. Skip.")
        return {'count': 0, 'failed_codes': [], 'status': 'skipped', 'message': f'Already up to date ({global_latest_date})'}
    
    # 5. 确定同步策略
    if global_latest_date >= last_trade_day:
        logger.info(f"Global date is up to date, but found {len(new_codes)} new stocks. Syncing new stocks only.")
        sync_targets = new_codes
        # 新股只获取最近1年数据（而非10年）
        start_dt = (get_beijing_time() - timedelta(days=365)).strftime('%Y-%m-%d')
    else:
        logger.info(f"Syncing daily data from {global_latest_date} to {last_trade_day}...")
        sync_targets = targets
        start_dt = (pd.to_datetime(global_latest_date) + timedelta(days=1)).strftime('%Y-%m-%d')
    
    # 设置每只股票的start_dt
    pending = []
    for t in sync_targets:
        t['start_dt'] = start_dt
        pending.append(t)

    # 应用 SYNC_LIMIT 限制
    sync_limit = int(os.getenv("SYNC_LIMIT", -1))
    if sync_limit > 0 and len(pending) > sync_limit:
        logger.info(f"Limiting sync to first {sync_limit} targets (out of {len(pending)})")
        pending = pending[:sync_limit]

    if not pending:
        return {'count': 0, 'failed_codes': [], 'status': 'skipped', 'message': 'No pending targets'}
    
    logger.info(f"Syncing {len(pending)} targets...")

    all_new_data = []
    failed_codes = []
    success_codes = []
    
    with ThreadPoolExecutor(max_workers=get_thread_config_safe()['daily']) as executor:
        futures = {executor.submit(get_target_daily, t['code'], t['start_dt'], t['market']): t['code'] for t in pending}
        for i, future in enumerate(as_completed(futures), 1):
            code = futures[future]
            res = future.result()
            if res is not None:
                all_new_data.append(res)
                success_codes.append(code)
            else:
                failed_codes.append(code)
            if i % 500 == 0: logger.info(f"Progress: {i}/{len(pending)}")

    changed_files = []  # 记录变更的文件名
    
    if all_new_data:
        inc_df = pd.concat(all_new_data, ignore_index=True)
        # 方案3：及时释放内存
        del all_new_data
        total_records = len(inc_df)
        # 识别变动月份
        changed = inc_df.assign(yr=inc_df['trade_date'].dt.year, mo=inc_df['trade_date'].dt.month)[['yr', 'mo']].drop_duplicates().values
        
        for yr, mo in changed:
            yr, mo = int(yr), int(mo)
            filename = f"{yr}-{mo:02d}.parquet"
            local_path = Path(f"/tmp/data/parquet/{filename}")  # Sync Space 使用 /tmp
            local_path.parent.mkdir(parents=True, exist_ok=True)
            
            # 增量核心：先检查本地是否有，没有再从云端拉取
            old_df = None
            if local_path.exists():
                try:
                    old_df = pd.read_parquet(local_path)
                    logger.info(f"Using local cache for {filename}")
                except Exception: pass
            
            if old_df is None:
                repo_id = os.getenv("DATASET_REPO_ID")
                if repo_id:
                    try:
                        old_file = hf_hub_download(repo_id=repo_id, filename=f"data/parquet/{filename}", repo_type="dataset")
                        old_df = pd.read_parquet(old_file)
                        logger.info(f"Downloaded {filename} from cloud")
                    except Exception:
                        pass

            # 合并新数据
            month_inc = inc_df[(inc_df['trade_date'].dt.year == yr) & (inc_df['trade_date'].dt.month == mo)]
            if old_df is not None:
                final_month_df = pd.concat([old_df, month_inc]).drop_duplicates(subset=['code', 'trade_date'])
                # 方案3：释放旧数据内存
                del old_df, month_inc
            else:
                final_month_df = month_inc
                
            final_month_df.to_parquet(local_path)
            changed_files.append(filename)  # 记录变更的文件
            logger.info(f"Saved updated data for {filename}")
            # 方案3：释放最终数据内存
            del final_month_df
        
        # 方案3：循环结束后释放inc_df并触发GC
        del inc_df
        gc.collect()
    else:
        total_records = 0
            
    return {
        'count': len(success_codes),
        'failed_codes': failed_codes,
        'status': 'success' if not failed_codes else 'partial_fail',
        'record_count': total_records,
        'success_rate': len(success_codes) / len(pending) if pending else 0,
        'changed_files': changed_files  # 返回变更文件列表
    }


# ==================== 新增：资金流向数据同步 ====================

def get_stock_fund_flow(code: str, market: str) -> Optional[pd.DataFrame]:
    """获取单只股票资金流向数据（优先使用 efinance，失败则回退 AkShare）"""
    
    # 标准化列名映射
    standard_cols = ['code', 'trade_date', 'close', 'pct_chg',
                     'main_net_inflow', 'main_net_inflow_pct',
                     'huge_net_inflow', 'huge_net_inflow_pct',
                     'large_net_inflow', 'large_net_inflow_pct',
                     'medium_net_inflow', 'medium_net_inflow_pct',
                     'small_net_inflow', 'small_net_inflow_pct']
    
    # 1. 优先尝试 efinance
    try:
        df = get_fund_flow_efinance(code, '20000101', '20991231')
        if df is not None and not df.empty:
            # efinance 字段映射
            rename_map = {
                '日期': 'trade_date', '收盘价': 'close', '涨跌幅': 'pct_chg',
                '主力净流入': 'main_net_inflow',
                '主力净流入占比': 'main_net_inflow_pct',
                '超大单净流入': 'huge_net_inflow',
                '超大单流入净占比': 'huge_net_inflow_pct',
                '大单净流入': 'large_net_inflow',
                '大单流入净占比': 'large_net_inflow_pct',
                '中单净流入': 'medium_net_inflow',
                '中单流入净占比': 'medium_net_inflow_pct',
                '小单净流入': 'small_net_inflow',
                '小单流入净占比': 'small_net_inflow_pct',
            }
            df = df.rename(columns=rename_map)
            if 'trade_date' in df.columns:
                df['trade_date'] = pd.to_datetime(df['trade_date'])
            df['code'] = code
            
            result_cols = [c for c in standard_cols if c in df.columns]
            logger.debug(f"Got fund flow from efinance for {code}")
            return df[result_cols]
    except Exception as e:
        logger.debug(f"efinance fund flow failed for {code}: {e}")
    
    # 2. 回退到 AkShare
    max_retries = 3
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            # 确定 market 参数
            if market == '北交所' or code.startswith(('8', '4', '920')):
                mk = 'bj'
            elif code.startswith(('6', '9')):
                mk = 'sh'
            else:
                mk = 'sz'
            
            df = ak.stock_individual_fund_flow(stock=code, market=mk)
            if df is not None and not df.empty:
                # AkShare 字段映射
                rename_map = {
                    '日期': 'trade_date', '收盘价': 'close', '涨跌幅': 'pct_chg',
                    '主力净流入-净额': 'main_net_inflow',
                    '主力净流入-净占比': 'main_net_inflow_pct',
                    '超大单净流入-净额': 'huge_net_inflow',
                    '超大单净流入-净占比': 'huge_net_inflow_pct',
                    '大单净流入-净额': 'large_net_inflow',
                    '大单净流入-净占比': 'large_net_inflow_pct',
                    '中单净流入-净额': 'medium_net_inflow',
                    '中单净流入-净占比': 'medium_net_inflow_pct',
                    '小单净流入-净额': 'small_net_inflow',
                    '小单净流入-净占比': 'small_net_inflow_pct',
                }
                df = df.rename(columns=rename_map)
                df['trade_date'] = pd.to_datetime(df['trade_date'])
                df['code'] = code
                
                result_cols = [c for c in standard_cols if c in df.columns]
                logger.debug(f"Got fund flow from AkShare for {code}")
                return df[result_cols]
        except Exception as e:
            if attempt > 0:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
    
    return None


def sync_fund_flow(targets: List[Dict[str, str]], last_trade_day: str) -> Dict[str, Any]:
    """同步资金流向数据（按月分表版），返回详细结果"""
    logger.info("Syncing fund flow data...")
    
    # 1. 从所有本地 parquet 文件计算全局最新日期
    flow_dir = Path("/tmp/data/fund_flow")
    flow_dir.mkdir(parents=True, exist_ok=True)
    
    global_latest_date = "2000-01-01"
    existing_codes = set()
    
    # 扫描本地已有的月度文件
    for f in flow_dir.glob("*.parquet"):
        try:
            df = pd.read_parquet(f)
            if not df.empty:
                max_date = df['trade_date'].max().strftime('%Y-%m-%d')
                if max_date > global_latest_date:
                    global_latest_date = max_date
                existing_codes.update(df['code'].unique())
        except Exception:
            pass
    
    # 如果本地没有数据，尝试从云端下载
    if global_latest_date == "2000-01-01":
        repo_id = os.getenv("DATASET_REPO_ID")
        if repo_id:
            try:
                # 获取云端文件列表
                files = list_repo_files(repo_id=repo_id, repo_type="dataset")
                flow_files = sorted([f for f in files if f.startswith("data/fund_flow/") and f.endswith(".parquet")])

                for ff in flow_files[-3:]:  # 先下载最近3个月的数据作为基准
                    try:
                        local_file = hf_hub_download(repo_id=repo_id, filename=ff, repo_type="dataset")
                        df = pd.read_parquet(local_file)
                        if not df.empty:
                            # 提取月份
                            filename = Path(ff).stem  # 如 "2026-02"
                            local_path = flow_dir / f"{filename}.parquet"
                            df.to_parquet(local_path)
                            max_date = df['trade_date'].max().strftime('%Y-%m-%d')
                            if max_date > global_latest_date:
                                global_latest_date = max_date
                            existing_codes.update(df['code'].unique())
                    except Exception:
                        pass
                logger.info(f"Downloaded fund flow from cloud, latest date: {global_latest_date}")
            except Exception as e:
                logger.info(f"No existing fund flow data in cloud: {e}")

    # 2. 过滤目标（排除 ETF/LOF/REITs/可转债）
    stock_targets = [t for t in targets if t['market'] not in ['ETF', 'LOF', 'REITs', '可转债']]
    new_codes = [t for t in stock_targets if t['code'] not in existing_codes]
    
    # 3. 全局水位线拦截
    if global_latest_date >= last_trade_day and not new_codes:
        logger.info(f"Fund flow data is already up to date ({global_latest_date}) and no new stocks. Skip.")
        return {'count': 0, 'failed_codes': [], 'status': 'skipped', 'message': f'Already up to date ({global_latest_date})'}

    # 4. 增量获取
    if global_latest_date >= last_trade_day:
        logger.info(f"Global date is up to date, but found {len(new_codes)} new stocks. Syncing new stocks only.")
        sync_targets = new_codes
    else:
        logger.info(f"Syncing fund flow from {global_latest_date} to {last_trade_day}...")
        sync_targets = stock_targets

    all_data = []
    success_codes = []
    failed_codes = []
    
    with ThreadPoolExecutor(max_workers=get_thread_config_safe()['fund']) as executor:
        futures = {executor.submit(get_stock_fund_flow, t['code'], t['market']): t['code'] for t in sync_targets}
        for i, future in enumerate(as_completed(futures), 1):
            code = futures[future]
            res = future.result()
            if res is not None and not res.empty:
                if code in existing_codes:
                    res = res[res['trade_date'] > pd.to_datetime(global_latest_date)]
                if not res.empty:
                    all_data.append(res)
                    success_codes.append(code)
            else:
                failed_codes.append(code)
            if i % 500 == 0:
                logger.info(f"Fund flow progress: {i}/{len(sync_targets)}, success: {len(success_codes)}")
    
    total_records = 0
    changed_files = []  # 记录变更的文件名
    
    # 5. 按月分表保存
    if all_data:
        new_df = pd.concat(all_data, ignore_index=True)
        # 方案3：及时释放内存
        del all_data
        total_records = len(new_df)
        
        # 确定需要更新的月份
        if not new_df.empty:
            min_date = new_df['trade_date'].min()
            max_date = new_df['trade_date'].max()
            
            current = min_date.to_period('M')
            end_period = max_date.to_period('M')
            
            while current <= end_period:
                yr, mo = current.year, current.month
                month_data = new_df[(new_df['trade_date'].dt.year == yr) & (new_df['trade_date'].dt.month == mo)]
                
                if not month_data.empty:
                    filename = f"{yr}-{mo:02d}.parquet"
                    local_path = flow_dir / filename
                    
                    # 合并已有数据
                    old_month_df = None
                    if local_path.exists():
                        try:
                            old_month_df = pd.read_parquet(local_path)
                        except Exception:
                            pass
                    
                    if old_month_df is not None:
                        final_month_df = pd.concat([old_month_df, month_data]).drop_duplicates(subset=['code', 'trade_date'])
                        # 方案3：释放旧数据内存
                        del old_month_df, month_data
                    else:
                        final_month_df = month_data
                    
                    final_month_df.to_parquet(local_path)
                    changed_files.append(filename)  # 记录变更的文件
                    logger.info(f"Saved fund flow data for {filename}")
                    # 方案3：释放最终数据内存
                    del final_month_df
                
                current += 1
        
        logger.info(f"Fund flow updated: {len(new_df)} new records")
        # 方案3：释放new_df并触发GC
        del new_df
        gc.collect()
    
    return {
        'count': len(success_codes),
        'failed_codes': failed_codes,
        'status': 'success' if not failed_codes else 'partial_fail',
        'record_count': total_records,
        'success_rate': len(success_codes) / len(sync_targets) if sync_targets else 0,
        'changed_files': changed_files  # 返回变更文件列表
    }


# ==================== 新增：估值指标数据同步 ====================

def get_stock_valuation(code: str) -> Optional[pd.DataFrame]:
    """获取单只股票估值指标数据"""
    max_retries = 5
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            df = ak.stock_a_lg_indicator(symbol=code)
            if df is not None and not df.empty:
                # 标准化列名
                rename_map = {
                    '日期': 'trade_date',
                    '市盈率': 'pe_ttm',
                    '市盈率TTM': 'pe_ttm',
                    '静态市盈率': 'pe_static',
                    '市净率': 'pb',
                    '市销率': 'ps_ttm',
                    '股息率': 'dv_ratio',
                    '总市值': 'total_mv',
                    '流通市值': 'circ_mv',
                }
                df = df.rename(columns=rename_map)
                df['trade_date'] = pd.to_datetime(df['trade_date'])
                df['code'] = code
                
                cols = ['code', 'trade_date', 'pe_ttm', 'pe_static', 'pb', 
                        'ps_ttm', 'dv_ratio', 'total_mv', 'circ_mv']
                available_cols = [c for c in cols if c in df.columns]
                return df[available_cols]
        except Exception:
            if attempt > 0:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
    return None


def sync_valuation(targets: List[Dict[str, str]], last_trade_day: str) -> Dict[str, Any]:
    """同步估值指标数据（按月分表版）"""
    logger.info("Syncing valuation data...")
    
    # 1. 从所有本地 parquet 文件计算全局最新日期
    val_dir = Path("/tmp/data/valuation")
    val_dir.mkdir(parents=True, exist_ok=True)
    
    global_latest_date = "2000-01-01"
    existing_codes = set()
    
    # 扫描本地已有的月度文件
    for f in val_dir.glob("*.parquet"):
        try:
            df = pd.read_parquet(f)
            if not df.empty:
                max_date = df['trade_date'].max().strftime('%Y-%m-%d')
                if max_date > global_latest_date:
                    global_latest_date = max_date
                existing_codes.update(df['code'].unique())
        except Exception:
            pass
    
    # 如果本地没有数据，尝试从云端下载
    if global_latest_date == "2000-01-01":
        repo_id = os.getenv("DATASET_REPO_ID")
        if repo_id:
            try:
                files = list_repo_files(repo_id=repo_id, repo_type="dataset")
                val_files = sorted([f for f in files if f.startswith("data/valuation/") and f.endswith(".parquet")])

                for vf in val_files[-3:]:  # 先下载最近3个月
                    try:
                        local_file = hf_hub_download(repo_id=repo_id, filename=vf, repo_type="dataset")
                        df = pd.read_parquet(local_file)
                        if not df.empty:
                            filename = Path(vf).stem
                            local_path = val_dir / f"{filename}.parquet"
                            df.to_parquet(local_path)
                            max_date = df['trade_date'].max().strftime('%Y-%m-%d')
                            if max_date > global_latest_date:
                                global_latest_date = max_date
                            existing_codes.update(df['code'].unique())
                    except Exception:
                        pass
                logger.info(f"Downloaded valuation from cloud, latest date: {global_latest_date}")
            except Exception as e:
                logger.info(f"No existing valuation data in cloud: {e}")

    # 2. 过滤目标（排除 ETF/LOF/REITs/可转债）
    stock_targets = [t for t in targets if t['market'] not in ['ETF', 'LOF', 'REITs', '可转债']]
    new_codes = [t for t in stock_targets if t['code'] not in existing_codes]

    # 3. 全局水位线拦截
    if global_latest_date >= last_trade_day and not new_codes:
        logger.info(f"Valuation data is already up to date ({global_latest_date}) and no new stocks. Skip.")
        return {'count': 0, 'failed_codes': [], 'status': 'skipped', 'record_count': 0, 'changed_files': [], 'message': f'Already up to date ({global_latest_date})'}

    # 4. 增量获取
    if global_latest_date >= last_trade_day:
        logger.info(f"Global date is up to date, but found {len(new_codes)} new stocks. Syncing new stocks only.")
        sync_targets = new_codes
    else:
        logger.info(f"Syncing valuation from {global_latest_date} to {last_trade_day}...")
        sync_targets = stock_targets

    all_data = []
    success_count = 0
    
    with ThreadPoolExecutor(max_workers=get_thread_config_safe()['valuation']) as executor:
        futures = {executor.submit(get_stock_valuation, t['code']): t['code'] for t in sync_targets}
        for i, future in enumerate(as_completed(futures), 1):
            res = future.result()
            if res is not None and not res.empty:
                code = futures[future]
                if code in existing_codes:
                    res = res[res['trade_date'] > pd.to_datetime(global_latest_date)]
                if not res.empty:
                    all_data.append(res)
                    success_count += 1
            if i % 500 == 0:
                logger.info(f"Valuation progress: {i}/{len(sync_targets)}, success: {success_count}")
    
    total_records = 0
    changed_files = []  # 记录变更的文件名
    
    # 5. 按月分表保存
    if all_data:
        new_df = pd.concat(all_data, ignore_index=True)
        # 方案3：及时释放内存
        del all_data
        total_records = len(new_df)
        
        if not new_df.empty:
            min_date = new_df['trade_date'].min()
            max_date = new_df['trade_date'].max()
            
            current = min_date.to_period('M')
            end_period = max_date.to_period('M')
            
            while current <= end_period:
                yr, mo = current.year, current.month
                month_data = new_df[(new_df['trade_date'].dt.year == yr) & (new_df['trade_date'].dt.month == mo)]
                
                if not month_data.empty:
                    filename = f"{yr}-{mo:02d}.parquet"
                    local_path = val_dir / filename
                    
                    old_month_df = None
                    if local_path.exists():
                        try:
                            old_month_df = pd.read_parquet(local_path)
                        except Exception:
                            pass
                    
                    if old_month_df is not None:
                        final_month_df = pd.concat([old_month_df, month_data]).drop_duplicates(subset=['code', 'trade_date'])
                        # 方案3：释放旧数据内存
                        del old_month_df, month_data
                    else:
                        final_month_df = month_data
                    
                    final_month_df.to_parquet(local_path)
                    changed_files.append(filename)  # 记录变更的文件
                    logger.info(f"Saved valuation data for {filename}")
                    # 方案3：释放最终数据内存
                    del final_month_df
                
                current += 1
        
        logger.info(f"Valuation updated: {len(new_df)} new records")
        # 方案3：释放new_df并触发GC
        del new_df
        gc.collect()
    
    return {
        'count': success_count,
        'failed_codes': [],
        'status': 'success',
        'record_count': total_records,
        'changed_files': changed_files  # 返回变更文件列表
    }


# ==================== 新增：融资融券数据同步 ====================

def get_stock_margin(code: str) -> Optional[pd.DataFrame]:
    """获取单只股票融资融券数据"""
    max_retries = 5
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            # 尝试上交所
            if code.startswith('6'):
                df = ak.stock_margin_detail_sh(symbol=code)
            else:
                df = ak.stock_margin_detail_sz(symbol=code)
            
            if df is not None and not df.empty:
                # 标准化列名
                rename_map = {
                    '日期': 'trade_date',
                    '融资余额': 'rzye',
                    '融资买入额': 'rzmre',
                    '融资偿还额': 'rzche',
                    '融券余额': 'rqye',
                    '融券卖出量': 'rqmcl',
                    '融资融券余额': 'rzrqye',
                }
                df = df.rename(columns=rename_map)
                if 'trade_date' in df.columns:
                    df['trade_date'] = pd.to_datetime(df['trade_date'])
                df['code'] = code
                
                cols = ['code', 'trade_date', 'rzye', 'rzmre', 'rzche', 'rqye', 'rqmcl', 'rzrqye']
                available_cols = [c for c in cols if c in df.columns]
                return df[available_cols]
        except Exception as e:
            if attempt == max_retries - 1:
                pass
            if attempt > 0:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
    return None


def sync_margin(targets: List[Dict[str, str]], last_trade_day: str) -> Dict[str, Any]:
    """同步融资融券数据（按月分表版）"""
    logger.info("Syncing margin trading data...")
    
    # 1. 从所有本地 parquet 文件计算全局最新日期
    margin_dir = Path("/tmp/data/margin")
    margin_dir.mkdir(parents=True, exist_ok=True)
    
    global_latest_date = "2000-01-01"
    existing_codes = set()
    
    # 扫描本地已有的月度文件
    for f in margin_dir.glob("*.parquet"):
        try:
            df = pd.read_parquet(f)
            if not df.empty:
                max_date = df['trade_date'].max().strftime('%Y-%m-%d')
                if max_date > global_latest_date:
                    global_latest_date = max_date
                existing_codes.update(df['code'].unique())
        except Exception:
            pass
    
    # 如果本地没有数据，尝试从云端下载
    if global_latest_date == "2000-01-01":
        repo_id = os.getenv("DATASET_REPO_ID")
        if repo_id:
            try:
                files = list_repo_files(repo_id=repo_id, repo_type="dataset")
                margin_files = sorted([f for f in files if f.startswith("data/margin/") and f.endswith(".parquet")])

                for mf in margin_files[-3:]:  # 先下载最近3个月
                    try:
                        local_file = hf_hub_download(repo_id=repo_id, filename=mf, repo_type="dataset")
                        df = pd.read_parquet(local_file)
                        if not df.empty:
                            filename = Path(mf).stem
                            local_path = margin_dir / f"{filename}.parquet"
                            df.to_parquet(local_path)
                            max_date = df['trade_date'].max().strftime('%Y-%m-%d')
                            if max_date > global_latest_date:
                                global_latest_date = max_date
                            existing_codes.update(df['code'].unique())
                    except Exception:
                        pass
                logger.info(f"Downloaded margin from cloud, latest date: {global_latest_date}")
            except Exception as e:
                logger.info(f"No existing margin data in cloud: {e}")

    # 2. 过滤目标（只保留主板、创业板、科创板）
    stock_targets = [t for t in targets if t['market'] in ['主板', '创业板', '科创板']]
    new_codes = [t for t in stock_targets if t['code'] not in existing_codes]

    # 3. 全局水位线拦截
    if global_latest_date >= last_trade_day and not new_codes:
        logger.info(f"Margin data is already up to date ({global_latest_date}) and no new stocks. Skip.")
        return {'count': 0, 'failed_codes': [], 'status': 'skipped', 'record_count': 0, 'changed_files': [], 'message': f'Already up to date ({global_latest_date})'}

    # 4. 增量获取
    if global_latest_date >= last_trade_day:
        logger.info(f"Global date is up to date, but found {len(new_codes)} new stocks. Syncing new stocks only.")
        sync_targets = new_codes
    else:
        logger.info(f"Syncing margin data from {global_latest_date} to {last_trade_day}...")
        sync_targets = stock_targets

    all_data = []
    success_count = 0
    
    with ThreadPoolExecutor(max_workers=get_thread_config_safe()['margin']) as executor:
        futures = {executor.submit(get_stock_margin, t['code']): t['code'] for t in sync_targets}
        for i, future in enumerate(as_completed(futures), 1):
            res = future.result()
            if res is not None and not res.empty:
                code = futures[future]
                if code in existing_codes:
                    res = res[res['trade_date'] > pd.to_datetime(global_latest_date)]
                if not res.empty:
                    all_data.append(res)
                    success_count += 1
            if i % 500 == 0:
                logger.info(f"Margin progress: {i}/{len(sync_targets)}, success: {success_count}")
    
    total_records = 0
    changed_files = []  # 记录变更的文件名
    
    # 5. 按月分表保存
    if all_data:
        new_df = pd.concat(all_data, ignore_index=True)
        total_records = len(new_df)
        
        if not new_df.empty:
            min_date = new_df['trade_date'].min()
            max_date = new_df['trade_date'].max()
            
            current = min_date.to_period('M')
            end_period = max_date.to_period('M')
            
            while current <= end_period:
                yr, mo = current.year, current.month
                month_data = new_df[(new_df['trade_date'].dt.year == yr) & (new_df['trade_date'].dt.month == mo)]
                
                if not month_data.empty:
                    filename = f"{yr}-{mo:02d}.parquet"
                    local_path = margin_dir / filename
                    
                    old_month_df = None
                    if local_path.exists():
                        try:
                            old_month_df = pd.read_parquet(local_path)
                        except Exception:
                            pass
                    
                    if old_month_df is not None:
                        final_month_df = pd.concat([old_month_df, month_data]).drop_duplicates(subset=['code', 'trade_date'])
                    else:
                        final_month_df = month_data
                    
                    final_month_df.to_parquet(local_path)
                    changed_files.append(filename)  # 记录变更的文件
                    logger.info(f"Saved margin data for {filename}")
                
                current += 1
        
        logger.info(f"Margin updated: {len(new_df)} new records")
    
    return {
        'count': success_count,
        'failed_codes': [],
        'status': 'success',
        'record_count': total_records,
        'changed_files': changed_files  # 返回变更文件列表
    }


# ==================== 新增：财务指标数据同步 ====================

def get_stock_financial_indicator(code: str) -> Optional[pd.DataFrame]:
    """获取单只股票财务指标数据"""
    max_retries = 5
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            df = ak.stock_financial_analysis_indicator(symbol=code)
            if df is not None and not df.empty:
                # 标准化列名
                rename_map = {
                    '日期': 'trade_date',
                    '净资产收益率': 'roe',
                    '总资产净利率': 'roa',
                    '销售毛利率': 'gross_margin',
                    '销售净利率': 'net_margin',
                    '资产负债率': 'debt_ratio',
                    '流动比率': 'current_ratio',
                    '速动比率': 'quick_ratio',
                    '存货周转率': 'inventory_turnover',
                    '应收账款周转率': 'receivable_turnover',
                    '总资产周转率': 'total_asset_turnover',
                }
                df = df.rename(columns=rename_map)
                if 'trade_date' in df.columns:
                    df['trade_date'] = pd.to_datetime(df['trade_date'])
                df['code'] = code
                
                cols = ['code', 'trade_date', 'roe', 'roa', 'gross_margin', 'net_margin',
                        'debt_ratio', 'current_ratio', 'quick_ratio', 
                        'inventory_turnover', 'receivable_turnover', 'total_asset_turnover']
                available_cols = [c for c in cols if c in df.columns]
                return df[available_cols]
        except Exception:
            if attempt > 0:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
    return None


def sync_financial_indicator(targets: List[Dict[str, str]]) -> Dict[str, Any]:
    """同步财务指标数据（极致增量版），返回详细结果"""
    logger.info("Syncing financial indicator data...")
    fi_path = Path("/tmp/data/financial_indicator.parquet")
    fi_path.parent.mkdir(parents=True, exist_ok=True)
    
    old_df = None
    old_count = 0
    global_latest_date = "2000-01-01"
    existing_codes = set()

    # 1. 优先读取本地缓存
    if fi_path.exists():
        try:
            old_df = pd.read_parquet(fi_path)
            old_count = len(old_df)
            global_latest_date = old_df['trade_date'].max().strftime('%Y-%m-%d')
            existing_codes = set(old_df['code'].unique())
            logger.info(f"Local financial cache found, latest date: {global_latest_date}, records: {old_count}")
        except Exception as e:
            logger.warning(f"Failed to read local financial cache: {e}")

    # 2. 本地无缓存，尝试从云端拉取
    if old_df is None:
        repo_id = os.getenv("DATASET_REPO_ID")
        if repo_id:
            try:
                old_file = hf_hub_download(repo_id=repo_id, filename="data/financial_indicator.parquet", repo_type="dataset")
                old_df = pd.read_parquet(old_file)
                old_count = len(old_df)
                global_latest_date = old_df['trade_date'].max().strftime('%Y-%m-%d')
                existing_codes = set(old_df['code'].unique())
                old_df.to_parquet(fi_path)
                logger.info(f"Downloaded financial from cloud, latest date: {global_latest_date}, records: {old_count}")
            except Exception:
                logger.info("No existing financial data found in cloud.")

    # 3. 财务指标特殊拦截 + 新股检测
    stock_targets = [t for t in targets if t['market'] not in ['ETF', 'LOF', 'REITs', '可转债']]
    new_codes = [t for t in stock_targets if t['code'] not in existing_codes]
    
    today = get_beijing_time()
    is_recent = False
    if global_latest_date != "2000-01-01":
        days_diff = (today - pd.to_datetime(global_latest_date)).days
        if days_diff < 90:
            is_recent = True

    if is_recent and not new_codes:
        logger.info(f"Financial data is recent ({global_latest_date}) and no new stocks. Skip.")
        return {'count': 0, 'status': 'skipped', 'record_count': old_count, 'new_records': 0, 'message': f'Already up to date ({global_latest_date})'}

    # 4. 增量获取
    if is_recent:
        logger.info(f"Financial data is recent, but found {len(new_codes)} new stocks. Syncing new stocks only.")
        sync_targets = new_codes
    else:
        logger.info(f"Syncing financial indicators (last update: {global_latest_date})...")
        sync_targets = stock_targets

    all_data = []
    success_count = 0
    
    with ThreadPoolExecutor(max_workers=get_thread_config_safe()['financial']) as executor:
        futures = {executor.submit(get_stock_financial_indicator, t['code']): t['code'] for t in sync_targets}
        for i, future in enumerate(as_completed(futures), 1):
            res = future.result()
            if res is not None and not res.empty:
                code = futures[future]
                if code in existing_codes:
                    res = res[res['trade_date'] > pd.to_datetime(global_latest_date)]
                
                if not res.empty:
                    all_data.append(res)
                    success_count += 1
            if i % 500 == 0:
                logger.info(f"Financial indicator progress: {i}/{len(sync_targets)}, success: {success_count}")
    
    # 5. 合并保存
    new_records = 0
    final_count = old_count
    if all_data:
        new_df = pd.concat(all_data, ignore_index=True)
        new_records = len(new_df)
        final_df = pd.concat([old_df, new_df]) if old_df is not None else new_df
        final_df = final_df.drop_duplicates(subset=['code', 'trade_date'])
        final_count = len(final_df)
        final_df.to_parquet(fi_path)
        logger.info(f"Financial updated: {final_count} total records ({new_records} new)")
    
    return {
        'count': success_count,
        'status': 'success' if success_count > 0 or old_count > 0 else 'fail',
        'record_count': final_count,
        'new_records': new_records,
        'previous_count': old_count
    }


# ==================== 新增：股东户数数据同步 ====================

def sync_holder_num() -> Dict[str, Any]:
    """同步股东户数数据（批量获取），返回详细结果"""
    logger.info("Syncing holder number data...")
    
    hn_path = Path("/tmp/data/holder_num.parquet")
    old_count = 0
    
    # 读取现有数据
    if hn_path.exists():
        try:
            old_df = pd.read_parquet(hn_path)
            old_count = len(old_df)
        except Exception:
            pass
    
    try:
        # 获取最近报告期的股东户数数据
        df = ak.stock_zh_a_gdhs(symbol="全部")
        if df is not None and not df.empty:
            # 标准化列名
            rename_map = {
                '代码': 'code',
                '股东户数': 'holder_num',
                '户均持股数量': 'avg_share',
                '户均持股金额': 'avg_value',
                '总股本': 'total_share',
                '总市值': 'total_value',
                '日期': 'trade_date',
            }
            df = df.rename(columns=rename_map)
            if 'trade_date' in df.columns:
                df['trade_date'] = pd.to_datetime(df['trade_date'])
            
            new_count = len(df)
            
            # 保存到 Parquet
            hn_path.parent.mkdir(parents=True, exist_ok=True)
            df.to_parquet(hn_path)
            
            # 判断是否有变化
            is_changed = new_count != old_count
            
            logger.info(f"Holder number data saved: {new_count} records (previous: {old_count}, changed: {is_changed})")
            return {
                'count': new_count,
                'status': 'success',
                'record_count': new_count,
                'previous_count': old_count,
                'is_changed': is_changed
            }
    except Exception as e:
        logger.warning(f"Failed to sync holder number data: {e}")
    
    return {
        'count': 0,
        'status': 'fail',
        'record_count': old_count,
        'previous_count': old_count,
        'is_changed': False
    }


# ==================== 新增：分红数据同步 ====================

def get_stock_dividend(code: str) -> Optional[pd.DataFrame]:
    """获取单只股票分红数据"""
    max_retries = 5
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            df = ak.stock_history_dividend(symbol=code)
            if df is not None and not df.empty:
                # 标准化列名
                rename_map = {
                    '公告日期': 'trade_date',
                    '分红方案': 'dividend_type',
                    '分红金额': 'dividend_amount',
                    '股权登记日': 'record_date',
                    '除权除息日': 'ex_date',
                    '派息日': 'pay_date',
                }
                df = df.rename(columns=rename_map)
                df['trade_date'] = pd.to_datetime(df['trade_date'])
                df['code'] = code
                
                cols = ['code', 'trade_date', 'dividend_type', 'dividend_amount', 'record_date', 'ex_date', 'pay_date']
                available_cols = [c for c in cols if c in df.columns]
                return df[available_cols]
        except Exception:
            if attempt > 0:
                delay = base_delay * (2 ** attempt)
                time.sleep(delay)
    return None


def sync_dividend(targets: List[Dict[str, str]]) -> Dict[str, Any]:
    """同步分红数据（极致增量版），返回详细结果"""
    logger.info("Syncing dividend data...")
    div_path = Path("/tmp/data/dividend.parquet")
    div_path.parent.mkdir(parents=True, exist_ok=True)
    
    old_df = None
    old_count = 0
    global_latest_date = "2000-01-01"
    existing_codes = set()

    if div_path.exists():
        try:
            old_df = pd.read_parquet(div_path)
            old_count = len(old_df)
            global_latest_date = old_df['trade_date'].max().strftime('%Y-%m-%d')
            existing_codes = set(old_df['code'].unique())
        except Exception: pass

    if old_df is None:
        repo_id = os.getenv("DATASET_REPO_ID")
        if repo_id:
            try:
                old_file = hf_hub_download(repo_id=repo_id, filename="data/dividend.parquet", repo_type="dataset")
                old_df = pd.read_parquet(old_file)
                old_count = len(old_df)
                global_latest_date = old_df['trade_date'].max().strftime('%Y-%m-%d')
                existing_codes = set(old_df['code'].unique())
                old_df.to_parquet(div_path)
            except Exception: pass

    # 90天检查一次 + 新股检测
    stock_targets = [t for t in targets if t['market'] not in ['ETF', 'LOF', 'REITs', '可转债']]
    new_codes = [t for t in stock_targets if t['code'] not in existing_codes]

    today = get_beijing_time()
    is_recent = False
    if global_latest_date != "2000-01-01":
        if (today - pd.to_datetime(global_latest_date)).days < 90:
            is_recent = True

    if is_recent and not new_codes:
        logger.info(f"Dividend data is recent and no new stocks. Skip.")
        return {'count': 0, 'status': 'skipped', 'record_count': old_count, 'new_records': 0, 'message': f'Already up to date ({global_latest_date})'}

    # 4. 增量获取
    if is_recent:
        logger.info(f"Dividend data is recent, but found {len(new_codes)} new stocks. Syncing new stocks only.")
        sync_targets = new_codes
    else:
        logger.info(f"Syncing dividend data (last update: {global_latest_date})...")
        sync_targets = stock_targets
    
    all_data = []
    success_count = 0
    
    with ThreadPoolExecutor(max_workers=get_thread_config_safe()['dividend']) as executor:
        futures = {executor.submit(get_stock_dividend, t['code']): t['code'] for t in sync_targets}
        for i, future in enumerate(as_completed(futures), 1):
            res = future.result()
            if res is not None and not res.empty:
                code = futures[future]
                if code in existing_codes:
                    res = res[res['trade_date'] > pd.to_datetime(global_latest_date)]

                if not res.empty:
                    all_data.append(res)
                    success_count += 1
            if i % 500 == 0:
                logger.info(f"Dividend progress: {i}/{len(sync_targets)}, success: {success_count}")
    
    new_records = 0
    final_count = old_count
    if all_data:
        new_df = pd.concat(all_data, ignore_index=True)
        new_records = len(new_df)
        final_df = pd.concat([old_df, new_df]) if old_df is not None else new_df
        final_df = final_df.drop_duplicates(subset=['code', 'trade_date', 'dividend_type'])
        final_count = len(final_df)
        final_df.to_parquet(div_path)
        logger.info(f"Dividend updated: {final_count} total records ({new_records} new)")
    
    return {
        'count': success_count,
        'status': 'success' if success_count > 0 or old_count > 0 else 'fail',
        'record_count': final_count,
        'new_records': new_records,
        'previous_count': old_count
    }


# ==================== 新增：十大股东数据同步 ====================

def sync_top_holders() -> Dict[str, Any]:
    """同步十大股东数据（批量获取），返回详细结果"""
    logger.info("Syncing top holders data...")
    
    path = Path("/tmp/data/top_holders.parquet")
    old_count = 0
    
    # 读取现有数据
    if path.exists():
        try:
            old_df = pd.read_parquet(path)
            old_count = len(old_df)
        except Exception:
            pass
    
    try:
        today = get_beijing_time()
        df = ak.stock_gdfx_holding_analyse_em(date=today.strftime('%Y%m%d'))
        if df is not None and not df.empty:
            rename_map = {
                '股票代码': 'code',
                '公告日期': 'trade_date',
                '股东名称': 'holder_name',
                '持股数量': 'hold_num',
                '持股比例': 'hold_ratio',
                '持股变动': 'hold_change',
            }
            df = df.rename(columns=rename_map)
            df['trade_date'] = pd.to_datetime(df['trade_date'])
            
            new_count = len(df)
            
            path.parent.mkdir(parents=True, exist_ok=True)
            df.to_parquet(path)
            
            # 判断是否有变化
            is_changed = new_count != old_count
            
            logger.info(f"Top holders data saved: {new_count} records (previous: {old_count}, changed: {is_changed})")
            return {
                'count': new_count,
                'status': 'success',
                'record_count': new_count,
                'previous_count': old_count,
                'is_changed': is_changed
            }
    except Exception as e:
        logger.warning(f"Failed to sync top holders: {e}")
    
    return {
        'count': 0,
        'status': 'fail',
        'record_count': old_count,
        'previous_count': old_count,
        'is_changed': False
    }


# ==================== 新增：限售解禁数据同步 ====================

def sync_restricted_unlock() -> Dict[str, Any]:
    """同步限售解禁数据（批量获取），返回详细结果"""
    logger.info("Syncing restricted unlock data...")
    path = Path("/tmp/data/restricted_unlock.parquet")
    path.parent.mkdir(parents=True, exist_ok=True)
    
    old_count = 0
    
    # 读取现有数据
    if path.exists():
        try:
            old_df = pd.read_parquet(path)
            old_count = len(old_df)
        except Exception:
            pass
    
    try:
        # 获取全市场限售解禁数据
        df = ak.stock_restricted_shares(stock="all")
        if df is not None and not df.empty:
            rename_map = {
                '代码': 'code',
                '名称': 'name',
                '解禁日期': 'unlock_date',
                '解禁数量': 'unlock_num',
                '解禁股本占总股本比例': 'unlock_ratio',
            }
            df = df.rename(columns=rename_map)
            df['unlock_date'] = pd.to_datetime(df['unlock_date'])
            df['trade_date'] = get_beijing_time() # 记录同步日期
            
            new_count = len(df)
            
            df.to_parquet(path)
            
            # 判断是否有变化
            is_changed = new_count != old_count
            
            logger.info(f"Restricted unlock data saved: {new_count} records (previous: {old_count}, changed: {is_changed})")
            return {
                'count': new_count,
                'status': 'success',
                'record_count': new_count,
                'previous_count': old_count,
                'is_changed': is_changed
            }
    except Exception as e:
        logger.warning(f"Failed to sync restricted unlock: {e}")
    
    return {
        'count': 0,
        'status': 'fail',
        'record_count': old_count,
        'previous_count': old_count,
        'is_changed': False
    }


def main() -> int:
    """
    主函数 - 执行完整的数据同步流程（每类指标完成后即时上传，并记录状态）
    
    Returns:
        int: 退出码，0 表示成功，1 表示失败
    """
    logger.info("=" * 60)
    logger.info("Stock Data Sync Started")
    logger.info("=" * 60)
    
    try:
        # 初始化线程配置
        init_thread_config()
        
        db = get_db()
        db.init_db()
        
        # 获取状态管理器
        status = get_sync_status()
        
        # 获取最后交易日
        last_day = get_last_trading_day()
        logger.info(f"Last trading day: {last_day}")
        
        # 1. 列表同步
        logger.info("-" * 40)
        logger.info("Syncing stock list...")
        target_list = get_stock_list()
        list_parquet = Path("/tmp/data/stock_list.parquet")
        list_parquet.parent.mkdir(parents=True, exist_ok=True)
        target_list.to_parquet(list_parquet)
        db.upload_indicator("Stock List", list_parquet, "data")
        status.update('stock_list', 
                     last_trade_date=last_day,
                     record_count=len(target_list),
                     status='success')

        # 2. 行情同步
        logger.info("-" * 40)
        logger.info("Syncing daily data...")
        daily_result = sync_stock_daily(target_list.to_dict('records'), last_day)
        # 智能上传日K行情数据（根据变更文件数量选择策略）
        parquet_dir = Path("/tmp/data/parquet")
        if parquet_dir.exists() and any(parquet_dir.glob("*.parquet")):
            db.upload_indicator_smart("Daily Data", parquet_dir, "data/parquet",
                                     daily_result.get('changed_files', []))
        status.update('daily',
                     last_trade_date=last_day,
                     record_count=daily_result.get('record_count', 0),
                     status=daily_result.get('status', 'unknown'),
                     failed_codes=daily_result.get('failed_codes', []),
                     success_rate=daily_result.get('success_rate', 0),
                     message=daily_result.get('message', ''))
        
        # 3. 指数同步
        logger.info("-" * 40)
        logger.info("Syncing index data...")
        idx_df = get_index_daily('000300')
        idx_count = 0
        if idx_df is not None:
            idx_path = Path("/tmp/data/parquet/index_000300.parquet")
            idx_path.parent.mkdir(parents=True, exist_ok=True)
            idx_df.to_parquet(idx_path)
            db.upload_indicator("Index Data", idx_path, "data/parquet")
            idx_count = len(idx_df)
        status.update('index',
                     last_trade_date=last_day,
                     record_count=idx_count,
                     status='success' if idx_count > 0 else 'fail')

        # 4. 资金流向同步
        logger.info("-" * 40)
        logger.info("Syncing fund flow...")
        fund_flow_result = sync_fund_flow(target_list.to_dict('records'), last_day)
        # 智能上传资金流向数据（根据变更文件数量选择策略）
        fund_flow_dir = Path("/tmp/data/fund_flow")
        if fund_flow_dir.exists() and any(fund_flow_dir.glob("*.parquet")):
            db.upload_indicator_smart("Fund Flow", fund_flow_dir, "data/fund_flow",
                                     fund_flow_result.get('changed_files', []))
        status.update('fund_flow',
                     last_trade_date=last_day,
                     record_count=fund_flow_result.get('record_count', 0),
                     status=fund_flow_result.get('status', 'unknown'),
                     failed_codes=fund_flow_result.get('failed_codes', []),
                     success_rate=fund_flow_result.get('success_rate', 0),
                     message=fund_flow_result.get('message', ''))
        
        # 5. 估值指标同步
        logger.info("-" * 40)
        logger.info("Syncing valuation...")
        valuation_result = sync_valuation(target_list.to_dict('records'), last_day)
        # 智能上传估值指标数据（根据变更文件数量选择策略）
        valuation_dir = Path("/tmp/data/valuation")
        if valuation_dir.exists() and any(valuation_dir.glob("*.parquet")):
            db.upload_indicator_smart("Valuation", valuation_dir, "data/valuation",
                                     valuation_result.get('changed_files', []))
        status.update('valuation',
                     last_trade_date=last_day,
                     record_count=valuation_result.get('record_count', 0),
                     status=valuation_result.get('status', 'success'))
        
        # 6. 融资融券同步
        logger.info("-" * 40)
        logger.info("Syncing margin...")
        margin_result = sync_margin(target_list.to_dict('records'), last_day)
        # 智能上传融资融券数据（根据变更文件数量选择策略）
        margin_dir = Path("/tmp/data/margin")
        if margin_dir.exists() and any(margin_dir.glob("*.parquet")):
            db.upload_indicator_smart("Margin", margin_dir, "data/margin",
                                     margin_result.get('changed_files', []))
        status.update('margin',
                     last_trade_date=last_day,
                     record_count=margin_result.get('record_count', 0),
                     status=margin_result.get('status', 'success'))
        
        # 7. 财务指标同步
        logger.info("-" * 40)
        logger.info("Syncing financial indicator...")
        financial_result = sync_financial_indicator(target_list.to_dict('records'))
        fi_path = Path("/tmp/data/financial_indicator.parquet")
        if fi_path.exists() and financial_result.get('new_records', 0) > 0:
            upload_success = db.upload_indicator("Financial Indicator", fi_path, "data")
            financial_status = 'success' if upload_success else 'upload_fail'
        else:
            financial_status = financial_result.get('status', 'skipped')
        status.update('financial',
                     last_trade_date=last_day,
                     record_count=financial_result.get('record_count', 0),
                     status=financial_status,
                     new_records=financial_result.get('new_records', 0))
        
        # 8. 股东户数同步
        logger.info("-" * 40)
        logger.info("Syncing holder num...")
        holder_result = sync_holder_num()
        holder_path = Path("/tmp/data/holder_num.parquet")
        if holder_result.get('is_changed', False) and holder_path.exists():
            upload_success = db.upload_indicator("Holder Num", holder_path, "data")
            holder_status = 'success' if upload_success else 'upload_fail'
        else:
            holder_status = 'skipped' if not holder_result.get('is_changed', False) else holder_result.get('status', 'fail')
        status.update('holder_num',
                     last_trade_date=last_day,
                     record_count=holder_result.get('record_count', 0),
                     status=holder_status,
                     is_changed=holder_result.get('is_changed', False))
        
        # 9. 分红数据同步
        logger.info("-" * 40)
        logger.info("Syncing dividend...")
        dividend_result = sync_dividend(target_list.to_dict('records'))
        div_path = Path("/tmp/data/dividend.parquet")
        if div_path.exists() and dividend_result.get('new_records', 0) > 0:
            upload_success = db.upload_indicator("Dividend", div_path, "data")
            dividend_status = 'success' if upload_success else 'upload_fail'
        else:
            dividend_status = dividend_result.get('status', 'skipped')
        status.update('dividend',
                     last_trade_date=last_day,
                     record_count=dividend_result.get('record_count', 0),
                     status=dividend_status,
                     new_records=dividend_result.get('new_records', 0))
        
        # 10. 十大股东同步
        logger.info("-" * 40)
        logger.info("Syncing top holders...")
        top_holders_result = sync_top_holders()
        top_holders_path = Path("/tmp/data/top_holders.parquet")
        if top_holders_result.get('is_changed', False) and top_holders_path.exists():
            upload_success = db.upload_indicator("Top Holders", top_holders_path, "data")
            top_holders_status = 'success' if upload_success else 'upload_fail'
        else:
            top_holders_status = 'skipped' if not top_holders_result.get('is_changed', False) else top_holders_result.get('status', 'fail')
        status.update('top_holders',
                     last_trade_date=last_day,
                     record_count=top_holders_result.get('record_count', 0),
                     status=top_holders_status,
                     is_changed=top_holders_result.get('is_changed', False))

        # 11. 限售解禁同步
        logger.info("-" * 40)
        logger.info("Syncing restricted unlock...")
        restricted_result = sync_restricted_unlock()
        restricted_path = Path("/tmp/data/restricted_unlock.parquet")
        if restricted_result.get('is_changed', False) and restricted_path.exists():
            upload_success = db.upload_indicator("Restricted Unlock", restricted_path, "data")
            restricted_status = 'success' if upload_success else 'upload_fail'
        else:
            restricted_status = 'skipped' if not restricted_result.get('is_changed', False) else restricted_result.get('status', 'fail')
        status.update('restricted_unlock',
                     last_trade_date=last_day,
                     record_count=restricted_result.get('record_count', 0),
                     status=restricted_status,
                     is_changed=restricted_result.get('is_changed', False))
        
        # 12. 上传状态文件
        logger.info("-" * 40)
        logger.info("Uploading sync status...")
        status_path = Path("/tmp/data/sync_status.json")
        status_upload_success = db.upload_indicator("Sync Status", status_path, "data")
        if not status_upload_success:
            logger.warning("Failed to upload sync status file")
        
        logger.info("=" * 60)
        logger.info("Sync Completed Successfully!")
        summary = (f"Daily={daily_result.get('count', 0)}, FundFlow={fund_flow_result.get('count', 0)}, "
                   f"Valuation={valuation_result.get('count', 0)}, Margin={margin_result.get('count', 0)}, Financial={financial_result.get('count', 0)}, "
                   f"Holder={holder_result.get('count', 0)}, Dividend={dividend_result.get('count', 0)}, "
                   f"TopHolders={top_holders_result.get('count', 0)}, Restricted={restricted_result.get('count', 0)}")
        logger.info(f"Summary: {summary}")
        logger.info("=" * 60)
        return 0
        
    except Exception as e:
        logger.error(f"Sync failed with error: {e}")
        return 1

if __name__ == "__main__":
    sys.exit(main())