文章同时发布在个人博客：基于SQL的实时股票分析
概述
这次分享的是在给某券商做实时流处理案例，基于我们公司的StreamSQL流处理功能和他们的业务数据，做的股票实时分析功能。
主要功能，采集DBF实时股票交易数据并写入消息队列，使用StreamSQL组件，实时接受消息队列中数据，并进行统计分析。
做的流处理功能有：
- 保存实时明细数据，每天4个小时的时间，约为500-600万条。
- 计算当前成交额排名靠前的股票（实际效果跟同花顺看到的成交额一致）
- 实时计算5分钟内，成交额靠前股票
流程架构图
</a> 流程架构图[/caption]
由于时间限制，并没有做更深入的功能。其实结合券商其他数据和实时数据，可以做很多有价值的实时分析（例如：离线数据模型训练，实时放量股票和模型数据结合分析做股票推荐），给券商产品提供基础数据支撑。
Stream SQL介绍
Transwarp Stream是星环专为企业级用户打造的流计算引擎，主要应用于实时性较强的应用场景。比如，金融行业需要对市场波动进行实时预警；交通行业需要将卡口数据实时入库，并在线使用图像识别鉴别套牌车进行预警等；银行业务需要在线分析业务，及时鉴别欺诈等违规行为；采用复杂物联网的行业，如机场、风电等，需要将大量传感器数据进行实时分析和数据挖掘。
企业级用户往往对流处理产品在实时性、吞吐量、高可用性、易用性、安全性和稳定性等方面有着极其苛刻的要求。星环凭借自身强大的技术实力，以及国内最多最复杂的流处理案例经验，开发出满足这些苛刻要求的Transwarp Stream流计算引擎：
Transwarp Stream架构图
更多StreamSQL产品的介绍，可以参考星环科技Transwarp Stream：业界SQL支持最强 可实时数据挖掘的流计算引擎
数据采集
股票实时数据交易数据一般会存入DBF文件，这个格式证券行业已经沿用20多年。
数据采集部分，包含功能：
- 定时解析dbf文件（判断dbf文件是否变化）
- 写入kafka消息队列，为后续streamSQL提供数据
注意，需要引入解析dbf格式的jar包，dbf.jar
部分代码如下：
InputStream fis = null;
// 读取文件的输入流
fis = new FileInputStream(path);
// 根据输入流初始化一个DBFReader实例，用来读取DBF文件信息
DBFReader reader = new DBFReader(fis);
reader.setCharactersetName("gbk");
// 调用DBFReader对实例方法得到path文件中字段的个数
int fieldsCount = reader.getFieldCount();
// 取出字段信息
// for (int i = 0; i < fieldsCount; i++) {
// DBFField field = reader.getField(i);
// // logger.info(field.getName() + "\t");
// }
Object[] rowValues;
int num = 1;
String time = null;
A: while ((rowValues = reader.nextRecord()) != null) {
//提出DBF中的当前时间，同时判断文件是否修改
if (num == 1 && rowValues[0].equals("000000")) {
Double t = (Double) rowValues[7];
time = rowValues[1] + "" + t.intValue();
if (now == null) {
now = time;
continue A;
} else if (now.equals(time)) {
break A;
} else {
now = time;
continue A;
}
}
num++;
StringBuffer message = new StringBuffer();
message.append(time + ",");
for (int i = 0; i < rowValues.length; i++) {
message.append(rowValues[i] + ",");
}
logger.info(message.toString());
producer.send(new KeyedMessage<Integer, String>(topic, message.toString()));
}
流处理
流处理部分，使用StreamSQL组件，编写SQL完成。
明细保存StreamJOB
由于明细后续需要做统计分析，所以把数据存储在基于内存和SSD存储的列式存储holodesk组件中，语句
create streamjob holo_detail_stream_job as ("insert into holo_stream_zq_detail select * from stream_demo") JOBPROPERTIES('stream.number.receivers'='4');
5分钟成交量StreamJOB
基于明细数据，实时计算5分钟内成交额
create streamjob holo_count_stream_job as ("insert into hb_stream_holo select concat(TDH_TODATE(created,'yyyyMMddHHmmss','yyyyMMdd'),row_number() OVER(ORDER BY cjl desc)),created,hqzqdm,HQZQJC,cjl from (select max(created) as created,hqzqdm,HQZQJC,sum(HQZJCJ*HQCJBS) as cjl from (select created,hqzqdm,HQZQJC,HQZJCJ,HQCJBS from holo_stream_zq_detail union select created,hqzqdm,HQZQJC,HQZJCJ,HQCJBS from stream_demo)holo_stream_zq_detail where TDH_TODATE(created,'yyyyMMddHHmmss','yyyy-MM-dd HH:mm:ss:SSS')>CAST(sysdate-TO_MINUTE_INTERVAL(5) AS STRING) group by hqzqdm,HQZQJC order by cjl desc limit 20) t") JOBPROPERTIES('stream.number.receivers'='4');
start streamjob holo_count_stream_job;
计算结果展示
- 计算时，只采集了深圳A股和创业板的数据，计算当前成交额排名靠前的股票，实际效果跟同花顺看到的成交额一致。
- 实时计算5分钟内，成交额靠前股票。这个在同花顺也没有统计，有一点点的参考价值，基于这个，还可以做实时放量分析。
</a>
<a href="http://infullstack.com/?attachment_id=186" rel="attachment wp-att-186">
</a>
报表展示
有了实时分析结果，再结合报表工具的股价图，做了实现的展现（图中数据，缺失盘高盘底数据）。
<a href="http://infullstack.com/?attachment_id=185" rel="attachment wp-att-185">
</a>
总结
基于StreamSQL对SQL的完整支持，和实时性、吞吐量、高可用性、易用性等特性，让实时分析变的更简单。简单的实现数据采集工作，根据需求，随时调整统计SQL，就可以完成实时的分析。
当然这次案例，让我对股票有了更深的认识。
参考链接
更多StreamSQL产品的介绍，可以参考
网友评论