scalaz stream TA

package finance;

import scalaz.stream._
import scalaz.stream.Exchange._
import scalaz.concurrent.Task
import scalaz.stream.async.mutable._
import scalaz.stream.wye._
import scalaz.stream.Process._
import scalaz.stream.ReceiveY._
import scalaz.stream.process1.Await1

import Cause._
import scala.annotation.tailrec
import scala.concurrent.duration.Duration
import scalaz.\/._

object TA {

    case class Ohlc(o: BigDecimal, h: BigDecimal, l: BigDecimal, c:BigDecimal, v:BigDecimal)

    //http://en.wikipedia.org/wiki/Average_directional_movement_index
    //UpMove for ADX
    class upMove  
    {
        //we send out zero on day 1
        private var day1: Boolean = true
        private var yesterday: BigDecimal = 0.0

        val p  =  process1.lift[Ohlc, BigDecimal] (day =>{
            val upmove : BigDecimal = day1 match {
                case true => {day1 = false; 0.0 };
                case false => day.h - yesterday
                }

            //set yesterday to this day
            yesterday = day.h
            upmove
        })

    }

    val upmove = new upMove p

    //http://en.wikipedia.org/wiki/Average_directional_movement_index
    //DownMove for ADX
    class downMove  
    {
        //we send out zero on day 1
        private var day1: Boolean = true
        private var yesterday: BigDecimal = 0.0

        val p  =  process1.lift[Ohlc, BigDecimal] (day =>{

            val downmove : BigDecimal = day1 match {
                case true => {day1 = false; 0.0 };
                case false =>  yesterday - day.l
                }

            //set yesterday to this day
            yesterday = day.l
            downmove
        })

    }

    val downmove = new downMove p

    //http://en.wikipedia.org/wiki/Average_directional_movement_index
    //+DM for ADX
    class pDM  
    {
        private var storedDay:BigDecimal = 0.0
        private var haveUp = false
        private var haveDown = false
        private val _downmove = downmove
        private val _upmove = upmove

        private def recUp(up: BigDecimal):Option[BigDecimal] = {
            if (haveDown == true)
            {
                haveDown = false
                Some(if(up > storedDay && up > 0.0) up else 0.0)
            } else
            {
                storedDay = up
                haveUp = true
                None
            }
        } 

        private def recDown(down: BigDecimal):Option[BigDecimal] = {
            if (haveUp == true)
            {
                haveUp = false
                Some(if(storedDay > down && storedDay > 0.0) storedDay else 0.0)
            } else
            {
                storedDay = down
                haveDown = true
                None
            }
        } 

            

        private def com: Wye[BigDecimal,BigDecimal,BigDecimal] =
            receiveBoth {
              case ReceiveL(i) => recUp(i) match {
                    case Some(v) => emit(v);
                }
              case ReceiveR(i) => recDown(i) match {
                    case Some(v) => emit(v);
                }
              case HaltL(End)   => awaitR.repeat
              case HaltR(End)   => awaitL.repeat
              case HaltOne(rsn) => Halt(rsn)
            }
        val p =  _upmove.wye(_downmove)(com)


    }

    val pdm = new pDM p
}