如何用mclapply [parallelized]替换Quantstrat 'for loop‘？

Question

我想将quantstrat并行化。我的代码并不完全像这样，但这说明了这个问题。我认为问题在于.blotter环境被初始化为指针内存地址，并且我无法初始化new.env()的数组/矩阵。

我想做的是用mclapply替换for循环，这样我就可以用不同的日期/符号运行多个applyStrategies (这里只显示不同的符号)。我的最终目标是一个beowulf集群(makeCluster)，并计划使用多达252个交易日(滚动窗口)并行运行这些集群，每次迭代使用不同的符号(但我不需要所有这些)。我只是在问，是否有一种方法可以使用mclapply)来分配portfolio和后续的.blotter内存对象

#Load quantstrat in your R environment.

rm(list = ls())

local()

library(quantstrat) 
library(parallel)

# The search command lists all attached packages.
search()

symbolstring1 <- c('QQQ','GOOG')
#symbolstring <- c('QQQ','GOOG')

#for(i in 1:length(symbolstring1))
  mlapply(symbolstring1, function(symbolstring)
{
  #local()
  #i=2
  #symbolstring=as.character(symbolstring1[i])
  
  .blotter <- new.env()
  .strategy <- new.env()
  
  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)

# The initDate should be lower than the startDate. The initDate will be used later while initializing the strategy.

initDate <- '2010-01-01'

startDate <- '2011-01-01'

endDate <- '2019-08-10'

init_equity <- 50000

# Set UTC TIME

Sys.setenv(TZ="UTC")

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

# Define names for portfolio, account and strategy. 

#portfolioName <- accountName <- strategyName <- "FirstPortfolio"
portfolioName <- accountName <- strategyName <- paste0("FirstPortfolio",symbolstring)

print(portfolioName)
# The function rm.strat removes any strategy, portfolio, account, or order book object with the given name. This is important

#rm.strat(strategyName)

print("port")
initPortf(name = portfolioName,
          symbols = symbolstring,
          initDate = initDate)

initAcct(name = accountName,
         portfolios = portfolioName,
         initDate = initDate,
         initEq = init_equity)

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")
add.indicator(strategy = strategyName, 
              name = "EMA", 
              arguments = list(x = quote(Cl(mktdata)), 
                               n = 10), label = "nFast")

add.indicator(strategy = strategyName, 
              name = "EMA", 
              arguments = list(x = quote(Cl(mktdata)), 
                               n = 30), 
              label = "nSlow")

# Add long signal when the fast EMA crosses over slow EMA.

print("sig")
add.signal(strategy = strategyName,
           name="sigCrossover",
           arguments = list(columns = c("nFast", "nSlow"),
                            relationship = "gte"),
           label = "longSignal")

# Add short signal when the fast EMA goes below slow EMA.

add.signal(strategy = strategyName, 
           name = "sigCrossover",
           arguments = list(columns = c("nFast", "nSlow"),
                            relationship = "lt"),
           label = "shortSignal")

# go long when 10-period EMA (nFast) >= 30-period EMA (nSlow)

print("rul")
add.rule(strategyName,
         name= "ruleSignal",
         arguments=list(sigcol="longSignal",
                        sigval=TRUE,
                        orderqty=100,
                        ordertype="market",
                        orderside="long",
                        replace = TRUE, 
                        TxnFees = -10),
         type="enter",
         label="EnterLong") 

# go short when 10-period EMA (nFast) < 30-period EMA (nSlow)

add.rule(strategyName, 
         name = "ruleSignal", 
         arguments = list(sigcol = "shortSignal", 
                          sigval = TRUE, 
                          orderside = "short", 
                          ordertype = "market", 
                          orderqty = -100, 
                          TxnFees = -10,                     
                          replace = TRUE), 
         type = "enter", 
         label = "EnterShort")

# Close long positions when the shortSignal column is True

add.rule(strategyName, 
         name = "ruleSignal", 
         arguments = list(sigcol = "shortSignal", 
                          sigval = TRUE, 
                          orderside = "long", 
                          ordertype = "market", 
                          orderqty = "all", 
                          TxnFees = -10, 
                          replace = TRUE), 
         type = "exit", 
         label = "ExitLong")

# Close Short positions when the longSignal column is True

add.rule(strategyName, 
         name = "ruleSignal", 
         arguments = list(sigcol = "longSignal", 
                          sigval = TRUE, 
                          orderside = "short", 
                          ordertype = "market", 
                          orderqty = "all", 
                          TxnFees = -10, 
                          replace = TRUE), 
         type = "exit", 
         label = "ExitShort")

print("summary")
summary(getStrategy(strategyName))

# Summary results are produced below

print("results")
results <- applyStrategy(strategy= strategyName, portfolios = portfolioName,symbols=symbolstring)

# 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)

dateRange <- time(getPortfolio(portfolioName)$summary)[-1]

updateAcct(portfolioName,dateRange)

updateEndEq(accountName)

print(plot(tail(getAccount(portfolioName)$summary$End.Eq,-1), main = "Portfolio Equity"))

#cleanup
for (name in symbolstring) rm(list = name)
#rm(.blotter)
rm(.stoploss)
rm(.txnfees)
#rm(.strategy)
rm(symbols)

}
)

但在get(symbol，envir = envir)中抛出错误:找不到对象'QQQ‘

具体地说，问题是FinancialInstrument:::.instrument指向的内存地址没有用我的封装变量调用(symbolstring)更新

Answer 1

quantstrat中的apply.paramset已经使用foreach构造来并行化applyStrategy的执行。

apply.paramset需要做大量的工作，以确保workers中的环境可用于执行工作，并收集适当的结果以将它们发送回调用进程。

您要做的最简单的事情可能是使用apply.paramset。设置日期和符号参数，并使函数正常运行。

或者，我建议您查看在apply.paramset中使用并行foreach构造所需的步骤，以将其修改为您建议的情况。

还要注意，您的问题是关于使用Beowulf集群和mclapply的。这行不通的。mclapply只能在单个内存空间中工作。Beowulf集群通常不会共享单个内存和进程空间。它们通常通过MPI等并行库分发作业。通过使用foreach的doMPI后端，apply.paramset已经可以在Beowulf集群上分发。这就是我们使用foreach的原因之一:可用的大量不同的并行后端。foreach的doMC后端实际上在幕后使用mclapply。

Answer 2

我相信这会使代码并行化。我已经交换了指示符和符号，但是使用不同符号和日期的逻辑在这里

基本上，我添加了

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()
  
}
)

它看起来是并行的，但它不能正确更新init_equity