文章同时发布在个人博客：基于SQL的实时股票分析

概述

这次分享的是在给某券商做实时流处理案例，基于我们公司的StreamSQL流处理功能和他们的业务数据，做的股票实时分析功能。

主要功能，采集DBF实时股票交易数据并写入消息队列，使用StreamSQL组件，实时接受消息队列中数据，并进行统计分析。

做的流处理功能有：

1. 保存实时明细数据，每天4个小时的时间，约为500-600万条。
2. 计算当前成交额排名靠前的股票（实际效果跟同花顺看到的成交额一致）
3. 实时计算5分钟内，成交额靠前股票

流程架构图

</a> 流程架构图[/caption]

由于时间限制，并没有做更深入的功能。其实结合券商其他数据和实时数据，可以做很多有价值的实时分析（例如：离线数据模型训练，实时放量股票和模型数据结合分析做股票推荐），给券商产品提供基础数据支撑。

Stream SQL介绍

Transwarp Stream是星环专为企业级用户打造的流计算引擎，主要应用于实时性较强的应用场景。比如，金融行业需要对市场波动进行实时预警；交通行业需要将卡口数据实时入库，并在线使用图像识别鉴别套牌车进行预警等；银行业务需要在线分析业务，及时鉴别欺诈等违规行为；采用复杂物联网的行业，如机场、风电等，需要将大量传感器数据进行实时分析和数据挖掘。

企业级用户往往对流处理产品在实时性、吞吐量、高可用性、易用性、安全性和稳定性等方面有着极其苛刻的要求。星环凭借自身强大的技术实力，以及国内最多最复杂的流处理案例经验，开发出满足这些苛刻要求的Transwarp Stream流计算引擎：

Transwarp Stream架构图

更多StreamSQL产品的介绍，可以参考星环科技Transwarp Stream：业界SQL支持最强 可实时数据挖掘的流计算引擎

数据采集

股票实时数据交易数据一般会存入DBF文件，这个格式证券行业已经沿用20多年。

数据采集部分，包含功能：

1. 定时解析dbf文件（判断dbf文件是否变化）
2. 写入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;

计算结果展示

1. 计算时，只采集了深圳A股和创业板的数据，计算当前成交额排名靠前的股票，实际效果跟同花顺看到的成交额一致。
2. 实时计算5分钟内，成交额靠前股票。这个在同花顺也没有统计，有一点点的参考价值，基于这个，还可以做实时放量分析。

</a>

</a>

报表展示

有了实时分析结果，再结合报表工具的股价图，做了实现的展现（图中数据，缺失盘高盘底数据）。

</a>

总结

基于StreamSQL对SQL的完整支持，和实时性、吞吐量、高可用性、易用性等特性，让实时分析变的更简单。简单的实现数据采集工作，根据需求，随时调整统计SQL，就可以完成实时的分析。

当然这次案例，让我对股票有了更深的认识。

参考链接

更多StreamSQL产品的介绍，可以参考

• 星环科技Transwarp Stream：业界SQL支持最强 可实时数据挖掘的流计算引擎
• 星环官网