easyTLDataImport 通联历史数据自动化导入功能模块

自 2.00.10 版本开始，DolphinDB 安装包的 server/modules 目录下已预装 easyTLDataImport 模块（模块版本 1.0）。

由于 1.0 版本的 easyTLDataImport 模块存在以下问题：

无法适配上交所从 2023.11.27 启用的逐笔合并数据文件（对应通联的 mdl_4_24_0.csv 文件）。

内存使用较高，至少需要 64 GB 以上内存才能执行导入任务。

对于导入任务的进度以及每天的任务执行结果无法监控。

模块结构不清晰，二次开发难度较大。

因此，开发了 2.0 版本的 easyTLDataImport 模块。对原来的模块结构进行了调整，对导入流程进行了优化。

自 2.00.15 和 3.00.3 版本开始，DolphinDB 安装包的 server/modules 目录下预装 easyTLDataImport 模块（模块版本 2.0）。

若需要安装 1.0 版本，也可以从 9.1 的附录章节下载对应的模块包。解压后，覆盖节点所在服务器上 [home]/modules 路径下的对应文件夹，从而对模块进行升级。（[home] 目录可以通过 getHomeDir 函数查看。）

本文将对 easyTLDataImport 模块（模块版本 2.0）进行详细介绍。

在使用本教程功能模块时，用户需要自行解压通联数据，并确保在主目录下创建了以日期命名的文件夹（"yyyyMMdd" 的日期格式）。每个日期目录下包含该日期的 .csv 文件。

例如，主目录是 level2，则文件结构可以设置如下：

|—— level2
|      |—— 20211201
|      |      |—— xxx.csv
|      |      |—— …
|      |—— 20211202
|      |      |—— xxx.csv
|      |      |—— …
|      |—— …

数据也可以按年组织，则文件结构如下
|—— level2
|      |—— 2021
|      |      |—— 20211201
|      |      |      |—— xxx.csv
|      |      |      |—— …
|      |      |—— …
|      |—— 2022
|      |      |—— 20221201
|      |      |      |—— xxx.csv
|      |      |      |—— …
|      |      |—— …
|      |—— …

数据也可以按年组织，则文件结构如下

|—— level2

|      |—— 2021

|      |      |—— 20211201

|      |      |      |—— xxx.csv

|      |      |      |—— …

|      |      |—— …

|      |—— 2022

|      |      |—— 20221201

|      |      |      |—— xxx.csv

|      |      |      |—— …

|      |      |—— …

|      |—— …

注意：日期这一层目录之前的文件结构和日期目录下的文件结构都没有要求。

easyTLDataImportPro 模块主要包含主函数（loadTLData.dos）、数据表结构 (tbSchema)、分布式库表创建 (createDbAndTb) 和数据导入 (loadData) 四个部分。

• 第一步：用户按照第 2 章文件结构中的要求解压并准备好数据。

假设数据放在 /hdd/hdd1/data/TLData/ 目录下，文件结构如下图：

• 第二步：将模块同步到 DolphinDB 的 sever/modules 的目录下。
  • VS Code 插件：连接对应的节点 ——> 右键 DolphinDBModules 文件夹 ——> 选择 DolphinDB: Upload Module
  • GUI 客户端：连接对应的节点 ——> 右键 DolphinDBModules 文件夹 ——> 选择 Synchronize module to server

• 第三步：载入模块和导入数据方法如下：

// 加载模块
use DolphinDBModules::easyTLDataImport::loadTLData

// 配置参数
fileDir = "/hdd/hdd1/data/TLData/"
dataSource = "TLAll"
dbName = {
	"TLSnapshot": "dfs://Level2",
	"TLEntrust": "dfs://Level2",
	"TLTrade": "dfs://Level2"
}
tableName = {
	"TLSnapshot": "snapshot",
	"TLEntrust": "entrust",
	"TLTrade": "trade"
}
isDataDuplicated = false
isConcatTime = false
isNeedOrderQueue = true

// 调用函数，导入通联数据
jobid = autoLoadTongLianData(fileDir=fileDir, dataSource=dataSource, dbName=dbName, tableName=tableName, 
                              isDataDuplicated=isDataDuplicated, isConcatTime=isConcatTime, isNeedOrderQueue=isNeedOrderQueue)

（1）使用以下代码可以查询后台任务状态：

getTLJobStatus(jobid)             // 函数名发生变更，1.0 版本中为 getJobStatus(jobid)

• jobid 为 autoLoadTongLianData 函数的返回值，对应 getRecentJobs() 结果里的 jobDesc 字段值。
• 通过函数里的参数 parallel 可以设置提交的后台任务的个数。
• 通过配置文件里的参数 maxBatchJobWorker 可以设置处理后台任务的最大工作线程数。
• 返回的结果表里，startTime 不是空值表示任务已经开始执行；endTime 不是空值表示任务已经结束执行。

（2）使用以下代码可以查询后台任务导入进度：

getTLJobDetails(jobid)          // 函数名发生变更，1.0 版本中为 getJobDetails(jobid)

• jobid 为 autoLoadTongLianData 函数的返回值，对应 getRecentJobs() 结果里的 jobDesc 字段值。
• 通过输出的信息，可以了解哪些日期的任务已经执行完毕，哪些日期的任务正在执行。

（3）使用以下代码可以查询导入子任务状态：

共享内存表 TLDataImportStatusTable 记录了查询导入子任务的状态。

select * from TLDataImportStatusTable

• 模块会自动生成一个共享表，记录每个导入子任务的状态。
• 脚本拆分的最小导入任务是：一天一个数据源的导入任务。
• 如果 TLDataImportStatusTable 没有对应的日期对应数据源的任务信息，表示该任务还没有结束。只有等一个最小导入任务执行结束，才会在共享表里写入任务状态。
• 每次执行 autoLoadTongLianData 函数都会重置任务状态的共享表 TLDataImportStatusTable。
• 会出现的报错信息详情可以查看 9.2 章节。

• fileDir

模块中对数据文件查找的方法如下：

（1）从 fileDir 目录下递归查找所有由 8 位数字组成的文件夹路径，即 "yyyyMMdd" 格式的日期文件夹。

（2）再将找到的所有日期和 startDate、endDate 进行字符串比较，从而达到导入数据日期筛选的目的。

（3）最后在每个日期文件夹下，递归查找所有以 csv 结尾的文件，从而获取该日期下的所有 .csv 文件的绝对路径。

因为第一步是查找所有 8 位数字组成的文件夹，所以参数 fileDir 指定的文件路径下必须有一层目录是以 "yyyyMMdd" 格式的日期命名。

所以如果用户想要导入一天的数据，需要通过 startDate、endDate 参数控制导入日期，而不是通过 fileDir 参数。

以图 4-1中的文件结构为例，假设只想导入 2021.12.01 这一天的数据，不是配置参数 fileDir 为 "/hdd/hdd1/data/TLData/20211201"，而是配置 startDate 和 endDate 为 "20211201"。

• isNeedOrderQueue

参数 isNeedOrderQueue 为 true 时，对最优报价 50 档委托队列的数据处理逻辑如下：

（1）先读取对应的委托队列的 .csv 文件到内存中。

（2）对委托队列的内存表进行格式处理，生成 OfferOrders 和 BidOrders 列分别存储卖方和买方的委托队列。

（3）用 textChunkDS 函数将快照数据的文件划分为 512 MB 大小的数据源。

（4）通过 mr 函数对快照数据源和全量的委托队列数据进行左连接，连接列为 "TradeTime", "SecurityID", "ImageStatus"。并将关联后的结果写入对应的数据库。

因为需要将委托队列的数据全量读到内存并进行关联操作，所以这种方式会使用更多的内存。

可用内存小于 64 GB 的不推荐设置参数isNeedOrderQueue 为 true。

• isDataDuplicated

测试发现，2016 年左右的数据中，快照数据和 50 档委托数据里都会有大量的重复数据（以"TradeTime", "SecurityID", "ImageStatus"为键值的重复数据）。

如果直接做左连接，会让结果表的数据量大幅增加，导致内存使用暴增，经常出现 OOM。

因此导入 2016 年左右的快照数据且需要合并最优报价的 50 档委托数据（即参数 isNeedOrderQueue 为 true）时，推荐设置参数 isDataDuplicated 为 true。

导入近期的快照行情数据时，推荐设置参数 isDataDuplicated 为 false。

• parallel

当 parallel 参数设置很大时，可能会出现后台导入任务部分处于等待状态并未执行的情况，这可能与配置文件里的 maxBatchJobWorker 参数有关。函数里的 parallel 参数只能设置提交的后台任务的个数。同时执行的后台任务个数需要通过配置文件里的 maxBatchJobWorker 参数进行设置。

parallel 越大使用的内存会越多。在资源有限的情况下，parallel 参数不宜设置的过大。参数 parallel 对导入性能以及资源使用的影响可以参考 7.2 章节。

目前，easyTLDataImport 模块只支持导入快照行情、逐笔成交和逐笔委托。如果想要基于 easyTLDataImport 模块导入更多数据源的数据，用户可以按照以下流程对模块进行二次开发。

模块结构可以查看第 3 章节。下面将增加 dataSource="TLIndex" 的数据源，使模块能导入上交所指数数据（Index.csv）。

（1）在 DolphinDBModules/easyTLDataImport/tbSchema/csvSchema 文件夹下新增一个 indexCsvSchema.dos 文件，用于指定目标导入数据的结构。

module DolphinDBModules::easyTLDataImport::tbSchema::csvSchema::indexCsvSchema

def indexCsvSchema(){
	name = ["UpdateTime", "DataStatus", "SecurityID", "PreCloseIndex", "OpenIndex", "Turnover", "HighIndex", 
            "LowIndex", "LastIndex", "TradTime", "TradVolume", "CloseIndex", "LocalTime", "SeqNo"]
	typeString = ["TIME", "INT", "SYMBOL", "DOUBLE", "DOUBLE", "DOUBLE", "DOUBLE", 
                "DOUBLE", "DOUBLE", "TIME", "DOUBLE", "DOUBLE", "TIME", "LONG"]
	return table(name, typeString)
}

（2）在 DolphinDBModules/easyTLDataImport/tbSchema/dfsSchema 文件夹下新增一个 indexSchema.dos 文件，用于指定分布式表的表结构。

module DolphinDBModules::easyTLDataImport::tbSchema::dfsSchema::indexSchema

def indexSchemaTb(isConcatTime){
	if(isConcatTime==true){
        name = ["TradeTime", "DataStatus", "SecurityID", "PreCloseIndex", "OpenIndex", "Turnover", "HighIndex", 
                "LowIndex", "LastIndex", "TradTime", "TradVolume", "CloseIndex", "LocalTime", "SeqNo"]
        typeString = ["TIMESTAMP", "INT", "SYMBOL", "DOUBLE", "DOUBLE", "DOUBLE", "DOUBLE", 
                        "DOUBLE", "DOUBLE", "TIME", "DOUBLE", "DOUBLE", "TIME", "LONG"]
    }else{
        name = ["TradeDate", "TradeTime", "DataStatus", "SecurityID", "PreCloseIndex", "OpenIndex", "Turnover", "HighIndex", 
                "LowIndex", "LastIndex", "TradTime", "TradVolume", "CloseIndex", "LocalTime", "SeqNo"]
        typeString = ["DATE", "TIME", "INT", "SYMBOL", "DOUBLE", "DOUBLE", "DOUBLE", "DOUBLE", 
                    "DOUBLE", "DOUBLE", "TIME", "DOUBLE", "DOUBLE", "TIME", "LONG"]
    }
	return table(1:0, name, typeString)
}

（3）在 DolphinDBModules/easyTLDataImport/createDbAndTb/createDB.dos 的 createDb 函数里增加创建数据库的语句。

def createDb(dbName, dataSource, merge){
    if(dataSource in ["TLSnapshot", "TLTrade", "TLEntrust"]){
        // ... ...
    }else if(dataSource=="TLIndex"){                            // 新增  TLIndex 的建库语句
        db = database(dbName, VALUE, 2021.01.01..2021.01.03)      
    }
}

（4）在 DolphinDBModules/easyTLDataImport/createDbAndTb/createTB.dos 的 createTb 函数里增加创建分布式表的语句。

use DolphinDBModules::easyTLDataImport::tbSchema::dfsSchema::indexSchema   // 新增

def createTb(dbName, tbName, dataSource, market, isConcatTime){
    db = database(dbName)
    // Handle different data types for the table
    if(dataSource=="TLSnapshot"){
       // ... ...
    }else if(dataSource=="TLEntrust"){
       // ... ...
    }else if(dataSource=="TLTrade"){
       // ... ...
    }else if(dataSource=="TLIndex"){                                      // 新增  TLIndex 的建表语句
        t = indexSchemaTb(isConcatTime)
        partitionColumns=iif(isConcatTime, `TradeTime, `TradeDate)
        db.createPartitionedTable(table=t, tableName=tbName, partitionColumns=partitionColumns)
    }
}

（5）在 DolphinDBModules/easyTLDataImport/loadData/loadOneDayData 文件夹下新增一个 loadOneDayIndex.dos 文件，它是包含数据转化规则的导入脚本。

module DolphinDBModules::easyTLDataImport::loadData::loadOneDayData::loadOneDayIndex

use DolphinDBModules::easyTLDataImport::tbSchema::csvSchema::indexCsvSchema

def appendIndex(day, t, dbName, tbName, isConcatTime){
    // 转换数据格式
    tmpData = select day as TradeDate, UpdateTime as TradeTime, DataStatus as DataStatus, SecurityID as SecurityID,PreCloseIndex as PreCloseIndex,
                    OpenIndex as OpenIndex, Turnover as Turnover,HighIndex as HighIndex,LowIndex as LowIndex,LastIndex as LastIndex,
                    TradTime as TradTime,TradVolume as TradVolume,CloseIndex as CloseIndex,LocalTime as LocalTime,SeqNo as SeqNo 
                from t
    if(isConcatTime==true){
        tmpData.replaceColumn!(`TradeTime, concatDateTime(tmpData["TradeDate"], tmpData["TradeTime"]))
        tmpData.dropColumns!(`TradeDate)
    }

	// 写入数据库
	loadTable(dbName, tbName).append!(tmpData)
}

def loadOneDayIndexSH(csvFiles, day, dbName, tbName, isConcatTime){
	csvNames = "Index.csv"
	csvPath = csvFiles[endsWith(csvFiles, csvNames)][0]
    schemaTB = indexCsvSchema()
	try{
		ds = textChunkDS(filename=csvPath, chunkSize=512, schema=schemaTB)
		mr(ds, appendIndex{day, , dbName, tbName, isConcatTime}, parallel=false)
	}catch(ex){
		errorMessage = dict(`message`code, [string(ex[0])+":"+string(ex[1]), "error"])
		return errorMessage
	}
	return true
}

（6）在 DolphinDBModules/easyTLDataImport/loadData/loadOneDayData.dos 的 loadOneDayDataSH 函数里增加指数数据的导入规则。

use DolphinDBModules::easyTLDataImport::loadData::loadOneDayData::loadOneDayIndex  // 新增

def loadOneDayDataSH(csvFilesSH, day, dbNames, tbNames, merge, dtypes, isConcatTime, isDataDuplicated, isNeedOrderQueue) {
    csvFiles = csvFilesSH
    statTable = objByName(`TLDataImportStatusTable)     // 状态监控表
    // 定义数据源和导入函数的映射关系
    if("mdl_4_24_0.csv" in each(last, csvFiles.split("/"))){
        loadFuncs = ["TLSnapshot", "TLTradeEntrust", "TLIndex"]         // 新增  TLIndex 数据源
        loadTypes = ["TLSnapshot", ["TLTrade", "TLEntrust"], "TLIndex"] // 新增  TLIndex 数据源
    }else{
        loadFuncs = ["TLSnapshot", "TLEntrust", "TLTrade", "TLIndex"]   // 新增  TLIndex 数据源
        loadTypes = ["TLSnapshot", "TLEntrust", "TLTrade", "TLIndex"]   // 新增  TLIndex 数据源
    }
    // 串行导入多天的数据
    for(i in 0:size(loadFuncs)){       // i = 0
        if(any(loadTypes[i] in dtypes)){
            dtype = loadTypes[i]
            loadFunc = loadFuncs[i]
            if(loadFunc=="TLTradeEntrust"){
                dbName = dbNames
                tbName = tbNames
            }else{
                dbName = dbNames[dtypes==dtype][0]
                tbName = tbNames[dtypes==dtype][0]
            }
            // 导入开始时间
            start_time = now()
            // 使用新增的自定义函数导入指数数据
            if(loadFunc=="TLSnapshot"){
                // ... ... 
            }else if(loadFunc=="TLEntrust"){
                // ... ... 
            }else if(loadFunc=="TLTrade"){
                // ... ... 
            }else if(loadFunc=="TLTradeEntrust"){
                // ... ... 
            }else if(loadFunc=="TLIndex"){                              // 新增  TLIndex 数据源
                errorMessage = loadOneDayIndexSH(csvFiles, day, dbName, tbName, isConcatTime)
            }
            // 导入结束时间
            end_time = now()
            // 向子任务状态表记录任务情况
            try{
                if(typestr(errorMessage)=="BOOL")   errorMsg = ""
                else                                errorMsg = toStdJson(errorMessage)
                insert into statTable values(concat(dtype, ""), "SH", day, start_time, end_time, errorMsg)
            }catch(ex){print(ex)}
        }
    }
}

（7）在 DolphinDBModules/easyTLDataImport/loadTLData.dos 的主函数里增加新的数据源类型 "TLIndex"。

def autoLoadTongLianData(fileDir, dataSource, 
						dbName="dfs://TL_Level2", tableName=NULL, market="ALL", 
						startDate=NULL, endDate=NULL, parallel=1, initialDB=false, initialTB=false,
						isDataDuplicated=true, isConcatTime=true, isNeedOrderQueue=true){
    // ... ... (无需修改)
    
	// 对 dataSource 做参数校验时，增加数据源 TLIndex
	if(!(string(dataSource) in ["TLAll", "TLTradeEntrust", "TLSnapshot", "TLEntrust", "TLTrade", "TLIndex"])){
		errorMessage = dict(`message`code, ["数据源 [" + string(dataSource) + "] 暂时不支持", "error"])
		return printLog(errorMessage)
	}

	// 对数据类型做处理时，根据需求增加数据源 TLIndex 
	if(dataSource=="TLTradeEntrust")	dtypes = ["TLTrade", "TLEntrust"]
	else if(dataSource=="TLAll")		dtypes = ["TLSnapshot", "TLTrade", "TLEntrust", "TLIndex"]
	else								dtypes = [dataSource]

    // ... ... (无需修改)
}

（8）重新同步模块后，即可导入上交所指数数据。（模块同步步骤查看第 4 章节）

// 加载模块
use DolphinDBModules::easyTLDataImport::loadTLData

// 配置参数
fileDir = "/hdd/hdd1/data/TLData/"
dataSource = "TLIndex"
dbName = {"TLIndex": "dfs://Level2Index"}
tableName = {"TLIndex": "index"}
isDataDuplicated = false
isConcatTime = false

// 调用函数，导入通联数据
jobid = autoLoadTongLianData(fileDir=fileDir, dataSource=dataSource, dbName=dbName, tableName=tableName, 
                              isDataDuplicated=isDataDuplicated, isConcatTime=isConcatTime)