量化金融范例

中国股票市场每3秒更新一条 level 2 的行情数据，一般包括股票代码、日期、时间、交易量、交易价格、交易次数、买方与卖方的10档报价与量等常用信息，以及其它信息等数据。本例中所用数据为上海证券交易所A股股票2020年6月的 level 2 数据，每天的数据是一个约2.5GB的CSV文件，共68列数据。所有数据文件均存于同一个文件夹下。若其中一半的列为常用数据，遵循每个表每个分区中的常用数据压缩前为100MB左右的原则，可将数据库设计为复合分区。按天（date列）进行值分区，并按照股票代码（symbol列）分为10个HASH分区。

建库以及导入数据的脚本如下。使用loadTextEx导入分布式数据库，其详情请参阅文本数据加载教程。若您已有高频数据，可使用以下脚本建库，但请注意数据中列名与本例中列名一致。

dbDate = database("", VALUE, 2020.01.01..2020.12.31)
dbSymbol=database("", HASH, [SYMBOL, 10])
db = database("dfs://level2", COMPO, [dbDate, dbSymbol])

//请注意更换目录dataDir
dataDir="/hdd/hdd1/data/Level2TextFiles/"

def importTxtFiles(dataDir, db){
    dataFiles = exec filename from files(dataDir) where isDir=false
    for(f in dataFiles){
        loadTextEx(db, `quotes, `date`symbol, dataDir + f)
    }
}
importTxtFiles(dataDir, db);

选择合适的数据库分区机制，对确保数据库最优性能非常重要。具体细节请参阅分区数据库教程。

本例中导入的数据是上海证券交易所A股股票2020年6月1日的 level 2 数据。与csv导入时能自动生成分布式表的shema不同，二进制导入前需要显式定义分布式表，具体建库建表的脚本如下：

login("admin", "123456")
dbDate = database("", VALUE, 2020.01.01..2020.12.31)
dbSymbol=database("", HASH, [SYMBOL, 10])
db = database("dfs://level2", COMPO, [dbDate, dbSymbol])
schemaTable=table(	
	array(SYMBOL,0) as  symbol,
	array(SYMBOL,0) as  market,
	array(DATE,0) as  date,
	array(TIME,0) as  time,
	array(DOUBLE,0) as  preClose,
	array(DOUBLE,0) as  open,
	array(DOUBLE,0) as  high,
	array(DOUBLE,0) as  low,
	array(DOUBLE,0) as  last,
	array(INT,0) as  numTrades,
	array(INT,0) as  curNumTrades,
	array(INT,0) as  volume,
	array(INT,0) as  curVol,
	array(DOUBLE,0) as  turnover,
	array(INT,0) as  curTurnover,
	array(INT,0) as  peratio1,
	array(INT,0) as  peratio2,
	array(INT,0) as  totalAskVolume,
	array(DOUBLE,0) as  wavgAskPrice,
	array(INT,0) as  askLevel,
	array(INT,0) as  totalBidVolume,
	array(DOUBLE,0) as  wavgBidPrice,
	array(INT,0) as  bidLevel,
	array(DOUBLE,0) as  iopv,
	array(INT,0) as  ytm,
	array(DOUBLE,0) as  askPrice1,
	array(DOUBLE,0) as  askPrice2,
	array(DOUBLE,0) as  askPrice3,
	array(DOUBLE,0) as  askPrice4,
	array(DOUBLE,0) as  askPrice5,
	array(DOUBLE,0) as  askPrice6,
	array(DOUBLE,0) as  askPrice7,
	array(DOUBLE,0) as  askPrice8,
	array(DOUBLE,0) as  askPrice9,
	array(DOUBLE,0) as  askPrice10,
	array(DOUBLE,0) as  bidPrice1,
	array(DOUBLE,0) as  bidPrice2,
	array(DOUBLE,0) as  bidPrice3,
	array(DOUBLE,0) as  bidPrice4,
	array(DOUBLE,0) as  bidPrice5,
	array(DOUBLE,0) as  bidPrice6,
	array(DOUBLE,0) as  bidPrice7,
	array(DOUBLE,0) as  bidPrice8,
	array(DOUBLE,0) as  bidPrice9,
	array(DOUBLE,0) as  bidPrice10,
	array(INT,0) as  askVolume1,
	array(INT,0) as  askVolume2,
	array(INT,0) as  askVolume3,
	array(INT,0) as  askVolume4,
	array(INT,0) as  askVolume5,
	array(INT,0) as  askVolume6,
	array(INT,0) as  askVolume7,
	array(INT,0) as  askVolume8,
	array(INT,0) as  askVolume9,
	array(INT,0) as  askVolme10,
	array(INT,0) as  bidVolume1,
	array(INT,0) as  bidVolume2,
	array(INT,0) as  bidVolume3,
	array(INT,0) as  bidVolume4,
	array(INT,0) as  bidVolume5,
	array(INT,0) as  bidVolume6,
	array(INT,0) as  bidVolume7,
	array(INT,0) as  bidVolume8,
	array(INT,0) as  bidVolume9,
	array(INT,0) as  bidVolume10,
	array(LONG,0) as  unixTime,
	array(DOUBLE,0) as  upperLimit,
	array(DOUBLE,0) as  lowerLimit
)
db.createPartitionedTable(schemaTable,`quotes,`date`symbol)

对于二进制格式的文件，DolphinDB提供了2个函数用于导入：readRecord!函数和loadRecord函数。二者的区别是，前者不支持导入字符串类型的数据，后者支持。在二进制文件中，date列和time列的数据以数值形式存储，可以使用temporalParse函数进行日期和时间类型数据的格式转换。再使用replaceColumn!函数替换表中原有的列。symbol和market列在二进制文件中也是数值形式存储，处理的方式类似。具体代码如下所示：

schema = [
("symbol", INT), ("market", INT), ("date", INT), ("time", INT), ("preClose", DOUBLE),
("open", DOUBLE), ("high", DOUBLE), ("low", DOUBLE), ("last", DOUBLE), ("numTrades", INT),
("curNumTrades", INT), ("volume", INT), ("curVol", INT), ("turnover", DOUBLE), ("curTurnover", INT),
("peratio1", INT), ("peratio2", INT), ("totalAskVolume", INT), ("wavgAskPrice", DOUBLE), ("askLevel", INT),
("totalBidVolume", INT), ("wavgBidPrice", DOUBLE), ("bidLevel", INT), ("iopv", DOUBLE), ("ytm", INT),
("askPrice1", DOUBLE), ("askPrice2", DOUBLE), ("askPrice3", DOUBLE), ("askPrice4", DOUBLE), ("askPrice5", DOUBLE),
("askPrice6", DOUBLE), ("askPrice7", DOUBLE), ("askPrice8", DOUBLE), ("askPrice9", DOUBLE), ("askPrice10", DOUBLE),
("bidPrice1", DOUBLE), ("bidPrice2", DOUBLE), ("bidPrice3", DOUBLE), ("bidPrice4", DOUBLE), ("bidPrice5", DOUBLE),
("bidPrice6", DOUBLE), ("bidPrice7", DOUBLE), ("bidPrice8", DOUBLE), ("bidPrice9", DOUBLE), ("bidPrice10", DOUBLE),
("askVolume1", INT), ("askVolume2", INT), ("askVolume3", INT), ("askVolume4", INT), ("askVolume5", INT),
("askVolume6", INT), ("askVolume7", INT), ("askVolume8", INT), ("askVolume9", INT), ("askVolme10", INT),
("bidVolume1", INT), ("bidVolume2", INT), ("bidVolume3", INT), ("bidVolume4", INT), ("bidVolume5", INT),
("bidVolume6", INT), ("bidVolume7", INT), ("bidVolume8", INT), ("bidVolume9", INT),("bidVolume10", INT),
("unixTime", LONG), ("upperLimit", DOUBLE), ("lowerLimit", DOUBLE) ]

dataDir="/hdd/hdd1/data/Level2BinFiles/"

def importBinFiles(dataDir, schema){
    tick1 = loadTable("dfs://level2","quotes")
    dataFiles = exec filename from files(dataDir)
    for(f in dataFiles){
        t=loadRecord(dataDir + f, schema)
        t.replaceColumn!(`date, t.date.string().datetimeParse("yyyyMMdd"))
        t.replaceColumn!(`time, t.time.format("000000000").datetimeParse("HHmmssSSS"))
        t.replaceColumn!(`symbol, t.symbol.format("000000"))
        t.replaceColumn!(`market, iif(t.market==0,"SH","SZ"))
        tick1.append!(t)
    }
}
importBinFiles(dataDir, schema);

若您尚无高频数据，可下载20200601.csv（或20200601.bin），采用上两节中的脚本，将这天数据载入数据库，然后通过修改日期，生成多天的高频数据以供测试。

下列代码通过sqlDS函数将之前导入的2020.06.01这一天的数据，按分布式表一个分区生成一个数据源的方式，共分成10个数据源，然后通过mr函数将这10份数据先取到内存更新日期，再写入数据库。mr函数的parallel参数可设为true，即采用并行执行，以加块生成模拟数据的速度。若服务器内存不足容纳一天的数据，则需要设为false，即采用串行执行以尽量少占用内存。

def writeData(mutable t,dbName,tableName, days){
	pt = loadTable(dbName,tableName)
	for(day in days){
		update t set date = day
		pt.append!(t)		
	}
}
def main(dbName,tableName,days){
	rds = sqlDS(<select * from loadTable(dbName,tableName) where date=2020.06.01>)
	mr(ds=rds, mapFunc=writeData{,dbName,tableName,days}, parallel=true)
}
login(`admin,`123456)
days=2020.06.01..2020.06.30
days=days[weekday(days) between 1:5]
main("dfs://level2","quotes",days)

将quotes表元数据载入内存：

db = database("dfs://level2")
quotes = loadTable(db, `quotes);

计算某只股票某天中，每分钟内的平均价差(bid-ask spread)，并将结果在GUI以图形展示：

avgSpread = select max((askPrice1-bidPrice1)/(askPrice1+bidPrice1)*2) as avgSpread from quotes where date=2020.06.01, symbol=`600600, time between 09:30:00.000 : 15:00:00.000, askPrice1>=bidPrice1 group by minute(time) as minute
plot(avgSpread.avgSpread, avgSpread.minute, "Average bid-ask spread per minute")

context by是DolphinDB独有的功能，是对标准SQL语句的拓展。使用context by子句可以简化对面板数据的操作。

context by与group by类似，都对数据进行分组。但是，用group by时，每一组返回一个标量值，而用context by时，每一组返回一个和组内元素数量相同的向量。group by只能配合聚合函数使用，而context by既可以配合聚合函数使用，也可以与移动窗口函数或累积函数等其它函数结合使用。

计算每只股票在2020.06.01中截止到当前时刻的每次数据更新（绝大部分情况下为3秒）中最大交易量：

t = select symbol, date, time, cummax(curVol) as volume_cummax from quotes where date=2020.06.01 context by symbol, date

context by子句亦可结合滑动窗口函数使用。下例计算每只股票在2020.06.01中过去20次数据更新中的平均价格（约为过去一分钟的平均价格）：

t = select symbol, date, time, mavg(last, 20) as price_mavg1min from quotes where date=2020.06.01 context by symbol, date

context by子句还可结合聚合函数使用，将聚合函数的分组计算结果赋予组内每一行：

t = select symbol, date, time, max(curVol) as volume_dailyMax from quotes where date=2020.06.01 context by symbol, date

使用某天数据，计算高频因子：

t=select symbol, date, time, (2*bidVolume1+bidVolume2-2*askVolume1-askVolume2)\(2*bidVolume1+bidVolume2+2*askVolume1+askVolume2) as factor1, (2*bidVolume1+bidVolume2-2*askVolume1-askVolume2)\mavg((2*bidVolume1+bidVolume2+2*askVolume1+askVolume2),100) as factor2 from quotes where date=2020.06.01, symbol>=`600000, time between 09:30:00.000 : 15:00:00.000 context by symbol order by symbol, date, time

移动窗口函数可嵌套使用。下例中，首先分别计算第一档买量与卖量100次数据更新（约为5分钟）的移动平均，再计算两者60次数据更新（约为3分钟）的移动相关性。仅用一行代码即可完成。

t=select symbol, date, time, mcorr(mavg(bidVolume1,100), mavg(askVolume1,100), 60) from quotes where date=2020.06.01, symbol>=`600000, time between 09:30:00.000 : 15:00:00.000 context by symbol order by symbol, date, time

pivot by是DolphinDB的独有功能，是对标准SQL语句的拓展。它将表中某列的内容按照两个维度重新排列，亦可配合数据转换函数使用。

下例中的投资组合的股票代码为向量syms，其中各股票的持仓数量分别为100, 200, 400, 800, 600, 400, 300。本例的目标是计算某天中每个时刻的投资组合价值。请注意，两行DolphinDB代码即可实现目标，但为了读者理解方便，这里分为三步逐步介绍如何实现。

首先对数据使用pivot by子句进行重新排列：

db = database("dfs://level2")
quotes = loadTable(db, `quotes)
syms = `600000`600300`600400`600500`600600`600800`600900

pv = select last from quotes where date=2020.06.01, symbol in syms, time between 09:30:00.000 : 15:00:00.000 pivot by time, symbol
select top 10 * from pv;

结果为：

time         C600000 C600300 C600400 C600500 C600600 C600800 C600900
------------ ------- ------- ------- ------- ------- ------- -------
09:30:00.000 10.63   3.09    3.28    5.03    63.5    4.68    17.43
09:30:03.000 10.62   3.08    3.28    5.03    63.5    4.68    17.4
09:30:05.000 10.61                                   4.68    17.41
09:30:06.000         3.08    3.28    5.02    63.51
09:30:08.000 10.62                                   4.68    17.41
09:30:09.000         3.07    3.28    5.03    63.74
09:30:11.000 10.63                                   4.68    17.42
09:30:12.000         3.07    3.27    5.03    63.95
09:30:14.000 10.64                                   4.68    17.43
09:30:15.000         3.06            5.03    63.95

可见并不是所有股票均在同一时刻更新数据。若要计算每个时刻的投资组合价值，首先应在以上结果中使用ffill函数向前填充last列的空值：

tmp = select ffill(last) from quotes where date=2020.06.01, symbol in syms, time between 09:30:00.000 : 15:00:00.000 pivot by time, symbol
select top 10 * from tmp;

结果为：

time         C600000 C600300 C600400 C600500 C600600 C600800 C600900
------------ ------- ------- ------- ------- ------- ------- -------
09:30:00.000 10.63   3.09    3.28    5.03    63.5    4.68    17.43
09:30:03.000 10.62   3.08    3.28    5.03    63.5    4.68    17.4
09:30:05.000 10.61   3.08    3.28    5.03    63.5    4.68    17.41
09:30:06.000 10.61   3.08    3.28    5.02    63.51   4.68    17.41
09:30:08.000 10.62   3.08    3.28    5.02    63.51   4.68    17.41
09:30:09.000 10.62   3.07    3.28    5.03    63.74   4.68    17.41
09:30:11.000 10.63   3.07    3.28    5.03    63.74   4.68    17.42
09:30:12.000 10.63   3.07    3.27    5.03    63.95   4.68    17.42
09:30:14.000 10.64   3.07    3.27    5.03    63.95   4.68    17.43
09:30:15.000 10.64   3.06    3.27    5.03    63.95   4.68    17.43

最后计算每一时刻的投资组合价值：

select time, 100*C600000+200*C600300+400*C600400+800*C600500+600*C600600+400*C600800+300*C600900 from tmp;

计算某天分钟级K线：

minuteBar = select first(last) as open, max(last) as high, min(last) as low, last(last) as last, sum(curVol) as volume from quotes where date=2020.06.01, symbol>=`600000 group by symbol, date, minute(time) as minute;

以上为仅计算一天数据的K线。若需使用全部数据计算分钟级K线，并将结果存入数据库，由于全部数据有可能超过内存，同时为了充分利用系统硬件资源进行并行计算，可使用MapReduce函数mr。

model=select top 1 symbol,date, minute(time) as minute, open, high, low, last, curVol as volume from quotes where date=2020.06.01,symbol=`600000
if(existsTable("dfs://level2", "minuteBar"))
	db.dropTable("minuteBar")
db.createPartitionedTable(model, "minuteBar", `date`symbol)

def saveMinuteBar(t){
	minuteBar=select first(last) as open, max(last) as high, min(last) as low, last(last) as last, sum(curVol) as volume from t where symbol>=`600000, time between 09:30:00.000 : 15:00:00.000 group by symbol, date, minute(time) as minute
	loadTable("dfs://level2", "minuteBar").append!(minuteBar)
	return minuteBar.size()
}

ds = sqlDS(<select symbol, date, time, last, curVol from quotes>)
mr(ds,saveMinuteBar,+)

有关K线计算的更多场景及范例，例如指定K线窗口的起始时刻、重叠K线窗口、使用交易量划分K线窗口等等，请参考K线计算教程。

DolphinDB提供性能极佳的非同时连接函数aj及wj。函数aj(asof join)为左表中每条记录，在右表中获取符合指定条件的组中该时刻之前（包括该时刻）的最后一条记录；函数wj(window join)为左表中每条记录，在右表中获取符合指定条件的组基于该时刻的指定时间范围的记录并进行计算。DolphinDB的asof join比pandas中的asof join速度快约200倍。

Asof join 在国际证券市场的最典型应用场景为，报价与交易信息处于不同的数据表中，使用asof join可以高效的寻找每一笔交易之前的最近一笔报价。在国内证券市场，asof join 亦有许多应用场景。

每天交易结束后，若要将交易价格与交易前的最新报价进行对比，可将交易记录与报价记录进行asof join。如下所示：

trades=table(`600000`600300`600800 as symbol, take(2020.06.01,3) as date, [14:35:18.000, 14:30:30.000, 14:31:09.000] as time, [10.63, 3.12, 4.72] as tradePrice)
quotesTmp=select symbol, date, time, bidPrice1, askPrice1 from quotes where symbol in `600000`600300`600800 and date=2020.06.01
select * from aj(trades, quotesTmp, `symbol`date`time)

结果为：

symbol date       time         tradePrice quotesTmp_time bidPrice1 askPrice1
------ ---------- ------------ ---------- -------------- --------- ---------
600000 2020.06.01 14:35:18.000 10.63      14:35:17.000   10.62     10.63
600300 2020.06.01 14:30:30.000 3.12       14:30:29.000   3.12      3.13
600800 2020.06.01 14:31:09.000 4.72       14:31:09.000   4.72      4.73

从4.3节中，可见不同的股票行情数据更新的时刻可能不同。若要为某只股票每一个数据更新的时刻，提供另外一只股票在此时刻的最新报价数据，可使用asof join。

t1 = select symbol, date, time, askPrice1, bidPrice1 from quotes where date=2020.06.01, symbol=`600000, time between 09:30:00.000 : 15:00:00.000 
C600300 = select date, time, askPrice1, bidPrice1 from quotes where date=2020.06.01, symbol=`600300, time between 09:30:00.000 : 15:00:00.000 
t = aj(t1, C600300, `date`time)

若要对某天中所有记录，计算之前一分钟之内的总交易量，可使用window join。以下脚本中，假设只保留原表中的symbol, date, time, curVol, askPrice1, bidPrice1这几列。

t=select symbol, date, time, curVol, askPrice1, bidPrice1 from quotes where date=2020.06.01, symbol>=`600000, time between 09:30:00.000 : 15:00:00.000 order by symbol, date, time
t1 = wj(t, t, -60000:-1, <sum(curVol) as sumVolumePrev1m>, `symbol`date`time)

若要对某天中所有记录，寻找一分钟之后第一笔第一档报价，可使用以下脚本。假设只保留原表中的symbol, date, time, askPrice1, bidPrice1这几列。

t=select symbol, date, time, askPrice1, bidPrice1 from quotes where date=2020.06.01, symbol>=`600000, time between 09:30:00.000 : 15:00:00.000 order by symbol, date, time
t1 = wj(t, t, 60000:70000, <[first(askPrice1) as firstAskPrice1In1m, first(bidPrice1) as firstBidPrice1In1m]>, `symbol`date`time)

高阶函数与其它函数的不同之处在于，高阶函数的变量之一为一个函数。输入的数据被分解成多个数据块（可能重叠），然后将函数应用于每个数据块，最后将所有的结果组合为一个对象返回。某些复杂的分析任务如果使用高阶函数，可大大减少代码量。

以下代码计算2020年6月1日交易量最大的100只股票的分钟级收益率的两两相关性。

minuteBar = select first(last) as open, max(last) as high, min(last) as low, last(last) as last, sum(curVol) as volume from quotes where date=2020.06.01, symbol>=`600000 group by symbol, date, minute(time) as minute;
syms = (exec sum(volume) from minuteBar group by symbol order by sum_volume desc).symbol[0:100]
priceMatrix = exec last from minuteBar where symbol in syms pivot by minute, symbol
retMatrix = each(def(x):ratios(x)-1, priceMatrix)
corrMatrix = pcross(corr, retMatrix);

首先计算分钟级K线，然后获取交易量最大的100只股票。将分钟级K线数据整理为每分钟价格矩阵（priceMatrix），其中每列为一只股票，每行为一分钟。然后对价格矩阵使用高阶函数each，对每列应用函数ratios(x)-1，将价格矩阵转化为收益率矩阵（retMatrix）。最后对收益率矩阵使用高阶函数pcross，对其每两列应用函数corr以计算其两两相关性。最终结果为100*100的相关性矩阵（corrMatrix）。

有关其它高阶函数以及更多细节，请参考第八章中的高阶函数部分内容。

DolphinDB提供Python, C++, Java, C#, Go, JavaScript, Excel等常用系统的API。

Python是目前最流行的数据分析语言，pandas是Python在金融行业最常用的库。本小节讲述如何使用python API 写入分布式数据库以及从DolphinDB数据库中读取数据。

• 使用Python API将数据写入分布式数据库

import dolphindb as ddb
import dolphindb.settings as keys
import pandas as pd
import numpy as np
dbPath = "dfs://level2API"
tableName = "quotes"
s = ddb.session()
s.connect("127.0.0.1",8848,"admin","123456")

if(s.existsDatabase(dbPath)):
    s.dropDatabase(dbPath)
s.database('dbDate',partitionType=keys.VALUE,partitions="2020.01.01..2020.12.31")
s.database('dbSymbol',partitionType=keys.HASH,partitions="[SYMBOL,10]")
s.database('db', partitionType=keys.COMPO, partitions="[dbDate, dbSymbol]", dbPath=dbPath)

data=pd.DataFrame({"symbol":['600007','600104'],"dt":[np.datetime64('2020-01-01T20:01:01'),np.datetime64('2020-01-01T20:01:02')],"last":[100.36,99.3],"askPrice1":[100.36,101.22], "bidPrice1":[100.35,100.45],"askVolume1":[4138,2],"bidVolume1":[20,39],"volume":[1,5]})
s.run("db.createPartitionedTable(table(1:0, `symbol`datetime`last`askPrice1`bidPrice1`askVolume1`bidVolume1`volume, [SYMBOL,DATETIME,DOUBLE,DOUBLE,DOUBLE,INT,INT,INT]),`quotes, `datetime`symbol)")
s.upload({"tmpdata":data})
s.run("data=select symbol, datetime(dt) as datetime, last, askPrice1, bidPrice1, askVolume1, bidVolume1, volume from tmpdata")
s.run("tableInsert(loadTable('{db}','{tb}'),data)".format(db=dbPath,tb=tableName))

• 使用Python API将从分布式数据库中读取数据

import dolphindb as ddb
import pandas as pd
dbPath = "dfs://level2API"
tableName = "quotes"
s = ddb.session()
s.connect("127.0.0.1",8848,"admin","123456")
t = s.run("select * from loadTable('{db}','{tb}') where symbol=`600007".format(db=dbPath,tb=tableName))
print(t)

• orca 是基于Python API的完全实现pandas接口的库，下面的例子展示在 orca 中如何查询数据：

import dolphindb.orca as orca
orca.connect("127.0.0.1", 8848, "admin", "123456")
odf = orca.read_table("dfs://level2API", "quotes")
t = odf[(odf["bidVolume1"]>5*odf["askVolume1"]) & (odf["askVolume1"]>1000000)].compute()
print(t)

更多信息与范例，请参考 Python API教程 与 orca教程。

• DolphinDB中建立流数据表。以下为DolphinDB脚本：

share streamTable(100:0, `symbol`datetime`last`askPrice1`bidPrice1`askVolume1`bidVolume1`volume, [SYMBOL,DATETIME,DOUBLE,DOUBLE,DOUBLE,INT,INT,INT]) as level2

• 使用API将实时数据写入DolphinDB流数据表

在实际使用环境中，通常是先通过程序从数据提供商订阅实时数据，然后通过API写入到DolphinDB。本例使用Python API来展示这一过程，实际使用中用实时数据代替示例中的data变量。以下为Python代码。

import dolphindb as ddb
import pandas as pd
import numpy as np
data = [['600007','600104'],[np.datetime64('2019-01-01T20:01:01'),np.datetime64('2019-01-01T20:01:02')],[100.36,99.3],[100.36,101.22], [100.35,100.45],[4138,2],[20,39],[1,5]]
s = ddb.session()
s.connect("127.0.0.1",8848,"admin","123456")
s.run("tableInsert{level2}",data)

• 使用数据回放功能将历史数据以指定速率写入流表

使用replay函数，可将历史数据以一定的速率注入到流数据表中，实现数据回放。结合流数据处理引擎或自定义函数使用，可实现基于历史数据的高频策略回测。具体回放函数的使用可参考数据回放教程。

在本例中的后续示例中，会使用历史数据回放来代替实时数据。即使在未实现三方数据源接口的情况下，也可以快速体验DolphinDB实时数据处理的功能。

在使用回放功能之前，首先需要准备好供回放的历史数据，本例中利用3.1导入的历史数据来生成回放的分布式表。

dbDate = database("", VALUE, 2020.01.01..2020.12.31)
dbSymbol=database("", HASH, [SYMBOL, 10])
db = database("dfs://level2Replay", COMPO, [dbDate, dbSymbol])
modal = table(1:0, `symbol`datetime`last`askPrice1`bidPrice1`askVolume1`bidVolume1`volume, [SYMBOL,DATETIME,DOUBLE,DOUBLE,DOUBLE,INT,INT,INT])
db.createPartitionedTable(modal,`quotes, `datetime`symbol)
data = select symbol, datetime(datetime(date(date))+second(time)) as datetime, last, askPrice1, bidPrice1, askVolume1, bidVolume1, curVol as volume from loadTable("dfs://level2","quotes")
loadTable("dfs://level2Replay","quotes").append!(data)

使用以下脚本，可以以10倍的速率将已保存到分布式表的报价数据回放到level2流数据表中：

quotes = loadTable("dfs://level2Replay","quotes")

//设置每次提取到内存数据量=1小时
repartitionSchema = time(cutPoints(08:00:00..18:00:00,10))
inputDS = replayDS(<select * from quotes>, `datetime, `datetime,  repartitionSchema)
submitJob("replay_quotes", "replay_quotes_stream",  replay,  [inputDS],  [`level2], `datetime, `datetime, 10, false, 2)

当回放函数被提交后，它在后台运行，可以通过getRecentJobs()函数查看已经提交的回放任务情况，通过cancelJob(jobid)取消回放任务。

本示例脚本下载

本小节介绍如何使用流数据时序聚合引擎createTimeSeriesAggregator函数实时计算K线。

时间窗口不重合，可将createTimeSeriesAggregator函数的windowSize参数和step参数设置为相同值。时间窗口部分重合，可将windowSize参数设为大于step参数。请注意，windowSize必须是step的整数倍。

场景一：每隔5分钟计算过去5分钟的K线数据

modal = table(100:0, `symbol`datetime`last`askPrice1`bidPrice1`askVolume1`bidVolume1`volume, [SYMBOL,DATETIME,DOUBLE,DOUBLE,DOUBLE,INT,INT,INT])
share streamTable(100:0, `datetime`symbol`open`high`low`close`volume,[DATETIME,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG]) as OHLC1
tsAggr1 = createTimeSeriesAggregator(name="tsAggr1", windowSize=300, step=300, metrics=<[first(last),max(last),min(last),last(last),sum(volume)]>, dummyTable=modal, outputTable=OHLC1, timeColumn=`datetime, keyColumn=`symbol)
subscribeTable(tableName="level2", actionName="act_tsAggr1", offset=0, handler=append!{tsAggr1}, msgAsTable=true);

场景二：每隔1分钟计算过去5分钟的K线数据

modal = table(100:0, `symbol`datetime`last`askPrice1`bidPrice1`askVolume1`bidVolume1`volume, [SYMBOL,DATETIME,DOUBLE,DOUBLE,DOUBLE,INT,INT,INT])
share streamTable(100:0, `datetime`symbol`open`high`low`close`volume,[DATETIME,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG]) as OHLC2
tsAggr2 = createTimeSeriesAggregator(name="tsAggr2", windowSize=300, step=60, metrics=<[first(last),max(last),min(last),last(last),sum(volume)]>, dummyTable=modal, outputTable=OHLC2, timeColumn=`datetime, keyColumn=`symbol)
subscribeTable(tableName="level2", actionName="act_tsAggr2", offset=0, handler=append!{tsAggr2}, msgAsTable=true);

本示例脚本下载

除了流数据聚合引擎，用户亦可使用自定义函数进行更复杂的实时计算。下面的例子是将实时数据与历史数据结合来计算因子。

本例中计算的因子定义为：前100笔记录与前200笔记录的资金净流入（net_amount）之比。计算流程为：订阅level2实时数据，调用因子计算函数factorHandler计算每只股票的因子值。在算法设计中，为了优化历史数据的存取性能，设计了一个字典（d）来存储每个symbol的历史资金净流入值（net_amount）。类似如下格式：

//字典定义 {key:symbol, value: [net_amount]}, 
600000->[-85,-69,-32,-57,80,-54,71,87,43,45,...]
600300->[-33,49,16,21,-82,-30,68,-44,-58,-66,...]
600400->[13,35,-67,-30,49,44,85,-10,-42,96,...]
600500->[-15,78,18,17,13,-70,67,-31,-19,-78,...]
600600->[65,-64,-90,-75,85,-99,15,83,-85,60,...]
600800->[14,-78,96,93,-89,92,99,-96,21,93,...]

当实时数据到达后，通过dictUpdate!将最新计算得到的net_amount更新到每个symbol的数据向量中，得到一个包含当天历史和最新net_amount数据的字典。针对这个字典通过循环调用自定义函数calcNetAmountRatio来计算每只股票的因子值。

• 核心计算脚本如下

//定义函数calcNetAmountRatio，对一个向量求前n个与前2n个元素之和的比值：
defg calcNetAmountRatio(x,n){
	size = x.size()
	return x.subarray((size - n):size).sum()\x.subarray((size - 2*n):size).sum()
}

//因子计算函数
def factorHandler(mutable factorTable,mutable d, facName,msg){
		codeList = msg.symbol.distinct()
		symbolCount = codeList.size()
		//资金净流入（net_amount）= volume * iif(bidPrice1>=askPrice1, 1, -1)
		t2 = select symbol, volume * iif(bidPrice1>=askPrice1, 1, -1) as net_amount from msg
		//将本次数据的计算net_amount追加更新字典
		dictUpdate!(d,append!, t2.symbol, t2.net_amount)
		//计算因子
	  	factorValue = array(DOUBLE,symbolCount)
		for(i in 0:symbolCount){
			factorValue[i] = calcNetAmountRatio(d[codeList[i]],100)
		}
		//添加时间戳，写入因子结果表
		factorTable.append!(table(take(now(),symbolCount) as timestamp, codeList as symbol,factorValue as value, take(facName,symbolCount) as factorName))
}

• 下列脚本完成数据的定义以及预处理

  • 定义保存因子的结果表

  share(streamTable(100:0, `timestamp`symbol`value`factorName,[TIMESTAMP,SYMBOL,DOUBLE,SYMBOL]),"FACTOR")

  • 预先构造历史的net_amount数据字典，此处利用了3.1的历史数据表。

  	d = dict(STRING, ANY)
  	his = select symbol,volume * iif(bidPrice1>=askPrice1, 1, -1) as net_amount from loadTable("dfs://level2","quotes") context by symbol limit -200
  	for(id in his[`symbol].distinct())
  		d[id]= exec net_amount from his where symbol == id

• 设置level2数据的订阅，建立实时数据和因子处理函数的关系。

subscribeTable(tableName="level2", actionName="act_factor", offset=0, handler=factorHandler{FACTOR,d,"factor1"}, msgAsTable=true, batchSize=4000, throttle=1)

订阅函数的调用需要注意以下几点：

• 部分应用的使用。因为订阅的handler必须是一个一元函数，仅接受一个msg参数(msg即订阅到的实时数据)，所以当因子处理函数需要多个参数时，需要将除msg之外的参数通过部分应用来固化，生成一个符合要求的一元函数。
• batchSize和throttle参数的使用。若消息的数量达到batchSize或者距离上次计算的时间间隔达到throttle这两个条件任意一个符合，都会触发handler。

本示例脚本下载

某些场景只需要每只股票最新的一条level 2数据。为了保证此类场景的最佳查询性能，可使用键值表订阅流数据表，并指定键值为股票代码。

newestLevel2 = keyedTable(`symbol, 100:0, `symbol`datetime`last`askPrice1`bidPrice1`askVolume1`bidVolume1`volume, [SYMBOL,DATETIME,DOUBLE,DOUBLE,DOUBLE,INT,INT,INT])
subscribeTable(tableName="level2", actionName="newestLevel2data", offset=0, handler=append!{newestLevel2}, msgAsTable=true)

键值表newestLevel2中，每只股票的数据仅为最新的一条level 2数据。

本示例脚本下载