diff --git a/scripts/data_platform/backfill_15min_baostock.py b/scripts/data_platform/backfill_15min_baostock.py index 2f76c51..891d667 100644 --- a/scripts/data_platform/backfill_15min_baostock.py +++ b/scripts/data_platform/backfill_15min_baostock.py @@ -222,6 +222,8 @@ def main(): parser.add_argument("--codes", help="指定股票代码,逗号分隔") parser.add_argument("--limit", type=int, default=0, help="限制处理数量(测试用)") parser.add_argument("--force", action="store_true", help="强制重建(覆盖已有BaoStock数据)") + parser.add_argument("--shard", type=int, default=None, help="分片号 (0-indexed,并行回填用)") + parser.add_argument("--total", type=int, default=1, help="总分片数(与 --shard 配合)") args = parser.parse_args() start_date = f"{args.start[:4]}-{args.start[4:6]}-{args.start[6:8]}" @@ -247,9 +249,15 @@ def main(): if args.limit > 0: codes = codes[:args.limit] + # 分片过滤(并行回填:每个 worker 处理 i % total == shard 的子集) + if args.shard is not None and args.total > 1: + codes = [s for i, s in enumerate(codes) if i % args.total == args.shard] + logger.info("=" * 60) logger.info("BaoStock 15min全量重建开始") logger.info(f" 股票数: {len(codes)}") + if args.shard is not None: + logger.info(f" 分片: shard={args.shard}/{args.total}") logger.info(f" 日期范围: {start_date} ~ {end_date}") logger.info(f" 数据目录: {MINUTE_15_DIR}") logger.info(f" 旧数据备份: {BACKUP_DIR}")