长期反转选股策略

DeBondt-Thaler 长期反转策略关注过去较长时期表现落后的股票，尝试捕捉后续价格修复。本文在沪深300 历史成分股范围内，通过 Backtest 引擎按约三年的复权累计收益率进行选股、建立等权多头组合、定期调仓和结果分析。

1. 策略介绍

策略将长期收益排名转化为股票选择和目标股数。其核心是买入历史相对弱势股票，而非预测单只股票的合理价格。下面分别介绍因子含义、交易标的和交易规则。

1.1 核心因子

本策略以股票过去 3 年（约 756 个交易日）的累计价格变化率作为核心选股因子。采用收盘价乘复权因子作为价格序列，其计算公式如下：

$R_{i, t} = \frac{C_{i, t} A_{i, t}}{C_{i, t - 756} A_{i, t - 756}} - 1$

其中，C 为未复权收盘价，单位为元/股；A 为无量纲复权因子；R 为价格变化率。每只股票分别按时间排序，3 年的交易日约为 756 天，因此对比 756 个交易日前的价格。计算一个有效值至少需要当前记录及之前第 756 条记录，并要求价格、复权因子和分母有效。

对当前交易日所有满足条件的 M 只股票，按因子值升序排序（买入跌幅最大/反转效应最强的标的），最终选取的标的数量 N 为：

N = min(M, max(1, floor(0.1M)))

本文提供的案例数据中共 300 只股票。若 300 只股票均有当日行情、成交量大于 0 且指标有效，则按 10% 比例（向下取整）买入 30 只；若仅有 285 只有效，则选出 28 只；若无有效候选标的，则不进行选股。

1.2 交易标的

交易资产为中国 A 股普通股票，账户仅做多，行情频率为日频。股票代码沿用数据源的 .SH、.SZ 后缀，股票池来自样例数据中的 000300.SH 的历史成分股。

本策略实现采用单一复权累计收益排名、单一多头组合和约一年调仓周期，未构造赢家空头组合，未计算形成期的市场调整超额收益，未有实现多批重叠组合，亦未在形成期与持有期之间的设置跳过窗口。因此，本文实现仅作工程化演示，不可等同于经典长期反转研究的完整实证复现。

1.3 交易规则

首个有效交易日（Bar 回调）生成首次信号；此后每隔 252 个交易日（即第 1、253、505 个有效交易日）再次生成调仓信号。若收到整日合成分红行情，onBar 回调将直接返回，不计入调仓周期。
剔除因子缺失的标的，按收益率升序选取最低的 10% 的标的，以实际入选数量等权分配目标权益。
使用信号日收盘价及手续费计算并锁定目标股数。在次日（T+1 日）行情回调触发时，按“先卖后买”原则执行——先卖出落选标的及超配持仓，再买入新入选及低配标的。
卖出数量受昨日可用持仓（扣除当日已卖数量）限制；买入金额取预估现金与引擎实际可用现金的较小值，并按整手规则向下取整。
单次调仓流程尝试完毕后即重置调仓标志。若因停牌无行情、涨跌停限制或资金不足导致未完全成交，后续交易日不进行自动补单，亦不进行主动撤单或回报重试。

若候选池不为空但所有标的因子均缺失，则判定本次选股无效，维持原有持仓不变，不触发调仓；首次选股标志在触发后即行清除，不会因因子缺失而保留。若候选池本身为空，则视为清仓信号，在下一有效日尝试清仓。

2. 策略回测

以下按照准备数据、定义策略、设置参数、创建引擎、回放行情和读取结果的顺序展开。本案例已在 DolphinDB Server 3.00.6.1、Backtest 3.00.6.6 与 MatchingEngineSimulator 3.00.6.6 环境下验证通过。运行脚本前，请确保服务器已安装版本匹配的插件，并按顺序先加载 MatchingEngineSimulator 再加载 Backtest（案例脚本本身不包含插件加载与自定义数据安装代码）。插件安装与配置详情请参见 Backtest。

2.1 准备数据

下载本文附件中的 deBondtThaler.csv 模拟数据，该文件包含了 300 只股票的日频行情数据，回测区间为 2025.08.07 至 2026.08.06。使用前请先导入对应区间的 CSV 数据，并将代码中的 <path> 替换为 CSV 文件的实际存储路径。

actualStartDate = 2025.08.07
actualEndDate = 2026.08.06
csvSchema = table(
    ["symbol", "tradeTime", "open", "low", "high", "close", "volume", "amount",
     "upLimitPrice", "downLimitPrice", "adj_factor", "vwap"] as name,
    ["STRING", "TIMESTAMP", "DOUBLE", "DOUBLE", "DOUBLE", "DOUBLE", "LONG", "DOUBLE",
     "DOUBLE", "DOUBLE", "DOUBLE", "DOUBLE"] as type)
data = loadText("<path>/deBondtThaler.csv", ",", csvSchema, containHeader=true)
update data set tradeTime = timestamp(date(tradeTime))

数据字段如下：

表 1. 表 2-1 数据字段
字段	类型	说明
symbol	STRING	模拟代码，例如 000001.SH 或 000300.SH
tradeTime	TIMESTAMP	交易日零点时间戳，仅作为日线标识
open	DOUBLE	模拟开盘价（OHLC 价格）
low	DOUBLE	模拟最低价（OHLC 价格）
high	DOUBLE	模拟最高价（OHLC 价格）
close	DOUBLE	模拟收盘价（OHLC 价格）
volume	LONG	模拟成交量
amount	DOUBLE	模拟成交额
upLimitPrice	DOUBLE	本例均为空值，不模拟涨停
downLimitPrice	DOUBLE	本例均为空值，不模拟跌停
adj_factor	DOUBLE	固定为 1.0，不模拟复权调整
vwap	DOUBLE	`amount / volume × adj_factor`，成交量无效时为空

按照 756 个交易日，计算价格变化因子：

data = select *, double((close * adj_factor) / move(close * adj_factor, 756) - 1) as ret_3y
    from data context by symbol csort tradeTime
data = select * from data where date(tradeTime) between actualStartDate:actualEndDate
    order by tradeTime, symbol
stockPool = exec distinct symbol from data

2.2 编写策略逻辑

策略由 roundLot、normalizeSellQty、isBuyableAt、isSellableAt、initialize 和 onBar 组成。前四个函数处理申报数量和成交条件，initialize 写入策略状态，onBar 执行延迟调仓、市场状态判断和横截面选股。

初始化函数： initialize

initialize(mutable context) 是回测引擎的初始化回调。每次创建新引擎时调用一次，用于建立本次回测需要的可变状态。

context: 可变 DICTIONARY，由回测引擎在回调时传入。策略在其中写入调仓计数、待执行标记、信号日期、入选股票和目标数量。

initialize 中的关键初始状态如下：

def initialize(mutable context) {
    ind = dict(STRING, ANY)
    ind["ret_3y"] = <ret_3y>
    Backtest::subscribeIndicator(context.engine, "ohlc", ind)
    context["commission"] = 0.00015
    context["tax"] = 0.0005
    context["rebalanceThreshold"] = 0.0
    context["dayCount"] = -1
    context["initialRebalancePending"] = true
    context["selectedStocks"] = []$STRING
    context["stockWeights"] = dict(STRING, DOUBLE)
    context["selectionDataValid"] = true
    context["currentPortfolioPct"] = 1.0
    context["pendingRebalance"] = false
    context["rebalanceDate"] = NULL
    context["pendingTargetQtys"] = dict(STRING, LONG)
}

函数没有显式返回值，而是更新 context。dayCount = -1 使第一次日线回调成为第0次计数，从而立即产生首次信号；pendingRebalance = false 表示初始化时没有待执行调仓。

回调函数： onBar

onBar 负责跳过合成分红日、执行待调仓、处理延迟卖出、更新计数并在调仓日生成新信号。代码依赖 context、msg 和 indicator，并通过 Backtest 引擎读取账户、提交订单和更新状态。

待调仓执行使用下一有效日开盘价，先卖出未入选或超配股票，再尝试买入目标股票。
卖出数量受昨日持仓减当日已卖数量约束。
买入数量受目标股数、预计现金、实际可用现金和申报单位共同约束。
调仓尝试后即清除 pendingRebalance，不会自动补单。
选股阶段使用最新指数成分、ST 状态、上市日期、成交量和 ret_3y 有效性过滤。

def onBar(mutable context, msg, indicator) {
    curBarTs = msg.values()[0]["tradeTime"]
    curDate = date(curBarTs)
    // === 延迟调仓执行（次日开盘：信号日锁定目标下单数量，执行日 open 撮合 + cash 兜底）===
    if (context["pendingRebalance"]) {
        if ((curDate != context["rebalanceDate"])) {
            signalEquity_ = context["signalEquity"]
            weights = context["stockWeights"]
            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"])
                }
                isSelected = iif(istock in context["selectedStocks"], true, false)
                if (!isSelected && pos > 0 && pos_sellable > 0 && isSellableAt(bar["open"], bar)) {
                    Backtest::submitOrder(context.engine, (istock, context.tradeTime, 5, bar["open"], pos_sellable, 3), "rebalance")
                    expectedCash_ = expectedCash_ + pos_sellable * bar["open"] * sellCashMult_
                } else if (isSelected && pos > 0 && pos_sellable > 0 && isSellableAt(bar["open"], bar)) {
                    targetQty = iif(istock in tgtQtys_, tgtQtys_[istock], 0)
                    if (pos > targetQty && (pos - targetQty) > 0 && 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, bar["open"], 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 && targetQty > 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) {
                        Backtest::submitOrder(context.engine, (istock, context.tradeTime, 5, bar["open"], 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)) {
                // sell limit 用 min(open, upLim) 兜底:涨停日 bar.open 浮点精度可能微超 prev*1.1=upLim,
                // DDB 引擎按 sell limit > upLim 直接拒(A 股规则:sell 价不能高于涨停价)
                sellLimit_ = iif(isNull(bar["upLimitPrice"]), bar["open"], min(bar["open"], bar["upLimitPrice"]))
                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"] % 252 == 0)
    initialRebalance_ = context["initialRebalancePending"]
    isRebalanceDay = (isRebalanceDay || initialRebalance_)
    context["initialRebalancePending"] = false

    if (isRebalanceDay) {
        // 本例中全部标的参与候选，不另查成分股、ST 或上市资料，实际情况应当进行筛选。
        // --- 横截面选股 ---
        pipelineDataValid_ = true
        selectedStocks = []$STRING
        stockWeights_ = dict(STRING, DOUBLE)
        symbols = []$STRING
        m_ret_3y_list = []$DOUBLE
        for (istock in msg.keys()) {
            bar = msg[istock]
            if (bar["volume"] <= 0) { continue }  // 跳过停牌股

            symbols.append!(istock)
            m_ret_3y_list.append!(double(indicator[istock]["ret_3y"]))
        }

        // 组装并处理pipeline
        poolT = table(symbols as symbol, symbols as code, m_ret_3y_list as `ret_3y)
        if (poolT.size() > 0) {

            rankInputSize0_ = poolT.size()
            rankValidT0_ = select * from poolT where isValid(ret_3y)
            if (rankInputSize0_ > 0 && rankValidT0_.size() == 0) { pipelineDataValid_ = false }
            limitN = iif(int(rankValidT0_.size() * 0.1) > 1, int(rankValidT0_.size() * 0.1), 1)
            poolT = select * from rankValidT0_ order by (ret_3y) asc limit limitN
            selectedStocks = exec distinct symbol from poolT
            for (s in selectedStocks) { stockWeights_[s] = 1.0 / iif(selectedStocks.size() > 0, selectedStocks.size(), 1) }
        }
        if (pipelineDataValid_) {
            context["selectedStocks"] = selectedStocks
            context["stockWeights"] = stockWeights_
        }
        context["selectionDataValid"] = pipelineDataValid_
        // --- 横截面选股结束 ---
        if (context["selectionDataValid"]) {
        // 信号日：用信号日 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)
        context["currentPortfolioPct"] = current_portfolio_pct_
        context["lastPortfolioPct"] = current_portfolio_pct_
        weights_ = context["stockWeights"]
        tgtQtys_ = dict(STRING, LONG)
        for (s_ in context["selectedStocks"]) {
            w_ = iif(s_ in weights_, weights_[s_], 0.0)
            tgtQtys_[s_] = roundLot(s_, signalEquity_ * current_portfolio_pct_ * w_ / (msg[s_]["close"] * (1 + context["commission"])))
        }
        context["pendingTargetQtys"] = tgtQtys_
        context["signalEquity"] = signalEquity_
        context["pendingRebalance"] = true
        context["rebalanceDate"] = curDate
        }
    }
}

2.3 设置回测参数

回测使用日频股票策略组、100 万元初始资金、万分之 1.5 双边手续费和千分之 0.5 卖出税。订单以次日 open 字段提交，保留订单状态信息，不启用单笔最低手续费。

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

2.4 创建回测引擎

创建引擎前，需要先将策略定义的 initialize 和 onBar 函数注册到回调字典 callbacks 中。

callbacks = dict(STRING, ANY)
callbacks["initialize"] = initialize
callbacks["onBar"] = onBar

engineName = "Backtest_DeBondtThaler_test"
try { Backtest::dropBacktestEngine(engineName) } catch(ex) { }
engine = Backtest::createBacktester(engineName, config, callbacks, false)

创建前脚本尝试删除同名引擎，异常被忽略；同名并发任务可能冲突，应使用独立会话与不冲突的名称。

2.5 执行回测

将排好时间顺序的 data 送入引擎，回调按交易日推进，依次执行信号计算和后续交易。第一天只产生目标股数，第二天才尝试成交。

回放结束时，通过 Backtest::appendEndMarker(engine) 向引擎发送结束标记，以通知引擎行情已全部发送完毕，之后引擎不再接收行情。

engine = Backtest::createBacktester(engineName, config, callbacks, false)
Backtest::appendQuotationMsg(engine, data)
Backtest::appendEndMarker(engine)

2.6 获取回测结果

回放结束后读取收益汇总、交易明细、每日持仓和账户权益：

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

返回表说明：

表 2. 表 2-2 回测结果
结果对象	用途与关键内容
returnSummary	引擎绩效概述；
tradeDetails	订单与成交明细；需按实际状态和正成交数量辨认成交
dailyPosition	每日证券持仓，按 tradeDate 保留实际起始日之后记录
totalPortfolios	每日账户权益，含 tradeDate、totalEquity；基准有效时关联基准列

3. 回测结果分析

使用上述版本、非 JIT 模式、附件提供的数据，回测处理 72,600 条行情，初始权益为 1,000,000.00 元，区间总收益为 1.4877%。由于数据来自模拟数据，这一指标不能用于评价真实市场中的相对表现。

表 3. 表 3-1 回测结果分析
指标	本次结果	解读
区间总收益	1.4877%	期末权益相对 100 万元初始资金的累计变化
年化收益	1.5496%	引擎按回测日收益年化后的结果
年化波动率	1.6611%	价格路径波动很高
夏普比率	0.9329	去掉风险波动后，收益处于普通水平
最大回撤	1.066%	期内净值相对先前高点的最大下跌，风险暴露一般
回撤比率	1.4536	引擎返回的收益与回撤相关统计
日胜率	49.7925%	每日收益为正的交易日占比

这些结果仅描述一组模拟价格和随机因子上的资金变化。输入没有赋予“长期被低估的股票，应有更高未来收益”的关系，也没有模拟真实财报公告和修订。不能根据本次收益判断策略在真实市场中是否有效，也不能仅凭一组模拟结果宣称优于其他策略。评价长期反转还需要更长样本、多轮调仓与样本外数据，并考虑行业集中、长期基本面恶化、幸存者偏差和交易成本的影响。

4. 总结

该策略将 756 条复权价格收益排序、最低 10% 等权选择、信号日锁定股数及下一有效日开盘调仓串联起来。每 252 个有效回调更新一次目标，实际账户会因数量取整、现金和成交条件偏离目标权重。

使用时最关键的是历史数据覆盖及可用时点、运行日期相关的初始股票池、合成分红行情口径和单次调仓不重试机制。

5. 附录

完整脚本包含策略函数、配置和引擎执行流程。

deBondtThaler.dos

deBondtThaler.csv