Influxdb 用于金融应用程序

Question

我正在将我的财务分析应用程序数据从 MongoDB 迁移到 InfluxDB，因为数据和分析呈指数级增长。

我目前的情况是：

1) 每秒从交易所获取一次报价，并将其存储在称为“报价”的度量中；

2) 每 10 秒运行一次连续查询，将这个“刻度”数据按分钟分组为一个称为“ohlc”（烛台数据）的度量；

这就是我的疑问.. 当我使用 Mongo 作为我的数据库时，在我得到刻度的那一刻，我已经将它转换为烛台数据并计算了一些指标（MACD、EMA、BB、RSI）并存储它。

我看到 InfluxDB 有 Kapacitor 作为它的数据处理器，有一种方法可以在 Kapacitor 中编写一些脚本来计算这个指标，还是我应该将数据流式传输到 NodeJS 并自己计算？

如果我必须流式传输数据，那么最好的做法是什么？

Answer 1

使用 InfluxDB 时有几个选项。使用 Kapacitor，您可以将用户定义的函数合并到任何支持协议缓冲区的语言中，或者您可以编写 TICKscript 来进行数据转换。

您还可以使用数据库的连续查询功能，尽管它们有时可能是昂贵的查询，具体取决于查询和间隔。

如果您想在 NodeJS 中编写自己的函数，您基本上只需编写一些侦听 unix 域套接字的代码，Kapacitor 连接到该套接字，然后可以通过该套接字连接写入数据（完整文档请参见此处）。

如果您想编写 TICKscript，这里有几个示例：

// {alert_name}

// metric: {alert_metric}
// available_fields: [[other_telegraf_fields]]

// TELEGRAF CONFIGURATION
// [inputs.{plugin}]
//   # full configuration

// DEFINE: kapacitor define {alert_name} -type batch -tick 
//{plugin}/{alert_name}.tick -dbrp telegraf.autogen
// ENABLE: kapacitor enable {alert_name}

// Parameters
var info = {info_level} 
var warn = {warn_level}
var crit = {crit_level}
var infoSig = 2.5
var warnSig = 3
var critSig = 3.5
var period = 10s
var every = 10s

// Dataframe
var data = stream
  |from()
    .database('telegraf')
    .retentionPolicy('autogen')
    .measurement({plugin})
    .groupBy('host')
  |window()
    .period(period)
    .every(every)
  |mean({alert_metric})
    .as("stat")

// Thresholds
var alert = data
  |eval(lambda: sigma("stat"))
    .as('sigma')
    .keep()
  |alert()
    .id('{{ index .Tags "host"}}/{alert_metric}')
    .message('{{ .ID }}:{{ index .Fields "stat" }}')
    .info(lambda: "stat" > info OR "sigma" > infoSig)
    .warn(lambda: "stat" > warn OR "sigma" > warnSig)
    .crit(lambda: "stat" > crit OR "sigma" > critSig)

// Alert
alert
  .log('/tmp/{alert_name}_log.txt')

我希望这有帮助！