How do I replace Quantstrat 'for loop' with mclapply [parallelized]?

apply.paramset in quantstrat already uses a foreach construct to parallelize execution of applyStrategy.

apply.paramset needs to do a fair amount of work to make sure that the environments are available in the workers to do the work, and to collect the proper results to send them back to the calling process.

The simplest thing for you to do would probably be to use apply.paramset. Make your dates and symbols parameters, and have the function run normally.

Alternately, I suggest you look at the steps required to use a parallel foreach construction in apply.paramset to modify it to your suggested case.

Also note that your question asks about using a Beowulf cluster and mclapply. This won't work. mclapply only works in a single memory space. Beowulf clusters don't normally share a single memory and process space. They typically distribute jobs via parallel libraries such as MPI. apply.paramset could already distribute on a Beowulf cluster by using a doMPI backend to foreach. That is one of the reasons we used foreach: the multitude of different parallel backends that are available. The doMC backend for foreach actually uses mclapply behind the scenes.

I believe this parallelizes the code. I've swapped the indicators out as well as as symbols, but the logic of using different symbols and dates is in there

Basically I added

Dates=paste0(startDate,"::",endDate)

rm(list = ls())

library(lubridate)
library(parallel)

autoregressor1  = function(x){
  if(NROW(x)<12){ result = NA} else{
    y = Vo(x)*Ad(x)
    #y = ROC(Ad(x))
    y = ROC(y)
    y = na.omit(y)
    step1 = ar.yw(y)
    step2 = predict(step1,newdata=y,n.ahead=1)
    step3 = step2$pred[1]+1
    step4 = (step3*last(Ad(x))) - last(Ad(x))
    
    result = step4
  }
  return(result)
}

autoregressor = function(x){
  ans = rollapply(x,26,FUN = autoregressor1,by.column=FALSE)
  return (ans)}

########################indicators#############################

library(quantstrat) 
library(future.apply)
library(scorecard)

reset_quantstrat <- function() {
  if (! exists(".strategy")) .strategy <<- new.env(parent = .GlobalEnv)
  if (! exists(".blotter")) .blotter <<- new.env(parent = .GlobalEnv)
  if (! exists(".audit")) .audit <<- new.env(parent = .GlobalEnv)
  suppressWarnings(rm(list = ls(.strategy), pos = .strategy))
  suppressWarnings(rm(list = ls(.blotter), pos = .blotter))
  suppressWarnings(rm(list = ls(.audit), pos = .audit))
  FinancialInstrument::currency("USD")
}

reset_quantstrat()

initDate <- '2010-01-01'

endDate <- as.Date(Sys.Date())
startDate <- endDate %m-% years(3)

symbolstring1 <- c('SSO','GOLD')

getSymbols(symbolstring1,from=startDate,to=endDate,adjust=TRUE,src='yahoo')

#symbolstring1 <- c('SP500TR','GOOG')

.orderqty <- 1
.txnfees <- 0

#random <- sample(1:2, 2, replace=FALSE)

random <- (1:2)

equity <- lapply(random, function(x)
{#x=1
  try(rm("account.Snazzy","portfolio.Snazzy",pos=.GlobalEnv$.blotter),silent=TRUE)
  rm(.blotter)
  rm(.strategy)
  portfolioName <- accountName <- strategyName <- paste0("FirstPortfolio",x+2)
  #endDate <- as.Date(Sys.Date())
  startDate <- endDate %m-% years(1+x)
 
  #Load quantstrat in your R environment.
  reset_quantstrat()
  
  # The search command lists all attached packages.
  search()

  symbolstring=as.character(symbolstring1[x])
  print(symbolstring)
  
  try(rm.strat(strategyName),silent=TRUE)
  try(rm(envir=FinancialInstrument:::.instrument),silent=TRUE)
  for (name in ls(FinancialInstrument:::.instrument)){rm_instruments(name,keep.currencies = FALSE)}
  print(symbolstring)
  
  currency('USD')
  
  stock(symbolstring,currency='USD',multiplier=1)
  
  # Currency and trading instrument objects stored in the 
  # .instrument environment
  
  print("FI")
  ls(envir=FinancialInstrument:::.instrument)
  
  # blotter functions used for instrument initialization 
  # quantstrat creates a private storage area called .strategy
  
  ls(all=T)
  
  init_equity <- 10000
  
  Sys.setenv(TZ="UTC")
  
  print(portfolioName)
 
  print("port")

  try(initPortf(name = portfolioName,
            symbols = symbolstring,
            initDate = initDate))
  
 
  try(initAcct(name = accountName,
           portfolios = portfolioName,
           initDate = initDate,
           initEq = init_equity))
  
  try(initOrders(portfolio = portfolioName,
             symbols = symbolstring,
             initDate = initDate))
  
  # name: the string name of the strategy
  
  # assets: optional list of assets to apply the strategy to.  
  
  # Normally these are defined in the portfolio object
  
  # contstrains: optional portfolio constraints
  
  # store: can be True or False. If True store the strategy in the environment. Default is False
  print("strat")
  strategy(strategyName, store = TRUE)
  
  ls(all=T)
  
  # .blotter holds the portfolio and account object 
  
  ls(.blotter)
  
  # .strategy holds the orderbook and strategy object
  
  print(ls(.strategy))
  
  print("ind")
  #ARIMA
    
    add.indicator(
      strategy  =   strategyName, 
      name      =   "autoregressor", 
      arguments =   list(
        x       =   quote(mktdata)),
      label     =   "arspread")
    
    ################################################ Signals #############################
    
    add.signal(
      strategy          = strategyName,
      name              = "sigThreshold",
      arguments         = list(
        threshold       = 0.25,
        column          = "arspread",
        relationship    = "gte",
        cross           = TRUE),
      label             = "Selltime")
    
    add.signal(
      strategy          = strategyName,
      name              = "sigThreshold",
      arguments         = list(
        threshold       = 0.1,
        column          = "arspread",
        relationship    = "lt",
        cross           = TRUE),
      label             = "cashtime")
    
    add.signal(
      strategy          = strategyName,
      name              = "sigThreshold",
      arguments         = list(
        threshold       = -0.1,
        column          = "arspread",
        relationship    = "gt",
        cross           = TRUE),
      label             = "cashtime")
    
    add.signal(
      strategy          = strategyName,
      name              = "sigThreshold",
      arguments         = list(
        threshold       = -0.25,
        column          = "arspread",
        relationship    = "lte",
        cross           = TRUE),
      label             = "Buytime")
    
    ######################################## Rules #################################################
    
    #Entry Rule Long
    add.rule(strategyName,
             name               =   "ruleSignal",
             arguments          =   list(
               sigcol           =   "Buytime",
               sigval           =   TRUE,
               orderqty     =   .orderqty,
               ordertype        =   "market",
               orderside        =   "long",
               pricemethod      =   "market",
               replace          =   TRUE,
               TxnFees              =   -.txnfees
               #,
               #osFUN               =   osMaxPos
             ), 
             type               =   "enter",
             path.dep           =   TRUE,
             label              =   "Entry")
    
    #Entry Rule Short
    
    add.rule(strategyName,
             name           =   "ruleSignal",
             arguments          =   list(
               sigcol           =   "Selltime",
               sigval           =   TRUE,
               orderqty     =   .orderqty,
               ordertype        =   "market",
               orderside        =   "short",
               pricemethod      =   "market",
               replace          =   TRUE,
               TxnFees              =   -.txnfees
               #,
               #osFUN               =   osMaxPos
             ), 
             type               =   "enter",
             path.dep           =   TRUE,
             label              =   "Entry")
    
    #Exit Rules
    
  print("summary")
  summary(getStrategy(strategyName))
  
  # Summary results are produced below
  
  print("results")
  
  results <- applyStrategy(strategy= strategyName, portfolios = portfolioName)
  
  # The applyStrategy() outputs all transactions(from the oldest to recent transactions)that the strategy sends. The first few rows of the applyStrategy() output are shown below
  
  getTxns(Portfolio=portfolioName, Symbol=symbolstring)
  
  mktdata
  
  updatePortf(portfolioName,Dates=paste0(startDate,"::",endDate))
  
  dateRange <- time(getPortfolio(portfolioName)$summary)
  
  updateAcct(portfolioName,dateRange[which(dateRange >= startDate & dateRange <= endDate)])
  
  updateEndEq(accountName, Dates=paste0(startDate,"::",endDate))
  
  print(plot(tail(getAccount(portfolioName)$summary$End.Eq,-1), main = symbolstring))
  
  tStats <- tradeStats(Portfolios = portfolioName, use="trades", inclZeroDays=FALSE,Dates=paste0(startDate,"::",endDate))
  
  final_acct <- getAccount(portfolioName)
  
  #final_acct
  #View(final_acct)
  
  options(width=70)
  
  print(plot(tail(final_acct$summary$End.Eq,-1), main = symbolstring))
  #dev.off()
  
  tail(final_acct$summary$End.Eq)
  
  rets <- PortfReturns(Account = accountName)
  
  #rownames(rets) <- NULL
  
  tab.perf <- table.Arbitrary(rets,
                              metrics=c(
                                "Return.cumulative",
                                "Return.annualized",
                                "SharpeRatio.annualized",
                                "CalmarRatio"),
                              metricsNames=c(
                                "Cumulative Return",
                                "Annualized Return",
                                "Annualized Sharpe Ratio",
                                "Calmar Ratio"))
  tab.perf
  
  tab.risk <- table.Arbitrary(rets,
                              metrics=c(
                                "StdDev.annualized",
                                "maxDrawdown"
                              ),
                              metricsNames=c(
                                "Annualized StdDev",
                                "Max DrawDown"))
  
  tab.risk
  
  return (as.numeric(tail(final_acct$summary$End.Eq,1))-init_equity)

  #reset_quantstrat()
  
}
)

it appears to be parallized but it doesn't update init_equity correctly