DolphinDB 在深度学习中的应用：股票实时波动率预测

// 定义聚合函数
defg featureEngineering(DateTime, BidPrice, BidOrderQty, OfferPrice, OfferOrderQty, aggMetaCode){
	wap = (BidPrice * OfferOrderQty + BidOrderQty * OfferPrice) \ (BidOrderQty + OfferOrderQty)
	wapBalance = abs(wap[0] - wap[1])
	priceSpread = (OfferPrice[0] - BidPrice[0]) \ ((OfferPrice[0] + BidPrice[0]) \ 2)
	BidSpread = BidPrice[0] - BidPrice[1]
	OfferSpread = OfferPrice[0] - OfferPrice[1]
	totalVolume = OfferOrderQty.rowSum() + BidOrderQty.rowSum()
	volumeImbalance = abs(OfferOrderQty.rowSum() - BidOrderQty.rowSum())
	logReturnWap = logReturn(wap)
	logReturnOffer = logReturn(OfferPrice)
	logReturnBid = logReturn(BidPrice)
	subTable = table(DateTime as `DateTime, BidPrice, BidOrderQty, OfferPrice, OfferOrderQty, wap, wapBalance, priceSpread, BidSpread, OfferSpread, totalVolume, volumeImbalance, logReturnWap, logReturnOffer, logReturnBid)
	colNum = 0..9$STRING
	colName = `DateTime <- (`BidPrice + colNum) <- (`BidOrderQty + colNum) <- (`OfferPrice + colNum) <- (`OfferOrderQty + colNum) <- (`Wap + colNum) <- `WapBalance`PriceSpread`BidSpread`OfferSpread`TotalVolume`VolumeImbalance <- (`logReturn + colNum) <- (`logReturnOffer + colNum) <- (`logReturnBid + colNum)
	subTable.rename!(colName)
	subTable['BarDateTime'] = bar(subTable['DateTime'], 10m)
	result = sql(select = aggMetaCode, from = subTable).eval().matrix()
	result150 = sql(select = aggMetaCode, from = subTable, where = <time(DateTime) >= (time(BarDateTime) + 150*1000) >).eval().matrix()
	result300 = sql(select = aggMetaCode, from = subTable, where = <time(DateTime) >= (time(BarDateTime) + 300*1000) >).eval().matrix()
	result450 = sql(select = aggMetaCode, from = subTable, where = <time(DateTime) >= (time(BarDateTime) + 450*1000) >).eval().matrix()
	return concatMatrix([result, result150, result300, result450])
}

// 元编程生产查询语句并查询
whereConditions = [<date(DateTime) between 2021.01.04 : 2021.12.31>, <SecurityID in stockList>, <(time(DateTime) between 09:30:00.000 : 11:29:59.999) or (time(DateTime) between 13:00:00.000 : 14:56:59.999)>]
result = sql(select = sqlColAlias(<featureEngineering(DateTime, matrix(BidPrice), matrix(BidOrderQty), matrix(OfferPrice), matrix(OfferOrderQty), aggMetaCode)>, metaCodeColName <- (metaCodeColName+"_150") <- (metaCodeColName+"_300") <- (metaCodeColName+"_450")), from = snapshot, where = whereConditions, groupBy = [<SecurityID>, <bar(DateTime, 10m) as DateTime>]).eval()
复制

create database "dfs://tenMinutesFactorDB"
partitioned by VALUE(2021.01M..2021.12M), VALUE(`f1`f2)
engine = 'TSDB'

create table "dfs://tenMinutesFactorDB"."tenMinutesFactorTB"(
	SecurityID SYMBOL,
	DateTime TIMESTAMP[comment="时间列", compress="delta"],
	FactorNames SYMBOL,
	FactorValues DOUBLE
)
partitioned by DateTime, FactorNames,
sortColumns=[`SecurityID, `DateTime],
keepDuplicates=ALL, 
sortKeyMappingFunction=[hashBucket{, 500}]
复制

pip install dolphindb
pip install dolphindb-tools
复制

# 数据预处理，并切分训练集和测试集
conn = ddb.session("192.168.100.201", 8848, "admin", "123456")
conn.run("""startDay = 2021.01.01
            endDay = 2021.12.31
            splitDay = (startDay..endDay)[((endDay-startDay)*0.8).floor()]
            Data = select FactorValues from loadTable("dfs://tenMinutesFactorDB", "tenMinutesFactorTB") where date(DateTime) >= objByName('startDay') and date(DateTime) <= objByName('endDay') and SecurityID=`600030 pivot by DateTime, SecurityID, FactorNames 
            Data = Data[each(isValid, Data.values()).rowAnd()]
            """)
复制

# Dataloader 参数
targetColumns = ["LogReturn0_realizedVolatility"]
excludedColumns = ["SecurityID", "DateTime", "LogReturn0_realizedVolatility"]
batchSize = 256
windowSize = [120, 1]
windowStride=[1, 1]
offset = 120
trainSql = """select * from objByName('Data') where date(DateTime) >= objByName('startDay') and date(DateTime) <= objByName('splitDay')"""
testSql = """select * from objByName('Data') where date(DateTime) > objByName('splitDay') and date(DateTime) <= objByName('endDay')"""  

# 实例化 DDBDataloader
trainLoader = DDBDataLoader(ddbSession=conn, sql=trainSql, targetCol=targetColumns, excludeCol=excludedColumns, batchSize=batchSize, device=device, windowSize=windowSize, windowStride=windowStride, offset=offset)
testLoader = DDBDataLoader(ddbSession=conn, sql=testSql, targetCol=targetColumns, excludeCol=excludedColumns, batchSize=batchSize, device=device, windowSize=windowSize, windowStride=windowStride, offset=offset)
print("Using DDBDataLoader, data is ready")
复制

# 模型定义
class LSTMModel(nn.Module):
    def __init__(self, inputSize, units):
        super(LSTMModel, self).__init__()
        self.lstm1 = nn.LSTM(input_size=inputSize, hidden_size=units, batch_first=True, dropout=0.4)
        self.lstm2 = nn.LSTM(input_size=units, hidden_size=128, batch_first=True, dropout=0.3)
        self.lstm3 = nn.LSTM(input_size=128, hidden_size=32, batch_first=True, dropout=0.1)
        self.fc = nn.Linear(32, 1)

    def forward(self, x):
        out, _ = self.lstm1(x)
        out, _ = self.lstm2(out)
        out, _ = self.lstm3(out)
        out = out[:, -1, :]  
        out = self.fc(out)
        return out
复制

# 训练阶段
model.train()
trainLoss = 0.0
trainLen = 0
for inputs, targets in trainLoader:
    inputs = inputs.float().to(device)
    targets = targets.float().to(device)
    optimizer.zero_grad()
    outputs = model(inputs)
    loss = criterion(outputs, targets)
    trainLoss += loss.item()
    loss.backward()
    optimizer.step()
    trainLen += 1
trainLoss /= trainLen
复制

torchScriptModel = torch.jit.script(model)
torchScriptModel.save("/your_path/LSTMmodel.pt")
复制

name = loadTable("dfs://l2StockSHDB", "snapshot").schema().colDefs.name
type = loadTable("dfs://l2StockSHDB", "snapshot").schema().colDefs.typeString
share(streamTable(100000:0, name, type), `SnapshotStream)
复制

testSnapshot = select * from loadTable("dfs://l2StockSHDB", "snapshot") where SecurityID=`600030 and date(DateTime)=mdDate
submitJob("replaySnapshot", "replay 1 day snapshot", replay{testSnapshot, SnapshotStream, `DateTime, `DateTime, 100, false, 1, , true})
复制

// 创建实时因子表 
share(streamTable(100000:0 , `DateTime`SecurityID`ReceiveTime <- metaCodeColName <- (metaCodeColName+"_150") <- (metaCodeColName+"_300") <- (metaCodeColName+"_450") <- `HandleTime,`TIMESTAMP`SYMBOL`NANOTIMESTAMP <- take(`DOUBLE, 676) <- `NANOTIMESTAMP) , `aggrFeatures10min)
go
// 创建时序聚合引擎
metrics=sqlColAlias(<featureEngineering(DateTime,
		matrix(BidPrice[0],BidPrice[1],BidPrice[2],BidPrice[3],BidPrice[4],BidPrice[5],BidPrice[6],BidPrice[7],BidPrice[8],BidPrice[9]),
		matrix(BidOrderQty[0],BidOrderQty[1],BidOrderQty[2],BidOrderQty[3],BidOrderQty[4],BidOrderQty[5],BidOrderQty[6],BidOrderQty[7],BidOrderQty[8],BidOrderQty[9]),
		matrix(OfferPrice[0],OfferPrice[1],OfferPrice[2],OfferPrice[3],OfferPrice[4],OfferPrice[5],OfferPrice[6],OfferPrice[7],OfferPrice[8],OfferPrice[9]),
		matrix(OfferOrderQty[0],OfferOrderQty[1],OfferOrderQty[2],OfferOrderQty[3],OfferOrderQty[4],OfferOrderQty[5],OfferOrderQty[6],OfferOrderQty[7],OfferOrderQty[8],OfferOrderQty[9]), aggMetaCode)>, metaCodeColName <- (metaCodeColName+"_150") <- (metaCodeColName+"_300") <- (metaCodeColName+"_450"))	
		
createTimeSeriesEngine(name="aggrFeatures10min", windowSize=600000, step=600000, metrics=[<now(true) as ReceiveTime>, metrics, <getHandleTime() as HandleTime>], useSystemTime=false ,dummyTable=SnapshotStream, outputTable=aggrFeatures10min, timeColumn=`DateTime, useWindowStartTime=true, keyColumn=`SecurityID)
// 订阅原始行情数据，实时写入时序聚合引擎
subscribeTable(tableName="SnapshotStream", actionName="aggrFeatures10min", offset=-1, handler=getStreamEngine("aggrFeatures10min"), msgAsTable=true, batchSize=2000, throttle=0.01, hash=0, reconnect=true)
复制

installPlugin("GpuLibtorch")
loadPlugin("GpuLibtorch")
复制

// 加载模型
modelPath = "/home/lnfu/ytxie/LSTMmodel.pt"
model = Libtorch::load(modelPath)
Libtorch::setDevice(model, "CUDA")
复制

// 创建实时预测结果表 
share(streamTable(100000:0 , `Predicted`SecurityID`DateTime`ReceiveTime`HandleTime`PredictedTime, `FLOAT`SYMBOL`TIMESTAMP`NANOTIMESTAMP`NANOTIMESTAMP`NANOTIMESTAMP), `result10min)
go
// 定义实时处理方法：数据预处理和模型推理
def predictRV(model, window, msg){
	tmp = select * from msg 
	tmp.dropColumns!(`ReceiveTime`HandleTime)
	tmp.reorderColumns!(objByName(`historyData).columnNames())
	predictedtSet = []
	for(row in tmp){
		objByName(`historyData).tableInsert(row)
		data = tail(objByName(`historyData), window) 
		data.dropColumns!(`SecurityID`DateTime`LogReturn0_realizedVolatility)		
		data = data[each(isValid, data.values()).rowAnd()]
		input = tensor([matrix(data)$FLOAT])
		predict = Libtorch::predict(model, input)
		predictedtSet.append!(predict[0][0])
	}
	ret = select predictedtSet as predicted, SecurityID, DateTime, ReceiveTime, HandleTime, now(true) as PredictedTime from msg 
	objByName("result10min").append!(ret)
}
// 获取当天往前 20 日的十分钟因子，为滑动窗口准备历史数据
data = select FactorValues from loadTable("dfs://tenMinutesFactorDB", "tenMinutesFactorTB") where date(DateTime) between (mdDate-20):(mdDate-1) and SecurityID=`600030 pivot by DateTime, SecurityID, FactorNames 
data = data[each(isValid, data.values()).rowAnd()]
share(data, `historyData)
// 时间窗口大小，用 120 个时间点的因子预测下一个时间点的波动率
window = 120
// 预热模型，避免第一次调用模型推理时的较大耗时
warmupData = tail(objByName(`historyData), window)
warmupData.dropColumns!(`SecurityID`DateTime`LogReturn0_realizedVolatility)
input = tensor([matrix(warmupData)$FLOAT])
out = Libtorch::predict(model, input)
// 订阅实时因子表
subscribeTable(tableName="aggrFeatures10min", actionName="predictRV", offset=-1, handler=predictRV{model, window}, msgAsTable=true, batchSize=100, throttle=0.001, hash=1, reconnect=true)
复制

# 订阅实时计算结果流表
import dolphindb as ddb

s = ddb.session(keepAliveTime=60)
s.connect("192.168.100.201", 8848, "admin", "123456")
s.enableStreaming()

def handler(lst):
    print(lst)

s.subscribe(host="192.168.100.201", port=8848, handler=handler, tableName="result10min", actionName="result10min", offset=-1, resub=True, msgAsTable=True, batchSize=100, throttle=0.001)
复制

stat = select (HandleTime-ReceiveTime)/1000/1000 as factorDelay, (PredictedTime-HandleTime)/1000/1000  as predictDelay, (PredictedTime-ReceiveTime)/1000/1000  as totalDelay  from result10min	
select avg(factorDelay), percentile(factorDelay, 95), min(factorDelay), max(factorDelay), avg(predictDelay), percentile(predictDelay, 95), min(predictDelay), max(predictDelay), avg(totalDelay), percentile(totalDelay, 95), min(totalDelay), max(totalDelay)  from stat
复制