fix：修复了财务类数据NAN值报错的bug。

quantitative_data/importer.py

@@ -98,6 +98,26 @@ def safe_float(val):
         return None
 
 
+def _get_table_columns(table_name: str, conn) -> set:
+    """
+    查询数据库表的实际列名集合
+    """
+    cursor = conn.cursor()
+    try:
+        cursor.execute(
+            """
+            SELECT column_name
+            FROM information_schema.columns
+            WHERE table_name = %s
+            """,
+            (table_name,),
+        )
+        cols = {row[0] for row in cursor.fetchall()}
+    finally:
+        cursor.close()
+    return cols
+
+
 def batch_insert(table_name: str, df: pd.DataFrame, conn, conflict_columns: List[str]):
     """
     使用 execute_values 批量 UPSERT (INSERT ... ON CONFLICT)
@@ -105,15 +125,90 @@ def batch_insert(table_name: str, df: pd.DataFrame, conn, conflict_columns: List
     - df: 待导入 DataFrame
     - conn: psycopg2 连接
     - conflict_columns: 冲突列 (唯一约束列)，冲突时更新其他列
+
+    自动过滤 DataFrame 中数据库表不存在的列，避免
+    'column "xxx" of relation "yyy" does not exist' 错误。
     """
     if df.empty:
         logger.warning(f"  {table_name}: 空数据，跳过")
         return 0
 
-    # pd.NaT / pd.NaT / numpy NaN 等无法被 psycopg2 识别，统一替换为 Python None
+    # ---- 动态过滤：只保留数据库表中存在的列 ----
-    df = df.where(pd.notna(df), None)
+    db_columns = _get_table_columns(table_name, conn)
+    if not db_columns:
+        # 表可能不存在，回退到不过滤（后续会抛出真实错误）
+        logger.warning(f"  {table_name}: 未查询到表结构，使用原始列")
+    else:
+        extra_cols = [c for c in df.columns if c not in db_columns]
+        if extra_cols:
+            logger.warning(
+                f"  {table_name}: 过滤掉表中不存在的列 {extra_cols}"
+            )
+        df = df[[c for c in df.columns if c in db_columns]]
+
+    if df.empty:
+        logger.warning(f"  {table_name}: 过滤后无可用列，跳过")
+        return 0
 
     columns = list(df.columns)
+
+    # 过滤后验证冲突列仍存在
+    missing_conflict = [c for c in conflict_columns if c not in columns]
+    if missing_conflict:
+        logger.error(
+            f"  {table_name}: 冲突列 {missing_conflict} 不在可用列中，跳过"
+        )
+        return 0
+
+    # ---- 按冲突列去重 ----
+    # PostgreSQL 的 ON CONFLICT DO UPDATE 不允许同一命令中出现重复冲突键：
+    #   "ON CONFLICT DO UPDATE command cannot affect row a second time"
+    # Tushare 财务接口 (income/balancesheet/cashflow/fina_indicator) 对同一股票
+    # 同一报告期可能返回多行数据（如多次公告修正，ann_date 不同但冲突键相同），
+    # 必须先在批内去重。
+    #
+    # 为保证保留的是"最新公告"的数据而非仅依赖 Tushare 返回顺序：
+    # 若存在公告日期列 (ann_date / f_ann_date)，先按公告日期升序排序，
+    # 再 drop_duplicates(keep="last") 即可稳定保留最新一条 (公告日期最大)，
+    # 且缺失公告日期的行 (NaT) 会排在最后，仅当无公告日期时才被保留。
+    #
+    # 注意：去重必须在此处 (NaN->None 替换之前) 执行，
+    # 此时日期列仍为 datetime64 类型，sort_values(na_position="last")
+    # 能正确处理 NaT；若在替换之后排序，object 类型混合日期/None 排序不可靠。
+    #
+    # 关键修复：PostgreSQL 对 VARCHAR/数值列会做隐式类型转换，因此数据库眼里
+    #   report_type = '1' (str) 与 report_type = 1 (int) 是"同一个约束值"；
+    # 但 pandas 的 drop_duplicates 认为 '1' != 1，导致去重不彻底、仍报错。
+    # 因此去重前先将非日期冲突列统一转成 str 再判断重复 (不修改原始 DataFrame)。
+    dedup_cols = [c for c in conflict_columns if c in columns]
+    before_dedup = len(df)
+    date_cols = [c for c in ["f_ann_date", "ann_date"] if c in columns]
+    if date_cols:
+        df = df.sort_values(date_cols, na_position="last")
+    if dedup_cols:
+        # 规范化去重键：非 datetime64 列统一 astype(str)，消除类型不一致的隐患
+        key_df = df[dedup_cols].copy()
+        for c in key_df.columns:
+            if not pd.api.types.is_datetime64_any_dtype(key_df[c]):
+                key_df[c] = key_df[c].astype(str)
+        dup_mask = key_df.duplicated(keep="last")
+        df = df[~dup_mask]
+    after_dedup = len(df)
+    if after_dedup < before_dedup:
+        logger.warning(
+            f"  {table_name}: 检测到 {before_dedup - after_dedup} 行重复冲突键，"
+            f"已保留最新公告记录 (去重后 {after_dedup} 行)"
+        )
+
+    # pd.NaT / numpy NaN 等无法被 psycopg2 识别，统一替换为 Python None。
+    # 关键：必须先 astype(object)！pandas 的 datetime64 列无法存储 None，
+    # 直接 where(..., None) 时 pandas 会把 None 自动提升回 NaT，
+    # 导致 psycopg2 生成 'NaT'::timestamp 非法 SQL
+    # (典型报错: fina_indicator 的 ann_date 为空时报
+    #  "invalid input syntax for type timestamp: \"NaT\"")。
+    # 转为 object 后 None 可正常存储，NaT/NaN 会被真正替换为 NULL。
+    df = df.astype(object).where(pd.notna(df), None)
+
     rows = [tuple(row) for row in df.itertuples(index=False)]
 
     # 构建 ON CONFLICT 子句（使用 sql.Identifier 防止注入/语法错误）
@@ -153,6 +248,25 @@ def batch_insert(table_name: str, df: pd.DataFrame, conn, conflict_columns: List
         return len(rows)
     except Exception as e:
         conn.rollback()
+        # 兜底：若批内仍存在重复冲突键 (如极端类型差异未被识别)，
+        # 回退为逐行 INSERT。单行命令永远不会触发
+        # "ON CONFLICT DO UPDATE command cannot affect row a second time"。
+        if "cannot affect row a second time" in str(e):
+            logger.warning(
+                f"  {table_name}: 批内仍存在重复冲突键，回退为逐行导入 ({len(rows)} 行)"
+            )
+            try:
+                inserted = 0
+                for r in rows:
+                    execute_values(cursor, upsert_sql.as_string(cursor), [r], page_size=1)
+                    inserted += 1
+                conn.commit()
+                logger.info(f"  {table_name}: 逐行成功导入 {inserted} 条记录")
+                return inserted
+            except Exception as e2:
+                conn.rollback()
+                logger.error(f"  {table_name}: 逐行导入失败 - {e2}")
+                raise
         logger.error(f"  {table_name}: 批量导入失败 - {e}")
         raise
     finally:
@@ -886,6 +1000,62 @@ def import_index_daily(
 # 8. 初始化数据库 Schema
 # ============================================================
 
+# ---- 表结构迁移映射: 为已存在的旧表补充新列 ----
+# key: 表名, value: 需要确保存在的列 -> (列定义类型, 注释)
+_SCHEMA_MIGRATIONS = {
+    "income": {
+        "fv_value_chg_gain": "NUMERIC(20,4)",
+    },
+    "balancesheet": {
+        "total_share": "NUMERIC(20,4)",
+    },
+    "cashflow": {
+        "finan_exp": "NUMERIC(20,4)",
+    },
+    "fina_indicator": {
+        "ca_turn": "NUMERIC(16,4)",
+    },
+}
+
+
+def _migrate_schema(conn):
+    """
+    迁移已存在的旧表: 使用 ALTER TABLE ... ADD COLUMN IF NOT EXISTS
+    补齐 Tushare API 返回但旧 schema 缺失的列。
+
+    - 对已存在的表生效 (CREATE TABLE IF NOT EXISTS 不会修改旧表)
+    - ADD COLUMN IF NOT EXISTS 幂等，可安全重复执行
+    - 与其他存储引擎不同，PostgreSQL 的 ADD COLUMN 是 O(1) 元数据操作
+    """
+    cursor = conn.cursor()
+    try:
+        for table_name, columns in _SCHEMA_MIGRATIONS.items():
+            # 先检查表是否存在
+            cursor.execute(
+                "SELECT to_regclass(%s)",
+                (table_name,),
+            )
+            if cursor.fetchone()[0] is None:
+                continue
+
+            for col_name, col_type in columns.items():
+                cursor.execute(
+                    sql.SQL("ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {}").format(
+                        sql.Identifier(table_name),
+                        sql.Identifier(col_name),
+                        sql.SQL(col_type),
+                    )
+                )
+                logger.info(f"  迁移: {table_name}.{col_name} 列已确认 ({col_type})")
+
+        conn.commit()
+    except Exception as e:
+        conn.rollback()
+        logger.warning(f"  表结构迁移失败 (可忽略，导入时自动过滤): {e}")
+    finally:
+        cursor.close()
+
+
 def init_database():
     """
     执行 DDL，创建所有表结构
@@ -946,6 +1116,9 @@ def init_database():
         conn.commit()
         cursor.close()
         logger.info("  Schema 初始化完成")
+
+        # 迁移已存在的旧表: 补齐 Tushare API 新增列
+        _migrate_schema(conn)
     except Exception as e:
         conn.rollback()
         logger.error(f"  Schema 初始化失败: {e}")

quantitative_data/schema.sql

@@ -216,7 +216,8 @@ CREATE TABLE IF NOT EXISTS income (
     n_sec_uw_income NUMERIC(20,4),              -- 证券承销业务净收入
     n_asset_mg_income NUMERIC(20,4),            -- 受托客户资产管理业务净收入
     oth_b_income    NUMERIC(20,4),              -- 其他业务收入
-    fv_value_chg    NUMERIC(20,4),              -- 加:公允价值变动净收益
+    fv_value_chg    NUMERIC(20,4),              -- 加:公允价值变动净收益 (兼容旧版字段)
+    fv_value_chg_gain NUMERIC(20,4),            -- 加:公允价值变动净收益 (Tushare income 实际字段)
     invest_income   NUMERIC(20,4),              -- 加:投资净收益
     ass_invest_income NUMERIC(20,4),            -- 其中:对联营企业和合营企业的投资收益
     forex_gain      NUMERIC(20,4),              -- 加:汇兑净收益
@@ -325,6 +326,9 @@ CREATE TABLE IF NOT EXISTS balancesheet (
     decr_in_disbur   NUMERIC(20,4),              -- 待处理流动资产损益
     oth_nca          NUMERIC(20,4),              -- 其他非流动资产
 
+    -- 股东权益相关
+    total_share      NUMERIC(20,4),              -- 总股本 (Tushare balancesheet 返回字段)
+
     -- 负债
     total_liab       NUMERIC(20,4),              -- 负债合计
     total_cur_liab   NUMERIC(20,4),              -- 流动负债合计
@@ -363,7 +367,8 @@ CREATE TABLE IF NOT EXISTS cashflow (
 
     -- 经营活动
     net_profit       NUMERIC(20,4),              -- 净利润
-    fin_exp          NUMERIC(20,4),              -- 财务费用
+    fin_exp          NUMERIC(20,4),              -- 财务费用 (兼容旧版字段)
+    finan_exp        NUMERIC(20,4),              -- 财务费用 (Tushare cashflow 实际字段)
     c_fr_sale_sg     NUMERIC(20,4),              -- 销售商品、提供劳务收到的现金
     recp_tax_rends   NUMERIC(20,4),              -- 收到的税费返还
     n_depos_incr_fi  NUMERIC(20,4),              -- 客户存款和同业存放款项净增加额
@@ -429,6 +434,7 @@ CREATE TABLE IF NOT EXISTS fina_indicator (
     arturn_days      NUMERIC(16,4),              -- 应收账款周转天数
     inv_turn         NUMERIC(16,4),              -- 存货周转率
     ar_turn          NUMERIC(16,4),              -- 应收账款周转率
+    ca_turn          NUMERIC(16,4),              -- 流动资产周转率 (Tushare fina_indicator 实际字段)
     assets_turn      NUMERIC(16,4),              -- 总资产周转率
     -- 盈利能力
     roe              NUMERIC(16,4),              -- 净资产收益率