历史贸易数据的时间序列重采样答案

【问题标题】：Time series ressampling of historical trade data历史贸易数据的时间序列重采样
【发布时间】：2014-01-21 20:32:50
【问题描述】：

我在 csv 文件中有一些历史交易日期，格式为：unixtime, price, volume 我想分析这些数据。

我设法在 Python 中做到了，但速度非常慢（我需要大约 2 天的时间来运行算法以进行 30 天的数据测试）。

我正在尝试在 c/c++ 甚至 Java 或 Scala 中执行此操作，但我的主要问题是我无法重新采样数据。我需要将此数据重新采样为以下格式：日期时间、打开、高、低、关闭、音量，间隔 15 分钟，但我在 c/c++ 中找不到任何方法

在 Python 中，这可以满足我的要求（它使用 pandas 数据框）：

def resample_data(raw_data, time_frame):
    # resamples the ticker data in ohlc
    resampledData = raw_data.copy()
    ohlc_dict = {
        'open':'first',
        'high':'max',
        'low':'min',
        'close':'last',
        'price':'first'
        }

    resampledData = resampledData.resample(time_frame, how={'price':ohlc_dict, 'amount':'sum'})
    resampledData.amount = resampledData['amount']['sum'].fillna(0.0)
    resampledData['price']['close'] = resampledData['price']['close'].fillna(method='pad')
    resampledData = resampledData.apply(lambda x: x.fillna(resampledData['price']['close']))

    return resampledData

在 c/c++/Java/scala 中有什么想法（或库）吗？

【问题讨论】：

如果需要，您为什么不按日期对数据进行排序，然后将其分成 15 分钟的块并聚合每个块。这可以迭代完成，因此您不必在内存中保存太多数据。您可以使用标准的 Scala 函数。也许你也可以并行运行。
问题是“手动”执行此操作对我来说似乎太容易出错。考虑到不断变化的小时、天、月、年以保持检查 15 分钟（或不同）的时间间隔，这只是需要进行大量测试以检查所有计算是否正确。最好使用一些已经通过所需质量控制的库（如 Python 的 pandas 或 R 的 Zoo）。
很遗憾，这不是图书馆推荐网站。如果您开始做某事，我们可以帮助您解决问题。就个人而言，在处理数据集时，我倾向于使用 SQL，尤其是当它看起来应该是简单的聚合时；使用日历/时间表，不同的时间段变得微不足道。否则，准备将数据放入的“桶”应该可以工作，并且可能是可并行的。所需的任何测试都分为两部分；根据时差创建桶，并放入这些桶中。
我认为您可以定义一个从流数据中产生间隔的函数。如果您处理以分钟、秒、小时表示的任何内容，您可以将日期转换为这些单位，然后拆分。如果你处理几天、几周等，那么你将不得不使用 JodaTime 之类的东西来提取这些单位。闰秒或许可以忽略。
熊猫有重采样功能。由于您使用的是 pandas DataFrame，因此与使用提供的重采样函数相比，您的解决方案很复杂。

标签： java c++ scala time-series resampling

【解决方案1】：

只是一个简单的示例，说明您可以使用标准 Scala 库做什么。这段代码可以在 Scala REPL 中运行：

// not importing external libraries like Joda time and its Scala wrappers
import java.util.Date
import scala.annotation.tailrec

case class Sample(value: Double, timeMillis: Long)
case class SampleAggregate(startTimeMillis: Long, endTimeMillis: Long,
  min: Sample, max: Sample)

val currentMillis = System.currentTimeMillis
val inSec15min = 15 * 60
val inMillis15min = inSec15min * 1000
// sample each second:
val data = (1 to inSec15min * 100).map { i =>
  Sample(i, currentMillis + i*1000) }.toList

@tailrec
def aggregate(xs: List[Sample], intervalDurationMillis: Long,
  accu: List[SampleAggregate]): List[SampleAggregate] =
  xs match {
    case h :: t =>
      val start = h.timeMillis
      val (slice, rest) = xs.span(_.timeMillis < (start + intervalDurationMillis))
      val end = slice.last.timeMillis
      val aggr = SampleAggregate(start, end, slice.minBy(_.value),
        slice.maxBy(_.value))
      aggregate(rest, intervalDurationMillis, aggr :: accu)
    case Nil =>
      accu.reverse
  }

val result = aggregate(data, inMillis15min, Nil)

假数据：

data.take(10).foreach(println)
Sample(1.0,1388809630677)
Sample(2.0,1388809631677)
Sample(3.0,1388809632677)
Sample(4.0,1388809633677)
Sample(5.0,1388809634677)
Sample(6.0,1388809635677)
Sample(7.0,1388809636677)
Sample(8.0,1388809637677)
Sample(9.0,1388809638677)
Sample(10.0,1388809639677)

结果：

result.foreach(println)
SampleAggregate(1388809630677,1388810529677,Sample(1.0,1388809630677),Sample(900.0,1388810529677))
SampleAggregate(1388810530677,1388811429677,Sample(901.0,1388810530677),Sample(1800.0,1388811429677))
SampleAggregate(1388811430677,1388812329677,Sample(1801.0,1388811430677),Sample(2700.0,1388812329677))
SampleAggregate(1388812330677,1388813229677,Sample(2701.0,1388812330677),Sample(3600.0,1388813229677))
SampleAggregate(1388813230677,1388814129677,Sample(3601.0,1388813230677),Sample(4500.0,1388814129677))
SampleAggregate(1388814130677,1388815029677,Sample(4501.0,1388814130677),Sample(5400.0,1388815029677))
SampleAggregate(1388815030677,1388815929677,Sample(5401.0,1388815030677),Sample(6300.0,1388815929677))

我们可以将一个函数传递给span，它将定义时间间隔（小时或天）。这也可以在从文件中读取时转换为 Stream。

【讨论】：

非常感谢，我会试试后者，让你知道结果。

【解决方案2】：

尝试查看Saddle，进行数据操作。我自己才发现这个，所以不确定它的全部功能，但它是受到 Pandas 的启发。

【讨论】：

您好，Saddle 实际上是我的第一次尝试。但是虽然它实现了像 DataFrame 这样的 Pandas，但它没有处理 TimeSeries 的重采样函数。