Aggregator.scala

package com.ovoenergy

import java.util.{Calendar, Date}

object Aggregator {

  case class DailySnapshot(date: Date, amount: Double)
  case class MonthlyConsumption(date: Date, amount: Double)

  def parseData(data: Seq[(String, Double)]): Seq[DailySnapshot] = {
    val format = new java.text.SimpleDateFormat("yyyy-MM-dd")
    data.map { case (date, amount) => DailySnapshot(format.parse(date), amount) }
  }

  // Fix the sequence of snapshots to a only increasing list of snapshots
  def makeIncreasing(snaps: Seq[DailySnapshot]): Seq[DailySnapshot] = {

    def getJumps(ls: List[Double]): List[Double] = ls match {
      case Nil => Nil
      case x :: xs => 0 :: ls.zip(xs).map { case (p, n) => if (n < p) { p } else { 0 } }
    }

    def sumJumps(ls: List[Double]): List[Double] = ls match {
      case Nil => Nil
      case x :: xs => x :: ls.zip(xs).map { case (p, c) => p + c }
    }
    val differences = sumJumps(getJumps(snaps.toList.map(_.amount)))
    snaps.zip(differences).map { case (s, diff) => DailySnapshot(s.date, diff + s.amount)  }
  }

  // Only take the value at the begining of the month. Take the last recoded value as well
  // This function assumes the amounts are correcly aggergated in increasing order
  def makeMonthly(snaps: Seq[DailySnapshot]): Seq[DailySnapshot] = {
    def isFirstDay(date: Date): Boolean = {
      val c = Calendar.getInstance()
      c.setTime(date) // Thanks java's stateful API
      c.get(Calendar.DAY_OF_MONTH) == 1
    }
    val monthlies = snaps.filter { d => isFirstDay(d.date) }
    snaps.lastOption.map { last => monthlies :+ last }.getOrElse(monthlies)
  }

  def aggregateMonthlyConsumption(snapshots: Seq[DailySnapshot]): Seq[MonthlyConsumption] = {
    val increasing = makeIncreasing(snapshots)
    val monthly = makeMonthly(increasing)
    monthly.map { daily => MonthlyConsumption(daily.date, daily.amount) }
  }

  val Snapshots = Seq(
    ("2015-07-15T00:00:00Z", 2407.15),
    ("2015-07-16T00:00:00Z", 2410.75),
    ("2015-07-17T00:00:00Z", 2416.15),
    ("2015-07-18T00:00:00Z", 2423.85),
    ("2015-07-19T00:00:00Z", 2429.55),
    ("2015-07-20T00:00:00Z", 2434.95),
    ("2015-07-21T00:00:00Z", 2446.3),
    ("2015-07-22T00:00:00Z", 2451.95),
    ("2015-07-23T00:00:00Z", 2463.75),
    ("2015-07-24T00:00:00Z", 2471.25),
    ("2015-07-25T00:00:00Z", 2475.05),
    ("2015-07-26T00:00:00Z", 2476.6),
    ("2015-07-27T00:00:00Z", 2478.1),
    ("2015-07-28T00:00:00Z", 2479.75),
    ("2015-07-29T00:00:00Z", 2481.3),
    ("2015-07-30T00:00:00Z", 2482.85),
    ("2015-07-31T00:00:00Z", 2484.55),
    ("2015-08-01T00:00:00Z", 2486.1),
    ("2015-08-02T00:00:00Z", 2487.6),
    ("2015-08-03T00:00:00Z", 2489.25),
    ("2015-08-04T00:00:00Z", 2490.85),
    ("2015-08-05T00:00:00Z", 2492.45),
    ("2015-08-06T00:00:00Z", 2494.15),
    ("2015-08-07T00:00:00Z", 2495.75),
    ("2015-08-08T00:00:00Z", 2497.4),
    ("2015-08-09T00:00:00Z", 2501.95),
    ("2015-08-10T00:00:00Z", 2507.2),
    ("2015-08-11T00:00:00Z", 2513.85),
    ("2015-08-12T00:00:00Z", 2521.4),
    ("2015-08-13T00:00:00Z", 2530.65),
    ("2015-08-14T00:00:00Z", 2546.45),
    ("2015-08-15T00:00:00Z", 2550.2),
    ("2015-08-16T00:00:00Z", 2554.1),
    ("2015-08-17T00:00:00Z", 2560.95),
    ("2015-08-18T00:00:00Z", 2568.25),
    ("2015-08-19T00:00:00Z", 2575.7),
    ("2015-08-20T00:00:00Z", 2588.55),
    ("2015-08-21T00:00:00Z", 3.0)
  )
}

object Main extends App {
  println(Aggregator.aggregateMonthlyConsumption(Aggregator.parseData(Aggregator.Snapshots)))
}