查询以明确获取每个股票代码的最新报价

Question

在 Meteor 中，我有一个价格报价的 Quotes 集合，它是所有跟踪符号的组合。 Quotes 集合如下所示：

{ "sym" : "CL", "bid" : 49.29, "ask" : 49.31, "timestamp" : 1424835506105, "_id" : "2yPrQzoZ8MTad8viw" }
{ "sym" : "CL", "bid" : 49.3, "ask" : 49.31, "timestamp" : 1424835512114, "_id" : "oMcsLm2qqkLqWDxpG" }
{ "sym" : "ZB", "bid" : 146.0625, "ask" : 146.09375, "timestamp" : 1424835529244, "_id" : "7XSDNAR8Hx6DmjWhg" }
{ "sym" : "NQ", "bid" : 4445.25, "ask" : 4445.75, "timestamp" : 1424835533854, "_id" : "3AgfZWYAbnWW7pGcc" }
{ "sym" : "YM", "bid" : 18181, "ask" : 18182, "timestamp" : 1424835540869, "_id" : "AQupJnnx2mNMLwSq5" }
{ "sym" : "ZB", "bid" : 146.0625, "ask" : 146.125, "timestamp" : 1424835540870, "_id" : "85eemh5pPQd469cMK" }
{ "sym" : "NQ", "bid" : 4445.25, "ask" : 4445.5, "timestamp" : 1424835540883, "_id" : "7cMmsJqoAkct2CPtm" }
{ "sym" : "NQ", "bid" : 4445.25, "ask" : 4445.75, "timestamp" : 1424835546572, "_id" : "r7zNAuqeQXL7jvnDv" }
{ "sym" : "YM", "bid" : 18181, "ask" : 18183, "timestamp" : 1424835546573, "_id" : "7Dxf2WmhibBuuZrg2" }
{ "sym" : "ES", "bid" : 2112.5, "ask" : 2113, "timestamp" : 1424835577420, "_id" : "yWKtqNjbtSoN2dMzc" }
{ "sym" : "YM", "bid" : 18181, "ask" : 18182, "timestamp" : 1424835577438, "_id" : "bWmmtvdAafxgaNRNJ" }

我只想检索每个 sym 的最新报价，这意味着我需要将 distinct on sym 与 max on timestamp 结合起来。我正在 Meteor 中这样做。

是否有一个 Quotes.find() 命令将返回此结果集？

{ "sym" : "CL", "bid" : 49.3, "ask" : 49.31, "timestamp" : 1424835512114, "_id" : "oMcsLm2qqkLqWDxpG" }
{ "sym" : "ZB", "bid" : 146.0625, "ask" : 146.125, "timestamp" : 1424835540870, "_id" : "85eemh5pPQd469cMK" }
{ "sym" : "NQ", "bid" : 4445.25, "ask" : 4445.75, "timestamp" : 1424835546572, "_id" : "r7zNAuqeQXL7jvnDv" }
{ "sym" : "ES", "bid" : 2112.5, "ask" : 2113, "timestamp" : 1424835577420, "_id" : "yWKtqNjbtSoN2dMzc" }
{ "sym" : "YM", "bid" : 18181, "ask" : 18182, "timestamp" : 1424835577438, "_id" : "bWmmtvdAafxgaNRNJ" }

Answer 1

从 Meteor 1.0.3.2 开始，aggregate function is not supported

但基于the meteor docs 中的示例，您可以在发布函数中观察服务器上集合的变化，并返回具有聚合值的动态集合。

在您的情况下，这是如何工作的：

// server: publish the current size of a collection
Meteor.publish("recentQuotes", function () {
  var self = this;

  // here we'll store information about which documents are used
  // as the most recent ones
  var onlyLastValues = {};

  var handle = Quotes.find({}, {sort: {timestamp: 1}}).observeChanges({

    // fired when a document is added to the Quotes collection
    added: function (id, fields) {

      if (onlyLastValues[fields.sym] === undefined)
        self.added("recentQuotesAll", fields.sym, fields);   
      else
        self.changed("recentQuotesAll", fields.sym, fields);  

      onlyLastValues[fields.sym] = id;

    },

    // fired when a document is removed from the Quotes collection
    // if that's one
    removed: function (id) {

      // as you only have id as a parameter you need to find  
      // which sym was changed

      onlyLastValuesById = _.invert(onlyLastValues);

      if (onlyLastValuesById[id] !== undefined){
        // the removed document was one of the last documents

        // search for last quote with that sym
        lastQuote = Quotes.findOne({sym: onlyLastValuesById[id]}, {sort: {timestamp: -1}});

        // remove if it was the last one
        if (lastQuote === undefined)
          self.removed("recentQuotesAll", onlyLastValuesById[id]); 

        else
          // modify the dynamic collection
          self.changed("recentQuotesAll", lastQuote.sym, lastQuote);    
      }

    },

    // fired when the document in Quotes collection is changed
    // need to check if that's one of the last quotes
    changed: function(id, fields){

      // this is fairly similar to removed

      onlyLastValuesById = _.invert(onlyLastValues);

      if (onlyLastValuesById[id] !== undefined){

        // this time we have all the fields already

        // modify the dynamic collection
        self.changed("recentQuotesAll", fields.sym, fields);    
      }
    }

  });

  self.ready();

  // Stop observing the cursor when client unsubs.
  // Stopping a subscription automatically takes
  // care of sending the client any removed messages.
  self.onStop(function () {
    handle.stop();
  });
});

}

if (Meteor.isClient){

// client: declare collection to hold aggregated object
RecentQuotesAll = new Mongo.Collection("recentQuotesAll");

// client: subscribe to the aggregated values
Tracker.autorun(function () {
  Meteor.subscribe("recentQuotes");
});

您还可以添加initializing 变量以在首次完全统计数据后发布数据。为此，您必须将所有字段存储在 onlyLastValues 对象中，并为每个存储的值运行 self.added self.ready(); 打电话。

Answer 2

我不知道如何在meteor 中查询，但下面的 mongo 聚合将解决您的问题

    db.collectionName.aggregate({
    "$sort": {
    "timestamp": -1
    }
}, {
    "$group": {
    "_id": "$sym",
    "bid": {
        "$first": "$bid"
    },
    "ask": {
        "$first": "$ask"
    },
    "timestamp": {
        "$first": "$timestamp"
    },
    "id": {
        "$first": "$_id"
    }
    }
}, {
    "$project": {
    "sym": "$_id",
    "bid": "$bid",
    "ask": "$ask",
    "timestamp": "$timestamp",
    "_id": "$id"
    }
})

Answer 3

试试这个 -

db.collection.aggregate({ "$group" : {"_id" : "$sym",
     "runtime" : {"$max" : "$timestamp"},
"bid":{$last:"$bid"},
"ask":{$last:"$ask"}
 }},

{ "$project" : {
     "sym" : "$_id" ,
     "runtime" : 1,
    "bid":1,
    "ask":1
 }
})