import scala.concurrent.Future
import akka.stream.{ActorMaterializer, KillSwitch, KillSwitches, ThrottleMode}
import akka.stream.scaladsl.{Keep, Sink, Source}
import akka.actor.ActorSystem

import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.Random

object TradeBot {

  def main(args: Array[String]): Unit = {

    implicit val system = ActorSystem("TradeBot")
    implicit val materializer = ActorMaterializer()
    var currentStockPrice = 0

    final case class StockQuote(symbol: String, price: Int)
    sealed trait Trade {
      val tickerSymbol: String
      val price: Int
      val shares: Int
    }
    final case class BuyTrade(tickerSymbol: String, shares: Int, price: Int) extends Trade
    final case class SellTrade(tickerSymbol: String, shares: Int, price: Int) extends Trade
    final case class TradeResult(success: Boolean, trade: Trade)

    var tradeBotKillSwitch: Option[KillSwitch] = None

    def sendEmail(emailAddress: String, emailMessage: String): Future[Boolean] = {
      println(emailMessage)
      Future.successful(true)
    }

    def sendTradeEmail(emailAddress: String)(tradeMade: Trade) = {
      sendEmail(emailAddress, s"Trade: $tradeMade")
    }

    def getNextStockQuote(tickerSymbol: String, priceChange: Int) = {
      if (priceChange % 2 == 0 && currentStockPrice - priceChange > 0) {
        currentStockPrice = currentStockPrice - priceChange
      } else {
        currentStockPrice = currentStockPrice + priceChange
      }
      val nextStockQuote = StockQuote(tickerSymbol, currentStockPrice)
      println(s"Stock Quote:  $nextStockQuote")
      nextStockQuote
    }

    def getQuoteStreamForStock(buyPrice: Int, sellPrice: Int)(tickerSymbol: String) = {
      currentStockPrice = (buyPrice + sellPrice) / 2
      Source.fromIterator(() => Iterator.continually(getNextStockQuote(tickerSymbol, Random.nextInt(10))))
        .throttle(1, 1.second, 1, ThrottleMode.shaping)
        .take(100)
    }

    def makeTrade(tradeToMake: Trade): Future[Boolean] = {
      Future.successful(true)
    }

    def createTrade(buyPrice: Int, sellPrice: Int)(stockQuote: StockQuote): Option[Trade] = {
      if (stockQuote.price < buyPrice) {
        Some(BuyTrade(stockQuote.symbol, 1, stockQuote.price))
      } else if (stockQuote.price > sellPrice) {
        Some(SellTrade(stockQuote.symbol, 1, stockQuote.price))
      } else {
        None
      }
    }

    def startTradeBot(tickerSymbol: String, tradesPerDayLimit: Int, buyPrice: Int, sellPrice: Int, notificationEmailAddress: String): Unit = {
      println(s"Starting new trade bot for $tickerSymbol at $tradesPerDayLimit per day, buy at $buyPrice sell at $sellPrice")
      tradeBotKillSwitch.foreach(_.shutdown())

      val tradeBotTradeCreator = createTrade(buyPrice, sellPrice)(_)
      val tradeBotEmailCreator = sendTradeEmail(notificationEmailAddress)(_)

      val newTradeBotKillSwitch = getQuoteStreamForStock(buyPrice, sellPrice)(tickerSymbol)
        .viaMat(KillSwitches.single)(Keep.right)
        .map(tradeBotTradeCreator)
        .mapConcat(_.toList)
        .throttle(tradesPerDayLimit, 1.day, tradesPerDayLimit, ThrottleMode.shaping)
        .mapAsync(tradesPerDayLimit)( trade => makeTrade(trade).map(TradeResult(_, trade)) )
        .filter(_.success)
        .mapAsync(tradesPerDayLimit) ( tradeResult => tradeBotEmailCreator(tradeResult.trade) )
        .toMat(Sink.ignore)(Keep.left)
        .run()

      tradeBotKillSwitch = Some(newTradeBotKillSwitch)
    }

    def stopTradeBot(): Unit = {
      println("Stopping trade bot")
      tradeBotKillSwitch.foreach(_.shutdown())
    }

    startTradeBot("GOOGL", 10, 900, 950, "foobar@gmail.com")
    Thread.sleep(5000)
    startTradeBot("TSLA", 10, 330, 340, "foobar@gmail.com")
    Thread.sleep(5000)
    stopTradeBot()
    startTradeBot("AAPL", 1, 163, 165, "foobar@gmail.com")

  }

}