一、业务背景
随着现代金融市场交易速度的加快以及交易量的大幅增加，随之而来的是多种交易风险的显著提升。对于交易所而言，对市场、交易等领域的监察能力提出了更高的要求，利用先进的大数据技术实现实时处理和分析交易数据，并识别潜在的、规则化的异常交易模式。因此，多个交易所完成了计算更加高效的具备实时能力的交易监察系统——新一代监察系统，借助实时的数据处理技术手段实现高效的风险控制系统，以便即时识别和管理市场波动带来的风险，如市场操纵、内幕交易或欺诈行为。
新一代监察系统借助高性能的计算资源和优化的数据结构，能够以秒级的时间延迟处理交易数据，实现对客户交易行为的实时监测。通过流处理框架允许系统持续不断地处理数据流，而无需等待批量数据处理周期，这极大地提高了数据处理的实时性和效率，并提供实时监控仪表板和报告功能力，使监管人员能够快速理解市场状况和潜在风险。
二、监察系统实时应用场景
新一代监察系统实时应用包括实时业务查询、实时风险管理、实时异常交易预警等多种类型：
- 第一类：业务查询
新一代监察系统通常依赖高速数据处理引擎，能够实时接收和处理交易委托、成交确认和持仓变动等信息，允许用户根据需要查询特定时间段内的交易委托详情、成交记录和持仓状态。
- 第二类：风险管理
如强平管理：当会员、境外特殊参与者或客户违规，比如未能及时追加保证金、持仓超过交易所规定的限仓标准、或是违反了其他交易规则时，交易所或经纪商有权对其持仓进行强行平仓。 在监察和执行强行平仓的过程中，交易所通过实时计算的能力完成强平量和盈亏额几个关键指标测算。
- 第三类：异常交易
期货异常交易包括以下等多种认定规则：
1）以自己为交易对象，多次进行自买自卖的行为（自成交）；
2）一组实际控制关系账户内的客户之间多次进行相互为对手方的交易；
3）日内出现频繁申报并撤销申报，可能影响期货交易价格或者误导期货市场其他参与者进行期货交易的行为（频繁报撤单）；
4）日内出现多次大额申报并撤销申报，可能影响期货交易价格或者误导期货市场其他参与者进行期货交易的行为（大额报撤单）；
5）单个交易日在某一上市品种或者合约上的开仓交易数量超过交易所制定的日内开仓交易量的；
6）采取程序化交易方式下达交易指令，可能影响交易所系统安全或者正常交易秩序的行为；
新一代监察系统能够借助实时计算的能力通过规则化构建实时应用的方式完成多类异常交易认定行为的即时分析和预警。
三、基于Dataphin的构建过程
本文将重点结合Dataphin介绍如何利用Flink实时计算引擎完成新一代监察系统各类实时应用的构建和落地。
- 监察应用的数据处理方法（以对敲为例）
在期货的对敲、约定交易等异常交易场景下，需要对数据进行一些特殊处理才能确保统计的完整性，对敲/约定交易的规则：A，B客户双开或双平，非主力合约/主力合约，交易价差和数量达到设定值。下文以期货异常交易—对敲为例介绍Flink实时数据处理的内容和常用方法。
假设现在有张三和李四两位期货投资人进行对敲交易，在理论情况下在一个合约下存在四种对敲交易（即互为多空的双开/双平）可能性，如下图所示：
两位投资人互为交易对手的情况下，存在多头开仓与空头开仓、空头平仓与多头平仓的互换交易关系。 常用的数据处理方式如下： 构建一个以张三和李四交易信息结合的编号，如将张三的交易账号（A000001）与李四的交易账号（A000002）构建出一个A000001-A000002的对敲交易编号，且该编号能够覆盖四种对敲交易可能性，如下图所示：
首先在处理期货交易流水时，需要通过按照合约品种、交易数量/金额、交易时间等条件（期货双开必须同时具备买卖双方都是新开仓合约品种相同、数量相等等几点要素）对两条存在对敲行为的交易数据进行合并处理，并通过交易一致性条件判断两个投资人存在的对敲风险交易行为：
1）将交易流水表中的数据按照交易条件一致性进行对敲交易合并。
2）数据合并的操作流程，以下图交易流水示例：
从交易流水表中客户看到一共有11条交易记录，包括有4组双开/双平交易以及1组多开交易，需要从数据中将4组双开/双平提炼出来并将一组交易按条件进行合并。 常规的通过FlinkSQL Join的方式将一组数据合并会出现以下情况，每组数据（左表和右表）会被重复拉通，如下图所示：
因此需要对数据进行精准一次处理，可以采用下述方法进行优化处理。
方法：对交易进行Row_Number处理，并将rank1的数据和rank2的数据进行join，由于双开/双平交易在全等值条件下仅会存在两条交易数据，因此可以采用这种处理方式。 这种方式下的优势是在Row_Number处理时对交易账号进行排序，有利于能够固定两个投资人之间的交易编号序列化顺序（rank1账号-rank2账号），避免出现两个投资人多组交易下的编号随机序列化，如账号2-账号1或者账号1-账号2的随机交易编号现象。
通过上述方法处理后，再对得到的数据通过交易时间、交易量、建仓类型、交易类型等条件进行判断处理得到每组交易的对敲标示以及对敲交易编号。
3）根据对敲交易编号和品种类型（非主力/主力）进行累计，并可以在Flink Job中设定对敲数量或者对敲价差预警值，一旦超过预警值能够推送告警信息值监察平台。 
在交易系统（数据库）中追加期货交易流水，查看计算效果。
通过上述数据处理方法实现完整的对敲异常交易监控的实时数据应用。
4）Flink状态以及窗口管理控制
另外，由于Flink一种是有状态的计算引擎，可以根据异常交易统计周期窗口设定对应的状态管理模式，避免出现跨窗口期的重复计算，Dataphin提供了Flink状态可视化管理的能力。 但是，由于期货交易时间存在一些特殊性，如日盘、夜盘、交易日等特定时间窗口，因此常规的FlinkSQL状态管理对于满足应用需求存在较大难度。 这里假设我们按交易日窗口级别来统计对敲异常交易行为：
假设设定State TTL为6个半小时，能够保证跨交易日的State不会被累计计算，但可能会出现交易时间内容State过期的问题出现在一个交易日窗口内状态时效无法基于State进行累计计算导致结果错误。
再假设把State TTL设定为17个半小时，能够保证整个交易日从9:00日盘开盘到2:30夜盘收盘的状态连续性，但由于17个半小时远远大于两个交易日之间的6个半小时间隔，因此会出现把上一个交易日的状态带入下一个的问题。
在这种情况下我们建议采用窗口模式，比如Flink的滚动窗口(Tumbling Windows)、累积窗口(Cumulate Windows)。 首先在滚动窗口模式下，我们可以把窗口大小为24小时，然后窗口的偏移时间可以不填写，那么起始时间2024-07-15T08:00；或者偏移时间设定1小时Time.hours(1L)，那么起始时间2024-07-15T09:00。但由于滚动窗口是在一个统计窗口期（比如一个完整交易日）内连续计算的，无法在发现异常交易行为即时输出结果，因此需要通过early-fire这类参数来控制窗口的提前触发。
另外，在累计窗口模式下，我们可以把窗口的INTERVAL定义为1Day，然后将统计步长Step可以设定为30秒（或者更少），这样就可以连续的对一个交易周期进行累计统计，而且也能够秒级的输出发常交易行为计算结果。
- Dataphin构建监察Flink实时任务
1）第一步：在Dataphin中构建Flink Catalog。
在Dataphin的表管理中将期货交易来源表（如kafka）、监察计算结果表（如ADB）统一构建为实时元表Catalog，便于FlinkSQL任务快速引用，即无需在FlinkSQL任务中重复的去Create Table，实时元表能够支持Catalog多任务复用的效果。
2）第二步：在Dataphin中创建FlinkSQL Job。
Dataphin提供了FlinkSQL IDE的能力，不仅有代码补全、SQL关键词提醒、自动联想等能力，还提供了代码异常检测的能力，能够降低在FlinkSQL开发过程中的错误率。
3）第三步：在Dataphin中对FlinkSQL进行预编译和在线测试。
Dataphin支持对FlinkSQL Job进行在线预编译，而且要求预编译通过后的作业才能够提交，预编译会对Job中引用的数据表权限、Catalog对象有效性、环境变量等的进行检测，确保提交后的作业能够正常运行降低运维错误率。
4）第四步：在Dataphin中对FlinkSQL开启流批校验。
在构建Flink实时作业时，尽管通过Row_Number、2PC提交等方式来确保数据处理的Exactly Once精准一次，但应用方往往都会顾虑Flink实时作业存在数据丢失、数据重复的问题导致结果差错。因此，Dataphin提供对一个Flink Job开启离线和实时两种计算模式，离线能够对业务数据进行全量计算，并通过离线计算和实时计算结果数据的一致性校验完成可信性验证。
Dataphin支持构建离线表和实时表映射的镜像Calalog实时元表，能够将离线数据和实时数据连接成一个整体元表，在FlinkSQL开发任务中通过引用镜像元表即可对FlinkSQL Job开启离线计算和实时计算两种处理模式。
对FlinkSQL的Batch作业定义调度属性。 将FlinkSQL的Batch离线示例定义成小时级自定义调度模式（可按照期货交易时间定义调度时间周期），并与离线入仓作业构建上下游依赖关系，在数据入仓完成后能够按调度周期自动完成离线全量数据计算。
在Dataphin质量中对离线及实时数据一致性校验。 Dataphin提供了数据一致性数据质量校验模式，能够将离线计算结果和实时计算结果表进行字段值的一致性校验，支持设定错误数据容忍阀值，一旦校验不通过可以触发质量告警。
在出现数据不一致的情况下，可以在Dataphin中构建一个触发式任务节点并在下游挂在一个数据集成任务（把离线数据数据集成写入到实时数据结果表中），业务方可以在出现数据一致性校验告警时来自动触发该任务节点以及下游的数据集成任务节点，完成数据结果异常修订。
5）第五步：FlinkSQL作业运维管理。 Dataphin提供了FlinkSQL作业可视化运维管理的能力，支持对JobManager、TaskManager、Metrics等信息进行可视化管理。
至此，在Dataphin平台上完成期货监察异常交易FlinkSQL任务、数据校验任务、数据修订任务的构建，可以提供一套完整的基于Flink的数据计算高时效性、结果高可靠性实时监察应用。
四、利用QuickBI打造交易监察看板和预警体系
对于Flink实时计算结果可以利用QuickBI的实时看板、监控告警、组件嵌入等能力打造一个具备数据实时分析、展示灵活多样、异常自动预警的期货交易监察系统嵌入式看板。
- 1）实时报表
在quickBI的仪表板中构建监察看板，并通过开启实时查询的能力打造实时动态更新的看板，实现监察异常交易数据实时展示。
- 2）监控告警
在对敲场景下需要对两个期货投资人之间的双开或双平交易，按照非主力/主力品种设定价差和数量的预警值（可以替代Flink中的监控规则）。
- 3）看板嵌入
QuickBi提供了安全嵌入的能力，可以将整个看板或者某个组件嵌入到期货监察系统中，快速搭建期货监察系统内部的监控可视化能力体系。
