feat: add cloud-to-dataset sync and local download script

README.md

@@ -39,6 +39,28 @@ pinned: false
 
 - **系统日志 (系统日志/)**: 统一存储各组件运行产生的日志。
 
+## 云端采集与数据同步 (New)
+
+本系统支持在 Hugging Face Spaces 上进行 7x24 小时自动行情采集，并自动同步到 Hugging Face Dataset。
+
+### 1. 云端配置 (Hugging Face)
+在 Space 的 **Settings** -> **Variables and secrets** 中添加以下 Secrets：
+- `SUPABASE_URL`: 你的 Supabase 项目 URL
+- `SUPABASE_ANON_KEY`: 你的 Supabase Anon Key
+- `HF_TOKEN`: **必须**。具有 `Write` 权限的 Hugging Face Token，用于将数据上传到 Dataset。
+
+### 2. 数据流向
+1. **采集**: `服务/数据采集/启动采集.py` 实时采集 Binance 数据并保存为 `.parquet` 碎片。
+2. **整理**: `app.py` 每 12 小时触发一次 `整理行情数据.py`，将碎片合并为每日文件。
+3. **同步**: 整理完成后，自动运行 `hf_sync.py` 将数据推送到数据集 `chenchuanshen/Quant_Market_Data`。
+
+### 3. 本地获取数据
+在本地项目根目录下运行：
+```bash
+python 下载云端数据.py
+```
+该脚本会自动对比云端与本地差异，只下载新增的行情数据。
+
 ## 开发指南
 
 1. **环境准备**: 建议使用 Python 3.14+ 环境。

app.py

@@ -48,9 +48,33 @@ def 启动后台采集():
     for line in process.stdout:
         print(f"[Collector] {line.strip()}")
 
+def 周期性整理与同步():
+    """每隔 12 小时执行一次数据整理与 HF 同步"""
+    import sys
+    organize_script = os.path.join("服务", "数据采集", "整理行情数据.py")
+    env = os.environ.copy()
+    env["PYTHONPATH"] = os.getcwd()
+    
+    while True:
+        # 等待一段时间再执行第一次（让采集运行一会儿）
+        time.sleep(60) # 启动后 1 分钟先跑一次
+        print("🕒 开始执行周期性数据整理与同步...")
+        try:
+            # 运行整理脚本，默认会触发 sync-hf (因为我们在脚本里改了默认值为 True)
+            subprocess.run([sys.executable, organize_script], env=env, check=True)
+            print("✅ 周期性整理与同步完成。")
+        except Exception as e:
+            print(f"❌ 周期性整理失败: {e}")
+        
+        # 每 12 小时运行一次
+        time.sleep(12 * 3600)
+
 # 启动后台线程
-thread = threading.Thread(target=启动后台采集, daemon=True)
-thread.start()
+thread_collector = threading.Thread(target=启动后台采集, daemon=True)
+thread_collector.start()
+
+thread_sync = threading.Thread(target=周期性整理与同步, daemon=True)
+thread_sync.start()
 
 # ==========================================
 # 2. UI 逻辑

requirements.txt

@@ -20,6 +20,7 @@ pydantic
 beautifulsoup4
 lxml
 plotly
+huggingface_hub
 SQLAlchemy
 # Other necessary libs from original list
 aiodns

下载云端数据.py

@@ -0,0 +1,48 @@
+"""
+Quant Unified 量化交易系统
+下载云端数据 (HF Dataset -> Local)
+"""
+import os
+from pathlib import Path
+from huggingface_hub import snapshot_download
+import logging
+
+# 配置日志
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# 配置区域
+# ---------------------------------------------------------
+# 数据集名称
+DATASET_REPO = "chenchuanshen/Quant_Market_Data"
+# 本地行情数据存放路径
+LOCAL_DATA_DIR = Path(__file__).resolve().parent / "data" / "行情数据_整理"
+# ---------------------------------------------------------
+
+def download_data():
+    """从 Hugging Face Dataset 下载/同步数据到本地"""
+    logger.info(f"🔍 正在检查云端数据集: {DATASET_REPO}...")
+    
+    # 确保本地目录存在
+    LOCAL_DATA_DIR.mkdir(parents=True, exist_ok=True)
+
+    try:
+        # 使用 snapshot_download 自动对比并下载增量数据
+        # ignore_patterns 可以排除一些不必要的文件
+        local_path = snapshot_download(
+            repo_id=DATASET_REPO,
+            repo_type="dataset",
+            local_dir=str(LOCAL_DATA_DIR),
+            local_dir_use_symlinks=False,  # 直接拷贝文件
+            # token=os.getenv("HF_TOKEN") # 如果是私有数据集需要 Token
+        )
+        
+        logger.info(f"✨ 同步完成！数据已保存至: {local_path}")
+        return True
+    except Exception as e:
+        logger.error(f"❌ 下载失败: {e}")
+        logger.info("💡 提示: 如果是私有数据集，请先运行 `huggingface-cli login` 或设置 HF_TOKEN 环境变量")
+        return False
+
+if __name__ == "__main__":
+    download_data()

服务/数据采集/hf_sync.py

@@ -0,0 +1,76 @@
+"""
+Quant Unified 量化交易系统
+Hugging Face Dataset 同步工具
+"""
+import os
+import shutil
+from pathlib import Path
+from huggingface_hub import HfApi, create_repo
+from datetime import datetime
+import logging
+
+# 配置日志
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# 配置区域
+# ---------------------------------------------------------
+# 数据集名称: 用户名/数据集名
+DATASET_REPO = "chenchuanshen/Quant_Market_Data"
+# 本地行情数据路径
+LOCAL_DATA_DIR = Path(__file__).resolve().parents[2] / "data" / "行情数据_整理"
+# ---------------------------------------------------------
+
+def sync_to_hf():
+    """将本地整理好的数据同步到 Hugging Face Dataset"""
+    token = os.getenv("HF_TOKEN")
+    if not token:
+        logger.error("❌ 未发现 HF_TOKEN 环境变量，请在 Space Settings 中添加 Secret: HF_TOKEN (需要 Write 权限)")
+        return False
+
+    api = HfApi(token=token)
+
+    # 1. 确保仓库存在
+    try:
+        create_repo(repo_id=DATASET_REPO, repo_type="dataset", exist_ok=True)
+        logger.info(f"✅ 数据集仓库已就绪: {DATASET_REPO}")
+    except Exception as e:
+        logger.error(f"❌ 创建/访问仓库失败: {e}")
+        return False
+
+    # 2. 扫描本地整理好的数据文件
+    if not LOCAL_DATA_DIR.exists():
+        logger.warning(f"⚠️ 本地整理目录不存在: {LOCAL_DATA_DIR}")
+        return False
+
+    files_to_upload = list(LOCAL_DATA_DIR.rglob("*.parquet"))
+    if not files_to_upload:
+        logger.info("ℹ️ 没有发现需要上传的 .parquet 文件")
+        return True
+
+    logger.info(f"🚀 准备同步 {len(files_to_upload)} 个文件到云端...")
+
+    # 3. 批量上传
+    try:
+        # 我们按日期分目录上传，保持目录结构
+        # 这里的 path_in_repo 会保持 LOCAL_DATA_DIR 之后的相对路径
+        for file_path in files_to_upload:
+            relative_path = file_path.relative_to(LOCAL_DATA_DIR)
+            path_in_repo = str(relative_path)
+            
+            logger.info(f"正在上传: {path_in_repo}")
+            api.upload_file(
+                path_or_fileobj=str(file_path),
+                path_in_repo=path_in_repo,
+                repo_id=DATASET_REPO,
+                repo_type="dataset",
+            )
+        
+        logger.info("✨ 所有文件同步完成！")
+        return True
+    except Exception as e:
+        logger.error(f"❌ 同步过程中出错: {e}")
+        return False
+
+if __name__ == "__main__":
+    sync_to_hf()

服务/数据采集/整理行情数据.py

@@ -42,6 +42,7 @@ import pandas as pd
 默认_DELETE_SOURCE = False   # (已弃用，建议用 MOVE_TO_BACKUP) 整理完后是否删除原始碎片文件？
 默认_DELETE_TODAY = False    # 是否移动/删除今天的碎片文件？(今天的还在采集，建议不移动)
 默认_CHECK_GAP = True        # 是否检查并生成空缺报告？
+默认_SYNC_HF = True          # 整理完成后是否自动同步到 Hugging Face Dataset
 默认_GAP_MS_DEPTH = 2000     # 深度数据超过 2 秒没数据就算小缺口
 默认_GAP_MS_TRADE = 10000    # 成交数据超过 10 秒没数据就算小缺口
 默认_GAP_SAMPLES = 50        # 每个文件最多记录多少个缺口样本
@@ -454,6 +455,7 @@ def main(argv: list[str]) -> int:
     parser.add_argument("--delete-source", action="store_true", default=bool(默认_DELETE_SOURCE))
     parser.add_argument("--delete-today", action="store_true", default=bool(默认_DELETE_TODAY))
     parser.add_argument("--check-gap", action="store_true", default=bool(默认_CHECK_GAP))
+    parser.add_argument("--sync-hf", action="store_true", default=bool(默认_SYNC_HF))
     parser.add_argument("--gap-ms-depth", type=int, default=int(默认_GAP_MS_DEPTH))
     parser.add_argument("--gap-ms-trade", type=int, default=int(默认_GAP_MS_TRADE))
     parser.add_argument("--gap-samples", type=int, default=int(默认_GAP_SAMPLES))
@@ -646,6 +648,16 @@ def main(argv: list[str]) -> int:
     md_path.write_text("\n".join(md_lines), encoding="utf-8")
     print(f"可读报告已生成: {md_path}")
 
+    if args.sync_hf:
+        try:
+            from hf_sync import sync_to_hf
+            print("\n🚀 正在触发云端同步...")
+            sync_to_hf()
+        except ImportError:
+            print("\n⚠️ 无法加载 hf_sync.py，跳过同步。")
+        except Exception as e:
+            print(f"\n❌ 同步过程中出错: {e}")
+
     return 0