Dolphindb Finance Dataimport

/dolphindb-finance-dataimport — 金融数据导入

适用于：已有建好的 DolphinDB 数据库和表，需要将 CSV 文件导入到表中。

本 skill 只负责导入数据，不负责建库建表。建库建表请调用 /dolphindb-finance-dbtbcreate。

执行流程

Step 1: 收集用户输入

用 AskUserQuestion 收集以下参数。每个问题独立调用，纯文本输入（不要设置 options），用户通过"Other"填写。

问题 1 — CSV 文件路径（必填）

• 直接提问："请输入 CSV 文件在 DDB 服务器上的绝对路径"
• 不设 options，用户通过"Other"填写路径

问题 2 — 数据库路径（必填）

• 直接提问："请输入目标数据库路径，例如 dfs://DB_split_SH"

问题 3 — 表名（必填）

• 直接提问："请输入目标表名，例如 TB_split_SH_entrust"

Step 2: 验证目标库表存在

2.1 执行验证

直接提交以下代码给 DolphinX 平台执行：

print("=== VERIFY ===");
print("DB_EXISTS=" + string(existsDatabase("{数据库路径}")));
print("TB_EXISTS=" + string(existsTable("{数据库路径}", "{表名}")));

读取 === VERIFY === 后的 DB_EXISTS 和 TB_EXISTS 值。

2.2 检查结果

• 任一为 False → 报错终止，提示用户先运行 /dolphindb-finance-dbtbcreate 建库建表

• 均为 True → 做 schema 对比，检查表结构是否与 CSV 兼容：

  直接提交以下代码给 DolphinX 平台执行：

  print("=== CSV_COLUMNS ===");
  print(extractTextSchema("{CSV路径}"));
  print("=== TABLE_COLUMNS ===");
  colDefs = schema(loadTable("{数据库路径}", "{表名}")).colDefs;
  print(select name, typeString from colDefs);
  print("=== TABLE_COLUMN_ORDER ===");
  print(select name from colDefs);
  

  AI 助手读取三条输出后推理判断兼容性：

  • order_mismatch=true → 表结构与 CSV 列顺序不兼容，说明库表名已存在但可能是旧表（公共列相对顺序不一致）
  • 其他情况（仅有 type_mismatches、leading_table_only、trailing_table_only）→ 直接进入 Step 3，这些差异由导入脚本自动处理

2.3 处理不兼容

当 order_mismatch=true 时，用 AskUserQuestion 询问用户：

目标库 {数据库路径} 和表 {表名} 已存在，但其结构与当前 CSV 不兼容。
请选择处理方式：
  1. 删除已有库表并重建（原数据将丢失）
  2. 更换库名/表名（输入新的名称）
  3. 取消导入

• 选 1 → 执行 dropDatabase("{数据库路径}")（通过 .dos 脚本）删除，然后运行 /dolphindb-finance-dbtbcreate 重建，重建后回到 Step 2
• 选 2 → 提示用户输入新库名/表名，重新从 Step 2.1 开始验证
• 选 3 → 报告后终止

Step 3: CSV 与表结构对比

3.1 执行结构对比

如 Step 2.2 已执行过 schema 对比则跳过此步。否则同 Step 2.2 直接提交对比代码给 DolphinX 平台执行。

AI 助手解析输出的三条标记段，向用户展示完整结构对比（含列顺序和列索引）：

CSV 与表结构对比：
  目标表列顺序 (14): TradeDate, ChannelNo, ApplSeqNum, ...
  CSV 列顺序 (13):   ChannelNo, ApplSeqNum, ...

  → 需要前置添加的列 (1): TradeDate（位于表的第一列，CSV 中不存在）
  → 需要后置填充的列 (0):
  → 列顺序匹配: ✓（公共列相对顺序一致）

  匹配的列 (13): ChannelNo, ApplSeqNum, ...
  CSV 独有的列 (0):
  表有但 CSV 没有的列 (0):
  类型不匹配 (2):
    [4] SecurityIDSource: INT → SYMBOL（CSV 列索引 4）
    [7] ExecType: INT → SYMBOL（CSV 列索引 7）

  注：列索引基于 CSV 文件列顺序，从 0 开始。AI 助手通过 `csv_column_order` 列表确定各列索引。

如果 type_mismatches 为空，则直接跳到 Step 3.4 确认导入。

3.3 确定列类型转换方案

如果存在类型不匹配，AI 助手根据 csv_column_order（CSV 列顺序列表）找到每个不匹配列在 CSV 中的索引位置，然后按以下映射表确定 array(ANY) 转换表达式。不再使用 update 语句做类型转换（DDB 禁止 update 更改列类型）。

生成以下转换配置变量（带入 Step 4）：

• symbol_cols = [索引列表] — 需要 symbol() 转换的列索引，如 [4, 5, 7, 8, 9, 29, 30, 31, 35]
• date_cols = [(索引, "格式"), ...] — 需要 date(temporalParse()) 的列索引及格式，如 [(25, "yyyy-MM-dd"), (26, "yyyy-MM-dd")]
• long_cols = [索引列表] — 需要 long() 转换的列索引
• double_cols = [索引列表] — 需要 double() 转换的列索引

向用户展示转换方案：

自动类型转换方案：
  [4] SecurityIDSource: INT → SYMBOL
  [7] ExecType: INT → SYMBOL
  [25] 上市交易日: STRING → DATE

3.4 用户确认

用 AskUserQuestion 确认是否继续，展示完整的转换方案（包含类型转换和列忽略信息），options 如下：

• 继续导入
• 取消

用户选择"取消"则报告后终止；选择"继续导入"则将 symbol_cols、date_cols、leading_exprs、trailing_exprs 等配置带入 Step 4。

Step 4: 生成并执行导入脚本

4.1 生成导入脚本

写入 generated/import_{表名}_{时间戳}.dos。使用 array(ANY) 模式：

// import_{表名}_{时间戳}.dos — CSV 导入（array(ANY) 模式）
raw = loadText("{CSV路径}");
cols = raw.columnNames();
rows_before = exec count(*) from loadTable("{数据库路径}", "{表名}");

// 在 array(ANY) 循环中逐列进行类型转换
v = array(ANY, 0, {总列数});
for (i in 0..(cols.size()-1)) {
    if (i == {列A} || i == {列B}) {                // ← 替换为 symbol_cols
        v.append!(symbol(raw[cols[i]]));
    } else if (i == {列C} || i == {列D}) {           // ← 替换为 date_cols
        v.append!(date(temporalParse(raw[cols[i]], "格式")));
    } else if (i == {列E}) {                         // ← 替换为 double_cols
        v.append!(double(raw[cols[i]]));
    } else {
        v.append!(raw[cols[i]]);
    }
}

// 增量构建内存表
t = table(v[0] as `c0);
for (i in 1..(cols.size()-1)) {
    t["c" + string(i)] = v[i];
}

// 过滤分区列中的 NULL 行
t_valid = select * from t where isValid(c{分区列索引});
n_valid = t_valid.size();

// 分批导入，每批 100 万行
batch_size = 1000000;
batch_no = 0;
start = 0;
while (start < n_valid) {
    end = min(start + batch_size, n_valid);
    batch_no += 1;
    batch = select {前置列} * {后置列} from t_valid[start:end];
    loadTable("{数据库路径}", "{表名}").append!(batch);
    start = end;
}

rows_after = exec count(*) from loadTable("{数据库路径}", "{表名}");
print("=== IMPORT_RESULT ===");
print("total_rows=" + string(n_valid));
print("rows_before=" + string(rows_before));
print("rows_after=" + string(rows_after));
print("imported=" + string(rows_after - rows_before));
print("batches=" + string(batch_no));

模板参数填充规则：

• symbol_cols 转换为一组 || 条件，如 i == 4 || i == 5 || i == 7 || i == 8 || i == 9
• date_cols 转换为 else if 条件，每组日期格式不同则拆多个 else if；格式在 temporalParse() 中指定
• 分区列索引 — 在 csv_column_order 中查找用户确认的分区列名所在位置
• 前置列 — 有前置列时填写 date("2023.02.01") as TradeDate,（末尾带逗号和空格）；无则留空
• 后置列 — 有后置列时填写 , "batch001" as BatchID（开头带逗号和空格）；无则留空
• 既无前置也无后置列时，整个 select 改为 select * from t_valid[start:end]

关键注意事项：

• append! 按列位置匹配，不是按列名，因此 select 的列顺序必须等于目标表列顺序
• 日期格式必须与 CSV 中实际格式匹配（如 "yyyy-MM-dd"、"yyyyMMdd"、"yyyy.MM.dd"）
• 中文列名通过 cols[i] 索引访问，无需在代码中直接书写中文字符

4.2 执行导入

AI 助手将 .dos 提交给 DolphinX 平台执行，读取 === IMPORT_RESULT === 后的结果。

Step 5: 报告结果

必须按以下顺序执行全部 3 个子步骤，缺一不可：

5.1 展示导入结果摘要

解析导入脚本的输出，以文本形式向用户展示导入结果摘要：

导入结果：
  CSV 文件: /hdd/hdd1/feb2023_csv/20230201/mdl_4_19_0.csv
  目标表: dfs://DB_split_SH.TB_split_SH_entrust
  状态: 成功
  新增行数: 1,234,567
  批次数: 2

5.2 执行验证查询

直接提交以下验证查询代码给 DolphinX 平台执行：

print("=== VERIFY_DATA ===");
print("行数: " + string(exec count(*) from loadTable("{数据库路径}", "{表名}")));
print("前10行:");
select top 10 * from loadTable("{数据库路径}", "{表名}");

5.3 展示查询代码与结果

AI 助手读取 === VERIFY_DATA === 后的行数和前 10 行数据，将查询代码和结果一并展示给用户：

验证查询：
  print("=== VERIFY_DATA ===");
  print("行数: " + string(exec count(*) from loadTable("{数据库路径}", "{表名}")));
  print("前10行:");
  select top 10 * from loadTable("{数据库路径}", "{表名}");

已导入表前 10 行预览（查询结果）：
  TradeDate   SecurityID   LastPrice   ...
  2023.02.01  000001.SZ    10.25
  ...

注意事项

• CSV 路径：路径必须在 DDB 服务器上可访问
• 导入前请确保表已存在：如不存在请先调用 /dolphindb-finance-dbtbcreate
• 列不匹配：CSV 有但表没有的列会被忽略；表有但 CSV 没有的列在 select 中通过前置/后置表达式合成
• array(ANY) 模式：所有类型转换在 array(ANY) 循环中按列索引完成
• 中文列名：通过 cols[i] 位置索引访问，无需在代码中直接书写中文

产出文件清单