关于后端:使用Flink-SQL传输市场数据1传输VWAP

本文是一个由多局部组成的系列文章的第一篇，展现了 FlinkSQL 利用于市场数据的弱小性能和可表白性。该系列的代码和数据可在 github 上取得。它由量化建模负责人 Simudyne 和 Krishnen Vytelingum 合着。
速度在金融市场上至关重要。无论指标是最大化 alpha 还是最大水平地缩小危险，金融技术人员都会投入大量资金，以获取无关市场情况以及行情的最新见解。事件驱动和流式解决体系结构可在事件产生时对事件进行简单的解决，使其很天然地适宜金融市场利用。

Flink SQL 是一种数据处理语言，可用于事件驱动和流应用程序的疾速原型设计和开发。Flink SQL 将 SQL 的简略性和可拜访性与 Apache Flink（一种风行的分布式流媒体平台）的性能和可伸缩性联合在一起。借助 Flink SQL，业务剖析人员、开发人员和量化人员都能够疾速建设流传输管道，以实时执行简单的数据分析。

在本文中，咱们将应用 Simudyne 开发的基于代理的模型（ABM）生成的综合市场数据。ABM 并不是自上而下的办法，而是在简单零碎中对自主参与者（或代理）进行建模，例如：金融市场中的各种买卖双方。能够捕捉这些交互，并能够针对许多应用程序剖析生成的综合数据集，例如用于检测紧急欺诈行为的训练模型，或摸索风险管理的“假如”场景。ABM 生成的综合数据在历史数据有余或不可用的状况下很有用。

1 流式 VWAP

咱们从一个简略的示例开始，该示例从一系列交易事件中计算成交量加权平均价格（VWAP）。VWAP 是交易中用来掂量证券的市场价格和将来方向的通用基准。在这里，咱们有一个 CSV 格局的数据集，该数据集显示了一个交易日（2020 年 10 月 22 日）的虚构证券（SIMUI）的交易事件。

sym,prc,vol,bid_id,ask_id,buyer_id,seller_id,step,time
SIMUl,149.86,2300,P|63-m-1,P|66-l-0,P|63,P|66,380,22-Oct-2020 08:00:07.600
SIMUl,149.86,1935,P|63-m-1,P|25-l-0,P|63,P|25,380,22-Oct-2020 08:00:07.600
SIMUl,149.74,582,P|18-l-0,P|98-m-0,P|18,P|98,428,22-Oct-2020 08:00:08.560
SIMUl,149.76,2475,P|27-l-0,P|42-m-1,P|27,P|42,1021,22-Oct-2020 08:00:20.420
SIMUl,149.84,21,P|5-m-0,P|42-l-0,P|5,P|42,1078,22-Oct-2020 08:00:21.560
SIMUl,149.76,2709,P|24-l-1,P|92-m-0,P|24,P|92,1200,22-Oct-2020 08:00:24.000
SIMUl,149.84,1653,P|8-m-1,P|24-l-0,P|8,P|24,1513,22-Oct-2020 08:00:30.260
SIMUl,149.84,400,P|19-m-0,P|24-l-0,P|19,P|24,1577,22-Oct-2020 08:00:31.540

这些列是：交易种类，价格，数量，出价 ID，要价 ID，买方 ID，卖方 ID，步骤和工夫戳。步骤列是离散步骤 ABM 市场模仿的伪像，出于咱们的目标能够疏忽；其余各栏不言自明。

要解决此数据，咱们须要通过收回 CREATE TABLE 语句来申明 Flink SQL 表。咱们的示例数据是基于文件系统的，然而能够轻松更改连接器类型以从其余起源（例如 Kafka 主题）读取数据。请留神，event_time 是派生的列，也用于水印。通过加水印，Flink 能够限度期待提早达到和故障事件的工夫，以便能够获得停顿。在这里，咱们申明，达到 event_time 超过水印一分钟以上的记录将被疏忽。

CREATE TABLE trades (
    symbol         STRING,
    price          DOUBLE,
    vol            INT,
    bid_id         STRING,
    ask_id         STRING,
    buyer_id       STRING,
    seller_id      STRING,
    step           INT,
    ts_str         STRING,
    event_time     AS TO_TIMESTAMP (ts_str, 'dd-MMM-yyyy HH:mm:ss.SSS'),
    WATERMARK FOR event_time AS event_time - INTERVAL '1' MINUTE
) WITH (
    'connector' = 'filesystem',
    'path' = '/path/to/varstream/data/trades_raw',
    'format' = 'csv'
); 

VWAP 的公式很简略：对于指定时间段内的每笔交易，将价格乘以交易股份数即可。将其总和除以该时间段内已交易的股票总数。上面的流查问将显示以后的 VWAP，它将随着新交易事件的到来而更新：

SELECT
  symbol,
  SUM (vol)                     AS cumulative_volume,
  SUM (price * vol)             AS cumulative_pv,
  SUM (price * vol) / SUM (vol) AS vwap
FROM
  trades
GROUP BY
  symbol
;

2 实时播放

因为 CSV 文件中一个符号中只有一天的数据价值，因而后果更新可能产生得太快了，您简直没有留神到。从源读取事件的速度比实时产生的速度要快。有时须要在准实时回放历史数据，就如同 Flink 当初正在接管历史事件数据（例如，用于演示或原型设计和开发过程中）。

为了解决这个问题，咱们提供了一个简略的 UDTF（用户定义的表函数），该数据以从行工夫戳派生的人工提早播放历史数据。UDTF 有两个参数：第二个参数指定行工夫戳（在咱们的示例中为 event_time），而第一个参数指定第一个行工夫戳之后的分钟持续时间（以分钟为单位），以开始利用提早。以下代码段显示了如何注册 UDTF 并在处理事件的前 120 分钟后将其用于视图中以利用提早。请留神 LATERAL TABLE 联接的应用，该联接将函数利用于主表中的每一行。

-- Register UDTF
CREATE FUNCTION replay_after AS 'varstream.ReplayAfterFunction' LANGUAGE JAVA ;
-- Create a view
CREATE VIEW trades_replay AS (
    SELECT * FROM trades
    LEFT JOIN LATERAL TABLE (replay_after (120, trades.event_time)) ON TRUE
) ;

您能够通过收回一个简略的查问来验证事件的重播形式：
SELECT * FROM trades_replay
应用此视图，咱们当初能够收回雷同的 VWAP 聚合查问，并察看对 VWAP 的流更新，就如同它们是实时产生的一样：

  symbol,
  SUM (vol)                     AS cumulative_volume,
  SUM (price * vol)             AS cumulative_pv,
  SUM (price * vol) / SUM (vol) AS vwap
FROM
  trades_replay
GROUP BY
  symbol
;

只管此 UDTF 在进行原型制作时十分有用，但从根本上没有打算把它用于生产用处。咱们在这里应用它只是为了演示 FlinkSQL 如何在事件以模仿实时达到时更新聚合后果。

3 Group Windows

后面的示例显示了如何计算当天的流式 VWAP。假如您要以每隔 1 分钟的工夫建设一个带有蜡烛图的交易仪表板。您可能须要计算每分钟的 VWAP、高价、高价和总体积。Flink SQL 通过组窗口使此操作变得容易，组窗口能够在 GROUP BY 工夫距离上利用聚合函数。

上面显示了如何获取每分钟的 VWAP：

CREATE VIEW vwap_1m AS (
    SELECT
        symbol,
        TUMBLE_START (event_time, INTERVAL '1' MINUTES) AS start_time,
        TUMBLE_ROWTIME (event_time, INTERVAL '1' MINUTES) AS row_time,
        MAX (price)          AS max_price,
        MIN (price)          AS min_price,
        SUM (price * vol)    AS total_price,
        SUM (vol)            AS total_vol,
        SUM (price * vol) / SUM (vol) AS vwap
    FROM
        trades
    GROUP BY
        TUMBLE (event_time, INTERVAL '1' MINUTES), symbol
);
SELECT symbol, start_time, total_price, total_vol, vwap FROM vwap_1m ;

后面的操作为每分钟内产生的交易计算了 VWAP。如果要在几分钟内计算挪动的 VWAP（MVWAP），则 Flink SQL 提供了一个跳跃的组窗口。上面显示了 5 分钟的挪动 VWAP，步长为 1 分钟。

CREATE VIEW vwap_5m AS (
    SELECT
        symbol,
        HOP_START (event_time, INTERVAL '1' MINUTES, INTERVAL '5' MINUTES) AS start_time,
        HOP_ROWTIME (event_time, INTERVAL '1' MINUTES, INTERVAL '5' MINUTES) AS row_time,
        MAX (price)          AS max_price,
        MIN (price)          AS min_price,
        SUM (price * vol)    AS total_price,
        SUM (vol)            AS total_vol,
        SUM (price * vol) / SUM (vol) AS vwap
    FROM
        trades
    GROUP BY
        HOP (event_time, INTERVAL '1' MINUTES, INTERVAL '5' MINUTES), symbol
);
SELECT symbol, start_time, total_price, total_vol, vwap FROM vwap_5m ;

4 论断

Flink SQL 能够极大地简化和放慢流数据流的开发。在本文中，咱们摸索了 SQL GROUP BY 子句的不同用法，以依据市场数据流计算 VWAP 的变动。在下一部分中，咱们将向您展现如何从市场数据中提取每分钟的流式采样，以计算日内危险价值（IVaR）。咱们心愿本系列文章能激励您尝试将 Flink SQL 用于流式市场数据应用程序。

原文作者：Patrick Angeles& Krishnen Vytelingum

原文链接：https://blog.cloudera.com/streaming-market-data-with-flink-sql-part-i-streaming-vwap/

关注微信公共号理解更多信息：

本文由 mdnice 多平台公布