当电力市场碰上量化技术，如何构建高效电力交易解决方案？

电力交易行情数据的管理面临以下挑战：

• 行情数据清洗：参与电力交易的市场主体整体信息化水平处于初级阶段，数据清洗绝大多数都是人工进行的，每一天都需要手工进行一次行情数据清洗，耗费了大量的时间。
• 行情数据存储：数据存储绝大多数都是基于 Excel。市场初期，通过 Excel 等简单工具可以勉力维持，但随着电力现货市场的常态化运行、交易流动性的持续增加以及跨省交易的不断发展，行情数据的量级会呈爆炸式增长，以 Excel 为载体便无法满足需求。
• 实时行情数据接入：现阶段，由于法律法规和信息披露的要求，电力交易中心并未向市场主体开放实时行情数据的接入接口，市场主体无法像金融市场主体一样获取实时的行情数据，只能获取到历史数据。

因此，一套高效的数据清洗、数据存储解决方案，能够极大地提高研究员的效率，更加专注于因子和策略本身。

电力交易市场的行情数据是一种带时间戳的结构化数据，从划分上来说，行情数据属于时序数据。引入数据库是行情数据管理信息化最优的解决方案。

DolphinDB 作为一款基于高性能时序数据库，支持复杂分析与流处理的实时计算平台，集成了功能强大的编程语言和高吞吐率的流数据分析系统，为海量结构化数据的快速存储、检索、分析及计算提供一站式解决方案。

上述电力交易行情数据在数据清洗与存储时面对的问题，可由 DolphinDB 在存储、分析和计算上支持的以下特性，以及基于这些特性的组合使用得以解决：

• 强大的数据 ETL 工具：DolphinDB 内置1500+函数，提供了丰富的转换和清洗功能，支持数据格式化、缺失值填充、异常值处理、离散化处理、数据整合、数据过滤、聚合和排序等操作。这些功能使得电力行情数据清洗和转换变得简单而高效。
• 高效存储方式：数据分区列式存储。针对电力行情数据可采用时间 + 标的日的方式组合分区来达成灵活分区存储，对于不同分区数据可以多线程并行操作。
• 高效压缩算法：支持 lz4、Delta（delta-of-delta encoding） 压缩算法，大大提升压缩率，节约硬件资源。
• 高效排序算法：sortColumn 对分区内数据排序存储，快速定位数据在分区中的位置。

电力交易中心交易平台中，连续运营交易多档行情数据界面如下图所示：

上图中包含了申报日、标的日和市场行情等信息。正常情况下，交易申报日（T 日）为周一至周三时，连续运营日滚动交易标的为 T+3日至 T+5日每日24个时段的电量；交易申报日（T 日）为周四至周五时，连续运营日滚动交易标的为 T+3日至 T+6日每日24个时段的电量。上图中，申报日为3月5日，标的日为3月8日、3月9日，3月10日。数据中没有申报日、标的日、交易类型等字段。因此在入库时，需要进行清洗，将这些字段补全。此外，还需要将买卖三档的量价数据组织成 ArrayVector 形式。

导入脚本详见附录1。在脚本中，定义了 importData 函数。importData 接收一个字符串参数 agrs，用于指定导入行情数据文件的存放路径。importData 函数将行情数据 csv 文件导入成 DolphinDB 内存表，经过一系列的数据清洗，最终转换成标准的行情数据，并落库存储。其实现步骤如下：

• 将 csv 文件数据加载到 DolphinDB 内存表
• 数据清洗，处理行情数据的量和价
• 增加申报日、标的日两个字段

行情数据导入后，在 DolphinDB 中查看数据如下图所示：

行情数据导入后，基于行情数据，可以计算一些简单的因子，如买方最高价/最低价，卖方最高价/最低价，买方市场总量，卖方市场总量等因子。如下图所示：

本文是电力交易市场主体信息化的开篇之作。后续，我们计划逐步发布一系列深入文章和教程，涵盖如何基于 DolphinDB 挖掘电力交易因子、进行因子管理、因子相关性分析，以及如何基于 DolphinDB 进行电力交易策略开发和回测等。随着电力交易市场信息化进程的持续推进和市场法规的逐步完善，我们还将探讨如何利用 DolphinDB 接入实时市场数据，以提高整个交易流程的效率和流畅性，实现市场主体信息化的全面贯通，从而打通信息化建设的“最后一公里”。

def importData(filePath){
//filePath：文件路径
    schema = extractTextSchema(filePath)
    totalNum = schema.size()
    if(totalNum==13){
        schema["type"] = `STRING`DOUBLE`STRING join take(`DOUBLE,10)
    }else{
        schema["type"] = `STRING`DOUBLE`STRING join take(`DOUBLE,totalNum-3)
    }
    t = loadText(filePath,,schema)
    update t set col2 = "查看"

    //行号为奇数的是量，偶数的是价
    data = select *,rowNo(col3)+1 as RN from t[,3:11]
    //前8行为量，后8行为价
    tmp = table(matrix(select * from data where RN%2 = 1) join matrix( select * from data where RN%2 = 0))
    dropColumns!(tmp,`col8`col17)
    askAndBid = select col0 as 成交量含线损,col9 as 成交均价含线损,col1 as 未成交量含线损,col10 as 未成交均价含线损,
                        fixedLengthArrayVector(col2,col3,col4) as 买三档量,
                        fixedLengthArrayVector(col5,col6,col7) as 卖三档量,
                        fixedLengthArrayVector(col11,col12,col13) as 买三档价格,
                        fixedLengthArrayVector(col14,col15,col16) as 卖三档价格
                from tmp
    
    //行号为奇数的是一条
    tmp1 = select *,rowNo(col0)+1 as RN from t
    tmp2 = select col0 as 交易时段,col1 as 最新价,col2 as 趋势,col11 as 持有量,col12 as 未成交量 from tmp1 where RN%2 = 1
    tmp3 = tmp2 join askAndBid


    //为数据打上当前日期，交易日期，是否为分时用户
    Date = temporalParse(left(right(filePath,12),8),"yyyyMMdd")

    //确定连续滚动的日期范围
    if(Date in 2024.04.01..2024.04.03){
        tradeDates = Date + (3..7)
    }else if( Date in 2024.04.28..2024.04.30){
        tradeDates = Date + (3..9)
    }else if(Date == 2024.04.07){
        tradeDates = Date + (3..5)
    }else if(weekday(Date) in (1,2,3)){
        tradeDates = Date + (3..5)
    }else{
        tradeDates = Date + (3..6)
    }
    DateList = take(Date,tmp3.size())
    tradeDateList = stretch(tradeDates,tradeDates.size()*48)
    isTime = take(stretch(`分时`非分时,48),tradeDates.size()*48)
    res = table(DateList as 当前日期,tradeDateList as 交易日期,isTime as 是否分时) join tmp3
    newColumn = `当前日期`交易日期`是否分时`交易时段`最新价`成交量含线损`成交均价含线损`未成交量含线损`未成交均价含线损`买三档价格`卖三档价格`买三档量`卖三档量`持有量`未成交量`趋势
    res.reorderColumns!(newColumn)

    res = select 
            take(2024.04.07T10:00:09.000,res.size()) as time,
            当前日期 as date,
            交易日期 as tradeDate,
            是否分时 as type,
            交易时段 as symbol,
            最新价 as lastPrice,
            成交量含线损 as turnOver,
            成交均价含线损 as turnOverPrice,
            未成交量含线损	as turnDown,
            未成交均价含线损 as tunrnDownPrice,
            买三档价格 as askPrice,
            卖三档价格	 as bidPrice,
            买三档量 as askQyt,
            卖三档量 as bidQyt
          from res
    pt1 = loadTable("dfs://snapshot","data")
    pt1.append!(res)
}

//文件路径
filePath = "./20240407.csv"
importData(filePath)