一、研究动机与问题

研究标的： 上期能源原油期货 SC 品种，采用主力合约连续。

（新手小白别担心我手把手教你们）

核心问题：

我们在 SC 原油期货上，使用简单的价格趋势（动量）策略，是否能在长期上取得相对于“单边多头持有”的超额收益？

原油这类大宗商品：

经常出现剧烈、持续的单边行情（如 2014–2015 暴跌、2020 疫情、2022 能源危机）；

直觉上，顺势交易（趋势策略）似乎更容易抓住大波段；

但市场参与者增多，趋势效应是否仍然存在、是否已经被套利掉，需要通过数据验证。

本研究采用极简的趋势定义：20 日价格动量，持有 5 日，通过回测检验趋势是否带来统计意义上的超额收益。

二、数据与实验设定

2.1 标的与数据来源

标的： SC 品种主力合约（通过主力映射动态切换合约）。

数据来源：

使用 panda_data.get_future_dominant 获取全区间 SC 主力合约映射；

使用 panda_data.get_market_data 获取相关合约日线收盘价；

在初始化阶段拼接成一条主力连续收盘价序列。

2.2 时间频率与区间

数据频率：日线。

样本区间：SC 上市以来的所有交易日。

在代码中，为了简化，不直接读回测起止时间，而是用一个很宽的区间：

start_date = ‘20000101’，end_date = ‘20991231’

实际回测区间由回测平台参数控制；策略内部按 context.now 使用当日数据。

2.3 回测环境

回测框架：panda_backtest（事件驱动）；

交易账户：期货账户 ID 5588；

回测模式：MODE = ‘backtest’（一次性预加载数据，盘中不访问网络）。

三、趋势策略设计

3.1 趋势（动量）定义

采用最基础的价格动量：

回看窗口 L：20 个交易日；

持有窗口 H：5 个交易日；

收盘价记为 (P_t)，动量定义为：
[
mom_t = \frac{P_t}{P_{t-L}} - 1
]

信号规则：

若 mom_t > 0：认为存在上升趋势，生成做多信号 +1；

若 mom_t < 0：认为存在下降趋势，生成做空信号 -1；

若 mom_t 缺失/为 0：视为趋势不明显，信号 0（空仓）。

3.2 交易与持有规则

每次生成趋势信号后：

持有该信号 H=5 个交易日，在这期间不切换；

第 6 个交易日再根据最新 mom_t 重新判断趋势方向；

重复上述过程，形成一条连续的多空切换策略曲线。

仓位设定：

为了聚焦在“趋势是否有效”，不引入复杂资金管理；

本研究固定每次开仓 1 手 SC 主力合约。

3.3 策略目标

这套简洁的策略主要为两个目的服务：

我们为了验证 SC 上是否存在明显的价格动量效应；
我们可以为后续加入资金管理、多因子过滤等提供一个干净又漂亮的基准框架。

四、代码框架与关键实现

我们可以根据下面基于你给出的代码，展示完整的数据预加载 + 初始化部分。后续在此基础上补充盘前 / 盘中 / 盘后逻辑即可形成可回测策略。

4.1 必要 import 与模式开关

panda_backtest.api.api：回测框架的核心 API，包括下单、账户访问等；

panda_backtest.api.stock_api：框架要求的 import，虽然本策略只做期货；

panda_data：外部数据接口，用于从数据库获取主力映射与行情。

MODE = ‘backtest’ 表示当前代码遵循“回测极致性能模式”，所有数据预加载在 initialize 中完成，handle_data 只读内存，不访问网络。

4.2 数据预加载 _preload_all_data(context)

说明：

调 get_future_dominant 获取 SC 在所有日期上的主力合约，构成 _dominant_map[(品种, 日期)] = symbol；
汇总所有涉及的合约（all_symbols），一次性用 get_market_data 拉出日线收盘价，整理成 _daily_close_map[symbol][date] = close；
遍历所有 SC 的日期：根据 _dominant_map 找到当日主力合约，再从 _daily_close_map 中拾取对应收盘价；
形成一条按日期排序的“SC 主力连续收盘价”序列 df，然后计算 20 日动量：mom = close / close.shift(L) - 1；
将结果存入 context.mom_df，后续只需按日期索引即可。

4.3 初始化 initialize(context)

要点：

将趋势研究的所有关键参数集中在 initialize：

lookback（L 日动量窗口）

holding（H 日持有时间）

fixed_hands（单次开仓手数）

context.products = [‘SC’] 清晰表明只研究原油；

调用 _preload_all_data 完成主力映射 + 连续价格 + 动量计算。

这部分就是整项研究的“数据与信号准备层”，在策略真正开始交易之前，我们就已经把趋势信号所需的所有信息准备好了。

五、如何在此框架上完成完整策略

在上述代码基础上，只需补齐三个部分即可完成完整回测：

盘前：确定当日主力合约

信号与交易执行：在 handle_data 中使用 mom_df

定义一个信号函数：

在 handle_data 中每持有 H 日切换一次趋势方向：
盘后：打印/记录绩效
完整实现时，将“平仓与开仓”部分补全即可形成可实测的趋势策略，用回测报告中收益曲线、年化收益、最大回撤等指标，与“单边长期多头 SC 主力”做对比，就能回答本文的核心问题。

六、总结与扩展方向

通过上述代码框架，我们已经搭建起一套以“SC 主力连续价格 + L 日动量”为核心的趋势研究平台：

优点：

信号定义简单透明，便于解释和复现；

数据预加载在初始化完成，回测性能友好；

参数集中管理，方便做 L/H 的稳健性测试。

下一步可扩展方向：
引入 ATR 波动率控制：用 ATR 决定每次开仓手数，控制风险敞口；
加入多因子过滤：例如成交量、持仓量、期限结构（近月-远月价差）；
多品种扩展：从 SC 推广到化工、黑色系等，验证趋势效应是否普遍存在。

你当前这份代码已经很好地完成了“预加载与动量计算”这一层，后面只要把
before_trading / handle_data / after_trading 补齐，就可以跑出完整的趋势策略结果。

八、这次做 SC 原油趋势策略的过程，对我最大的触动有两点：

1. 把“情绪化的感觉”拆成可以检验的东西

一开始对原油的印象，其实和大部分人差不多：

觉得它“波动大、很趋势”；

觉得“顺势做应该很好赚”；

但当真的动手写代码的时候，会发现：如果不把这些“感觉”拆成具体的参数（比如用什么周期的动量、持有多久、用哪种价格序列），你其实什么都检验不了，只是在原地空想。

这次我把问题拆到了可以尝试回测的层面：

标的：SC 主力连续；

频率：日线；

动量：20 日价格动量；

规则：mom>0 做多，mom<0 做空，信号持有 5 天；

当所有假设都变成一个个清晰的规则，并且落在代码里能跑之后，会有一种“把一团雾气揉成一颗小石子”的感觉：这颗小石子可能不完美，但它是实实在在的。

2. 发现“简单策略”真正的价值

做这类趋势策略，很容易一上来就想得很复杂：加 ATR、加过滤条件、加仓位管理、加风险控制……

这次刻意压住这种冲动，只留了最简单的一条线：

不做别的花哨的指标；

不考虑资金曲线上的小技巧；

固定 1 手，只看方向是不是有价值；

###在这个过程中我体会到一个点：

真的想搞清楚“一个效应存不存在”，越简单越好。

复杂的东西容易让人误以为“赚钱是因为我多聪明”，但简单的动量 + 多空切换，更容易看清楚：到底是原油本身有趋势，还是我在用各种手段掩盖策略的问题。

3. 把“工程习惯”养在一开始

这次写的代码虽然只是一个研究 demo，但我：

严格区分 initialize / before_trading / handle_data 的职责；

所有重计算都放在预加载；

盘中代码尽量短、尽量干净，不访问网络，不写复杂逻辑；

一开始会觉得有点“多此一举”，好像多写了很多结构性的代码，但写着写着会发现：

策略变得非常容易扩展：想换成 RB 或 M，只要改品种；

想加一个新信号，只要在预加载那一层加一列；

这和“写脚本”最大的差别是：你在构建一个长期能复用的研究框架，而不是为了一次性回测随手凑个程序。

4. 对“结果好不好”的态度更冷静了一些

在正式回测之前，我心里其实是带有期待的：

会不会这条简单的动量曲线就能完爆单边多头？

但在搭建好整套框架的过程中，心情反而慢慢冷静下来：

我开始更关心“结论是否可信、可复现”，而不是“收益率看起来能不能吹”；

开始接受这样一种可能：结果也许是趋势效应并不明显，那也同样是一个重要的结论；

这种心态上的变化，对我来说是一个不小的收获——

不再执着于“证明自己是对的”，而是更在意“用干净的方法，把问题问清楚”。

5. 下一步更有方向感了

做完这一步，其实反而更知道“接下来该做什么”：

可以系统地扫一遍不同的 L 和 H，看看动量效应是否只在某个窄区间成立；

可以扩展到其他品种，看看 SC 是特殊，还是趋势在大宗整体上都存在；

可以逐步引入风险控制，看在不牺牲太多收益的前提下，回撤能降多少；

以前想这些事情的时候，脑子里是很乱的；现在因为有了一套跑得动的、结构清晰的代码框架，这些“想法”都变成了下一步可以排期的实验。

总的来说，这次 SC 趋势策略的实现，对我来说不只是多了一段代码，而是让“怎么从直觉走到结论”这条路变得更清楚了一些：

先把问题说清楚；

再用最简单的规则把它落到数据上；

用规范的工程习惯维护这套实验；

接受任何结果，然后在这个基础上往前走。

这套体验，对后面做任何别的品种、别的因子，都是可以复用的。谢谢大家观看。

最后是完整的代码给大家方便复制取用：

from panda_backtest.api.api import *
from panda_backtest.api.stock_api import *
import panda_data
import pandas as pd
import numpy as np

###====================================================

回测目标：检验“原油（SC）趋势是否有超额收益”

思路：

- 标的：SC 品种主力合约

- 周期：日线

- 趋势定义：L 日动量 mom = P_t / P_{t-L} - 1

- 策略：mom>0 做多，mom<0 做空，持有 H 日

- 基准：同样持有 H 日的“单边多头”收益（可用平台基准或单独策略对比）

###====================================================

MODE = ‘backtest’

def _preload_all_data(context):
“”“backtest 模式专用：预加载全区间 SC 主力合约和合约乘数，并预先计算动量”""
# 这里不再依赖 context.run_info，改用一个宽泛区间，由 before_trading 实际截取
start_date = ‘20000101’
end_date = ‘20991231’

# 1）预加载主力合约映射 {(品种, 日期): symbol}
context._dominant_map = {}
try:
    dom_df = panda_data.get_future_dominant(
        underlying_symbol=context.products,
        start_date=start_date,
        end_date=end_date
    )
    if dom_df is not None and not dom_df.empty:
        for _, row in dom_df.iterrows():
            key = (row['underlying_symbol'], str(row['date']))
            context._dominant_map[key] = row['symbol']
except Exception:
    context._dominant_map = {}

# 2）预加载合约乘数 {symbol: 乘数}
context._mul_map = {}
all_symbols = list(set(context._dominant_map.values()))
if all_symbols:
    try:
        mul_df = panda_data.get_future_list(
            symbol=all_symbols,
            fields=["symbol", "contract_multiplier"]
        )
        if mul_df is not None and not mul_df.empty:
            for _, row in mul_df.iterrows():
                context._mul_map[row['symbol']] = float(row['contract_multiplier'])
    except Exception:
        context._mul_map = {}

# 3）预载日线收盘价，拼接 SC 主力连续价格序列（先整体加载，再在使用时按日期过滤）
context.price_series = {}
try:
    if all_symbols:
        mkt_df = panda_data.get_market_data(
            symbol=all_symbols,
            start_date=start_date,
            end_date=end_date,
            type="future",
            fields=["symbol", "date", "close"]
        )
    else:
        mkt_df = None
    if mkt_df is not None and not mkt_df.empty:
        mkt_df['date'] = mkt_df['date'].astype(str)
        grouped = mkt_df.groupby('symbol')
        context._daily_close_map = {}
        for symbol, g in grouped:
            context._daily_close_map[symbol] = {
                row['date']: float(row['close']) for _, row in g.iterrows()
            }
    else:
        context._daily_close_map = {}
except Exception:
    context._daily_close_map = {}

# 4）构造主力连续收盘价序列并计算动量（后续在 handle_data 中按 context.now 使用）
context.mom_df = None
try:
    all_dates = sorted({str(d) for (prod, d) in context._dominant_map.keys() if prod == 'SC'})
    records = []
    for d in all_dates:
        symbol = context._dominant_map.get(('SC', d))
        if not symbol:
            continue
        price_map = context._daily_close_map.get(symbol, {})
        close_p = price_map.get(d)
        if close_p is None or close_p <= 0:
            continue
        records.append({'date': d, 'symbol': symbol, 'close': close_p})
    if records:
        df = pd.DataFrame(records)
        df.sort_values('date', inplace=True)
        df.reset_index(drop=True, inplace=True)

        L = context.lookback
        df['mom'] = df['close'] / df['close'].shift(L) - 1.0

        context.mom_df = df
except Exception:
    context.mom_df = None

def initialize(context):
“”“策略初始化：设定账户、参数，预加载数据”""
# 期货账户
context.account = ‘5588’

# 模式
context.mode = MODE

# 交易品种：只研究 SC 原油
context.products = ['SC']

# 趋势策略参数
context.lookback = 20   # L：回看窗口（用于计算动量）
context.holding = 5     # H：持有窗口（信号更新周期）

# 风控参数：这里不做复杂仓位控制，只做 1 手多空切换
context.fixed_hands = 1

# 以下变量名固定
context.today_dominant = {}   # {品种: symbol}
context.contract_mul = {}     # {symbol: 乘数}

# 预加载用缓存
context._dominant_map = {}
context._mul_map = {}
context._daily_close_map = {}
context.price_series = {}
context.mom_df = None

# 记录信号和统计
context.day_index = 0
context.signal_series = []  # 每次信号变更的记录

# 当前信号与持有天数
context.current_signal = 0
context.holding_counter = 0

if context.mode == 'backtest':
    _preload_all_data(context)

def _before_trading_backtest(context):
“”“backtest 模式：从预加载字典查当日主力与乘数”""
today = str(context.now)
context.today_dominant = {}
context.contract_mul = {}

for product in context.products:
    symbol = context._dominant_map.get((product, today))
    if symbol:
        context.today_dominant[product] = symbol
        mul = context._mul_map.get(symbol, 1.0)
        context.contract_mul[symbol] = mul

def _before_trading_live(context):
“”“live 模式：每天查 panda_data，兼容仿真/实盘”""
today = str(context.now)
context.today_dominant = {}
context.contract_mul = {}

try:
    dom_df = panda_data.get_future_dominant(
        underlying_symbol=context.products,
        start_date=today,
        end_date=today
    )
    if dom_df is not None and not dom_df.empty:
        for _, row in dom_df.iterrows():
            context.today_dominant[row['underlying_symbol']] = row['symbol']
except Exception:
    pass

symbols = list(context.today_dominant.values())
if symbols:
    try:
        mul_df = panda_data.get_future_list(
            symbol=symbols,
            fields=["symbol", "contract_multiplier"]
        )
        if mul_df is not None and not mul_df.empty:
            for _, row in mul_df.iterrows():
                context.contract_mul[row['symbol']] = float(row['contract_multiplier'])
    except Exception:
        pass

def before_trading(context):
if context.mode == ‘backtest’:
_before_trading_backtest(context)
else:
_before_trading_live(context)

def _get_today_mom_signal(context):
“”“根据预先计算好的 mom_df，给出今日的趋势信号：+1 多，-1 空，0 无信号”""
if context.mom_df is None or context.mom_df.empty:
return 0
today = str(context.now)
row = context.mom_df[context.mom_df[‘date’] == today]
if row.empty:
return 0
mom = float(row[‘mom’].iloc[0])
if np.isnan(mom):
return 0
if mom > 0:
return 1
elif mom < 0:
return -1
else:
return 0

def _close_all_positions(context, fut_acct, symbol):
“”“平掉该合约所有多空头寸”""
if fut_acct is None:
return
positions = fut_acct.positions
if symbol not in list(positions.keys()):
return
pos = positions[symbol]
# 平多
if pos.closable_buy_quantity > 0:
try:
sell_close(context.account, symbol, pos.closable_buy_quantity, style=MarketOrderStyle)
except Exception:
pass
# 平空
if pos.closable_sell_quantity > 0:
try:
buy_close(context.account, symbol, pos.closable_sell_quantity, style=MarketOrderStyle)
except Exception:
pass

def _open_position_by_signal(context, fut_acct, symbol, signal):
“”“根据信号开仓：+1 开多，-1 开空”""
if fut_acct is None:
return
hands = context.fixed_hands
if hands <= 0:
return
try:
if signal > 0:
buy_open(context.account, symbol, hands, style=MarketOrderStyle)
elif signal < 0:
sell_open(context.account, symbol, hands, style=MarketOrderStyle)
except Exception:
pass

def handle_data(context, data):
“”“每个交易日执行：按照预载的动量信号进行多空切换，用于检验趋势是否带来超额收益”""
context.day_index += 1

product = 'SC'
symbol = context.today_dominant.get(product)
if not symbol:
    return

fut_acct = context.future_account_dict.get(context.account)
try:
    bar = data[symbol]
except Exception:
    return
if not bar or bar.close <= 0:
    return

# 简单的信号持有逻辑：持有 H 天后重新评估趋势
context.holding_counter += 1
need_new_signal = (context.holding_counter > context.holding)

if need_new_signal:
    # 1）平掉当前所有仓位
    _close_all_positions(context, fut_acct, symbol)

    # 2）计算今日趋势信号
    new_signal = _get_today_mom_signal(context)

    # 3）根据信号开新仓
    if new_signal != 0:
        _open_position_by_signal(context, fut_acct, symbol, new_signal)

    # 4）记录信号
    context.current_signal = new_signal
    context.signal_series.append((str(context.now), new_signal))

    # 5）重置计数
    context.holding_counter = 1

def after_trading(context):
“”“每日收盘后打印简单统计，方便观察趋势策略表现”""
fut_acct = context.future_account_dict.get(context.account)
if fut_acct:
pos_count = sum(1 for p in fut_acct.positions.values()
if p.buy_quantity > 0 or p.sell_quantity > 0)
print(f"[{context.now}] 权益={fut_acct.total_value:.0f} 持仓={pos_count}个 当前信号={context.current_signal}")

def on_future_trade_rtn(context, order):
“”“成交回报：打印基本信息，便于分析策略行为”""
if order.status == 2:
direction = ‘买’ if order.side == 1 else ‘卖’
action = ‘开’ if order.effect == 0 else ‘平’
print(f"成交 [{context.now}] {order.order_book_id} {direction}{action} {order.filled_quantity}手 价格={order.price}")