"feat:upload-after-each-indicator"

app/database.py

@@ -200,7 +200,7 @@ class DatabaseManager:
             
             # 5. 上传融资融券数据（按月分表）
             margin_dir = Path(os.path.dirname(DUCKDB_PATH)) / "margin"
-            if margindir.exists():
+            if margin_dir.exists():
                 for mar_file in margin_dir.glob("*.parquet"):
                     upload_file(
                         path_or_fileobj=str(mar_file),
@@ -271,6 +271,53 @@ class DatabaseManager:
         finally:
             _ = self.conn
     
+    def upload_indicator(self, indicator_name: str, local_path: Path, remote_path: str) -> bool:
+        """
+        上传单个指标数据到 HF Dataset
+        
+        Args:
+            indicator_name: 指标名称（用于日志）
+            local_path: 本地文件或目录路径
+            remote_path: 远程路径前缀（如 "data/fund_flow"）
+        
+        Returns:
+            bool: 是否上传成功
+        """
+        if not HF_TOKEN or not DATASET_REPO_ID:
+            logger.warning("HF_TOKEN or DATASET_REPO_ID not set, skipping upload")
+            return False
+        
+        try:
+            login(token=HF_TOKEN)
+            
+            if local_path.is_file():
+                # 单文件上传
+                upload_file(
+                    path_or_fileobj=str(local_path),
+                    path_in_repo=f"{remote_path}/{local_path.name}",
+                    repo_id=DATASET_REPO_ID,
+                    repo_type="dataset",
+                )
+                logger.info(f"{indicator_name} uploaded: {local_path.name}")
+            elif local_path.is_dir():
+                # 目录上传（上传所有 parquet 文件）
+                uploaded_count = 0
+                for p_file in local_path.glob("*.parquet"):
+                    upload_file(
+                        path_or_fileobj=str(p_file),
+                        path_in_repo=f"{remote_path}/{p_file.name}",
+                        repo_id=DATASET_REPO_ID,
+                        repo_type="dataset",
+                    )
+                    uploaded_count += 1
+                if uploaded_count > 0:
+                    logger.info(f"{indicator_name} uploaded: {uploaded_count} files")
+            
+            return True
+        except Exception as e:
+            logger.error(f"Failed to upload {indicator_name}: {e}")
+            return False
+    
     def _create_tables(self) -> None:
         """创建数据库表结构"""
         conn = self.conn

sync_data.py

@@ -1318,7 +1318,7 @@ def sync_restricted_unlock() -> int:
 
 def main() -> int:
     """
-    主函数 - 执行完整的数据同步流程
+    主函数 - 执行完整的数据同步流程（每类指标完成后即时上传）
     
     Returns:
         int: 退出码，0 表示成功，1 表示失败
@@ -1334,16 +1334,33 @@ def main() -> int:
         db = get_db()
         db.init_db()
         
+        # 统计变量
+        sync_count = 0
+        fund_flow_count = 0
+        valuation_count = 0
+        margin_count = 0
+        financial_count = 0
+        holder_count = 0
+        dividend_count = 0
+        top_holders_count = 0
+        restricted_count = 0
+        
         # 1. 列表同步
         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")
 
         # 2. 行情同步
         last_day = get_last_trading_day()
         logger.info(f"Last trading day: {last_day}")
         sync_count = sync_stock_daily(target_list.to_dict('records'), last_day)
+        # 上传日K行情数据
+        parquet_dir = Path("/tmp/data/parquet")
+        if parquet_dir.exists():
+            for p_file in parquet_dir.glob("*.parquet"):
+                db.upload_indicator("Daily Data", p_file, "data/parquet")
         
         # 3. 指数同步
         idx_df = get_index_daily('000300')
@@ -1351,36 +1368,74 @@ def main() -> int:
             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")
 
-        # 4-10. 各类指标同步
+        # 4. 资金流向同步
         logger.info("-" * 40)
         fund_flow_count = 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")):
+            for ff_file in fund_flow_dir.glob("*.parquet"):
+                db.upload_indicator("Fund Flow", ff_file, "data/fund_flow")
         
+        # 5. 估值指标同步
         logger.info("-" * 40)
         valuation_count = 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")):
+            for val_file in valuation_dir.glob("*.parquet"):
+                db.upload_indicator("Valuation", val_file, "data/valuation")
         
+        # 6. 融资融券同步
         logger.info("-" * 40)
         margin_count = 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")):
+            for mar_file in margin_dir.glob("*.parquet"):
+                db.upload_indicator("Margin", mar_file, "data/margin")
         
+        # 7. 财务指标同步
         logger.info("-" * 40)
         financial_count = sync_financial_indicator(target_list.to_dict('records'))
+        # 即时上传
+        fi_path = Path("/tmp/data/financial_indicator.parquet")
+        if fi_path.exists():
+            db.upload_indicator("Financial Indicator", fi_path, "data")
         
+        # 8. 股东户数同步
         logger.info("-" * 40)
         holder_count = sync_holder_num()
+        # 即时上传
+        holder_path = Path("/tmp/data/holder_num.parquet")
+        if holder_path.exists():
+            db.upload_indicator("Holder Num", holder_path, "data")
         
+        # 9. 分红数据同步
         logger.info("-" * 40)
         dividend_count = sync_dividend(target_list.to_dict('records'))
+        # 即时上传
+        div_path = Path("/tmp/data/dividend.parquet")
+        if div_path.exists():
+            db.upload_indicator("Dividend", div_path, "data")
         
+        # 10. 十大股东同步
         logger.info("-" * 40)
         top_holders_count = sync_top_holders()
+        # 即时上传
+        top_holders_path = Path("/tmp/data/top_holders.parquet")
+        if top_holders_path.exists():
+            db.upload_indicator("Top Holders", top_holders_path, "data")
 
+        # 11. 限售解禁同步
         logger.info("-" * 40)
         restricted_count = sync_restricted_unlock()
-
-        # 11. 上传
-        logger.info("-" * 40)
-        logger.info("Uploading to Hugging Face Dataset...")
-        db.upload_db()
+        # 即时上传
+        restricted_path = Path("/tmp/data/restricted_unlock.parquet")
+        if restricted_path.exists():
+            db.upload_indicator("Restricted Unlock", restricted_path, "data")
         
         logger.info("=" * 60)
         logger.info("Sync Completed Successfully!")