基于Storm、Esper CEP及Spark构建异常交易行为风险监控系统

1、 基于Storm、Esper CEP及Spark构建异常交易行为风险监控系统 1 概述2015年以来，随着股市的持续火爆，中国证监会、证券业协会先后发布相关文件，要求对证券公司外部信息系统接入风险加强行业监管和自律。其中要求证券公司应当建立投资者证券、资金账户风险监测模型，动态监测账户交易行为特征、客户交易终端定位信息、账户资金进出等情况，及时发现借助信息系统外部接入、非法接入网上交易接口等方式开展的违规拆分账户、出借账户或违反账户实名制等违法违规行为线索，并及时予以处理。同时，根据交易所的相关规定，对于出现某些异常交易行为的，交易所会对相关投资者发出书面警示，或者直接采取暂停投资者账户当日交易、限制投资者账户交易等措施。为了更好地应对此类风险，我们利用最新的大数据相关技术，实现对海量交易数据进行各类复杂的实时与离线相结合的异常交易行为分析。2 业务需求2.1 事前风控事前风控可对交易指令先进行风险监控，通过监控的交易指令才提交给交易系统进行处理，未通过监控的交易指令将直接予以拒绝。事前风控要求处理时间很短（5ms以内）。业务应用方面，事前风控一般用于对存在较高风险的特定接入渠道和特定

2、投资者，在满足合规要求的前提下进行风险监控。事前风控的特点是：1、与交易系统耦合极高需要通过对所有发送到交易系统的指令进行拦截，并进行筛选，通过监控的重新放回请求队列，不通过的直接拒绝。2、低延时、高可靠性要求由于事前风控会对所有交易指令进行过滤，对可靠性要求与交易系统一致。同时，事前风控模块需要对处理时间做严格控制，如果事前风控模块在指定时间没有返回结果，则默认通过风控，以避免对交易产生影响。2.2 事中风控事中风控是本系统最核心的风控手段。交易指令在提交给交易系统进行处理的同时，旁路同样的指令到事中风控子系统进行分析处理；当触发事中风控规则后，系统自动进行报警，由人工进行处置。事中风控一般需要在短时间内（50ms以内）对交易数据流做出分析结果。业务应用方面，主要分为两类：1、对于经评估和实际验证合格的特定接入渠道和特定投资者，公司与其签署专项交易风险管理协议，并通过事中风控子系统对其交易行为实时监控。2、针对所有用户，可对其在单个业务交易系统或者多个业务交易系统（一般是指集中交易和融资融券业务）上的交易行为实时监控。下表中列举了第2类应用中典型的事中风控规则。1.png事中风控的特

3、点是：1、与交易系统耦合低通过旁路的方式接入到交易系统中，对交易系统的耦合相对较低。2、可实现相对复杂的风控处理规则2.3 事后风控某些复杂的交易行为需要通过复杂的模型和规则进行分析，通过事前或事中风控不适合进行处理，这些风险模型需要通过多维度、较长时间的数据（如最近一个月的数据）进行分析，发现可疑的异常交易行为。下表中列举了部分典型的事后风控规则。2.png事后风控的特点是：1、离线分析，计算量大，依赖长时间用户交易数据。2、分析结果可用来作为事前、事中风控规则中的基准数据。3 技术实现3.1 系统架构本系统整体架构如下图所示：3.png核心流程简介：本系统通过交易指令处理对接实际的交易系统，将接收的交易指令解析后交由风控系统（本处主要指事前风控和事中风控）进行风控处理，处理时可能需要依赖hadoop数据仓库/行情数据/CDC（change data capture）数据库实时流水数据等外部数据协助分析。事后风控子系统是相对独立的系统，主要是通过Spark（SparkSQL、MLlib）分析hadoop数据仓库或ElasticSearch交易指令流水数据进行风控处理。风控处理后若出现

4、告警事件，则交由风控web管理子系统进行告警展示和后续处置，web管理子系统同时负责风控规则的管理。3.2 本系统使用的开源软件概述3.2.1 Apache StormStorm是一个免费开源、分布式、高容错的实时计算系统。Storm使持续不断的流计算变得容易，弥补了Hadoop批处理所不能满足的实时要求。Storm主要分为Nimbus和Supervisor两种组件。Nimbus负责在集群里面发送代码，分配工作给机器，并且监控状态。全局只有一个。Supervisor会监听分配给它那台机器的工作，根据需要启动/关闭工作进程Worker。每一个要运行Storm的机器上都要部署一个，并且，按照机器的配置设定上面分配的槽位数。Storm提交运行的程序称为Topology。Topology处理的最小的消息单位是一个Tuple，也就是一个任意对象的数组。Topology由Spout和Bolt构成。Spout是发出Tuple的结点。Bolt可以随意订阅某个Spout或者Bolt发出的Tuple。在本系统中，我们使用Storm作为事中风控的流式处理框架进行风控规则的分析计算及告警处理。3.2.2 Es

5、per CEPEsper是java开发的事件流处理和复杂事件处理引擎。其中事件流处理（ESP）是一种从大量事件数据流中过滤，分析有意义的事件，并能够实时取得这些有意义的信息的技术。而复杂事件处理（CEP）是一种实时事件处理并且从大量事件数据流中挖掘复杂模型的技术；Esper CEP可以结合多种数据源的数据对信息流进行监测、分析从推理出一些复杂的事件或模式，CEP的目的是识别出一些有意义的事件，例如：机遇、威胁，并且尽可能快的作出反应。Esper CEP目前支持大约30种数据窗口，下表列出常用的几种窗口类型：4.pngEsper CEP开发使用的事件处理语言（EPL）是SQL标准语言并做了扩展，提供了SELECT、 FROM、 WHERE、 GROUP BY、HAVING、ORDER BY及JOIN等功能。本系统中，我们使用Esper CEP用做事前风控的规则实时计算，并且在事中风控中将Esper CEP与Apache Strom集成，使规则实时计算集群化。3.2.3 Apache SparkApache Spark作为一个开源数据处理框架，它在数据计算过程中把中间数据直接缓存到内存里，

6、能大大提高处理速度，特别是复杂的迭代计算。Spark主要包括SparkSQL，SparkStreaming，Spark MLlib以及图计算。Spark SQL 是操作结构化数据的组件，可使用 SQL 访问分布式数据仓库。 Spark Streaming 是对实时数据进行流式计算的组件，适合分布式环境数据采集和分析。MLlib 是机器学习程序库，提供分类、回归、聚类、协同过滤、模型评估等一系列算法实现 。GraphX 是图计算程序库，适合于社交网络关系计算。事后风控子系统中主要应用Spark SQL及MLlib进行分析处理。3.3 技术实现3.3.1 交易指令处理指令处理用于处理与交易系统的请求/应答指令交互。其中，事前风控中该模块需要根据风控结果将过滤后的指令回放到交易系统指令请求队列或者将应答指令回扔到交易系统指令应答队列；事中风控中本模块仅用于请求指令的旁路复制。我们以金证集中交易系统的对接为例进行说明。金证集中交易系统逻辑架构图如下图所示：5.png交易指令处理模块通过对接KCXP Server实现指令拦截/旁路复制/指令应答等处理。与KCXP Server的交互通过KCXP

7、Client API编程实现。逻辑对接拓扑如下图所示：6.png各消息队列的说明如下表所示：7.png其他交易系统的对接也是类似的对接拓扑和实现方法。3.3.2 事前风控子系统用于对接收到的交易指令进行分析，该系统通过集成Esper CEP引擎可以分析相对复杂的风控规则，当然，如果风控规则特别简单，也可以不用CEP引擎实现，硬编码实现风控逻辑也是可行的。该子系统分析之后将风控分析结果传递给指令处理子系统。事前风控对响应时间有严格要求，所以通常采用单机模式而非集群模式进行部署，以减小网络交互，该系统的扩容主要通过垂直扩容来实现。考虑到Esper CEP引擎基于全内存计算，其处理能力受限于机器的内存资源，事前风控不建议处理特别复杂且消息缓存时间长的风控规则。3.3.3 事中风控子系统该子系统用过Apache Storm + Esper CEP对接收的请求做相对复杂的风控规则处理。从技术角度来说，事中风控是借助Storm的流式处理框架将Esper CEP处理从事前风控的单机模式扩展为集群模式，可从容应对海量的交易指令处理并且轻松扩容，并通过Storm框架内的失败重试/任务跟踪等简化开发工作。

8、现阶段在我们的系统中，事中风控是核心的风控手段。下面我们以风控规则盘中异常申报（拉抬打压）实现来描述实现的整个过程。3.3.3.1 规则描述8.png3.3.3.2 数据准备该规则依赖的数据如下所示：1、成交回报数据成交回报(Deal)消息数据格式如下表所示：9.png实际收到的json格式的成交回报消息示例如下：“ymt_code”:“100000000”,“orgid”:“0862”,“timestamp”:1481107919572,“gddm”:“0216840317”,“gdxm”:“”,“cjbh”:“00D0OHQI5P015Z4I”,“bcye”:“”,“gsdm”:“391188”,“cjjg”:“2.4400”,“cjje”:“854.0”,“bcrq”:“”,“cjsl”:“350.00”,“cjsj”:“20161207150000”,“mjbh”:“”,“market”:“0”,“sqbh”:“”,“bs”:“0S”,“sbsj”:“”,“custid”:“86262000596”,“zqdm”:“131810”,“tradeType”:“ptjy”2、行情数

9、据行情(Quotation)消息数据格式如下表所示：10.png实际收到的json格式的行情消息示例如下：“stkcode”:“300453”,“timestamp”:“1481091771203”,“tradevolume”:“881014”,“trademoney”:“22686285.120”,“currentprice”:“26.070”成交回报指令通过交易系统请求队列中实时复制出来，所以该指令的获取基本跟用户下单操作是同步进行。实时行情数据主要是最新成交价格以及成交数量，该数据可以通过对接行情服务获得。3.3.3.3 告警阈值本规则涉及到的阈值及其描述如下表所示：11.png3.3.3.4 Esper CEP 实现该规则的EPL语句实现如下：12.png3.3.3.5告警根据设定的EPL规则，当如下示例中的成交情况出现时，会立刻触发告警。某个交易日内14:42:0114:43:54时间范围内，一码通号为10000000的投资者买入股票600031且成交情况如下表所示，假如该股票最新成交价及成交数量如下：1、14:43:55这一刻，该股票的最新成交价上涨为19元2、14:41:5514:43:55该3分钟时间范围内该股票总成交量为10000000股那么该风控规则将会在14:43:55这一刻被触发。因为该时间段的成交满足如下条件：1、3分钟内累计成交量=30万或3分钟内累计成交量成交金额=200万3、3分钟内证券成交量比例值=30% 成交比例为4800000/10000000=48%13.png告警的指令详情可在风控web管理子系统中进行查看。3.3.4 事后风控子系统假若部分风控规则需要依赖相当长时间或海量的用户行为数据进行分析，则需

《基于Storm、Esper CEP及Spark构建异常交易行为风险监控系统》由会员Baige****0346分享，可在线阅读，更多相关《基于Storm、Esper CEP及Spark构建异常交易行为风险监控系统》请在金锄头文库上搜索。