elasticsearch - Elasticsearch汇总:按体积加权平均价格

Question

我需要绘制在特定时间范围内具有price_per_unit和quantitiy的交易的Volume-Weighted Average Prive (VWAP)。

作为聚合的结果，date_histogram的每个存储区都应包含到目前为止发生的所有交易的VWAP。

我不确定使用Elasticsearch是否可行，而且也不是正确的处理方式(例如使用脚本？)？
trade文档的基本映射非常简单:

"trade": {
  "properties": 
    "trade_id": {"type": "string", "index": "not_analyzed"},
    "product_id": {"type": "string", "index": "not_analyzed"},
    "quantity": {'type': 'double'}, // number of units
    "execution_time": {'type': 'date'},
    "price_per_unit": {'type': 'double'},
  }
}

而 execution_time应该使用 date_histogram，而总交易价格是 price_per_unit和 quantity的乘积。因此 VWAP = sum(price_per_unit * quantity) / sum(quantity)。

Answer 1

DELETE test
PUT test
{
  "mappings": {
    "trade": {
      "properties": {
        "trade_id": {
          "type": "string",
          "index": "not_analyzed"
        },
        "product_id": {
          "type": "string",
          "index": "not_analyzed"
        },
        "quantity": {
          "type": "double"
        },
        "execution_time": {
          "type": "date"
        },
        "price_per_unit": {
          "type": "double"
        }
      }
    }
  }
}

POST test/trade/_bulk
{"index":{}}
{"execution_time":"2016-11-18T22:45:27Z","quantity":10,"price_per_unit":5}
{"index":{}}
{"execution_time":"2016-11-18T22:45:27Z","quantity":10,"price_per_unit":5}
{"index":{}}
{"execution_time":"2016-11-19T22:45:27Z","quantity":10,"price_per_unit":5}
{"index":{}}
{"execution_time":"2016-11-20T22:45:27Z","quantity":10,"price_per_unit":5}
{"index":{}}
{"execution_time":"2016-11-20T22:45:27Z","quantity":10,"price_per_unit":5}
{"index":{}}
{"execution_time":"2016-11-20T22:45:27Z","quantity":10,"price_per_unit":5}
{"index":{}}
{"execution_time":"2016-11-21T22:45:27Z","quantity":10,"price_per_unit":5}
{"index":{}}
{"execution_time":"2016-11-21T22:45:27Z","quantity":10,"price_per_unit":5}

POST test/trade/_search
{
  "size": 0,
  "aggs": {
    "sales_per_day": {
      "date_histogram": {
        "field": "execution_time",
        "interval": "day"
      },
      "aggs": {
        "sales": {
          "sum": {
            "script": {
              "lang": "groovy",
              "inline": "doc['quantity'] * doc['price_per_unit']"
            }
          }
        },
        "cumulative_sales": {
          "cumulative_sum": {
            "buckets_path": "sales"
          }
        }
      }
    }
  }
}

并且您需要启用 inline scripting for groovy。