diff --git a/quantitative_data/.ipynb_checkpoints/数据批量导入-checkpoint.ipynb b/quantitative_data/.ipynb_checkpoints/数据批量导入-checkpoint.ipynb index 9d2ff3b..c3142f5 100644 --- a/quantitative_data/.ipynb_checkpoints/数据批量导入-checkpoint.ipynb +++ b/quantitative_data/.ipynb_checkpoints/数据批量导入-checkpoint.ipynb @@ -84,7 +84,6 @@ " init_database,\n", " import_stock_basic,\n", " import_trade_cal,\n", - " import_daily_batch,\n", " import_daily_by_year,\n", " import_daily_basic,\n", " import_daily_basic_by_date,\n", diff --git a/quantitative_data/importer.py b/quantitative_data/importer.py index 1fa5c3d..08f81b9 100644 --- a/quantitative_data/importer.py +++ b/quantitative_data/importer.py @@ -367,12 +367,16 @@ def import_daily_by_date( end_date: Optional[str] = None, conn=None, sleep_interval: float = 0.3, + skip_existing: bool = True, ): """ 按交易日批量导入日线行情 (高效模式) 使用 pro.daily(trade_date='YYYYMMDD') 一次性拉取全市场当日数据 大幅减少 API 调用次数: 约250交易日/年 × 16年 ≈ 4000次 (原来需要 5000股票 × 16年 = 80000次) + 参数: + - skip_existing: 是否跳过数据库中已有数据的交易日 (默认 True,避免重复导入) + 返回: 失败的交易日列表 """ if start_date is None: @@ -403,14 +407,51 @@ def import_daily_by_date( conn.close() conn = get_pg_connection() - total = len(trade_dates) - if total == 0: + total_cal = len(trade_dates) + if total_cal == 0: logger.warning(f" 日期范围 {start_date} ~ {end_date} 内无交易日") if own_conn: conn.close() return [] - logger.info(f" 日期范围 {start_date} ~ {end_date}: 共 {total} 个交易日") + # --- 跳过已有数据的交易日 --- + skipped_count = 0 + if skip_existing: + try: + cursor = conn.cursor() + cursor.execute( + """ + SELECT DISTINCT trade_date FROM daily + WHERE trade_date >= %s AND trade_date <= %s + """, + (start_date, end_date), + ) + existing_dates = set(row[0] for row in cursor.fetchall()) + cursor.close() + if existing_dates: + # 统一格式为 date 类型再比较 + trade_dates_filtered = [td for td in trade_dates if td not in existing_dates] + skipped_count = len(trade_dates) - len(trade_dates_filtered) + trade_dates = trade_dates_filtered + except Exception as e: + logger.warning(f" 查询已有交易日失败,将导入全部: {e}") + # 查询失败时回退到全量导入(安全策略) + + total = len(trade_dates) + if total == 0: + logger.info(f" 日期范围 {start_date} ~ {end_date}: 所有 {total_cal} 个交易日均已有数据,无需导入") + if own_conn: + conn.close() + return [] + + if skipped_count > 0: + logger.info( + f" 日期范围 {start_date} ~ {end_date}: " + f"共 {total_cal} 个交易日,跳过 {skipped_count} 个已有数据," + f"待导入 {total} 个" + ) + else: + logger.info(f" 日期范围 {start_date} ~ {end_date}: 共 {total} 个交易日") pro = get_ts_pro() success_count = 0