fix:添加断点续传功能。

quantitative_data/importer.py

@@ -795,19 +795,18 @@ def init_database():
     logger.info("=" * 60)
     logger.info("初始化数据库 Schema ...")
 
-    # 尝试创建数据库 (如果不存在)
+    # 尝试创建数据库（如果不存在）
     try:
         admin_conn = psycopg2.connect(
             host=DB_CONFIG["host"],
             port=DB_CONFIG["port"],
-            database="postgres",  # 连接默认 postgres 库来创建新库
+            database="postgres",
             user=DB_CONFIG["user"],
             password=DB_CONFIG["password"],
         )
         admin_conn.autocommit = True
         cursor = admin_conn.cursor()
 
-        # 检查数据库是否存在
         cursor.execute(
             "SELECT 1 FROM pg_database WHERE datname = %s",
             (DB_CONFIG["database"],),
@@ -828,82 +827,25 @@ def init_database():
         logger.warning(f"  创建数据库步骤跳过 (可能无权限): {e}")
 
     # 执行 DDL
+    import os as _os
+    import re as _re
+
+    schema_path = _os.path.join(
+        _os.path.dirname(_os.path.abspath(__file__)), "schema.sql"
+    )
+    with open(schema_path, "r", encoding="utf-8") as f:
+        ddl_sql = f.read()
+
+    # 先去掉单行注释，再按分号分句
+    ddl_sql = _re.sub(r"^\s*--.*$", "", ddl_sql, flags=_re.MULTILINE)
+    statements = [s.strip() for s in ddl_sql.split(";") if s.strip()]
+
     conn = get_pg_connection()
     try:
-        import os as _os
-
-        # 尝试使用 sqlparse 按语句拆分 (处理含分号的字符串字面量、函数体等)
-        try:
-            import sqlparse as _sqlparse
-            _USE_SQLPARSE = True
-        except ImportError:
-            _USE_SQLPARSE = False
-            logger.warning("  sqlparse 未安装，回退为简单分号拆分")
-
-        schema_path = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "schema.sql")
-        with open(schema_path, "r", encoding="utf-8") as f:
-            ddl_sql = f.read()
-
-        if _USE_SQLPARSE:
-            raw_statements = _sqlparse.split(ddl_sql)
-            statements = [
-                s.strip() for s in raw_statements
-                if s.strip() and not s.strip().startswith("--")
-            ]
-        else:
-            statements = [
-                s.strip() for s in ddl_sql.split(";")
-                if s.strip() and not s.strip().startswith("--")
-            ]
-
-        # 按依赖关系排序：CREATE TABLE → CREATE INDEX → COMMENT ON → CREATE VIEW
-        # 避免 sqlparse 拆分后语句乱序导致 UndefinedTable 错误
-        _priority = {
-            "CREATE TABLE": 1,
-            "CREATE INDEX": 2,
-            "COMMENT ON": 3,
-            "CREATE OR REPLACE VIEW": 4,
-            "CREATE VIEW": 4,
-        }
-
-        def _stmt_priority(s):
-            upper = s.upper()
-            for keyword, prio in _priority.items():
-                if upper.startswith(keyword):
-                    return prio
-            return 99  # 兜底：最后执行
-
-        statements.sort(key=_stmt_priority)
-
-        # 设置 autocommit 模式：每条 DDL 独立事务，互不影响
-        # 否则 rollback 会撤销之前已成功执行的 CREATE TABLE
-        conn.autocommit = True
-
         cursor = conn.cursor()
         for stmt in statements:
-            if stmt and not stmt.startswith("--"):
+            cursor.execute(stmt)
-                try:
+        conn.commit()
-                    cursor.execute(stmt)
-                except Exception as e:
-                    stmt_upper = stmt.upper()
-                    # 以下类型的语句失败视为可恢复的 warning，不中断整个初始化流程：
-                    #   1. 视图创建（依赖的基础表可能尚未创建）
-                    #   2. 索引创建（依赖的表可能尚未导入数据）
-                    #   3. COMMENT ON（依赖的表/视图可能尚未创建）
-                    is_recoverable = (
-                        "CREATE OR REPLACE VIEW" in stmt_upper
-                        or "CREATE VIEW" in stmt_upper
-                        or "CREATE INDEX" in stmt_upper
-                        or "COMMENT ON" in stmt_upper
-                    )
-                    if is_recoverable:
-                        logger.warning(f"  DDL 暂跳过（依赖尚未就绪）: {str(e)[:150]}\n  SQL: {stmt[:200]}")
-                    else:
-                        logger.error(f"  DDL 执行失败: {str(e)[:200]}\n  SQL: {stmt[:300]}")
-                        cursor.close()
-                        conn.close()
-                        raise
-
         cursor.close()
         logger.info("  Schema 初始化完成")
     except Exception as e:
@@ -913,7 +855,6 @@ def init_database():
     finally:
         conn.close()
 
-
 # ============================================================
 # 9. 获取所有股票列表 (辅助)
 # ============================================================
@@ -1041,6 +982,149 @@ def full_import(
     logger.info("=" * 70)
 
 
+# ============================================================
+# 11. 进度查询 & 断点续传辅助函数
+# ============================================================
+
+def check_daily_progress(conn=None):
+    """
+    查看 daily 表的导入进度（按年份 + 股票数统计）
+    返回各年份的记录数和独立股票数
+    """
+    own_conn = conn is None
+    if own_conn:
+        conn = get_pg_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute("""
+            SELECT 
+                EXTRACT(YEAR FROM trade_date)::int AS year,
+                COUNT(*) AS records,
+                COUNT(DISTINCT ts_code) AS stocks
+            FROM daily
+            GROUP BY year
+            ORDER BY year
+        """)
+        rows = cursor.fetchall()
+        if not rows:
+            logger.info("daily 表无数据")
+            return {}
+
+        logger.info(f"{'年份':<6} {'记录数':>12} {'股票数':>8}")
+        logger.info("-" * 30)
+        result = {}
+        for year, records, stocks in rows:
+            logger.info(f"{year:<6} {records:>12,} {stocks:>8,}")
+            result[int(year)] = {"records": int(records), "stocks": int(stocks)}
+        cursor.close()
+        return result
+    finally:
+        if own_conn:
+            conn.close()
+
+
+def check_table_summary(conn=None):
+    """
+    查看所有表的导入概览
+    """
+    own_conn = conn is None
+    if own_conn:
+        conn = get_pg_connection()
+    try:
+        tables = [
+            ("stock_basic", None),
+            ("trade_cal", (("trade_cal", "cal_date"),)),
+            ("daily", (("daily", "trade_date"),)),
+            ("daily_basic", (("daily_basic", "trade_date"),)),
+            ("adj_factor", (("adj_factor", "trade_date"),)),
+            ("income", (("income", "end_date"),)),
+            ("balancesheet", (("balancesheet", "end_date"),)),
+            ("cashflow", (("cashflow", "end_date"),)),
+            ("fina_indicator", (("fina_indicator", "end_date"),)),
+            ("index_daily", (("index_daily", "trade_date"),)),
+        ]
+        cursor = conn.cursor()
+        logger.info(f"{'表名':<20} {'记录数':>12} {'最早日期':>12} {'最晚日期':>12}")
+        logger.info("-" * 62)
+        for table_name, date_info in tables:
+            try:
+                cursor.execute(sql.SQL("SELECT COUNT(*) FROM {}").format(sql.Identifier(table_name)))
+                count = cursor.fetchone()[0]
+                if date_info:
+                    tbl, col = date_info
+                    cursor.execute(
+                        sql.SQL("SELECT MIN({}), MAX({}) FROM {}").format(
+                            sql.Identifier(col), sql.Identifier(col), sql.Identifier(tbl)
+                        )
+                    )
+                    min_d, max_d = cursor.fetchone()
+                    min_str = str(min_d)[:10] if min_d else "N/A"
+                    max_str = str(max_d)[:10] if max_d else "N/A"
+                    logger.info(f"{table_name:<20} {count:>12,} {min_str:>12} {max_str:>12}")
+                else:
+                    logger.info(f"{table_name:<20} {count:>12,}")
+            except Exception as e:
+                logger.warning(f"{table_name:<20} 查询失败: {e}")
+        cursor.close()
+    finally:
+        if own_conn:
+            conn.close()
+
+
+def resume_daily_by_year(stock_list, start_year=2010, end_year=2025):
+    """
+    从中断点恢复按年导入日线行情
+    自动跳过数据库已有的年份，只导入缺失年份的数据
+    """
+    conn = get_pg_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute("""
+            SELECT DISTINCT EXTRACT(YEAR FROM trade_date)::int AS year
+            FROM daily
+            ORDER BY year
+        """)
+        completed_years = set(row[0] for row in cursor.fetchall())
+        cursor.close()
+    finally:
+        conn.close()
+
+    logger.info(f"已完成年份: {sorted(completed_years)}")
+    logger.info(f"待导入年份: {[y for y in range(start_year, end_year+1) if y not in completed_years]}")
+
+    for year in range(start_year, end_year + 1):
+        if year in completed_years:
+            # 检查该年的股票覆盖是否完整
+            conn = get_pg_connection()
+            try:
+                cursor = conn.cursor()
+                cursor.execute(
+                    sql.SQL("""
+                        SELECT COUNT(DISTINCT ts_code)
+                        FROM {}
+                        WHERE EXTRACT(YEAR FROM trade_date) = %s
+                    """).format(sql.Identifier("daily")),
+                    (year,),
+                )
+                stock_count = cursor.fetchone()[0]
+                cursor.close()
+            finally:
+                conn.close()
+            logger.info(f"  {year} 年: 已有 {stock_count} 只股票, 跳过")
+            continue
+
+        year_start = f"{year}-01-01"
+        year_end = f"{year}-12-31"
+        logger.info(f"--- 导入 {year} 年日线行情 ---")
+        import_daily_batch(
+            stock_list,
+            start_date=year_start,
+            end_date=year_end,
+            sleep_interval=0.2,
+        )
+        logger.info(f"  {year} 年完成\n")
+
+
 if __name__ == "__main__":
     # 测试连接
     try:
@@ -1049,4 +1133,4 @@ if __name__ == "__main__":
         conn.close()
     except Exception as e:
         logger.error(f"无法连接到 PostgreSQL: {e}")
-        logger.error("请确认 Docker 容器已启动，且 config.py 中的连接参数正确")
+        logger.error("请确认 Docker 容器已启动，且 config.py 中的连接参数正确")

quantitative_data/数据批量导入.ipynb

@@ -134,6 +134,9 @@
     "    get_stock_codes_from_db,\n",
     "    full_import,\n",
     "    batch_insert,\n",
+    "    check_daily_progress,\n",
+    "    check_table_summary,\n",
+    "    resume_daily_by_year,\n",
     "    logger,\n",
     ")\n",
     "from config import DB_CONFIG, TUSHARE_TOKEN, START_DATE, END_DATE\n",
@@ -345,6 +348,76 @@
     ")"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### 4.2.1 查看日线导入进度\n",
+    "\n",
+    "按年份统计 daily 表中已导入的记录数和独立股票数，用于确认导入到哪个年份了。"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 查看每天的日线导入进度（按年份统计）\n",
+    "check_daily_progress()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 或者查看所有表的整体概览\n",
+    "check_table_summary()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### 4.2.2 断点续传 — 从中断处继续导入\n",
+    "\n",
+    "**方法一（推荐）：** 使用 `resume_daily_by_year` 自动跳过已完成的年份，只导入缺失年份。\n",
+    "\n",
+    "**方法二：** 手动修改 `start_year` 参数重新调用 `import_daily_by_year`（因为 UPSERT 幂等，重复导入不会造成数据问题）。"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 方法一：自动检测已有年份，只导入缺失年份（推荐）\n",
+    "stock_list = get_stock_codes_from_db()\n",
+    "resume_daily_by_year(stock_list, start_year=2010, end_year=2025)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 方法二：手动指定断点年份重新调用 import_daily_by_year\n",
+    "# 例如假设 2010~2020 已完成，从 2021 年开始继续\n",
+    "# stock_list = get_stock_codes_from_db()\n",
+    "# import_daily_by_year(stock_list, start_year=2021, end_year=2025)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "> **提示：** 以上两个方法都可以安全使用。因为 daily 表使用 `ON CONFLICT (ts_code, trade_date) DO UPDATE`，重复导入已存在的数据不会产生重复记录。另外也可查看 `import_data.log` 文件获取最后一次成功的日志输出。"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},