<commit_before>/*
 *  ____    ____    _____    ____    ___     ____ 
 * |  _ \  |  _ \  | ____|  / ___|  / _/    / ___|        Precog (R)
 * | |_) | | |_) | |  _|   | |     | |  /| | |  _         Advanced Analytics Engine for NoSQL Data
 * |  __/  |  _ <  | |___  | |___  |/ _| | | |_| |        Copyright (C) 2010 - 2013 SlamData, Inc.
 * |_|     |_| \_\ |_____|  \____|   /__/   \____|        All Rights Reserved.
 *
 * This program is free software: you can redistribute it and/or modify it under the terms of the 
 * GNU Affero General Public License as published by the Free Software Foundation, either version 
 * 3 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; 
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See 
 * the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License along with this 
 * program. If not, see <http://www.gnu.org/licenses/>.
 *
 */
package com.precog
package daze

import yggdrasil._
import yggdrasil.table._

import common._
import common.json._

import bytecode._
import TableModule._

import Jama._
import Jama.Matrix._

import blueeyes.json._

import scalaz._
import scalaz.syntax.monad._
import scalaz.std.list._
import scalaz.std.stream._
import scalaz.std.set._
import scalaz.syntax.traverse._

trait LinearRegressionLibModule[M[+_]] extends ColumnarTableLibModule[M] with EvaluatorMethodsModule[M] with PredictionLibModule[M] {
  trait LinearRegressionLib extends ColumnarTableLib with RegressionSupport with PredictionSupport with EvaluatorMethods {
    import trans._

    override def _libMorphism2 = super._libMorphism2 ++ Set(MultiLinearRegression, LinearPrediction)

    object MultiLinearRegression extends Morphism2(Stats2Namespace, "linearRegression") {
      val tpe = BinaryOperationType(JNumberT, JType.JUniverseT, JObjectUnfixedT)

      lazy val alignment = MorphismAlignment.Match(M.point(morph1))

      type Beta = Array[Double]
      type Result = Beta

      /**
       * http://adrem.ua.ac.be/sites/adrem.ua.ac.be/files/StreamFitter.pdf
       *
       * First slice size is made consistent. Then the slices are fed to the reducer, one by one.
       * The reducer calculates the Ordinary Least Squares regression for each Slice. 
       * The results of each of these regressions are then combined incrementally using `monoid`. 
       * `alpha` (a value between 0 and 1) is the paramater which determines the weighting of the
       * data in the stream. A value of 0.5 means that current values and past values
       * are equally weighted. The paper above outlines how `alpha` relates to the half-life
       * of the current window (i.e. the current Slice). In the future, we could let half-life,
       * or something related, be an optional parameter in the regression model.
       */

      val alpha = 0.5

      implicit def monoid = new Monoid[Result] {
        def zero = Array.empty[Double]
        def append(r1: Result, r2: => Result) = {
          lazy val newr1 = r1 map { _ * alpha }
          lazy val newr2 = r2 map { _ * (1.0 - alpha) }

          if (r1.isEmpty) r2
          else if (r2.isEmpty) r1
          else arraySum(newr1, newr2)
        }
      }

      implicit def resultMonoid = new Monoid[Option[Array[Result]]] {
        def zero = None
        def append(t1: Option[Array[Result]], t2: => Option[Array[Result]]) = {
          t1 match {
            case None => t2
            case Some(c1) => t2 match {
              case None => Some(c1)
              case Some(c2) => Some(c1 ++ c2)
            }
          }
        }
      }

      def reducer: Reducer[Result] = new Reducer[Result] {
        def reduce(schema: CSchema, range: Range): Result = {
          val features = schema.columns(JArrayHomogeneousT(JNumberT))

          val values: Set[Option[Array[Array[Double]]]] = features map {
            case c: HomogeneousArrayColumn[_] if c.tpe.manifest.erasure == classOf[Array[Double]] =>
              val mapped = range.toArray filter { r => c.isDefinedAt(r) } map { i => c.asInstanceOf[HomogeneousArrayColumn[Double]](i) }
              Some(mapped)
            case other => 
              logger.warn("Features were not correctly put into a homogeneous array of doubles; returning empty.")
              None
          }

          val arrays = {
            if (values.isEmpty) None
            else values.suml(resultMonoid)
          }

          val xs = arrays map { _ map { arr => 1.0 +: (java.util.Arrays.copyOf(arr, arr.length - 1)) } }
          val y0 = arrays map { _ map { _.last } }

          val matrixX = xs map { case arr => new Matrix(arr) }

          // FIXME ultimately we do not want to throw an IllegalArgumentException here
          // once the framework is in place, we will return the empty set and issue a warning to the user
          // this catches the case when the user runs regression on data when rows < (columns + 1)
          val inverseX = try {
            matrixX map { _.inverse() }
          } catch {
            case ex: RuntimeException if ex.getMessage == "Matrix is rank deficient." => 
              throw new IllegalArgumentException("More features than rows found in linear regression. Not enough information to determine model.", ex)
          }

          val matrixY = y0 map { case arr => new Matrix(Array(arr)) }

          val matrixProduct: Option[Matrix] = for {
            inverse <- inverseX
            y <- matrixY
          } yield {
            (inverse).times(y.transpose)
          }

          matrixProduct map { _.getArray flatten } getOrElse Array.empty[Double]
        }
      }

      def extract(res: Result, jtype: JType): Table = {
        val cpaths = Schema.cpath(jtype)

        val tree = CPath.makeTree(cpaths, Range(1, res.length).toSeq :+ 0)

        val spec = TransSpec.concatChildren(tree)
  
        val theta = Table.fromRValues(Stream(RArray(res.map(CNum(_)).toList)))

        val result = theta.transform(spec)

        val valueTable = result.transform(trans.WrapObject(Leaf(Source), paths.Value.name))
        val keyTable = Table.constEmptyArray.transform(trans.WrapObject(Leaf(Source), paths.Key.name))

        valueTable.cross(keyTable)(InnerObjectConcat(Leaf(SourceLeft), Leaf(SourceRight)))
      }
<commit_msg>Changed LinearRegression so that all data is weighted equally (instead of weighting the most recent data higher, which only makes sense when we have online model evaluation in place).<commit_after>/*
 *  ____    ____    _____    ____    ___     ____ 
 * |  _ \  |  _ \  | ____|  / ___|  / _/    / ___|        Precog (R)
 * | |_) | | |_) | |  _|   | |     | |  /| | |  _         Advanced Analytics Engine for NoSQL Data
 * |  __/  |  _ <  | |___  | |___  |/ _| | | |_| |        Copyright (C) 2010 - 2013 SlamData, Inc.
 * |_|     |_| \_\ |_____|  \____|   /__/   \____|        All Rights Reserved.
 *
 * This program is free software: you can redistribute it and/or modify it under the terms of the 
 * GNU Affero General Public License as published by the Free Software Foundation, either version 
 * 3 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; 
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See 
 * the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License along with this 
 * program. If not, see <http://www.gnu.org/licenses/>.
 *
 */
package com.precog
package daze

import yggdrasil._
import yggdrasil.table._

import common._
import common.json._

import bytecode._
import TableModule._

import Jama._
import Jama.Matrix._

import blueeyes.json._

import scalaz._
import scalaz.syntax.monad._
import scalaz.std.list._
import scalaz.std.stream._
import scalaz.std.set._
import scalaz.syntax.traverse._

trait LinearRegressionLibModule[M[+_]] extends ColumnarTableLibModule[M] with EvaluatorMethodsModule[M] with PredictionLibModule[M] {
  trait LinearRegressionLib extends ColumnarTableLib with RegressionSupport with PredictionSupport with EvaluatorMethods {
    import trans._

    override def _libMorphism2 = super._libMorphism2 ++ Set(MultiLinearRegression, LinearPrediction)

    object MultiLinearRegression extends Morphism2(Stats2Namespace, "linearRegression") {
      val tpe = BinaryOperationType(JNumberT, JType.JUniverseT, JObjectUnfixedT)

      lazy val alignment = MorphismAlignment.Match(M.point(morph1))

      type Beta = Array[Double]

      case class Accumulator(beta: Beta, count: Int)
      type Result = Accumulator

      /**
       * http://adrem.ua.ac.be/sites/adrem.ua.ac.be/files/StreamFitter.pdf
       *
       * First slice size is made consistent. Then the slices are fed to the reducer, one by one.
       * The reducer calculates the Ordinary Least Squares regression for each Slice. 
       * The results of each of these regressions are then combined incrementally using `monoid`. 
       * `alpha` (a value between 0 and 1) is the paramater which determines the weighting of the
       * data in the stream. A value of 0.5 means that current values and past values
       * are equally weighted. The paper above outlines how `alpha` relates to the half-life
       * of the current window (i.e. the current Slice). In the future, we could let half-life,
       * or something related, be an optional parameter in the regression model.
       */


      /**
       * This code here will necessary when we have online models and users want to weight
       * the most recent data with a higher (or lower) weight than the data already seen.
       * But since we don't have this capability yet, all data is weighted equally.

      val alpha = 0.5

      implicit def monoid = new Monoid[Result] {
        def zero = Array.empty[Double]
        def append(r1: Result, r2: => Result) = {
          lazy val newr1 = r1 map { _ * alpha }
          lazy val newr2 = r2 map { _ * (1.0 - alpha) }

          if (r1.isEmpty) r2
          else if (r2.isEmpty) r1
          else arraySum(newr1, newr2)
        }
      }
      */

      implicit def monoid = new Monoid[Result] {
        def zero = Accumulator(Array.empty[Double], 0)
        def append(r1: Result, r2: => Result) = {
          if (r1.beta isEmpty) r2
          else if (r2.beta isEmpty) r1
          else Accumulator(arraySum(r1.beta, r2.beta), r1.count + r2.count)
        }
      }

      implicit def betaMonoid = new Monoid[Option[Array[Beta]]] {
        def zero = None
        def append(t1: Option[Array[Beta]], t2: => Option[Array[Beta]]) = {
          t1 match {
            case None => t2
            case Some(c1) => t2 match {
              case None => Some(c1)
              case Some(c2) => Some(c1 ++ c2)
            }
          }
        }
      }

      def reducer: Reducer[Result] = new Reducer[Result] {
        def reduce(schema: CSchema, range: Range): Result = {
          val features = schema.columns(JArrayHomogeneousT(JNumberT))

          val values: Set[Option[Array[Array[Double]]]] = features map {
            case c: HomogeneousArrayColumn[_] if c.tpe.manifest.erasure == classOf[Array[Double]] =>
              val mapped = range.toArray filter { r => c.isDefinedAt(r) } map { i => c.asInstanceOf[HomogeneousArrayColumn[Double]](i) }
              Some(mapped)
            case other => 
              logger.warn("Features were not correctly put into a homogeneous array of doubles; returning empty.")
              None
          }

          val arrays = {
            if (values.isEmpty) None
            else values.suml(betaMonoid)
          }

          val xs = arrays map { _ map { arr => 1.0 +: (java.util.Arrays.copyOf(arr, arr.length - 1)) } }
          val y0 = arrays map { _ map { _.last } }

          val matrixX = xs map { case arr => new Matrix(arr) }

          // FIXME ultimately we do not want to throw an IllegalArgumentException here
          // once the framework is in place, we will return the empty set and issue a warning to the user
          // this catches the case when the user runs regression on data when rows < (columns + 1)
          val inverseX = try {
            matrixX map { _.inverse() }
          } catch {
            case ex: RuntimeException if ex.getMessage == "Matrix is rank deficient." => 
              throw new IllegalArgumentException("More features than rows found in linear regression. Not enough information to determine model.", ex)
          }

          val matrixY = y0 map { case arr => new Matrix(Array(arr)) }

          val matrixProduct: Option[Matrix] = for {
            inverse <- inverseX
            y <- matrixY
          } yield {
            (inverse).times(y.transpose)
          }

          val res = matrixProduct map { _.getArray flatten } getOrElse Array.empty[Double]
          
          Accumulator(res, 1)
        }
      }

      def extract(res: Result, jtype: JType): Table = {
        val cpaths = Schema.cpath(jtype)

        val tree = CPath.makeTree(cpaths, Range(1, res.beta.length).toSeq :+ 0)

        val spec = TransSpec.concatChildren(tree)

        val mean = res match { case Accumulator(beta, count) => beta map { _ / count } }

        val theta = Table.fromRValues(Stream(RArray(mean.map(CNum(_)).toList)))

        val result = theta.transform(spec)

        val valueTable = result.transform(trans.WrapObject(Leaf(Source), paths.Value.name))
        val keyTable = Table.constEmptyArray.transform(trans.WrapObject(Leaf(Source), paths.Key.name))

        valueTable.cross(keyTable)(InnerObjectConcat(Leaf(SourceLeft), Leaf(SourceRight)))
      }

{'fraction_non_alphanumeric': 0.09401586157173757, 'fraction_numerical': 0.0083633741888969, 'mean_word_length': 2.3651140223192626, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '978128', 'n_tokens_mistral': 4104, 'n_tokens_neox': 3914, 'n_words': 1315}