streaming_ta.md · tangxinhai/DolphinDB

# 如何利用技术分析（Technical Analysis）指标库中的函数实现实时流计算

在DolphinDB V1.30.3中发布的[响应式状态引擎（Reactive State Engine）](https://gitee.com/dolphindb/Tutorials_CN/blob/master/reactive_state_engine.md)是许多金融场景**流批统一计算**中的重要构件，[DolphinDB 技术分析（Technical Analysis）指标库](https://gitee.com/dolphindb/DolphinDBModules/tree/master/ta/README.md)在开发时也对其进行了适配，使得ta模块可以在响应式状态引擎中实现**流式增量计算**。本教程介绍如何使用ta模块、DolphinDB的时序聚合引擎、DolphinDB的响应式状态引擎、Grafana监控信息可视化工具，实现下面的**实时流计算**场景：

* 首先把股票快照数据实时合成1分钟bar
* 再使用响应式状态引擎计算ta中的技术分析指标
* 最后通过grafana对股票的技术分析指标进行可视化监控

> 提示：本教程基于DolphinDB V2.00.6, V1.30.18开发，需要下载2.00.6或1.30.18及以上版本进行教程内容测试。

**相关参考资料**

* [技术分析（Technical Analysis）指标库](https://gitee.com/dolphindb/DolphinDBModules/tree/master/ta/README.md)
* [DolphinDB的时序聚合引擎](https://www.dolphindb.cn/cn/help/200/FunctionsandCommands/FunctionReferences/c/createTimeSeriesEngine.html)
* [DolphinDB的响应式状态引擎](https://www.dolphindb.cn/cn/help/200/FunctionsandCommands/FunctionReferences/c/createReactiveStateEngine.html)
* [Grafana监控信息可视化工具](https://gitee.com/dolphindb/grafana-datasource/blob/master/README.zh.md)

本教程包含内容：

- [1. 流处理流程](#1-流处理流程)
- [2. 快速复现案例](#2-快速复现案例)
- [附录](#附录)

## 1. 流处理流程

![](images/streaming_ta/reactiveStateEnginee.png)

**Step1**：订阅实时获取`snapshotStream`表中的数据，利用DolphinDB内置的时序聚合计算引擎，进行窗口为1min，步长为1min的滚动窗口计算，核心代码为：

- 注册时间序列聚合计算引擎
```
createTimeSeriesEngine(name="aggr1min", windowSize=60000, step=60000, metrics=<[first(open),max(high),min(low),last(close),sum(vol)]>, dummyTable=snapshotStream, outputTable=getStreamEngine("taReactiveStateEngine"), timeColumn=`tradetime, useWindowStartTime=true, keyColumn=`SecurityID)
```
- 订阅snapshotStream流数据表中的实时增量数据

```
subscribeTable(tableName="snapshotStream", actionName="aggr1min", offset=-1, handler=getStreamEngine("aggr1min"), msgAsTable=true, batchSize=2000, throttle=1, hash=0, reconnect=true)
```

**Step2**：时序聚合计算引擎的计算结果直接输出至响应式状态计算引擎,使用**ta指标库**进行增量计算，并将最终结果写入`outputTable`表中核心代码为：

```
reactiveStateMetrics=<[
tradetime,
	ta::roc(close, 5) as `ROC,
	ta::rsi(close, 5) as `RSI,
	ta::cci(high, low, close, 5) as `CCI,
	ta::willr(high, low, close, 5) as `WILLR
]>
taReactiveStateEngine = createReactiveStateEngine("taReactiveStateEngine",metrics=reactiveStateMetrics,dummyTable=snapshotStream,outputTable=outputTable,keyColumn=`SecurityID,keepOrder=true)
```

**Step3**：实时将计算结果表`outputTable`中的数据推送给外部消费者消费，如：Grafana

## 2. 快速复现案例

复现数据为 600519,601319 一天的[快照数据](data/streaming_ta/taTestData.csv)，共9507条。

- 通过replay功能，回放snapshot数据至snapshotStream表
- 通过时间聚合引擎将快照数据聚合降采样至1min频率
- 通过响应式引擎计算roc, rsi, cci, willr指标
- 通过grafana进行数据可视化展示

### 2.1 DolphinDB Server
- 流处理引擎注册、流表订阅与订阅
```
def cleanEnvironment(){
	try{ unsubscribeTable(tableName = "snapshotStream",actionName = "aggr1min") } catch(ex){ print(ex) }
	try{ dropStreamEngine("taReactiveStateEngine") } catch(ex){ print(ex) }
	try{ dropStreamEngine("aggr1min") } catch(ex){ print(ex) }
	try{ dropStreamTable(`snapshotStream) } catch(ex){ print(ex) }
	try{ dropStreamTable(`outputTable) } catch(ex){ print(ex) }
	
	undef all
}
cleanEnvironment()
go
//load modules
use ta

//define stream table
schemaTable = table(`tradetime`SecurityID`high`low`open`close`vol as name,`TIMESTAMP`SYMBOL`DOUBLE`DOUBLE`DOUBLE`DOUBLE`INT as type)
share streamTable(1000000:0, schemaTable.name, schemaTable.type) as snapshotStream
share streamTable(1000000:0,`SecurityID`tradetime`ROC`RSI`CCI`WILLR,
`SYMBOL`TIMESTAMP`DOUBLE`DOUBLE`DOUBLE`DOUBLE) as outputTable

//register stream computing engine
reactiveStateMetrics = <[
tradetime,
ta::roc(close, 5) as `ROC,
ta::rsi(close, 5) as `RSI,
ta::cci(high, low, close, 5) as `CCI,
ta::willr(high, low, close, 5) as `WILLR
]>
taReactiveStateEngine = createReactiveStateEngine("taReactiveStateEngine", metrics = reactiveStateMetrics, dummyTable = snapshotStream, outputTable = outputTable, keyColumn = `SecurityID, keepOrder = true)
createTimeSeriesEngine(name = "aggr1min", windowSize = 60000, step = 60000, metrics = <[first(open), max(high), min(low), last(close), sum(vol)]>, dummyTable = snapshotStream, outputTable = getStreamEngine("taReactiveStateEngine"), timeColumn = `tradetime, useWindowStartTime = true, keyColumn = `SecurityID)
subscribeTable(tableName = "snapshotStream", actionName = "aggr1min", offset = -1, handler = getStreamEngine("aggr1min"), msgAsTable = true, batchSize = 2000, throttle = 1, hash = 0, reconnect = true)
```

- replay模拟实时产生的快照数据

```
csvPath = "/hdd/hdd9/data/streamTa/taTestData.csv"
snapshotdata = select * from loadText(csvPath, schema = schemaTable) order by tradetime, SecurityID
submitJob("replay_trade", "trade",  replay{snapshotdata, snapshotStream, `tradetime, `tradetime, 30, true, 1})
getRecentJobs()
```

> 提示：首先需要把教程的示例csv数据文件上传到DolphinDB server能够访问的路径，然后根据自己上传的路径修改上述代码中的csvPath参数。

### 2.2 Grafana可视化监控

Grafana配置DolphinDB数据源及监控DolphinDB数据表中数据的教程：[Grafana连接DolphinDB数据源](https://gitee.com/dolphindb/grafana-datasource)

- 设置变量进行股票筛选

![](images/streaming_ta/grafana1.png)

- 增加Panel对技术指标进行监控

![](images/streaming_ta/grafana2.png)

```
RSI Query:
select gmtime(timestamp(tradetime)), RSI from outputTable where SecurityID=`${SecurityID}

CCI Query:
select gmtime(timestamp(tradetime)), CCI from outputTable where SecurityID = `${SecurityID}

ROC Query:
select gmtime(timestamp(tradetime)), ROC from outputTable where SecurityID = `${SecurityID}

WILLR Query:
select gmtime(timestamp(tradetime)), WILLR from outputTable where SecurityID =  `${SecurityID}
```

- 可视化展示：

![](images/streaming_ta/grafana3.png)

## 附录

[DolphinDB Server](https://www.dolphindb.cn/alone/alone.php?id=10)

[ta.dos](https://gitee.com/dolphindb/DolphinDBModules/blob/master/ta/src/ta.dos)

[完整代码](script/streaming_ta/streaming_ta.txt)

[数据文件](data/streaming_ta/taTestData.csv)

tangxinhai/DolphinDB

简介

发行版

贡献者

近期动态

tangxinhai/DolphinDB .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

tangxinhai/DolphinDB