//设置package包名称以及导入依赖的类
package com.enriquegrodrigo.spark.crowd.aggregators 

import com.enriquegrodrigo.spark.crowd.types.GladPartial
import com.enriquegrodrigo.spark.crowd.types.GladParams
import com.enriquegrodrigo.spark.crowd.utils.Functions
import org.apache.spark.sql.{Encoder,Encoders}
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.broadcast.Broadcast

import scala.math.{log,exp}

private[crowd] class GladEAggregator(params: Broadcast[GladParams]) 
  extends Aggregator[GladPartial, GladEAggregatorBuffer, Double]{

  def zero: GladEAggregatorBuffer = GladEAggregatorBuffer(Vector.fill(2)(1)) //Binary
  
  def reduce(b: GladEAggregatorBuffer, a: GladPartial) : GladEAggregatorBuffer = {
    val alpha = params.value.alpha
    val sigmoidValue = Functions.sigmoid(alpha(a.annotator)*a.beta)
    val p0 = if (a.value == 0) sigmoidValue else (1 - sigmoidValue)
    val p1 = if (a.value == 1) sigmoidValue else (1 - sigmoidValue) 
    GladEAggregatorBuffer(Vector(Functions.logLim(p0) + b.aggVect(0), Functions.logLim(p1) + b.aggVect(1)))
  }

  def merge(b1: GladEAggregatorBuffer, b2: GladEAggregatorBuffer) : GladEAggregatorBuffer = { 
    GladEAggregatorBuffer(b1.aggVect.zip(b2.aggVect).map(x => x._1 * x._2))
  }

  def finish(reduction: GladEAggregatorBuffer) = {
    val w = params.value.w
    val negative = exp(reduction.aggVect(0) + Functions.logLim(w(0)))
    val positive = exp(reduction.aggVect(1) + Functions.logLim(w(1)))
    val norm = negative + positive
    positive/norm
  }

  def bufferEncoder: Encoder[GladEAggregatorBuffer] = Encoders.product[GladEAggregatorBuffer]

  def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

private[crowd] case class GladEAggregatorBuffer(aggVect: scala.collection.Seq[Double]) 

//设置package包名称以及导入依赖的类
package org.altic.spark.clustering.bitm

import scala.math.exp


trait TopoFactor extends Serializable {
  def gen(maxIter: Int, currentIter: Int, nbNeurons: Int): Array[Double]
}

object BiTMTopoFactor extends TopoFactor {
  def gen(maxIter: Int, currentIter: Int, nbNeurons: Int) = {
    val T: Double = (maxIter - currentIter + 1) / maxIter.toDouble
    //val T = 0.9
    Array.tabulate(nbNeurons * 2)(dist => exp(dist / T))
  }
}

object CroeucTopoFactor extends TopoFactor {
  def gen(maxIter: Int, currentIter: Int, nbNeurons: Int): Array[Double] = {
    val T: Double = (maxIter - currentIter + 1) / maxIter.toDouble
    //val T = 0.9
    Array.tabulate(nbNeurons * 2)(i => if (i == 0) exp(i / T) else Double.MaxValue)
  }
} 

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

import breeze.linalg.{DenseMatrix, DenseVector}

import scala.math.{exp, sqrt}

object FuturesPricing {
  def calculateForShares(sharePriceLattice: DenseMatrix[Double], termInYears: Double, volatility: Double,
                         numberOfPeriods: Int, interestRate: Double, dividendYield: Double): Double = {
    val result = calculatePricingMatrix(sharePriceLattice, termInYears, volatility, numberOfPeriods, interestRate, dividendYield)

    result.apply(0, 0)
  }

  def calculate(priceLattice: DenseMatrix[Double], numberOfPeriods: Int, q: Double, p: Double): Double = {
    val result = calculatePricingMatrix(priceLattice, numberOfPeriods, q, p)

    result.apply(0, 0)
  }

  private[week4] def calculatePricingMatrix(sharePriceLattice: DenseMatrix[Double], termInYears: Double, volatility: Double, numberOfPeriods: Int, interestRate: Double, dividendYield: Double): DenseMatrix[Double] = {
    val u: Double = exp(volatility * sqrt(termInYears / numberOfPeriods))
    val d: Double = 1 / u
    val q: Double = (exp((interestRate - dividendYield) * termInYears / numberOfPeriods) - d) / (u - d)
    val p: Double = 1 - q
    calculatePricingMatrix(sharePriceLattice, numberOfPeriods, q, p)
  }

  private def calculatePricingMatrix(priceLattice: DenseMatrix[Double], numberOfPeriods: Int, q: Double, p: Double): DenseMatrix[Double] = {
    val n = numberOfPeriods + 1

    val result = DenseMatrix.zeros[Double](n, n)
    val lastColumnIndex = numberOfPeriods
    result(::, lastColumnIndex) := priceLattice(::, lastColumnIndex).slice(0, n)

    val columnsFromPriorToLastOneDownToFirst = (lastColumnIndex - 1) to 0 by -1
    for (i <- columnsFromPriorToLastOneDownToFirst) {
      val previousColumn = result(::, i + 1)
      val previousColumnShiftedOneDown = DenseVector.vertcat(previousColumn(1 until previousColumn.length), DenseVector.zeros[Double](1))
      val currentColumn = (previousColumnShiftedOneDown * q) + (previousColumn * p)
      result(::, i) += currentColumn
    }
    result
  }
}