Python Parser 在金融量化分析场景的应用入门

在用 Python Parser 进行因子计算代码开发前，必须先把历史行情数据存入 DolphinDB，包括日 K 线、分钟 K 线、逐笔数据和 Level-1/Level-2 快照数据等。

关于历史数据存储的问题，DolphinDB 已经发布多篇相关教程，在此不再赘述：

• 通用行情数据导入教程：《国内股票行情数据导入实例》、《金融 PoC 用户历史数据导入指导手册之股票 level2 逐笔篇》
• 通联历史行情数据导入教程：《DolphinDBModules::easyTLDataImport 通联历史数据自动化导入功能模块使用教程》
• 文本数据导入教程：《数据导入教程》

上述教程代码都基于 DolphinDB 语法开发，在 DolphinDB 支持的编程 IDE 中执行代码时要选择 DolphinDB 语法解释器。选择方式如下：

1. GUI 客户端：建议下载官网的最新版本：DolphinDB GUI。

   使用前，应在 DolphinDB GUI 中确保语言下拉菜单已开启：点击 File"../modules/easyTLDataImport/easytl_data_import.md"guage dropdown** 复选框。

   
   

   此后，即可在 GUI 用户界面中的语言选择菜单中选择 DolphinDB 语法解释器。

   
   

2. Visual Studio Code（VS Code）编辑器 + DolphinDB VS Code 插件：建议下载最新版 VS Code 插件：Visual Studio Code extension for DolphinDB。本教程基于 V2.0.1041 开发。

   在 VS Code 的 Settings 界面搜索 @ext:dolphindb.dolphindb-vscode connections 后，编辑 settings.json，在 dolphindb.connections 一节中，指定“python”条目。

   "python": true 表示使用 Python Parser 语法解释器；"python": false 表示使用 DolphinDB 语法解释器。

   
   

为了能够成功运行本章节示例代码，可以先执行下述代码模拟生成少量逐笔成交数据：（需要将 csvDir 变量修改为 DolphinDB 部署服务器上 tradeData.csv 的实际目录，示例数据见附件）

def createTB():
    dbName, tbName = "dfs://TL_Level2", "trade"
    # 数据库如果存在，删除该数据库
    if existsDatabase(dbName):
        dropDatabase(dbName)
    # 创建数据库：时间维度按天 VALUE 分区 + 股票代码 SecurityID HASH 50 分区
    db1 = database("", ddb.VALUE, seq(2020.01.01, 2021.01.01))
    db2 = database("", ddb.HASH, [ddb.SYMBOL, 50].toddb())
    db = database(dbName, ddb.COMPO, [db1, db2].toddb(), engine="TSDB")
    schemaTB = table(array(ddb.INT, 0) as ChannelNo,
                    array(ddb.LONG, 0) as ApplSeqNum,
                    array(ddb.SYMBOL, 0) as MDStreamID,
                    array(ddb.LONG, 0) as BidApplSeqNum,
                    array(ddb.LONG, 0) as OfferApplSeqNum,
                    array(ddb.SYMBOL, 0) as SecurityID,
                    array(ddb.SYMBOL, 0) as SecurityIDSource,
                    array(ddb.DOUBLE, 0) as TradePrice,
                    array(ddb.LONG, 0) as TradeQty,
                    array(ddb.SYMBOL, 0) as ExecType,
                    array(ddb.TIMESTAMP, 0) as TradeTime,
                    array(ddb.TIME, 0) as LocalTime,
                    array(ddb.LONG, 0) as SeqNo,
                    array(ddb.INT, 0) as DataStatus,
                    array(ddb.DOUBLE, 0) as TradeMoney,
                    array(ddb.SYMBOL, 0) as TradeBSFlag,
                    array(ddb.LONG, 0) as BizIndex,
                    array(ddb.SYMBOL, 0) as OrderKind,
                    array(ddb.SYMBOL, 0) as Market)
    db.createPartitionedTable(schemaTB, tbName, partitionColumns=["TradeTime", "SecurityID"].toddb(), compressMethods={"TradeTime":"delta"}.toddb(), sortColumns=["SecurityID", "TradeTime"].toddb(), keepDuplicates=ddb.ALL)

def loadData(csvDir):
    # 创建存储逐笔成交的库表
    createTB()

    # 读示例数据
    name = ["ChannelNo", "ApplSeqNum", "MDStreamID", "BidApplSeqNum", "OfferApplSeqNum", "SecurityID", "SecurityIDSource", "TradePrice", "TradeQty", "ExecType", "TradeTime", "LocalTime", "SeqNo", "DataStatus", "TradeMoney", "TradeBSFlag", "BizIndex", "OrderKind", "Market"].toddb()
    type = ["INT", "LONG", "SYMBOL", "LONG", "LONG", "SYMBOL", "SYMBOL", "DOUBLE", "LONG", "SYMBOL", "TIMESTAMP", "TIME", "LONG", "INT", "DOUBLE", "SYMBOL", "LONG", "SYMBOL", "SYMBOL"].toddb()
    t = loadText(csvDir, schema=table(name, type))

    # append! 数据入库
    loadTable("dfs://TL_Level2", "trade").append!(t)

    # 统计库内数据量
    rowCount = select count(*) from loadTable("dfs://TL_Level2", "trade")      #	181,683
    print(rowCount)

# 执行 loadData 函数, 需要将 csvDir 变量修改为 ddb 部署服务器上 csv 的实际目录
csvDir = "/home/v2/下载/data/tradeData.csv"
loadData(csvDir)

本教程中的所有示例代码都是用 Python Parser 语法开发的，所以必须选择 Python Parser 语法解释器执行，GUI 编程工具解释器选择界面如下：

通过因子存储最佳实践，日频因子存储库表创建示例代码如下：

import pandas as pd
import dolphindb as ddb

dbName, tbName = "dfs://dayFactorDB", "dayFactorTB"

# 数据库如果存在，删除该数据库
if existsDatabase(dbName):
    dropDatabase(dbName)

# 创建数据库：时间维度按年 RANGE 分区 + 因子名维度 VALUE 分区
db1 = database("", ddb.RANGE, date(datetimeAdd(1980.01M,seq(0,80)*12,'M')))
db2 = database("", ddb.VALUE, ["f1","f2"].toddb())
db = database(dbName, ddb.COMPO, [db1, db2].toddb(), engine='TSDB', atomic='CHUNK')

# 创建分区表
schemaTB = table(array(ddb.DATE, 0) as tradetime, 
                array(ddb.SYMBOL, 0) as securityid, 
                array(ddb.SYMBOL, 0) as factorname, 
                array(ddb.DOUBLE, 0) as value)

db.createPartitionedTable(schemaTB, tbName, partitionColumns=["tradetime", "factorname"].toddb(), compressMethods={"tradetime":"delta"}.toddb(), 
                        sortColumns=["securityid", "tradetime"].toddb(), keepDuplicates=ddb.ALL, sortKeyMappingFunction=[lambda x:hashBucket(x, 500)].toddb())

# 查看分区表结构
pt = loadTable(dbName, tbName)
pt.schema()

存储结构

• 对于多因子数据场景，用户可以选择窄表和宽表两种模式。
• 窄表模式把因子名存为一列，而宽表模式则把每个因子存为一列。
• 经测试在数据查询方面两种模式相差不大，但在增加因子、删除因子和更新因子等数据运维方面窄表模式性能显著优于宽表模式。
• 本教程示例代码为窄表存储最佳方案。

分区存储

• DolphinDB 支持数据分区存储，对于不同分区数据支持多线程并行写入，对于同一个分区数据支持并发查询。
• 经测试日频因子数据采用“时间维度按年 + 因子名”的组合分区方式存储，综合性能最佳。

分区内分组排序存储

• DolphinDB 的 TSDB 存储引擎提供排序键设置，每一个分区的数据写在一个或多个 level file 中，每一个 level file 内部的数据按照指定的列进行排序且创建索引。
• 排序列中除了最后一列的其他列通常为在点查中过滤条件会用到的列，其唯一值组合称为 SortKeys。
• 为保证性能最优，每个分区的 SortKeys 建议不超过 1000 个。当 SortKeys 较多时，可以通过设置 sortKeyMappingFunction 对 SortKeys 降维。
• 经测试日频因子数据采用“securityid+tradetime”的方式进行排序，sortKeyMapping 设置为 500 综合性能最佳。
• DolphinDB 默认数据存储的压缩算法为 lz4，对于时间、日期类型的数据，建议指定采用 delta 压缩算法存储，提高存储的压缩比。

基于逐笔成交数据计算当日尾盘成交占比因子的计算公式如下：

其中 BCVP_t 表示 t 日期的尾盘成交占比；Vol_t 表示 t 日期的总成交量；Vol_{t,14:30-15:00} 表示 t 日期的 14:30—15:00 的成交量之和。

因子计算示例代码如下：

import pandas as pd
import dolphindb as ddb

# 定义因子函数
def beforeClosingVolumePercent(trade):
    tradeTime = trade["TradeTime"].astype(ddb.TIME)
    beforeClosingVolume = trade["TradeQty"][(tradeTime >= 14:30:00.000)&(tradeTime <= 15:00:00.000)].sum()
    totalVolume = trade["TradeQty"].sum()
    res = beforeClosingVolume / totalVolume
    return pd.Series([res], ["BCVP"])

# 指定计算某一天的因子
tradeTB = loadTable("dfs://TL_Level2", "trade")
df = pd.DataFrame(tradeTB, index="Market", lazy=True)
res = df[df["TradeTime"].astype(ddb.DATE)==2023.02.01][["TradeTime", "SecurityID", "TradeQty"]].groupby(["SecurityID"]).apply(beforeClosingVolumePercent)

示例代码解析：

• 因子函数 beforeClosingVolumePercent 的函数体是按照 Python 语法进行开发的。
• tradeTime = trade["TradeTime"].astype(ddb.TIME) 逐笔成交数据中的 TradeTime 以时间戳 TIMESTAMP 的数据类型存入数据库。数据类型的转化可以使用 astype() 函数。其中的类型关键字，需要加上 ddb 的前缀。
• DolphinDB 中的时间常量的格式：yyyy.MM.ddTHH:mm:ss.SSS，比如：14:30:00.000、2023.02.01、2023.02.01T14:30:00.000
• tradeTB = loadTable("dfs://TL_Level2", "trade") 通过 loadTable 函数，将 "dfs://TL_Level2" 数据库下的分布式表 "trade" 的元数据取回到内存。此时变量 tradeTB 只包含元数据，库内数据并未取到内存。
• df = pd.DataFrame(tradeTB, index="Market", lazy=True) 通过 pd.DataFrame() 函数，将 DolphinDB 的表转化为数据框。对于分布式表而言，index 为必填参数，可以指定表中的任意一列，该列仅作为索引，后续可以不参与计算；lazy 参数指定计算是否立即执行，必须指定为 True，表示该 DataFrame 会存储所有函数调用，尽可能延迟计算，以减少计算带来的性能消耗。
• 可以通过 df[过滤条件] 的形式选出库内指定范围的数据。比如：df[(df["TradeTime"].astype(ddb.DATE)==2023.02.01)&(df["SecurityID"]=="000001")] 指定取库内 2023.02.01 的“000001“这一天一只股票的数据。
• 建议在 groupby 执行计算函数之前，先对数据列进行过滤，只取出计算需要的列。可以降低内存使用，减少数据读取与拷贝的开销。
• 可以通过 .groupby(分组列).apply(函数) 的方式实现分组计算，Python Parser 内部对 groupby.apply 实现了并行计算。

上一节中的计算结果 res 是当前会话中的一个本地内存变量，关闭会话便会被释放，所以需要把计算结果存入数据库中的分区表中，达到持久化的目的。

把因子计算结果从内存存储到磁盘上的示例代码如下：

# res 是一个 Series, 需要将计算结果转化成窄表格式（共4列：tradetime, securityid, factorname, value）
result = res.reset_index().rename(columns={"SecurityID":"securityid"})
result["tradetime"] = 2023.02.01
result = result.melt(id_vars=["tradetime", "securityid"],value_vars=["BCVP"],var_name="factorname",value_name="value")

# 存入数据库
loadTable(dbName, tbName).append!(result.to_table())

# 查看库内数据量
select count(*) from loadTable(dbName, tbName)

示例代码解析：

• 因子库用窄表方式存储，需要先将 DataFrame 转化为和因子库统一的 4 列的格式：tradetime, securityid, factorname, value。
• 因为上述示例只计算了一天的因子，结果中没有日期信息，所以需要增加日期信息。
• 使用 melt 函数，将宽表转化为窄表。当计算出多个因子值时，可以修改参数 value_vars=[“factorname1“, “factorname2“, …]。
• result.to_table() 用 .to_table() 函数将 no-lazy 模式的 DataFrame 转化为 DolphinDB 中的内存表。
• 通过 append! 函数，将因子结果存入数据库。

日频因子数据库和分区表的创建示例代码：

import pandas as pd
import dolphindb as ddb

dbName, tbName = "dfs://dayFactorDB", "dayFactorTB"

# 数据库如果存在，删除该数据库
if existsDatabase(dbName):
    dropDatabase(dbName)

# 创建数据库：时间维度按年 RANGE 分区 + 因子名 VALUE 分区
db1 = database("", ddb.RANGE, date(datetimeAdd(1980.01M,seq(0,80)*12,'M')))
db2 = database("", ddb.VALUE, ["f1","f2"].toddb())
db = database(dbName, ddb.COMPO, [db1, db2].toddb(), engine='TSDB', atomic='CHUNK')

# 创建分区表
schemaTB = table(array(ddb.DATE, 0) as tradetime, 
                array(ddb.SYMBOL, 0) as securityid, 
                array(ddb.SYMBOL, 0) as factorname, 
                array(ddb.DOUBLE, 0) as value)

db.createPartitionedTable(schemaTB, tbName, partitionColumns=["tradetime", "factorname"].toddb(), compressMethods={"tradetime":"delta"}.toddb(), 
                        sortColumns=["securityid", "tradetime"].toddb(), keepDuplicates=ddb.ALL, sortKeyMappingFunction=[lambda x:hashBucket(x, 500)].toddb())

1 分钟频因子数据库和分区表的创建示例代码：

import pandas as pd
import dolphindb as ddb

dbName, tbName = "dfs://minuteFactorDB", "minuteFactorTB"

# 数据库如果存在，删除该数据库
if existsDatabase(dbName):
    dropDatabase(dbName)

# 创建数据库：时间维度按天 VALUE 分区 + 因子名 VALUE 分区
db1 = database("", ddb.VALUE, seq(2021.01.01, 2021.12.31))
db2 = database("", ddb.VALUE, ["f1","f2"].toddb())
db = database(dbName, ddb.COMPO, [db1, db2].toddb(), engine='TSDB', atomic='CHUNK')

# 创建分区表
schemaTB = table(array(ddb.DATE, 0) as tradetime, 
                array(ddb.SYMBOL, 0) as securityid, 
                array(ddb.SYMBOL, 0) as factorname, 
                array(ddb.DOUBLE, 0) as value)

db.createPartitionedTable(schemaTB, tbName, partitionColumns=["tradetime", "factorname"].toddb(), compressMethods={"tradetime":"delta"}.toddb(), 
                        sortColumns=["securityid", "tradetime"].toddb(), keepDuplicates=ddb.ALL, sortKeyMappingFunction=[lambda x:hashBucket(x, 500)].toddb())

10 分钟频因子数据库和分区表的创建示例代码：

import pandas as pd
import dolphindb as ddb

dbName, tbName = "dfs://tenMinutesFactorDB", "tenMinutesFactorTB"

# 数据库如果存在，删除该数据库
if existsDatabase(dbName):
    dropDatabase(dbName)

# 创建数据库：时间维度按月 VALUE 分区 + 因子名 VALUE 分区
db1 = database("", ddb.VALUE, seq(2023.01M, 2023.06M))
db2 = database("", ddb.VALUE, ["f1","f2"].toddb())
db = database(dbName, ddb.COMPO, [db1, db2].toddb(), engine='TSDB', atomic='CHUNK')

# 创建分区表
schemaTB = table(array(ddb.DATE, 0) as tradetime, 
                array(ddb.SYMBOL, 0) as securityid, 
                array(ddb.SYMBOL, 0) as factorname, 
                array(ddb.DOUBLE, 0) as value)

db.createPartitionedTable(schemaTB, tbName, partitionColumns=["tradetime", "factorname"].toddb(), compressMethods={"tradetime":"delta"}.toddb(), 
                        sortColumns=["securityid", "tradetime"].toddb(), keepDuplicates=ddb.ALL, sortKeyMappingFunction=[lambda x:hashBucket(x, 500)].toddb())

3 秒快照频因子数据库和分区表的创建示例代码：

import pandas as pd
import dolphindb as ddb

dbName, tbName = "dfs://level2FactorDB", "level2FactorTB"

# 数据库如果存在，删除该数据库
if existsDatabase(dbName):
    dropDatabase(dbName)

# 创建数据库：时间维度按天 VALUE 分区 + 因子名 VALUE 分区
db1 = database("", ddb.VALUE, seq(2022.01.01, 2022.12.31))
db2 = database("", ddb.VALUE, ["f1","f2"].toddb())
db = database(dbName, ddb.COMPO, [db1, db2].toddb(), engine='TSDB', atomic='CHUNK')

# 创建分区表
schemaTB = table(array(ddb.DATE, 0) as tradetime, 
                array(ddb.SYMBOL, 0) as securityid, 
                array(ddb.SYMBOL, 0) as factorname, 
                array(ddb.DOUBLE, 0) as value)

db.createPartitionedTable(schemaTB, tbName, partitionColumns=["tradetime", "factorname"].toddb(), compressMethods={"tradetime":"delta"}.toddb(), 
                        sortColumns=["securityid", "tradetime"].toddb(), keepDuplicates=ddb.ALL, sortKeyMappingFunction=[lambda x:hashBucket(x, 500)].toddb())

逐笔频因子数据库和分区表的创建示例代码：

import pandas as pd
import dolphindb as ddb

dbName, tbName = "dfs://tickFactorDB", "tickFactorTB"

# 数据库如果存在，删除该数据库
if existsDatabase(dbName):
    dropDatabase(dbName)

# 创建数据库：时间维度按天 VALUE 分区 + 因子名 VALUE 分区 + 股票代码 HASH 10 分区
db1 = database("", ddb.VALUE, seq(2022.01.01, 2022.12.31))
db2 = database("", ddb.VALUE, ["f1","f2"].toddb())
db3 = database("", ddb.HASH, [ddb.SYMBOL, 10].toddb())
db = database(dbName, ddb.COMPO, [db1, db2, db3].toddb(), engine='TSDB', atomic='CHUNK')

# 创建分区表
schemaTB = table(array(ddb.DATE, 0) as tradetime, 
                array(ddb.SYMBOL, 0) as securityid, 
                array(ddb.SYMBOL, 0) as factorname, 
                array(ddb.DOUBLE, 0) as value)

db.createPartitionedTable(schemaTB, tbName, partitionColumns=["tradetime", "factorname", "securityid"].toddb(), compressMethods={"tradetime":"delta"}.toddb(), 
                        sortColumns=["securityid", "tradetime"].toddb(), keepDuplicates=ddb.ALL)

1 秒频因子数据库和分区表的创建示例代码：

import pandas as pd
import dolphindb as ddb

dbName, tbName = "dfs://secondFactorDB", "secondFactorTB"

# 数据库如果存在，删除该数据库
if existsDatabase(dbName):
    dropDatabase(dbName)

# 创建数据库：时间维度按小时 VALUE 分区 + 因子名 VALUE 分区
db1 = database("", ddb.VALUE, seq(datehour(2022.01.01T00:00:00), datehour(2022.01.31T00:00:00)))
db2 = database("", ddb.VALUE, ["f1","f2"].toddb())
db = database(dbName, ddb.COMPO, [db1, db2].toddb(), engine='TSDB', atomic='CHUNK')

# 创建分区表
schemaTB = table(array(ddb.DATE, 0) as tradetime, 
                array(ddb.SYMBOL, 0) as securityid, 
                array(ddb.SYMBOL, 0) as factorname, 
                array(ddb.DOUBLE, 0) as value)

db.createPartitionedTable(schemaTB, tbName, partitionColumns=["tradetime", "factorname"].toddb(), compressMethods={"tradetime":"delta"}.toddb(), 
                        sortColumns=["securityid", "tradetime"].toddb(), keepDuplicates=ddb.ALL, sortKeyMappingFunction=[lambda x:hashBucket(x, 500)].toddb())

期货 500 毫秒频因子数据库和分区表的创建示例代码：

import pandas as pd
import dolphindb as ddb

dbName, tbName = "dfs://futuresFactorDB", "futuresFactorTB"
# 数据库如果存在，删除该数据库
if existsDatabase(dbName):
    dropDatabase(dbName)

# 创建数据库：时间维度按天 VALUE 分区 + 因子名 VALUE 分区
db1 = database("", ddb.VALUE, seq(2021.01.01, 2021.12.31))
db2 = database("", ddb.VALUE, ["f1","f2"].toddb())
db = database(dbName, ddb.COMPO, [db1, db2].toddb(), engine='TSDB', atomic='CHUNK')

# 创建分区表
schemaTB = table(array(ddb.DATE, 0) as tradetime, 
                array(ddb.SYMBOL, 0) as securityid, 
                array(ddb.SYMBOL, 0) as factorname, 
                array(ddb.DOUBLE, 0) as value)

db.createPartitionedTable(schemaTB, tbName, partitionColumns=["tradetime", "factorname"].toddb(), compressMethods={"tradetime":"delta"}.toddb(), 
                        sortColumns=["securityid", "tradetime"].toddb(), keepDuplicates=ddb.ALL, sortKeyMappingFunction=[lambda x:hashBucket(x, 500)].toddb())

双均线就是根据两条周期不一样的均线的相对位置来跟踪趋势的一种方法，本教程我们使用双均线交叉的方式来判断买卖方向：

• 短周期均线从下向上突破长周期均线时，意味着当前时间段具有上涨趋势，突破点就是常说的金叉，是股票的买入信号，在代码里，使用前一短均值小于前一长均值且当前短均值大于当前长均值的方式来判断金叉，赋予信号值为 1；
• 短周期均线从上向下跌破长周期均线时，意味着当前时间段具有下跌趋势，跌破点就是常说的死叉，是股票的卖出信号，在代码里，使用前一短均值大于前一长均值且当前短均值小于当前长均值的方式来判断金叉，赋予信号值为 -1。

在这里我们使用简单移动平均（Simple Moving Average），其中 n 为窗口大小：

在下边的例子中我们使用了全市场 2020.07.01 至 2023.07.19 的日频股票数据来进行计算，在实际使用中可以自由更换数据频率，如分钟频率，小时频率等。

因子计算示例代码如下：

import pandas as pd
import dolphindb as ddb

# 调用数据
df = loadTable("dfs://Daily_adj_price", "data")
df = pd.DataFrame(df, "TRADE_DATE", True)

# 定义计算 ma 金叉死叉信号生成的方法
def signal_ma(data_chunk, short, long):
    #分别计算出5日和20日均线及其前一根均线
    data_chunk['ma_5'] = data_chunk['CLOSE_PRICE_1'].fillna(0).rolling(int(short)).mean()
    data_chunk['ma_20'] = data_chunk['CLOSE_PRICE_1'].fillna(0).rolling(int(long)).mean()
    data_chunk['pre_ma5'] = data_chunk['ma_5'].shift(1)
    data_chunk['pre_ma20'] =  data_chunk['ma_20'].shift(1)
    # 通过df[contion]方式按列进行条件，判断出金叉死叉信号
    data_chunk['signal'] = 0
    data_chunk.loc[((data_chunk.loc[:,'pre_ma5']< data_chunk.loc[:,'pre_ma20'])& (data_chunk.loc[:,'ma_5'] > data_chunk.loc[:,'ma_20'])), "signal"] = 1
    data_chunk.loc[((data_chunk.loc[:,'pre_ma5']> data_chunk.loc[:,'pre_ma20']) & (data_chunk.loc[:,'ma_5'] < data_chunk.loc[:,'ma_20'])), "signal"] = -1
    return data_chunk

# 生成信号
combined_results = df.groupby('SECURITY_ID').apply(signal_ma,5,20)

示例代码解析：

• 通过使用 pandas 包中的 pd.groupby() + apply 的方式
• 实现了按照股票代码分组后对前复权的收盘价使用 rolling().mean() 求出移动平均线，并根据长短期均线之间的位置关系来判断出金叉死叉进而产生信号：1 代表多头信号，-1 代表空头信号，0 则为目前没有信号。
• 在计算双均线金叉死叉信号的时候，Python Pandas 版本的实现逻辑和代码同 Python Parser 保持了一致，因此在使用时无需额外操作即可在二者之间切换。此外，Python 解析器底层自动实现的批处理加速功能，省去了 Python 实现批处理所需的额外工作，更为高效。

本章节展示了如何基于 Level2 快照行情数据计算十档净委买增额。

十档净委买增额因子指的是在有效十档范围内买方资金总体增加量，即所有买价变化量的总和，计算公式如下：

其中 level10_Diff_t 表示 t 时刻的十档净委买增额；bid_i,t 表示 t 时刻的第 i 档买方报价；bidQty_i,t 表示 t 时刻的第 i 档买方挂单数量；指示函数 I 表示报价是否在有效释放范围内。

有效十档范围内表示不考虑已不在十档范围内的档位，即表示只考虑以下区间的档位：

最后，对过去 n 时间窗口内的十档净委买增额求和。

因子计算示例代码如下：

import pandas as pd
import dolphindb as ddb

# 定义因子函数
def level10Diff(df, lag=20):
    temp = df[["TradeTime", "SecurityID"]]
    temp["bid"] = df["BidPrice"].fillna(0)
    temp["bidAmt"] = df["BidOrderQty"].fillna(0) * df["BidPrice"].fillna(0)
    temp["prevbid"] = temp["bid"].shift(1).fillna(0)
    temp["prevbidAmt"] = temp["bidAmt"].shift(1).fillna(0)
    temp["bidMin"] = temp["bid"].apply("min")
    temp["bidMax"] = temp["bid"].apply("max")
    temp["prevbidMin"] = temp["bidMin"].shift(1).fillna(0)
    temp["prevbidMax"] = temp["bidMax"].shift(1).fillna(0)
    temp["pmin"] = temp[["bidMin", "prevbidMin"]].max(axis=1)
    temp["pmax"] = temp[["bidMax", "prevbidMax"]].max(axis=1)
    amount = temp["bidAmt"]*((temp["bid"]>=temp["pmin"])&(temp["bid"]<=temp["pmax"]))
    lastAmount = temp["prevbidAmt"]*((temp["prevbid"]>=temp["pmin"])&(temp["prevbid"]<=temp["pmax"]))
    temp["amtDiff"] = amount.apply("sum") - lastAmount.apply("sum")
    temp["amtDiff"] = temp["amtDiff"].rolling(lag, 1).sum()
    return temp[["TradeTime", "SecurityID", "amtDiff"]].fillna(0)

# 指定计算某一天一只股票的因子
snapshotTB = loadTable("dfs://TL_Level2", "snapshot")
df = pd.DataFrame(snapshotTB, index="Market", lazy=True)
df = df[(df["TradeTime"].astype(ddb.DATE)==2023.02.01)&(df["SecurityID"]=="000001")]
res = level10Diff(df.compute(), 20)

# 指定计算某一天的因子
snapshotTB = loadTable("dfs://TL_Level2", "snapshot")
df = pd.DataFrame(snapshotTB, index="Market", lazy=True)
res = df[df["TradeTime"].astype(ddb.DATE)==2023.02.01][["TradeTime", "SecurityID", "BidPrice", "BidOrderQty"]].groupby(["SecurityID"]).apply(lambda x:level10Diff(x, 20))

示例代码解析：

• Level2 快照行情数据拥有十档行情量价数据。针对这种类型相同、含义相近的数据，可以考虑使用 DolphinDB 的 Array Vector 类型来存储。在 DolphinDB 里 Array Vector 是一种特殊的向量，用于存储可变长度的二维数组。上述代码就是基于十档行情用 Array Vector 类型存储的数据库开发的，其中“BidPrice”、“BidOrderQty”都是 ArrayVector 的列。

• 表中 Array Vector 类型的列转化为 DataFrame 中的一列之后，其中每个元素为 List。
  

• 针对 Array Vector 转化的列，Python Parser 中支持四则运算、比较等基础运算（比如两列相乘 df["BidOrderQty"].fillna(0) * df["BidPrice"].fillna(0)）；其他运算需要使用 apply 函数（比如求每一行的最小值 temp["bid"].apply("min")）。

• 对于内置函数 (比如 max/min/sum 等)，使用 apply 函数时，传入字符串和传入函数是有区别的。

  • 传入字符串时，比如 temp["bid"].apply("max")，首先会去找 series 类里有没有实现这个函数，如果有，就用 series 的这个函数；如果没有，就去找内置函数。
  • 传入函数时，比如 temp["bid"].apply(max)，会直接使用内置函数。

​ 综上，对于 apply + 内置函数 的情况下，一般建议用户传入字符串。

• 使用 shift(1) 的方式获取上一笔快照的数据。
• 使用 rolling().sum() 的方式求滑动窗口内十档净委买增额的和。
• snapshotTB = loadTable("dfs://TL_Level2", "snapshot") 通过 loadTable 函数，将 "dfs://TL_Level2" 数据库下的分布式表 "snapshot" 的元数据取回到内存。此时变量 snapshotTB 只包含元数据，库内数据并未取到内存。
• df = pd.DataFrame(snapshotTB, index="Market", lazy=True) 通过 pd.DataFrame() 函数，将 DolphinDB 的表转化为数据框。对于分布式表而言，index 为必填参数，可以指定表中的任意一列，该列仅作为索引，后续可以不参与计算；lazy 参数指定计算是否立即执行，必须指定为 True，表示该 DataFrame 会存储所有函数调用，尽可能延迟计算，以减少计算带来的性能消耗。
• 可以通过 df[过滤条件] 的形式选出库内指定范围的数据。比如：df[(df["TradeTime"].astype(ddb.DATE)==2023.02.01)&(df["SecurityID"]=="000001")] 指定取库内 2023.02.01 的“000001“这一天一只股票的数据。
• 建议在 groupby 执行计算函数之前，先对数据列进行过滤，只取出计算需要的列。可以降低内存使用，减少数据读取与拷贝的开销。
• lazy 模式下不允许直接改变 DataFrame 的值。因为 level10Diff 函数里面有 temp["bid"] = df["BidPrice"].fillna(0) 的操作，所以对直接过滤出来的 df 直接调用函数 level10Diff(df, 20) 会报错： Lazy-model DataFrame does not support update value. 需要使用 df.compute() 将 lazy 模式的 DataFrame 强制触发计算，转化为 no-lazy 模式的 DataFrame。
• 可以通过 .groupby(分组列).apply(函数) 的方式实现分组计算，Python Parser 内部对 groupby.apply 实现了并行计算。