//设置package包名称以及导入依赖的类
package com.pygmalios.reactiveinflux.jawa

import java.util.Date

import com.pygmalios.{reactiveinflux => sc}
import org.joda.time.{DateTime, Instant}

class JavaPointTime(val underlying: sc.PointTime) extends PointTime {
  def this(seconds: Long, nano: Int) {
    this(sc.PointTime.ofEpochSecond(seconds, nano))
  }

  def this(dateTime: DateTime) {
    this(sc.PointTime(dateTime))
  }

  def this(instant: Instant) {
    this(sc.PointTime(instant))
  }

  def this(date: Date) {
    this(sc.PointTime(date))
  }

  override def getSeconds: Long = underlying.seconds
  override def getNano: Int = underlying.nanos

  override def toString: String = underlying.toString
  override def hashCode(): Int = underlying.hashCode()
  override def equals(obj: scala.Any): Boolean = underlying.equals(obj)
} 

//设置package包名称以及导入依赖的类
package com.pygmalios.reactiveinflux.jawa

import java.util
import java.util.Date

import com.pygmalios.reactiveinflux._
import com.pygmalios.reactiveinflux.jawa.Conversions._
import com.pygmalios.{reactiveinflux => sc}
import org.joda.time.{DateTime, Instant}

class JavaPoint(val underlyingPoint: sc.Point) extends JavaPointNoTime(underlyingPoint) with Point {
  def this(time: PointTime,
           measurement: String,
           tags: util.Map[String, String],
           fields: util.Map[String, Object]) {
    this(sc.Point(sc.PointTime.ofEpochSecond(time.getSeconds, time.getNano), measurement, tagsToScala(tags), fieldsToScala(fields)))
  }

  def this(dateTime: DateTime,
           measurement: String,
           tags: util.Map[String, String],
           fields: util.Map[String, Object]) {
    this(sc.Point(sc.PointTime(dateTime), measurement, tagsToScala(tags), fieldsToScala(fields)))
  }

  def this(instant: Instant,
           measurement: String,
           tags: util.Map[String, String],
           fields: util.Map[String, Object]) {
    this(sc.Point(sc.PointTime(instant), measurement, tagsToScala(tags), fieldsToScala(fields)))
  }

  def this(date: Date,
           measurement: String,
           tags: util.Map[String, String],
           fields: util.Map[String, Object]) {
    this(sc.Point(sc.PointTime(date), measurement, tagsToScala(tags), fieldsToScala(fields)))
  }

  override lazy val getTime: PointTime = new JavaPointTime(underlyingPoint.time)
} 

//设置package包名称以及导入依赖的类
package shapeless.datatype.tensorflow

import java.net.URI

import org.joda.time.Instant
import org.scalacheck.Prop.{all, forAll}
import org.scalacheck.ScalacheckShapeless._
import org.scalacheck._
import org.tensorflow.example.{Feature, Int64List}
import shapeless._
import shapeless.datatype.record._

import scala.collection.JavaConverters._

object TensorFlowTypeSpec extends Properties("TensorFlowType") {

  import shapeless.datatype.test.Records._
  import shapeless.datatype.test.SerializableUtils._

  implicit def compareByteArrays(x: Array[Byte], y: Array[Byte]) = java.util.Arrays.equals(x, y)
  implicit def compareIntArrays(x: Array[Int], y: Array[Int]) = java.util.Arrays.equals(x, y)
  implicit def compareDouble(x: Double, y: Double) = x.toFloat == y.toFloat

  def roundTrip[A, L <: HList](m: A)
                              (implicit
                               gen: LabelledGeneric.Aux[A, L],
                               fromL: FromFeatures[L],
                               toL: ToFeatures[L],
                               mr: MatchRecord[L]): Prop = {
    val t = ensureSerializable(TensorFlowType[A])
    val rm = RecordMatcher[A]
    all(
      t.fromExample(t.toExample(m)).exists(rm(_, m)),
      t.fromExampleBuilder(t.toExampleBuilder(m)).exists(rm(_, m)))
  }

  implicit val timestampTensorFlowMappableType = TensorFlowType.at[Instant](
    TensorFlowType.toLongs(_).map(new Instant(_)),
    xs => TensorFlowType.fromLongs(xs.map(_.getMillis)))
  property("required") = forAll { m: Required => roundTrip(m) }
  property("optional") = forAll { m: Optional => roundTrip(m) }
  property("repeated") = forAll { m: Repeated => roundTrip(m) }
  property("mixed") = forAll { m: Mixed => roundTrip(m) }
  property("seqs") = forAll { m: Seqs => roundTrip(m) }

  implicit val uriTensorFlowType = TensorFlowType.at[URI](
    TensorFlowType.toStrings(_).map(URI.create),
    xs => TensorFlowType.fromStrings(xs.map(_.toString)))
  property("custom") = forAll { m: Custom => roundTrip(m)}

} 

//设置package包名称以及导入依赖的类
package shapeless.datatype.test

import java.net.URI

import com.google.protobuf.ByteString
import org.joda.time.Instant
import org.scalacheck._

object Records {
  case class Required(booleanField: Boolean,
                      intField: Int, longField: Long, floatField: Float, doubleField: Double,
                      stringField: String,
                      byteStringField: ByteString, byteArrayField: Array[Byte],
                      timestampField: Instant)
  case class Optional(booleanField: Option[Boolean],
                      intField: Option[Int], longField: Option[Long],
                      floatField: Option[Float], doubleField: Option[Double],
                      stringField: Option[String],
                      byteStringField: Option[ByteString], byteArrayField: Option[Array[Byte]],
                      timestampField: Option[Instant])
  case class Repeated(booleanField: List[Boolean],
                      intField: List[Int], longField: List[Long],
                      floatField: List[Float], doubleField: List[Double],
                      stringField: List[String],
                      byteStringField: List[ByteString], byteArrayField: List[Array[Byte]],
                      timestampField: List[Instant])
  case class Mixed(longField: Long, doubleField: Double, stringField: String,
                   longFieldO: Option[Long], doubleFieldO: Option[Double], stringFieldO: Option[String],
                   longFieldR: List[Long], doubleFieldR: List[Double], stringFieldR: List[String])
  case class Nested(longField: Long, longFieldO: Option[Long], longFieldR: List[Long],
                    mixedField: Mixed, mixedFieldO: Option[Mixed], mixedFieldR: List[Mixed])
  case class Seqs(array: Array[Int], list: List[Int], vector: Vector[Int])
  case class Custom(uriField: URI, uriFieldO: Option[URI], uriFieldR: List[URI])

  implicit val arbByteString = Arbitrary(Gen.alphaStr.map(ByteString.copyFromUtf8))
  implicit val arbInstant = Arbitrary(Gen.chooseNum(0, Int.MaxValue).map(new Instant(_)))
  implicit val arbUri = Arbitrary(Gen.alphaStr.map(URI.create))
} 

//设置package包名称以及导入依赖的类
package org.nikosoft.oanda.instruments

import org.joda.time.Instant
import org.nikosoft.oanda.instruments.Model.CandleStick
import org.scalatest.{FunSpec, Matchers}

class Indicators$Test extends FunSpec with Matchers {

  describe("atr") {

    it("should calculate Average True Range when there is no previous value provided") {
      val inputHigh: Seq[BigDecimal] = Seq(50.19, 49.88, 49.66, 50.12, 50.19, 49.92, 49.35, 49.20, 49.05, 48.82, 48.87, 48.90, 48.72, 48.70)
      val inputLow: Seq[BigDecimal] = Seq(49.73, 49.43, 48.90, 49.20, 49.87, 49.50, 48.86, 48.94, 48.64, 48.24, 48.37, 48.39, 48.14, 47.79)
      val inputClose: Seq[BigDecimal] = Seq(50.03, 49.75, 49.50, 49.53, 50.13, 49.91, 49.32, 49.07, 49.03, 48.74, 48.63, 48.75, 48.61, 48.16)
      val input = (inputHigh, inputLow, inputClose).zipped.map(CandleStick(Instant.now, 0, _, _, _, 0, complete = true))
      val actual = Indicators.atr(14, input, None)
      val expected: BigDecimal = 0.56
      actual.fold(fail("should not fail here")) { actualValue =>
        actualValue shouldBe (expected +- 0.01)
      }
    }

    it("should calculate Average True Range when previous value provided") {
      val inputCandle = Seq(
        CandleStick(Instant.now(), 0, 50.36, 49.26, 50.31, 0, complete = true),
        CandleStick(Instant.now(), 0, 50.19, 49.73, 50.03, 0, complete = true))
      val previousAtr: BigDecimal = 0.56
      val actual = Indicators.atr(14, inputCandle, Some(previousAtr))
      val expected: BigDecimal = 0.59
      actual.fold(fail("should not fail here")) { actualValue =>
        actualValue shouldBe (expected +- 0.01)
      }
    }

    it("should return None if no previous value provided and amount of candles is less than period") {
      val inputCandle = Seq(
        CandleStick(Instant.now(), 0, 50.36, 49.26, 50.31, 0, complete = true),
        CandleStick(Instant.now(), 0, 50.19, 49.73, 50.03, 0, complete = true),
        CandleStick(Instant.now(), 0, 50.19, 49.73, 50.03, 0, complete = true),
        CandleStick(Instant.now(), 0, 50.19, 49.73, 50.03, 0, complete = true)
      )
      val actual = Indicators.atr(6, inputCandle, None)
      actual shouldBe None
    }

    it("should return None if previous value specified but amount of candles is less than 2") {
      val inputCandle = Seq(CandleStick(Instant.now(), 0, 50.36, 49.26, 50.31, 0, complete = true))
      val actual = Indicators.atr(14, inputCandle, Option(1))
      actual shouldBe None
    }

  }

} 

//设置package包名称以及导入依赖的类
package com.gaiam.gcsis.util

import org.joda.time.{Interval => JInterval}
import org.joda.time.Instant
import org.joda.time.ReadableInstant
import scala.util.Sorting

class IntervalSchedule(seq: Seq[JInterval]) {
    val intervals: Array[JInterval] = Sorting.stableSort(seq, (a: JInterval, b: JInterval) => a.getStart.compareTo(b.getStart) <= 0)

    private def combine(i1: JInterval, i2: JInterval) = {
        if (i1.contains(i2)) i1
        else if (i2.contains(i1)) i2
        else new JInterval(i1.getStart, i2.getEnd)
    }

    def continuousInterval(from: ReadableInstant) = {
        intervals.foldLeft[Option[JInterval]](None)(
            (result, interval) => {
             (result, interval) match {
                case (None, i: JInterval) if i.contains(from) => Some(i)
                case (None, _) => None
                case (Some(res: JInterval), i: JInterval) if res.overlaps(i) || res.abuts(i) => Some(combine(res, i))
                case _ => result
             }
            })
    }
} 

//设置package包名称以及导入依赖的类
package ch.becompany

import java.net.InetAddress

import _root_.akka.stream.scaladsl.{Sink, Source}
import _root_.akka.actor.ActorSystem
import _root_.akka.stream.ActorMaterializer
import org.joda.time.Instant

import scala.concurrent.duration._
import scala.util.Random

object LogGenerator {

  implicit val system = ActorSystem("log-analyzer")
  implicit val materializer = ActorMaterializer()

  case object Tick
  val interval = 200 millis

  val rand = new Random(System.currentTimeMillis)

  def ips = Seq("1.2.3.4", "2.3.4.5", "3.4.5.6").map(InetAddress.getByName)
  def urls = Seq("/foo", "/bar", "/baz")
  def userAgents = Seq("Firefox", "Chrome")

  def rnd[T](seq: Seq[T]) = seq(rand.nextInt(seq.size))

  def createLogEntry(): LogEntry =
    LogEntry(rnd(ips), Instant.now, rnd(urls), rnd(userAgents))

  def formatLogEntry(e: LogEntry): String =
    Seq(e.ip.getHostAddress, e.time.getMillis.toString, e.req, e.userAgent).
      map(s => s""""$s"""").
      mkString(",")

  def apply(): Unit = {
    val tickSource = Source.tick(0 seconds, interval, Tick)
    val logSource = Source.fromIterator(() => Iterator.continually(createLogEntry))
    Source.
      zipN(List(logSource, tickSource)).
      map(_.head.asInstanceOf[LogEntry]).
      map(formatLogEntry).
      runWith(Sink.foreach(println))
  }

} 

//设置package包名称以及导入依赖的类
package daos.doobie

import daos.SubjectDao
import daos.doobie.DoobieImports._
import doobie.imports._
import doobie.util.transactor.DataSourceTransactor
import javax.inject.Inject
import models.{ Role, User }
import models.security.MySubject
import org.joda.time.Instant
import play.api.db.Database

class SubjectDaoDoobie @Inject() (
    db: Database
) extends SubjectDao {
  import SubjectDaoDoobie.subjectQuery

  private[this] implicit def xa() = DataSourceTransactor[IOLite](db.dataSource)

  def subjectByIdentifier(identifier: String): Option[MySubject] = {
    subjectQuery(identifier).option.transact(xa()).unsafePerformIO map {
      case (id, login, connected, lastAct, rid, rname) ?
        MySubject(
          User(id, login, "", 0, rid, connected, lastAct),
          Role(rid, rname)
        )
    }
  }
}

object SubjectDaoDoobie {

  def subjectQuery(login: String): Query0[(Long, String, Boolean, Option[Instant], Int, String)] =
    sql"""
      Select u.id, u.login, u.connected, u.last_activity, r.id, r.name
      from users u join roles r on u.role_id = r.id
      where u.login = $login""".query[(Long, String, Boolean, Option[Instant], Int, String)]
} 

//设置package包名称以及导入依赖的类
package daos.doobie

import doobie.imports.Meta
import java.sql.Timestamp
import org.joda.time.{ DateTime, Instant }


object DoobieImports {
  implicit val InstantMeta: Meta[Instant] = Meta1653280882.nxmap(
    (t: Timestamp) ? new Instant(t.getTime),
    (i: Instant) ? new Timestamp(i.getMillis)
  )

  implicit val DateTimeMeta: Meta[DateTime] = Meta1653280882.nxmap(
    (t: Timestamp) ? new DateTime(t.getTime),
    (d: DateTime) ? new Timestamp(d.getMillis)
  )
} 

//设置package包名称以及导入依赖的类
package models

import slick.backend.DatabaseConfig
import slick.driver.JdbcProfile

import java.sql.Timestamp
import org.joda.time.Instant


case class User (
  id:           Long,
  login:        String,
  password:     String,
  salt:         Int,
  roleId:       Int,
  connected:    Boolean,
  lastActivity: Option[Instant]
)

trait UserTable {
  val dc: DatabaseConfig[JdbcProfile]

  import dc.driver.api._

  private[UserTable] class Usuarios(tag: Tag) extends Table[User](tag, "users") {
    def id           = column[Long]  ("id", O.PrimaryKey, O.AutoInc)
    def login        = column[String]("login")
    def password     = column[String]("password")
    def salt         = column[Int]("salt")
    def roleId       = column[Int]("role_id")
    def connected    = column[Boolean]("connected")
    def lastActivity = column[Option1653280882]("last_activity")

    def idxLogin = index("uk_login", login, unique = true)

    def * = (id, login, password, salt, roleId, connected, lastActivity).shaped <> (userTupled, userUnapply)
  }

  // User -> Option[Tuple]
  def userUnapply(u: User) =
    Some((u.id, u.login, u.password, u.salt, u.roleId, u.connected, u.lastActivity.map(instant2Timestamp)))

  // Tuple -> User
  def userTupled(row: (Long, String, String, Int, Int, Boolean, Option1653280882)): User = {
    val (id, login, pwd, salt, roleId, connected, oLastAct) = row
    User(id, login, pwd, salt, roleId, connected, oLastAct.map(timestamp2Instant))
  }

  // Conversions
  def instant2Timestamp(i: Instant): Timestamp = new Timestamp(i.getMillis())
  def timestamp2Instant(ts: Timestamp): Instant = new Instant(ts.getTime())


  lazy val users = TableQuery[Usuarios]
} 

//设置package包名称以及导入依赖的类
package highperfscala.clientreports.streams

import org.joda.time.Instant

case class Ticker(value: String) extends AnyVal
case class Price(value: BigDecimal) extends AnyVal
case class OrderId(value: Long) extends AnyVal
case class EventInstant(value: Instant) extends AnyVal
case class ClientId(value: Long) extends AnyVal

sealed trait Order {
  def created: EventInstant
  def id: OrderId
  def ticker: Ticker
  def price: Price
  def clientId: ClientId
}
case class BuyOrder(
  created: EventInstant, id: OrderId, ticker: Ticker, price: Price,
  clientId: ClientId) extends Order
case class SellOrder(
  created: EventInstant, id: OrderId, ticker: Ticker, price: Price,
  clientId: ClientId) extends Order

case class Execution(created: EventInstant, id: OrderId, price: Price)


sealed trait OrderBookEvent
case class BuyOrderSubmitted(
  created: EventInstant, id: OrderId, ticker: Ticker, price: Price,
  clientId: ClientId) extends OrderBookEvent
case class SellOrderSubmitted(
  created: EventInstant, id: OrderId, ticker: Ticker, price: Price,
  clientId: ClientId) extends OrderBookEvent
case class OrderCanceled(created: EventInstant, id: OrderId)
  extends OrderBookEvent
case class OrderExecuted(created: EventInstant, id: OrderId, price: Price)
  extends OrderBookEvent 

//设置package包名称以及导入依赖的类
package highperfscala.clientreports.views

import org.joda.time.Instant

sealed trait LastHourPnL
case object LastHourPositive extends LastHourPnL
case object LastHourNegative extends LastHourPnL

sealed trait LastDayPnL
case object LastDayPositive extends LastDayPnL
case object LastDayNegative extends LastDayPnL

sealed trait LastSevenDayPnL
case object LastSevenDayPositive extends LastSevenDayPnL
case object LastSevenDayNegative extends LastSevenDayPnL

case class Ticker(value: String) extends AnyVal

case class TradingPerformanceTrend(
  ticker: Ticker,
  lastHour: LastHourPnL,
  lastDay: LastDayPnL,
  lastSevenDay: LastSevenDayPnL)

case class Price(value: BigDecimal) extends AnyVal
object Price {
  def average(ps: List[Price]): Price = {
    val prices = ps.map(_.value)
    Price(prices.sum / prices.length)
  }
}
case class OrderId(value: Long) extends AnyVal
case class CreatedTimestamp(value: Instant) extends AnyVal
case class ClientId(value: Long) extends AnyVal
sealed trait Order {
  def created: CreatedTimestamp
  def id: OrderId
  def ticker: Ticker
  def price: Price
  def clientId: ClientId
}
case class BuyOrder(
  created: CreatedTimestamp, id: OrderId, ticker: Ticker, price: Price,
  clientId: ClientId) extends Order
case class SellOrder(
  created: CreatedTimestamp, id: OrderId, ticker: Ticker, price: Price,
  clientId: ClientId) extends Order

case class Execution(created: CreatedTimestamp, id: OrderId, price: Price)

case class PnL(value: BigDecimal) extends AnyVal
object PnL {
  val zero: PnL = PnL(BigDecimal(0))
}

sealed trait PeriodPnL
case object PeriodPositive extends PeriodPnL
case object PeriodNegative extends PeriodPnL

case class GenerateTradingPerformanceTrend(
  tickers: List[Ticker], clientId: ClientId) 

//设置package包名称以及导入依赖的类
package highperfscala.clientreports.views

import org.joda.time.{Duration, Instant, Interval}

object PerformanceReporting {

  def trend(
    now: () => Instant,
    findOrders: (Interval, Ticker) => List[Order],
    findExecutions: (Interval, Ticker) => List[Execution],
    request: GenerateTradingPerformanceTrend): List[TradingPerformanceTrend] = {
    def periodPnL(
      duration: Duration): Map[Ticker, PeriodPnL] = {
      val currentTime = now()
      val interval = new Interval(currentTime.minus(duration), currentTime)
      (for {
        ticker <- request.tickers
        orders = findOrders(interval, ticker)
        executions = findExecutions(interval, ticker)
        idToExecPrice = executions.groupBy(_.id).mapValues(es =>
          Price.average(es.map(_.price)))
        signedExecutionPrices = for {
          o <- orders
          if o.clientId == request.clientId
          price <- idToExecPrice.get(o.id).map(p => o match {
            case _: BuyOrder => Price(p.value * -1)
            case _: SellOrder => p
          }).toList
        } yield price
        trend = signedExecutionPrices.foldLeft(PnL.zero) {
          case (pnl, p) => PnL(pnl.value + p.value)
        } match {
          case p if p.value >= PnL.zero.value => PeriodPositive
          case _ => PeriodNegative
        }
      } yield ticker -> trend).toMap
    }

    val tickerToLastHour = periodPnL(Duration.standardHours(1)).mapValues {
      case PeriodPositive => LastHourPositive
      case PeriodNegative => LastHourNegative
    }
    val tickerToLastDay = periodPnL(Duration.standardDays(1)).mapValues {
      case PeriodPositive => LastDayPositive
      case PeriodNegative => LastDayNegative
    }
    val tickerToLastSevenDays = periodPnL(Duration.standardDays(7)).mapValues {
      case PeriodPositive => LastSevenDayPositive
      case PeriodNegative => LastSevenDayNegative
    }

    tickerToLastHour.zip(tickerToLastDay).zip(tickerToLastSevenDays).map({
      case (((t, lastHour), (_, lastDay)), (_, lastSevenDays)) =>
        TradingPerformanceTrend(t, lastHour, lastDay, lastSevenDays)
    }).toList
  }
} 

//设置package包名称以及导入依赖的类
package highperfscala.clientreports.views

import org.joda.time.{Duration, Instant, Interval}

object ViewPerformanceReporting {

  def trend(
    now: () => Instant,
    findOrders: (Interval, Ticker) => List[Order],
    findExecutions: (Interval, Ticker) => List[Execution],
    request: GenerateTradingPerformanceTrend): List[TradingPerformanceTrend] = {

    def periodPnL(
      duration: Duration): Map[Ticker, PeriodPnL] = {
      val currentTime = now()
      val interval = new Interval(currentTime.minus(duration), currentTime)
      (for {
        ticker <- request.tickers
        orders = findOrders(interval, ticker)
        executions = findExecutions(interval, ticker)
        idToExecPrice = executions.groupBy(_.id).mapValues(es =>
          Price.average(es.map(_.price)))
        signedExecutionPrices = for {
          o <- orders.view
          if o.clientId == request.clientId
          price <- idToExecPrice.get(o.id).map(p => o match {
            case _: BuyOrder => Price(p.value * -1)
            case _: SellOrder => p
          }).toList
        } yield price
        trend = signedExecutionPrices.foldLeft(PnL.zero) {
          case (pnl, p) => PnL(pnl.value + p.value)
        } match {
          case p if p.value >= PnL.zero.value => PeriodPositive
          case _ => PeriodNegative
        }
      } yield ticker -> trend).toMap
    }

    val tickerToLastHour = periodPnL(Duration.standardHours(1)).mapValues {
      case PeriodPositive => LastHourPositive
      case PeriodNegative => LastHourNegative
    }
    val tickerToLastDay = periodPnL(Duration.standardDays(1)).mapValues {
      case PeriodPositive => LastDayPositive
      case PeriodNegative => LastDayNegative
    }
    val tickerToLastSevenDays = periodPnL(Duration.standardDays(7)).mapValues {
      case PeriodPositive => LastSevenDayPositive
      case PeriodNegative => LastSevenDayNegative
    }

    tickerToLastHour.zip(tickerToLastDay).zip(tickerToLastSevenDays).map({
      case (((t, lastHour), (_, lastDay)), (_, lastSevenDays)) =>
        TradingPerformanceTrend(t, lastHour, lastDay, lastSevenDays)
    }).toList
  }

} 

//设置package包名称以及导入依赖的类
package highperfscala.orderbook

import org.joda.time.Instant
import org.scalacheck.Gen

// Model taken from chapter 2

case class Price(value: BigDecimal) extends AnyVal
object Price {
  implicit val genPrice: Gen[Price] = Gen.posNum[Double].map(d =>
    Price(BigDecimal(d)))
  implicit val ordering: Ordering[Price] = new Ordering[Price] {
    def compare(x: Price, y: Price): Int =
      Ordering.BigDecimal.compare(x.value, y.value)
  }
}

case class OrderId(value: Long)
object OrderId {
  implicit val genOrderId: Gen[OrderId] = Gen.posNum[Long].map(OrderId.apply)
}

sealed trait LimitOrder {
  def id: OrderId
  def price: Price
}
object LimitOrder {
  implicit val genLimitOrder: Gen[LimitOrder] = Gen.oneOf(
    BuyLimitOrder.genBuyLimitOrder, SellLimitOrder.genSellLimitOrder)
}
case class BuyLimitOrder(id: OrderId, price: Price) extends LimitOrder
object BuyLimitOrder {
  implicit val genBuyLimitOrder: Gen[BuyLimitOrder] = Gen.zip(
    OrderId.genOrderId, Price.genPrice).map(Function.tupled(BuyLimitOrder.apply))
}
case class SellLimitOrder(id: OrderId, price: Price) extends LimitOrder
object SellLimitOrder {
  implicit val genSellLimitOrder: Gen[SellLimitOrder] = Gen.zip(
    OrderId.genOrderId, Price.genPrice).map(Function.tupled(
    SellLimitOrder.apply))
}

case class Execution(orderId: OrderId, price: Price)

case class CommandInstant(value: Instant) extends AnyVal
object CommandInstant {
  def now(): CommandInstant =
    CommandInstant(new Instant(System.currentTimeMillis()))
  implicit val genCommandInstant: Gen[CommandInstant] =
    Gen.posNum[Long].map(l => CommandInstant(new Instant(l)))
}

case class EventInstant(value: Instant) extends AnyVal
object EventInstant {
  def now(): EventInstant =
    EventInstant(new Instant(System.currentTimeMillis()))
  implicit val genEventInstant: Gen[EventInstant] =
    Gen.posNum[Long].map(l => EventInstant(new Instant(l)))
} 

//设置package包名称以及导入依赖的类
package com.zendesk.scalaflow.sugar

import com.google.cloud.dataflow.sdk.coders.Coder
import com.google.cloud.dataflow.sdk.transforms._
import com.google.cloud.dataflow.sdk.values.{KV, PCollection, PCollectionList, POutput}
import com.zendesk.scalaflow.sugar.WrapperOps._
import org.joda.time.Instant

import scala.collection.JavaConverters._

trait CollectionOps {

  implicit class RichCollection[A: Coder](val collection: PCollection[A]) {

    def parDo[B](f: DoFn[A, B]#ProcessContext => Unit)(implicit coder: Coder[B]): PCollection[B] = {
      collection.apply(asParDo(f)).setCoder(coder)
    }

    def map[B : Coder](f: A => B): PCollection[B] = parDo {
      c => c.output(f(c.element))
    }

    def filter(f: A => Boolean): PCollection[A] = parDo {
      c => if (f(c.element)) c.output(c.element)
    }

    def collect[B : Coder](pf: PartialFunction[A, B]): PCollection[B] = parDo {
      c => if (pf.isDefinedAt(c.element)) c.output(pf(c.element))
    }

    def extractTimestamp(implicit c: Coder[(A, Instant)]): PCollection[(A, Instant)] = parDo {
      c => c.output((c.element, c.timestamp))
    }

    def flatMap[B : Coder](f: A => Iterable[B]): PCollection[B] = parDo {
      c => f(c.element).foreach(c.output)
    }

    def foreach(f: A => Unit): PCollection[A] = parDo {
      c => { f(c.element); c.output(c.element) }
    }

    def withKey[B : Coder](f: A => B)(implicit c: Coder[KV[B, A]]): PCollection[KV[B, A]] = parDo {
      c => c.output(KV.of(f(c.element), c.element))
    }

    def flattenWith(first: PCollection[A], others: PCollection[A]*): PCollection[A] = {
      val all  = collection :: first :: others.toList
      PCollectionList.of(all.asJava).apply(Flatten.pCollections[A])
    }

    def transformWith[B <: POutput](name: String)(f: PCollection[A] => B): B = {
      collection.apply(name, asPTransform(f))
    }
  }
}

object CollectionOps extends CollectionOps 

//设置package包名称以及导入依赖的类
package com.zendesk.scalaflow.sugar

import java.lang.{Iterable => JIterable}

import com.google.cloud.dataflow.sdk.coders.{Coder, IterableCoder}
import com.google.cloud.dataflow.sdk.transforms.{Combine, DoFn, GroupByKey, Top}
import com.google.cloud.dataflow.sdk.values.{KV, PCollection}
import com.zendesk.scalaflow._
import org.joda.time.Instant

import scala.collection.JavaConverters._

trait KVCollectionOps {

  implicit class RichKVCollection[K : Coder, A: Coder](val collection: PCollection[KV[K, A]]) {

    def parDo[B](f: DoFn[KV[K, A], KV[K, B]]#ProcessContext => Unit)(implicit coder: Coder[KV[K, B]]): PCollection[KV[K, B]] = {
      collection.apply(asParDo(f)).setCoder(coder)
    }

    def mapValue[B : Coder](f: A => B): PCollection[KV[K, B]] = parDo {
      c => c.output(KV.of(c.element.getKey, f(c.element.getValue)))
    }

    def flatMapValue[B : Coder](f: A => Iterable[B]): PCollection[KV[K, B]] = parDo {
      c => f(c.element.getValue).foreach { value => c.output(KV.of(c.element.getKey, value)) }
    }

    def extractTimestamp: PCollection[KV[K, (A, Instant)]] = parDo {
      c => c.output(KV.of(c.element.getKey, (c.element.getValue, c.timestamp)))
    }

    def combinePerKey(zero: A)(f: (A, A) => A): PCollection[KV[K, A]] = {
      val g = (input: JIterable[A]) => input.asScala.fold(zero)(f)
      collection.apply(Combine.perKey[K, A](asSimpleFn(g)))
    }

    def groupByKey: PCollection[KV[K, Iterable[A]]] = {
      collection.apply(GroupByKey.create[K, A]).mapValue(_.asScala)
    }

    def topPerKey(count: Int)(implicit ordered: Ordering[A]): PCollection[KV[K, List[A]]] = {
      collection.apply(Top.perKey(count, ordered)).mapValue(_.asScala.toList)
    }
  }
}

object KVCollectionOps extends KVCollectionOps