parSapply并行转换报错求助：AWS EC2上R函数并行化问题

首先咱们拆解你遇到的两个核心问题：并行代码的报错根源，以及原函数逻辑与预期不符的问题，一步步来解决。

一、并行代码报错的直接原因 #

你碰到的can only subtract from "POSIXt" objects错误，本质是并行集群的子节点没拿到必要的数据和变量，导致代码误操作了日期列（POSIXt类型）。具体问题点：

1. 只导出了myfun函数，但没传递EURUSD数据集、PipSize这些关键变量，子节点找不到这些对象，只能误碰日期列引发类型错误。
2. 函数里用了<<-全局赋值，并行环境中全局变量的作用域极其混乱，绝对要避免这种写法，应该直接返回结果，不要修改全局变量。

先修正并行代码的基础问题，同时改掉全局赋值：

library(parallel)

# 修正后的基础函数（先保留原逻辑，后续再调整）
PipSize <- 0.00886
myfun <- function(x, df, Limit, StopLoss) {
  highComp <- which(df$High - df$Open[x] > Limit)
  highCompMin <- if(length(highComp) == 0) 0 else min(highComp)
  lowComp <- which(df$Open[x] - df$Low > StopLoss)
  lowCompMin <- if(length(lowComp) == 0) 0 else min(lowComp)
  
  if(highCompMin == 0 & lowCompMin == 0) {
    result <- c(Limit = NA, Open = df$Open[x])
  } else if (highCompMin <= lowCompMin) {
    result <- c(Limit = 1, Open = df$Open[x])
  } else {
    result <- c(Limit= 0, Open = df$Open[x])
  }
  return(result)
}

# 并行执行的正确写法
n.cores <- detectCores()
cl <- makeCluster(n.cores, type="FORK")
# 导出所有需要的变量：数据集、止盈止损参数
clusterExport(cl, c("EURUSD", "PipSize"))
# 用parSapply传递参数
result <- parSapply(cl, 1:10, function(x) myfun(x, df=EURUSD, Limit=PipSize, StopLoss=PipSize))
stopCluster(cl)

# 转置结果，和原sapply输出格式一致
t(result)

这样修改后报错会消失，但这只是解决了并行语法问题，你的函数逻辑还和预期不符，接下来要修正核心逻辑。

二、修正函数逻辑以匹配预期结果 #

你的预期是：

若(High-Open > limit)则返回1，若(Open - Low > StopLoss)则返回0；若两者都不满足，则将同一开盘价与下一期的最高价和最低价比较；当返回1或0时，将开盘价索引加1并重复该过程。

但原函数存在两个关键问题：

• 它会检查所有行的High/Low，包括当前x之前的行，不符合时间顺序（应该只检查x之后的下一期及以后数据）
• 原函数是对每个x独立处理，而非迭代式地从x开始往后检查直到触发条件，再跳转到下一个起始点

1. 先实现正确的顺序逻辑（非并行） #

因为你的需求是顺序依赖的（下一个起始点依赖上一个触发的位置），这种场景下直接并行所有x不成立，先写出正确的顺序执行代码：

PipSize <- 0.00886
run_strategy <- function(df, Limit = PipSize, StopLoss = PipSize) {
  n <- nrow(df)
  results <- data.frame(Limit = logical(), Open = numeric(), stringsAsFactors = FALSE)
  i <- 1
  while(i <= n) {
    current_open <- df$Open[i]
    # 只检查当前行之后的所有行（符合时间顺序的下一期数据）
    check_rows <- (i+1):n
    if(length(check_rows) == 0) {
      # 没有后续行，返回NA
      results <- rbind(results, data.frame(Limit = NA, Open = current_open))
      break
    }
    # 找第一个满足止盈的行
    high_hit <- which(df$High[check_rows] - current_open > Limit)
    high_min <- if(length(high_hit) == 0) Inf else check_rows[min(high_hit)]
    # 找第一个满足止损的行
    low_hit <- which(current_open - df$Low[check_rows] > StopLoss)
    low_min <- if(length(low_hit) == 0) Inf else check_rows[min(low_hit)]
    
    if(is.infinite(high_min) & is.infinite(low_min)) {
      # 都不满足，返回NA，然后检查下一个开盘价
      results <- rbind(results, data.frame(Limit = NA, Open = current_open))
      i <- i + 1
    } else if(high_min <= low_min) {
      # 先触发止盈，跳转到触发行的下一行
      results <- rbind(results, data.frame(Limit = 1, Open = current_open))
      i <- high_min + 1
    } else {
      # 先触发止损，跳转到触发行的下一行
      results <- rbind(results, data.frame(Limit = 0, Open = current_open))
      i <- low_min + 1
    }
  }
  return(results)
}

# 测试执行
run_strategy(EURUSD)

2. 并行优化方案 #

因为核心逻辑是顺序依赖的，没法直接并行整个迭代过程，但如果数据集非常大，可以把数据拆分成多个独立块（比如按日期拆分，每个块内部顺序执行，块之间独立），再并行处理每个块，最后合并结果：

library(parallel)

# 拆分数据为多个独立块（示例按每1000行一个块）
split_df <- split(EURUSD, ceiling(seq(nrow(EURUSD))/1000))

n.cores <- detectCores()
cl <- makeCluster(n.cores, type="FORK")
clusterExport(cl, c("PipSize", "run_strategy"))

# 并行处理每个块
parallel_results <- parLapply(cl, split_df, run_strategy)
stopCluster(cl)

# 合并所有块的结果
final_results <- do.call(rbind, parallel_results)

这种方式既利用了多核资源，又保留了你的顺序迭代逻辑，适合大数据量场景。

三、关于Intel MKL的说明 #

你安装了Intel MKL但没用到多核，是因为MKL主要优化线性代数运算（比如矩阵乘法、线性回归等），而你的函数以循环和条件判断为主，MKL无法直接优化这类逻辑，必须用parallel包这类显式并行工具来实现多核利用。

内容的提问来源于stack exchange，提问作者BlueTurtle