<commit_before>package com.flowy.bexchange.trade

import akka.actor.{Actor, ActorLogging, ActorRef, PoisonPill, Props}
import com.flowy.common.database.TheEverythingBagelDao
import com.flowy.common.models.MarketStructures.MarketUpdate
import com.flowy.common.models._
import java.time.{Instant, ZoneOffset}
import java.util.UUID

import com.flowy.bexchange.trade.TrailingStopLossActor.TrailingStop
import com.flowy.common.Util

import scala.collection.mutable.ListBuffer
import scala.concurrent.{ExecutionContext, Future}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.tools.reflect.ToolBox


object TradeActor {
  def props(trade: Trade, bagel: TheEverythingBagelDao)(implicit context: ExecutionContext) =
    Props(new TradeActor(trade, bagel))

  case class Trigger(action: TradeAction.Value, price: Double, condition: String)
  case class Buy(price: Double, atCondition: String)
  case class Sell(price: Double, atCondition: String)
  case class Cancel(sender: ActorRef)
  case class Update(user: UserData, request: TradeRequest, sender: ActorRef)
}


/**
  * Services a single market: example BTC-NEO
  * @param trade
  * @param bagel
  */
class TradeActor(trade: Trade, bagel: TheEverythingBagelDao) extends Actor
  with ActorLogging {

  import TradeActor._

  import scala.reflect.runtime.currentMirror

  implicit val akkaSystem = context.system

  private var myTrade = trade
  private val dynamic = currentMirror.mkToolBox()

  private val buyConditions = scala.collection.mutable.ListBuffer[ActorRef]()
  private val lossConditions = scala.collection.mutable.ListBuffer[ActorRef]()
  private val profitConditions = scala.collection.mutable.ListBuffer[ActorRef]()

  override def preStart() = {
    loadConditions()
  }

  override def postStop() = {
    log.info(s"trade ${trade.id} is shutting down")
  }

  def receive: Receive = {
    case update: MarketUpdate =>
      context.system.actorSelection(s"${self.path}/*") ! update

    case Update(user, request, sender) =>
      updateTrade(user, request, sender)

    case Cancel(sender) =>
      cancelTrade(sender)

    case Trigger(TradeAction.Buy, price, condition) =>
      executeBuy(price, condition)

    case Trigger(TradeAction.Sell, price, condition) =>
      executeSell(price, condition)

    case x =>
      log.warning(s"received unknown message - $x")
  }


  private def executeBuy(price: Double, condition: String) = {
    log.info(s"buy ${myTrade.baseQuantity} ${myTrade.info.marketName} for user: ${myTrade.userId}")

    val currencyUnits = Util.roundDownPrecision4(myTrade.baseQuantity / price)
    val updatedTrade = myTrade.copy(
      stat = TradeStat(
        currencyQuantity = Some(currencyUnits.toDouble),
        boughtCondition = Some(condition),
        boughtPrice = Some(price),
        boughtTime = Some(Instant.now().atOffset(ZoneOffset.UTC))),
      status = TradeStatus.Bought
    )

    bagel.updateTrade(updatedTrade).map {
      case Some(updated) =>
        myTrade = updated
        balanceBuy(currencyUnits, price)
        loadConditions()
      case None =>
    }
  }

  private def loadConditions() = {
    def parseConditions(conditions: String, actorBuffer: ListBuffer[ActorRef]) = {
      conditions.split(" or ").foreach { cond =>
        val extractParams = """^.*?TrailingStop\((0\.\d{2,}),\s(\d+\.\d+).*?""".r
        cond match {
          case extractParams(percent, refPrice) =>
            val trail = TrailingStop(trade.userId, trade.id, trade.info.marketName, percent.toDouble, refPrice.toDouble)
            actorBuffer += context.actorOf(TrailingStopLossActor.props(TradeAction.Sell, trail))
          case sellConditions =>
            actorBuffer += context.actorOf(SimpleConditionActor.props(TradeAction.Sell, sellConditions))
        }
      }
    }

    // this actor can start a trade from any status
    myTrade.status match {
      case TradeStatus.Pending =>
        // pending trade must monitor buy conditions
        buyConditions += context.actorOf(SimpleConditionActor.props(TradeAction.Buy, myTrade.buyCondition))

      case TradeStatus.Bought =>
        // bought trade must monitor sell conditions
        val stopLoss = myTrade.stopLossCondition.getOrElse("")
        if (stopLoss != "") {
          parseConditions(stopLoss, lossConditions)
        }

        val stopProfit = myTrade.profitCondition.getOrElse("")
        if (stopProfit != "") {
          parseConditions(stopProfit, profitConditions)
        }
      case _ =>
        log.warning(s"encountered a trade status of ${myTrade.status}")
    }
  }

  private def executeSell(price: Double, condition: String) = {
    // TODO execute sell logic here

    log.info(s"sell ${myTrade.info.marketName} condition: $condition for user: ${myTrade.userId}")
    val updatedTrade = myTrade.copy(
      stat = myTrade.stat.copy(
        soldCondition = Some(condition),
        soldPrice = Some(price),
        soldTime = Some(Instant.now().atOffset(ZoneOffset.UTC))),
      status = TradeStatus.Sold
    )

    bagel.updateTrade(updatedTrade).map {
      case Some(updated) =>
        myTrade = updated

        updated.stat.currencyQuantity match {
          case Some(qty) =>
            balanceSell(qty, price)
          case None => ???
        }

        // this trade is finished kill this actor
        self ! PoisonPill
      case None =>
    }
  }

  private def cancelTrade(sender: ActorRef) = {
    // first stop all children condition
    // stop previous conditions
    lossConditions.foreach ( a => context stop a)
    profitConditions.foreach ( a => context stop a)
    buyConditions.foreach (a => context stop a)

    myTrade.status match {
      case TradeStatus.Sold =>
        // cannot cancel a sold trade because it is already finished.
        sender ! None

      case TradeStatus.Pending =>
        // delete the trade from the system if pending
        for {
          deletedOpt <- bagel.deleteTrade(trade)
          balanceOpt <- bagel.findBalance(trade.userId, trade.apiKeyId, trade.info.baseCurrency)
        } yield {

          (deletedOpt, balanceOpt) match {
            case (Some(deleted), Some(balance)) =>
              bagel
                .updateBalance(balance.copy(availableBalance = balance.availableBalance + deleted.baseQuantity))
                .map { x =>
                  sender ! Some(deleted)
                  self ! PoisonPill
                }

            case _ =>
              sender ! None
          }
        }

      case _ =>
        // all other trades statuses are cancellable
        bagel.updateTrade(trade.copy(status = TradeStatus.Cancelled)).map (sender ! _)
        self ! PoisonPill
    }
  }

  private def balanceBuy(purchasedQty: BigDecimal, atPrice: Double): Unit = {
    // ##Simulated case
    // #1 if buy the estimated qty will be  val currencyUnits = myTrade.baseQuantity / lastPrice
    val cost = atPrice * purchasedQty.toDouble

    // subtrade cost from the baseQuantity
    val remainingBase = myTrade.baseQuantity - cost<commit_msg>Use the qty in the myTrade reference when cancelling a trade to restore the available balance<commit_after>package com.flowy.bexchange.trade

import akka.actor.{Actor, ActorLogging, ActorRef, PoisonPill, Props}
import com.flowy.common.database.TheEverythingBagelDao
import com.flowy.common.models.MarketStructures.MarketUpdate
import com.flowy.common.models._
import java.time.{Instant, ZoneOffset}
import java.util.UUID

import com.flowy.bexchange.trade.TrailingStopLossActor.TrailingStop
import com.flowy.common.Util

import scala.collection.mutable.ListBuffer
import scala.concurrent.{ExecutionContext, Future}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.tools.reflect.ToolBox


object TradeActor {
  def props(trade: Trade, bagel: TheEverythingBagelDao)(implicit context: ExecutionContext) =
    Props(new TradeActor(trade, bagel))

  case class Trigger(action: TradeAction.Value, price: Double, condition: String)
  case class Buy(price: Double, atCondition: String)
  case class Sell(price: Double, atCondition: String)
  case class Cancel(sender: ActorRef)
  case class Update(user: UserData, request: TradeRequest, sender: ActorRef)
}


/**
  * Services a single market: example BTC-NEO
  * @param trade
  * @param bagel
  */
class TradeActor(trade: Trade, bagel: TheEverythingBagelDao) extends Actor
  with ActorLogging {

  import TradeActor._

  import scala.reflect.runtime.currentMirror

  implicit val akkaSystem = context.system

  private var myTrade = trade
  private val dynamic = currentMirror.mkToolBox()

  private val buyConditions = scala.collection.mutable.ListBuffer[ActorRef]()
  private val lossConditions = scala.collection.mutable.ListBuffer[ActorRef]()
  private val profitConditions = scala.collection.mutable.ListBuffer[ActorRef]()

  override def preStart() = {
    loadConditions()
  }

  override def postStop() = {
    log.info(s"trade ${trade.id} is shutting down")
  }

  def receive: Receive = {
    case update: MarketUpdate =>
      context.system.actorSelection(s"${self.path}/*") ! update

    case Update(user, request, sender) =>
      updateTrade(user, request, sender)

    case Cancel(sender) =>
      cancelTrade(sender)

    case Trigger(TradeAction.Buy, price, condition) =>
      executeBuy(price, condition)

    case Trigger(TradeAction.Sell, price, condition) =>
      executeSell(price, condition)

    case x =>
      log.warning(s"received unknown message - $x")
  }


  private def executeBuy(price: Double, condition: String) = {
    log.info(s"buy ${myTrade.baseQuantity} ${myTrade.info.marketName} for user: ${myTrade.userId}")

    val currencyUnits = Util.roundDownPrecision4(myTrade.baseQuantity / price)
    val updatedTrade = myTrade.copy(
      stat = TradeStat(
        currencyQuantity = Some(currencyUnits.toDouble),
        boughtCondition = Some(condition),
        boughtPrice = Some(price),
        boughtTime = Some(Instant.now().atOffset(ZoneOffset.UTC))),
      status = TradeStatus.Bought
    )

    bagel.updateTrade(updatedTrade).map {
      case Some(updated) =>
        myTrade = updated
        balanceBuy(currencyUnits, price)
        loadConditions()
      case None =>
    }
  }

  private def loadConditions() = {
    def parseConditions(conditions: String, actorBuffer: ListBuffer[ActorRef]) = {
      conditions.split(" or ").foreach { cond =>
        val extractParams = """^.*?TrailingStop\((0\.\d{2,}),\s(\d+\.\d+).*?""".r
        cond match {
          case extractParams(percent, refPrice) =>
            val trail = TrailingStop(trade.userId, trade.id, trade.info.marketName, percent.toDouble, refPrice.toDouble)
            actorBuffer += context.actorOf(TrailingStopLossActor.props(TradeAction.Sell, trail))
          case sellConditions =>
            actorBuffer += context.actorOf(SimpleConditionActor.props(TradeAction.Sell, sellConditions))
        }
      }
    }

    // this actor can start a trade from any status
    myTrade.status match {
      case TradeStatus.Pending =>
        // pending trade must monitor buy conditions
        buyConditions += context.actorOf(SimpleConditionActor.props(TradeAction.Buy, myTrade.buyCondition))

      case TradeStatus.Bought =>
        // bought trade must monitor sell conditions
        val stopLoss = myTrade.stopLossCondition.getOrElse("")
        if (stopLoss != "") {
          parseConditions(stopLoss, lossConditions)
        }

        val stopProfit = myTrade.profitCondition.getOrElse("")
        if (stopProfit != "") {
          parseConditions(stopProfit, profitConditions)
        }
      case _ =>
        log.warning(s"encountered a trade status of ${myTrade.status}")
    }
  }

  private def executeSell(price: Double, condition: String) = {
    // TODO execute sell logic here

    log.info(s"sell ${myTrade.info.marketName} condition: $condition for user: ${myTrade.userId}")
    val updatedTrade = myTrade.copy(
      stat = myTrade.stat.copy(
        soldCondition = Some(condition),
        soldPrice = Some(price),
        soldTime = Some(Instant.now().atOffset(ZoneOffset.UTC))),
      status = TradeStatus.Sold
    )

    bagel.updateTrade(updatedTrade).map {
      case Some(updated) =>
        myTrade = updated

        updated.stat.currencyQuantity match {
          case Some(qty) =>
            balanceSell(qty, price)
          case None => ???
        }

        // this trade is finished kill this actor
        self ! PoisonPill
      case None =>
    }
  }

  private def cancelTrade(sender: ActorRef) = {
    // first stop all children condition
    // stop previous conditions
    lossConditions.foreach ( a => context stop a)
    profitConditions.foreach ( a => context stop a)
    buyConditions.foreach (a => context stop a)

    myTrade.status match {
      case TradeStatus.Sold =>
        // cannot cancel a sold trade because it is already finished.
        sender ! None

      case TradeStatus.Pending =>
        // delete the trade from the system if pending
        for {
          deletedOpt <- bagel.deleteTrade(myTrade)
          balanceOpt <- bagel.findBalance(myTrade.userId, myTrade.apiKeyId, myTrade.info.baseCurrency)
        } yield {

          (deletedOpt, balanceOpt) match {
            case (Some(deleted), Some(balance)) =>
              bagel
                .updateBalance(balance.copy(availableBalance = balance.availableBalance + deleted.baseQuantity))
                .map { x =>
                  sender ! Some(deleted)
                  self ! PoisonPill
                }

            case _ =>
              sender ! None
          }
        }

      case _ =>
        // all other trades statuses are cancellable
        bagel.updateTrade(trade.copy(status = TradeStatus.Cancelled)).map (sender ! _)
        self ! PoisonPill
    }
  }

  private def balanceBuy(purchasedQty: BigDecimal, atPrice: Double): Unit = {
    // ##Simulated case
    // #1 if buy the estimated qty will be  val currencyUnits = myTrade.baseQuantity / lastPrice
    val cost = atPrice * purchasedQty.toDouble

    // subtrade cost from the baseQuantity
    val remainingBase = myTrade.baseQuantity - cost
{'fraction_non_alphanumeric': 0.10261115031757234, 'fraction_numerical': 0.0005645730416372619, 'mean_word_length': 2.890993959362987, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3525083', 'n_tokens_mistral': 4217, 'n_tokens_neox': 4082, 'n_words': 974}