ROE、经营现金流质量与资产负债率选股策略

ROE 用于衡量公司利用净资产创造利润的能力。本策略要求 ROE 为正，并在通过全部基础条件的股票中按 ROE 降序排序，保留前 30%。经营现金流与利润比值 ocf_to_profit 用于刻画现金流质量，策略要求该值为正。资产负债率 debt_to_assets 用于控制杠杆风险，策略要求其严格低于 60。

仅当一只股票的三个字段均为有效数值时，该股票才可参与筛选。脚本随后使用复权收盘价 adjClose 作为第二次排序字段，在 ROE 前30%的股票中优先选择价格较低的最多 10 只。这里的价格排序用于控制最终候选数量，不应解释为完整的估值模型或预期收益率。

本例交易的标的基于随机生成的模拟行情与分红数据。脚本会自动从数据中提取股票代码生成股票池，并据此构造 000300.SH 的模拟指数成分股快照。普通股票买入按 100 股整数倍处理。

策略采用普通股票现金账户，只建立多头持仓，不进行卖空、融资或借券。行情频率为日频，订单在下一交易日按开盘价撮合。脚本还预留了可转债、科创板、创业板、北交所和 ETF/LOF 的申报单位处理，但本例的模拟标的均按普通深市股票规则交易。

接入真实数据时，候选股票应来自具有历史时点信息的股票池。除日线和财务因子外，还需要准备当时有效的指数成分、ST 状态、上市日期、停复牌、复权和分红信息，不能使用当前成分股或事后财报数据替代过去各交易日的真实可用信息。

1. 在调仓信号日，从模拟沪深 300 成分股中剔除 ST、上市不足 90 天和成交量不大于 0 的股票。

2. 排除 ROE、经营现金流与利润比值或资产负债率缺失的股票，并应用 roe > 0、ocf_to_profit > 0 和 debt_to_assets < 60 三个严格条件。

3. 对合格股票按 ROE 降序排序，保留前 30%；样本数不足以得到 1 只时，最少保留 1 只。

4. 对 ROE 候选按复权收盘价升序排序，最多选择 10 只。实际入选 N 只时，每只目标权重为 1 / N。

5. 根据实际入选数量等权分配信号日账户权益，并使用信号日收盘价计算目标股数。目标数量向下取整到证券可申报单位，计算时预留买入手续费。

6. 下一交易日按开盘价调仓：先卖出未入选或超过目标数量的持仓，再买入低于目标数量的股票。

本例从第一个有效日线回调开始生成信号，之后每 20 个有效回调日重新选股。若候选股票存在但三项财务字段全部无效，策略将本轮信号视为数据链路无效并保留原持仓；若数据有效但没有股票通过筛选，策略生成空目标，并在下一交易日尝试卖出可卖持仓。

本例准备了 2024 年全年 300 只模拟股票的日线行情数据，并同时包含三个日频随机因子。数据基于交易日历生成，全年共计 242 个交易日。此外，为了测试回测引擎对公司行为的自动处理逻辑，数据中还额外构造了模拟的现金分红记录。

本例直接通过 loadText 函数从本地读取已准备好的 CSV 行情与因子文件，并从中提取不重复的股票代码生成回测所需的股票池（stockPool）

path = "/.../"
data = loadText(path + "piotroskiFScore.csv", ',')
stockPool = exec distinct symbol from data

行情表表已预先整合了 OHLC、成交量、成交额、涨跌停价格、复权收盘价（adjClose）及三项财务质量字段（roe、ocf_to_profit、debt_to_assets）。以此为基础，构建模拟指数成分股、上市日期等辅助快照表，供后续的除权除息处理与回测引擎使用。

策略由五个交易辅助函数和两个回调函数组成。

• roundLot 负责买入数量的申报单位处理；

• normalizeSellQty 负责卖出数量的规范化；

• isBuyableAt、isSellableAt 与 limitOrderPrice 处理停牌和涨跌停约束；

• initialize 初始化回测状态；onBar 执行调仓并产生新信号。

在 onBar 中，msg 是以股票代码为键、以当日日线和因子数据为值的字典。每次回调先执行此前信号日已经确定的目标，再更新计数器并判断是否生成新信号。这一顺序确保当日收盘价和因子不会用于同日开盘成交。

数量与价格辅助函数：

def isBuyableAt(price, bar) {
    return bar.volume > 0 && (isNull(bar.upLimitPrice) || price < bar.upLimitPrice)
}

def isSellableAt(price, bar) {
    return bar.volume > 0 && (isNull(bar.downLimitPrice) || price > bar.downLimitPrice)
}

def limitOrderPrice(price, bar) {
    lowerBound_ = iif(isNull(bar.downLimitPrice), price, bar.downLimitPrice)
    upperBound_ = iif(isNull(bar.upLimitPrice), price, bar.upLimitPrice)
    return min(max(price, lowerBound_), upperBound_)
}

roundLot 对普通股票按 100 股整数倍买入，并为其他证券类型保留相应申报规则。normalizeSellQty 在部分卖出时按可申报单位取整，但整仓退出时允许卖出不足一个申报单位的剩余持仓。

初始化函数：initialize

initialize(mutable context) 在创建引擎时调用一次。设置手续费、印花税、调仓计数、待执行目标和股票池对象。dayCount = -1 与 initialRebalancePending = true 使第一根日线直接产生首次信号。

def initialize(mutable context) {
    context["commission"] = 0.00015
    context["tax"] = 0.0005
    context["rebalanceThreshold"] = 0.0
    context["dayCount"] = -1
    context["initialRebalancePending"] = true
    context["selectedStocks"] = []$STRING
    context["pendingRebalance"] = false
    context["rebalanceDate"] = NULL
    context["pendingTargetQtys"] = dict(STRING, LONG)
    context["pendingSells"] = dict(STRING, INT)
    context["pendingSellsLabel"] = dict(STRING, STRING)
    context["pendingSellsDate"] = NULL
    // 基础信息表：股票池、指数成分快照、ST 历史、上市信息
    context["stockPool"] = objByName("stockPool")
    context["tpExplicit"] = objByName("tpExplicit_")
    context["tpIdxSnaps"] = objByName("tpIdxSnaps_")
    context["tpStHist"] = objByName("tpStHist_")
    context["tpListInfo"] = objByName("tpListInfo_")
    context["tradablePool"] = set(array(STRING, 0))
}

回调函数：onBar

onBar(mutable context, msg, indicator) 是日线行情回调，也是策略的主执行函数，其内部流程分为三部分：

1. 跳过合成除息日：检查当日是否为合成的除权除息日（synthetic_dividend_bar），若是则直接跳过不作处理。

2. 延迟调仓执行（次日开盘）：如果存在更早交易日生成的待执行目标，先检查可卖数量和开盘交易条件，卖出不在选中列表的持仓以及超配的持仓（偏差超出阈值才执行）；再检查可用现金，按目标数量买入低配股票。执行过程中统一使用当日开盘价（open）提单，并在执行完毕后清除待调仓标记。此外，函数还会对称处理独立的待卖出指令（pendingSells）。

3. 横截面选股与信号生成：调仓计数每达到 20 天（或遇到初始建仓日）时，触发重新选股。首先过滤出上市超 90 天且非 ST 的指数成分股作为基础池；接着要求 roe > 0、ocf_to_profit > 0 且 debt_to_assets < 60；随后按 roe 降序取前 30%，再按复权收盘价升序选出最多 10 只股票等权配置。最后，按信号日账户权益和收盘价计算目标股数，并保存至下一交易日执行。

策略逻辑如下：

def onBar(mutable context, msg, indicator) {
    curBarTs = msg.values()[0]["tradeTime"]
    curDate = date(curBarTs)
    syntheticDivDiag_ = objByName("piotroskiDiagnostics")
    if (size(syntheticDivDiag_) > 0) {
        syntheticDivDates_ = exec date(time) from syntheticDivDiag_
            where diagType == "synthetic_dividend_bar"
        if (curDate in syntheticDivDates_) { return }
    }

    // === 延迟调仓执行（次日开盘：信号日锁定目标下单数量，执行日 open 撮合 + cash 兜底）===
    if (context["pendingRebalance"]) {
        if ((curDate != context["rebalanceDate"])) {
            tgtQtys_ = context["pendingTargetQtys"]   // 读信号日锁定的 target_qty
            expectedCash_ = Backtest::getAvailableCash(context.engine)
            sellCashMult_ = 1.0 - context["commission"] - context["tax"]
            rebalThr_ = context["rebalanceThreshold"]
            // 第一轮：卖出不在选中列表的（清仓全卖）+ 超配的减仓（偏差超阈值才动）
            // T+1 资产 sellable = 昨日剩余持仓 - 当日已卖；T+0 = 当前总持仓
            for (istock in msg.keys()) {
                bar = msg[istock]
                posInfo = Backtest::getPosition(context.engine, istock)
                pos = 0
                pos_sellable = 0
                if (size(posInfo) > 0) {
                    pos = posInfo["longPosition"]
                    pos_sellable = max(0, posInfo["lastDayLongPosition"] - posInfo["todaySellVolume"])
                }
                if (!(pos_sellable > 0 && isSellableAt(bar["open"], bar))) { continue }
                sellLimit_ = limitOrderPrice(bar["open"], bar)
                if (!(istock in context["selectedStocks"])) {
                    // 不在最新入选名单中：整仓清仓
                    Backtest::submitOrder(context.engine, (istock, context.tradeTime, 5, sellLimit_, pos_sellable, 3), "rebalance")
                    expectedCash_ = expectedCash_ + pos_sellable * bar["open"] * sellCashMult_
                } else {
                    // 在名单中但持仓超过目标数：偏差超阈值才减仓
                    targetQty = iif(istock in tgtQtys_, tgtQtys_[istock], 0)
                    if (pos > targetQty && double(pos - targetQty) / double(pos) > rebalThr_) {
                        sellQty_ = normalizeSellQty(istock, pos, min(pos_sellable, pos - targetQty))
                        if (sellQty_ > 0) {
                            Backtest::submitOrder(context.engine, (istock, context.tradeTime, 5, sellLimit_, sellQty_, 3), "rebalance")
                            expectedCash_ = expectedCash_ + sellQty_ * bar["open"] * sellCashMult_
                        }
                    }
                }
            }
            // 第二轮：按排名顺序加仓+新买入（新建仓全额买入；加仓偏差超阈值才动）
            for (istock in context["selectedStocks"]) {
                if (!(istock in msg)) { continue }
                bar = msg[istock]
                if (!isBuyableAt(bar["open"], bar)) { continue }
                posInfo = Backtest::getPosition(context.engine, istock)
                pos = 0
                if (size(posInfo) > 0) { pos = posInfo["longPosition"] }
                targetQty = iif(istock in tgtQtys_, tgtQtys_[istock], 0)
                if (targetQty > pos) {
                    plannedQty = targetQty - pos
                    if (pos > 0 && double(plannedQty) / double(targetQty) <= rebalThr_) { continue }
                    affordableCash_ = min(expectedCash_, Backtest::getAvailableCash(context.engine))
                    affordableQty = roundLot(istock, affordableCash_ / (bar["open"] * (1 + context["commission"])))
                    buyQty = min(plannedQty, affordableQty)
                    buyQty = roundLot(istock, buyQty)
                    if (buyQty > 0) {
                        buyLimit_ = limitOrderPrice(bar["open"], bar)
                        Backtest::submitOrder(context.engine, (istock, context.tradeTime, 5, buyLimit_, buyQty, 1), "buy")
                        expectedCash_ = expectedCash_ - buyQty * bar["open"] * (1 + context["commission"])
                    }
                }
            }
            context["pendingRebalance"] = false
        }
    }
    // === 延迟卖出执行(下根 bar 入口取出意图,用 bar.open 价提单) — 跟 buy 对称(业界 next_open 标准) ===
    if (context["pendingSellsDate"] != NULL && curDate != context["pendingSellsDate"]) {
        for (istock in msg.keys()) {
            if (!(istock in context["pendingSells"])) { continue }
            bar = msg[istock]
            if (bar["volume"] <= 0) { continue }
            sellQty_pend = context["pendingSells"][istock]
            labelPend_ = context["pendingSellsLabel"][istock]
            posInfo_s = Backtest::getPosition(context.engine, istock)
            pos_total_s = 0
            pos_sellable_s = 0
            if (size(posInfo_s) > 0) {
                pos_total_s = posInfo_s["longPosition"]
                pos_sellable_s = max(0, posInfo_s["lastDayLongPosition"] - posInfo_s["todaySellVolume"])
            }
            sellQty_actual = normalizeSellQty(istock, pos_total_s, min(sellQty_pend, pos_sellable_s))
            if (sellQty_actual > 0 && isSellableAt(bar["open"], bar)) {
                // 确保卖单限价同时满足涨停和跌停约束。
                sellLimit_ = limitOrderPrice(bar["open"], bar)
                Backtest::submitOrder(context.engine, (istock, context.tradeTime, 5, sellLimit_, sellQty_actual, 3), labelPend_)
            }
        }
        context["pendingSells"] = dict(STRING, INT)
        context["pendingSellsLabel"] = dict(STRING, STRING)
        context["pendingSellsDate"] = NULL
    }

    // === 选股逻辑（信号日选股，下根 bar 开盘下单）===
    context["dayCount"] = context["dayCount"] + 1
    isRebalanceDay = (context["dayCount"] % 20 == 0)
    initialRebalance_ = context["initialRebalancePending"]
    isRebalanceDay = (isRebalanceDay || initialRebalance_)
    context["initialRebalancePending"] = false

    if (isRebalanceDay) {
        // 取当日最新一期指数成分股快照
        snapsT_ = context["tpIdxSnaps"]
        latestPerIdx_ = select index_code, max(snap_date) as latest_date from snapsT_ where snap_date <= curDate group by index_code
        idxMembers_ = set(exec con_code from ej(snapsT_, latestPerIdx_, `index_code`snap_date, `index_code`latest_date))
        tpExplicit_ = context["tpExplicit"]
        // 剔除当日处于 ST 状态的股票
        stHist_ = context["tpStHist"]
        stCodes_ = set(exec code from stHist_ where start_date <= curDate and (isNull(period_end) or period_end > curDate))
        // 剔除上市不足 N 天的新股
        listInfo_ = context["tpListInfo"]
        cutoffDate_ = date(curDate - 90)
        eligibleCodes_ = set(exec code from listInfo_ where list_date <= cutoffDate_)
        pool_ = array(STRING, 0)
        for (s_ in context["stockPool"]) { if (((s_ in idxMembers_) || (s_ in tpExplicit_)) && !(s_ in stCodes_) && (s_ in eligibleCodes_)) { pool_.append!(s_) } }
        context["tradablePool"] = set(pool_)
        // --- 横截面选股 ---
        pipelineDataValid_ = true
        selectedStocks = []$STRING
        stockWeights_ = dict(STRING, DOUBLE)
        symbols = []$STRING
        m_roe_list = []$DOUBLE
        m_ocf_to_profit_list = []$DOUBLE
        m_debt_to_assets_list = []$DOUBLE
        m_close_list = []$DOUBLE
        for (istock in msg.keys()) {
            if (!(istock in context["tradablePool"])) { continue }
            bar = msg[istock]
            if (bar["volume"] <= 0) { continue }  // 跳过停牌股

            symbols.append!(istock)
            m_roe_list.append!(double(bar["roe"]))
            m_ocf_to_profit_list.append!(double(bar["ocf_to_profit"]))
            m_debt_to_assets_list.append!(double(bar["debt_to_assets"]))
            m_close_list.append!(double(bar["adjClose"]))
        }

        // 组装并处理pipeline
        poolT = table(symbols as symbol, m_roe_list as `roe, m_ocf_to_profit_list as `ocf_to_profit, m_debt_to_assets_list as `debt_to_assets, m_close_list as `close)
        if (poolT.size() > 0) {
            // 候选有数据但三项因子整列缺失 → 本轮跳过选股
            filterValidN0_ = exec sum(iif(isValid(debt_to_assets) && isValid(ocf_to_profit) && isValid(roe), 1, 0)) from poolT
            if (filterValidN0_ == 0) { pipelineDataValid_ = false }
            // 过滤：三项因子有效，ROE > 0，经营现金流/净利润 > 0，资产负债率 < 60
            poolT = select * from poolT where isValid(roe) && roe > 0 && isValid(ocf_to_profit) && ocf_to_profit > 0 && isValid(debt_to_assets) && debt_to_assets < 60
            // 一级排序：ROE 降序取前 30%（至少保留 1 只）
            limitN = max(1, int(poolT.size() * 0.3))
            poolT = select * from poolT order by roe desc limit limitN
            // 二级排序：在一级结果中按收盘价升序取不超过 10 只
            rankValidT2_ = select * from poolT where isValid(close)
            if (poolT.size() > 0 && rankValidT2_.size() == 0) { pipelineDataValid_ = false }
            limitN = iif(10 < rankValidT2_.size(), 10, rankValidT2_.size())
            poolT = select * from rankValidT2_ order by close asc limit limitN
            selectedStocks = exec distinct symbol from poolT
            if (selectedStocks.size() > 0) {
                weightPerStock_ = 1.0 / selectedStocks.size()
                for (s in selectedStocks) { stockWeights_[s] = weightPerStock_ }
            }
        }
        if (pipelineDataValid_) {
            context["selectedStocks"] = selectedStocks
        }
        // --- 横截面选股结束 ---
        if (pipelineDataValid_) {
        // 信号日：用信号日 close 价锁定目标下单数量
        // target_qty = portfolio_value × pct × weight / (close × (1+commission))
        // 执行日次根 bar open 撮合 + cash 兜底（业务规则：信号日锁数量，避免跳空污染）
        signalEquity_ = Backtest::getTotalPortfolios(context.engine).totalEquity[0]
        current_portfolio_pct_ = double(1.0)


        tgtQtys_ = dict(STRING, LONG)
        for (s_ in context["selectedStocks"]) {
            w_ = iif(s_ in stockWeights_, stockWeights_[s_], 0.0)
            tgtQtys_[s_] = roundLot(s_, signalEquity_ * current_portfolio_pct_ * w_ / (msg[s_]["close"] * (1 + context["commission"])))
        }
        context["pendingTargetQtys"] = tgtQtys_
        context["pendingRebalance"] = true
        context["rebalanceDate"] = curDate
        }
    }
}

入选股票按实际数量等权。目标数量在信号日根据账户总权益、权重、收盘价和手续费锁定，下一交易日再按开盘价尝试成交。

config = {
    startDate: 2024.01.01,
    endDate: 2024.12.31,
    strategyGroup: "stock",
    cash: 1000000.0,
    commission: 0.00015,
    tax: 0.0005,
    dataType: 4,
    outputOrderInfo: true,
    matchingMode: 2,
    orderBookMatchingRatio: 0.2,
    enableMinimumPerTransactionFee: false
}

最大持仓 10 只、ROE 前 30%和 20 日调仓周期直接写在 onBar 中。修改参数时，建议将它们提取为命名变量，并同步记录每轮候选数、入选数和目标数量。

callbacks = dict(STRING, ANY)
callbacks["initialize"] = initialize
callbacks["onBar"] = onBar
engineName = "piotroskiFScoreEngine"
try { Backtest::dropBacktestEngine(engineName) } catch(ex) { }
engine = Backtest::createBacktester(engineName, config, callbacks, false)

创建前后均清理同名引擎，避免重复运行时保留旧持仓。第四个参数为 false，本例不启用 JIT 优化；如需开启，应在目标环境验证回调函数和数据类型兼容性。

将按 tradeTime、symbol 排序的 data 发送给引擎，回调按交易日推进。首个交易日只产生待执行目标，后一个交易日才尝试以开盘价成交。行情发送完毕后，脚本复制一行记录并将 symbol 改为 END，作为回放结束标记。

Backtest::appendQuotationMsg(engine, data)
endRow = select top 1 * from data
update endRow set symbol = "END"
Backtest::appendQuotationMsg(engine, endRow)

行情回放结束后，从回测引擎获取回测的各项核心结果表。返回的数据包括订单的详细成交记录、策略的总体收益概述、每日的个股持仓明细，以及每日的账户总权益变化

tradeDetails = Backtest::getTradeDetails(engine)
returnSummary = Backtest::getReturnSummary(engine)
dailyPosition = Backtest::getDailyPosition(engine)
totalPortfolios = Backtest::getDailyTotalPortfolios(engine)

其中 returnSummary 用于收益风险统计，tradeDetails 用于核对订单和成交，dailyPosition 用于检查每日持仓，totalPortfolios 用于追踪账户权益。