一、策略背景与研究目标

这次的研究目标，是做一套能从回测自然迁移到仿真/实盘的期货策略：

标的：螺纹钢期货（示例使用 RB2605.SHF，后续可替换为主力合约）

周期：分钟线（1 分钟 + 5 分钟）

策略类型：趋势跟随 + 固定点数止盈止损

核心诉求：

回测阶段在 panda_backtest 上验证策略有效性；

仿真阶段只需替换行情与交易接口，逻辑不变即可上线。

下面按“策略思路 → 回测实现 → 从回测到仿真联通方案 → 回测结果解读” 的顺序展开。

二、策略思路

1. 标的选择

为了降低复杂度，这里先只选一个标的：

螺纹钢期货 RB2605.SHF
在回测代码中，把它写成：

context.symbol = “RB2605.SHF”
后续可以很自然地扩展成“品种 + 主力合约”的模式，例如：

使用 panda_data.get_future_dominant(underlying_symbol=[“RB”], start_date, end_date) 在回测期内动态获取 RB 主力合约；

回测与仿真策略复用同一套逻辑，只是 symbol 的获取方式不同。

2. 信号周期设计

采用一个非常经典、易迁移的结构：

1 分钟线：刻画短期价格波动，用于确认入场的即时位置；

5 分钟线：过滤小级别噪音，作为中周期方向参考。

具体均线设置：

1 分钟：

取最近 20 根 1m K 线收盘价，计算简单移动平均：1m-20MA

5 分钟：

取最近 10 根 5m K 线收盘价，计算简单移动平均：5m-10MA

之所以用 20 和 10：

20 根 1 分钟 ≈ 20 分钟，给出一段较短的趋势参考；

10 根 5 分钟 ≈ 50 分钟，提供更平滑的中周期方向。

这些参数后续可以做网格/步进优化，但作为一个“回测–仿真贯通”的范例，这个配置足够直观。

3. 入场逻辑（只做多）

当满足以下条件时，考虑开多：

当前无多头持仓；
最新 1 分钟收盘价 current_price 同时高于两条均线：

current_price > 1m-20MA

current_price > 5m-10MA

解释：

1m-20MA 上方：短期价格已经“站上”近 20 分钟的平均水平，属于短期偏强；

5m-10MA 上方：中周期（约 50 分钟）也偏多，过滤掉大部分短期虚假突破；

两个条件同时满足，才开多 1 手，降低反复杂乱交易频率。

4. 出场逻辑（固定点数止盈 / 止损）

建仓后，对每一笔多头持仓跟踪浮动盈亏点数：

当前价：current_price

开仓价：entry_price

点数：profit_points = current_price - entry_price

出场规则：

当 profit_points >= +30 时：

认为顺势盈利达到预期，止盈平仓 1 手；

当 profit_points <= -30 时：

认为走势与预期偏离过大，止损平仓 1 手。

参数在代码中写入：

context.profit_target = 30 # 止盈 +30 点
context.stop_loss = -30 # 止损 -30 点
这样做的好处：

逻辑非常清晰，便于后续在仿真环境手动复核；

调参时只需要改一个地方，就可以测试不同的止盈止损组合。

三、回测实现要点（基于 panda_backtest + panda_data）

1. 使用分钟行情接口

在回测环境中，分钟线数据全部通过 panda_data 获取：
为了保证当日有足够数据，先用：

start_date = panda_data.get_previous_trading_date(date=trade_date, exchange=“SH”, n=2)
end_date = trade_date
向前取 2 个交易日的数据，然后在代码中只保留当日：

df_1m_today = df_1m[df_1m[“date”] == trade_date].copy()
df_5m_today = df_5m[df_5m[“date”] == trade_date].copy()
并做如下数据完整性检查：

len(df_1m_today) >= 20

len(df_5m_today) >= 10

不足则跳过该日，不随意“补数据”，保证回测结果更加可信。

2. 均线计算与信号生成

对当日的分钟数据按时间排序：

计算：

同时在日志中打印关键数值：

这些输出可以直接截图放在帖子里，用来说明策略在某天的具体行为。

3. 持仓与盈亏读取

在回测引擎内，期货账户信息通过 context.future_account_dict 提供：

这里只针对多头持仓：

buy_quantity == 0：无多头，考虑开多；

buy_quantity > 0：已有多头，跟踪浮动盈亏：

4. 开平仓指令

下单全部使用回测框架提供的期货 API：

开多：

order = buy_open(context.account, symbol, 1, style=MarketOrderStyle)[0]

平多：

order = sell_close(context.account, symbol, 1, style=MarketOrderStyle)[0]
并用 try/except 做防护，防止因为个别合约不可交易、价格异常等导致策略直接中断。

5. 生命周期日志

在 before_trading 中打印：

当前回测交易日、账户总权益、可用资金；

在 after_trading 中打印：

收盘后账户总权益、可用资金、各合约持仓信息（多空手数、开仓均价、总盈亏）。

这些日志能够让你在复盘某一天时，清晰地看到：

策略是否有交易；

持仓变化是否符合预期；

盈亏的贡献主要来自哪些波段。

四、从回测到仿真：联通思路

这套策略从一开始就按照“回测–仿真可对齐”的原则设计。

1. 不变的核心

无论回测还是仿真，策略的核心结构完全一致：

输入：分钟 K 线（1m / 5m）

运算：

计算 1m-20MA、5m-10MA

判断当前价是否站上双均线

计算当前价与开仓价的点数差

输出：

无多头 & 双均线突破 → buy_open(1 手)

有多头 & 点数达到阈值 → sell_close(1 手)

2. 需要替换的部分

从回测迁移到仿真，大致只改两块：

1）行情数据来源：

回测：

使用 panda_data.get_market_min_data 获取历史分钟线；

仿真：

使用框架提供的实时行情对象 data[symbol]，或仿真环境里的分钟聚合数据；

你可以将：

#回测里用 panda_data 拿历史 1m/5m 收盘价
替换为：

#仿真里直接使用当根 bar 的 close，再配合框架内部的“历史 bar 缓存”获取过去 N 根收盘价

2）交易 API：

回测端使用：buy_open / sell_close 来模拟成交；

仿真端对应换成：panda_trading 中的同名或等价接口，参数保持一致（账号、symbol、手数、订单类型）。

因为我们在回测阶段已经把交易接口抽象成：

buy_open(context.account, symbol, 1, style=MarketOrderStyle)
sell_close(context.account, symbol, 1, style=MarketOrderStyle)
所以迁移到仿真时，只需要保证仿真 API 提供类似的签名即可，无需动策略主逻辑。

五、调参思路（建议大家在跑出结果后补充）

参数敏感性

可以简单测试不同止盈止损参数：

如（20, -20）、（30, -30）、（40, -40），比较年化收益与最大回撤；

说明：当前版本选用 30 点，原因是兼顾了盈亏比和交易频率。

回测局限与优化方向

局限：

目前只做单一合约，未考虑主力移仓；

资金管理较为粗糙，每次固定 1 手。

优化方向：

接入 panda_data.get_future_dominant 做主力切换；

使用账户权益 × 风险比例动态计算开仓手数；

引入更严格的交易时段过滤（仅日盘，避开夜盘噪音）。

#六、完整回测代码（可直接运行）

可复制粘贴板：
from panda_backtest.api.api import *
from panda_backtest.api.stock_api import *
import panda_data

import pandas as pd
from datetime import datetime

“”"
主题示例：
《从回测到仿真：螺纹钢分钟多均线择时 + 固定点数止盈止损策略实战》

在社区分享时，你可以围绕以下结构展开：

一、策略思路
1）标的选择

• 选择螺纹钢主力合约（回测中暂用 RB2605.SHF 作为示例）。
  2）信号周期
• 使用 1 分钟线做短期走势刻画，5 分钟线做中周期滤波：
  • 1m：计算 20 根 K 线的均价（1m-20MA）
  • 5m：计算 10 根 K 线的均价（5m-10MA）
    3）入场逻辑
• 当最新 1 分钟收盘价同时高于：
  • 1m-20MA
  • 5m-10MA
• 且当前无多头持仓时，市价买入开仓 1 手。
  4）出场逻辑
• 建仓后跟踪浮动点数：profit = last_price - entry_price
• 当：profit ≥ +30 点（止盈），或 profit ≤ -30 点（止损）时，市价平掉 1 手多头。

二、回测实现要点
1）数据源

• 使用 panda_data.get_market_min_data 获取期货 1 分钟 / 5 分钟数据：
  • symbol_type=“future”
  • frequency=“1m” / “5m”
  • fields=[“date”, “minute”, “symbol”, “close”]
• 为避免数据不足，按当前回测日期向前取 2 个交易日的数据，然后只截取当日 (date == context.now) 的记录。

2）仓位与盈亏

• 从 context.future_account_dict[account].positions[symbol] 中读取持仓：
  • buy_quantity：当前多头手数
  • buy_avg_open_price：多头开仓均价
• 盈亏点数：current_close_1m - buy_avg_open_price。

3）风控与简化假设

• 固定开仓手数：每次信号触发只开 1 手，多次信号不会叠加加仓。
• 固定点数止盈止损：±30 点；不同品种可以在参数中调整。
• 未叠加资金管理（如权益占比开仓），方便先完成策略行为验证。

三、从回测到仿真
1）回测阶段（当前代码）

• 基于 panda_backtest 引擎，使用历史分钟数据验证：
  • 信号触发频率
  • 单笔盈亏分布
  • 最大回撤、年化收益等
    2）仿真阶段（下一步）
• 将 panda_data 获取历史行情的部分替换为仿真环境自带的实时分钟行情（如 data[symbol].close）。
• 交易接口在仿真环境中换成 panda_trading 的 buy_open / sell_close，保持逻辑一致。
  3）联通思路
• 策略核心：
  • 输入：分钟 K 线（1m/5m），
  • 运算：双均线 + 固定点数止盈止损，
  • 输出：期货多头开平仓指令。
• 只要保证回测和仿真在这三步上的数据口径与接口对齐，就能实现“回测–仿真”一键平移。

下面是对应的、可以直接在回测引擎中运行的完整代码。
“”"

def initialize(context):
“”"策略初始化

- 使用 context.account 保存期货账号
- 使用 context.symbol 保存需要交易的合约
- 使用 context.profit_target / context.stop_loss 控制止盈止损点数
"""
# 回测期货账号（平台约定一般为 '5588'）
context.account = '5588'

# 交易标的：这里先写死为 RB2605.SHF，后续可替换为主力连续合约
context.symbol = "RB2605.SHF"

# 固定点数止盈 / 止损参数（单位：价格点数）
context.profit_target = 30  # 止盈 +30 点
context.stop_loss = -30     # 止损 -30 点

# 记录策略启动时间（仅用于日志输出）
context.init_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f"[initialize] 策略启动时间: {context.init_time}, 回测当前日期: {context.now}")

def before_trading(context):
“”“开盘前回调：打印账户基本信息，便于回测调试与分享”""
print(f"[before_trading] 当前回测交易日: {context.now}")

futures_account = context.future_account_dict.get(context.account)
if futures_account:
    print(f"  账户总权益: {futures_account.total_value}")
    print(f"  可用资金: {futures_account.cash}")

def handle_data(context, data):
“”"策略主逻辑：双均线择时 + 固定点数止盈止损

步骤：
1. 读取当前回测日期 trade_date = context.now
2. 通过 panda_data 获取 1m / 5m 历史分钟线
3. 计算 1m-20MA 与 5m-10MA
4. 根据当前持仓决定是开多还是平多
"""
symbol = context.symbol
trade_date = str(context.now)

# 1. 为保证当日有足够分钟数据，向前取 2 个交易日的开始日期
start_date = panda_data.get_previous_trading_date(date=trade_date, exchange="SH", n=2)
if start_date is None:
    start_date = trade_date
end_date = trade_date

print(f"[handle_data] 日期: {trade_date}, 查询区间: {start_date} ~ {end_date}, 合约: {symbol}")

# 获取 1 分钟 K 线
try:
    df_1m = panda_data.get_market_min_data(
        symbol=symbol,
        start_date=start_date,
        end_date=end_date,
        symbol_type="future",
        fields=["date", "minute", "symbol", "close"],
        frequency="1m",
        time_zone=None,
    )
except Exception as e:
    print(f"  获取 1 分钟数据失败 {symbol}: {e}")
    return

# 获取 5 分钟 K 线
try:
    df_5m = panda_data.get_market_min_data(
        symbol=symbol,
        start_date=start_date,
        end_date=end_date,
        symbol_type="future",
        fields=["date", "minute", "symbol", "close"],
        frequency="5m",
        time_zone=None,
    )
except Exception as e:
    print(f"  获取 5 分钟数据失败 {symbol}: {e}")
    return

if df_1m is None or df_1m.empty or df_5m is None or df_5m.empty:
    print(f"  {symbol}: 无 1m 或 5m 数据，跳过")
    return

# 只保留当前交易日数据，避免跨日混入
df_1m_today = df_1m[df_1m["date"] == trade_date].copy()
df_5m_today = df_5m[df_5m["date"] == trade_date].copy()

if len(df_1m_today) < 20 or len(df_5m_today) < 10:
    print(f"  {symbol}: 当日 1m({len(df_1m_today)}) 或 5m({len(df_5m_today)}) 数据不足以计算均线")
    return

# 按时间排序
df_1m_today.sort_values(by=["date", "minute"], inplace=True)
df_5m_today.sort_values(by=["date", "minute"], inplace=True)

close_1m = df_1m_today["close"].reset_index(drop=True)
close_5m = df_5m_today["close"].reset_index(drop=True)

# 打印末尾部分收盘价，便于在社区文章中展示
print(f"  1m 收盘价({symbol}) 最近 10 条:\

{close_1m.tail(10)}")
print(f" 5m 收盘价({symbol}) 最近 10 条:
{close_5m.tail(10)}")

# 计算均线
ma_1m_20 = close_1m.rolling(window=20).mean()
ma_5m_10 = close_5m.rolling(window=10).mean()

ma_1m_20_last = ma_1m_20.iloc[-1]
ma_5m_10_last = ma_5m_10.iloc[-1]
current_price = close_1m.iloc[-1]

print(
    f"  {symbol}: 当前价={current_price:.2f}, "
    f"1m-20MA={ma_1m_20_last:.2f}, 5m-10MA={ma_5m_10_last:.2f}"
)

# 获取账户与持仓
futures_account = context.future_account_dict.get(context.account)
if not futures_account:
    print(f"  未找到期货账户 {context.account}")
    return

positions = futures_account.positions
position = positions.get(symbol)

buy_quantity = 0
entry_price = None
if position:
    buy_quantity = getattr(position, 'buy_quantity', 0) or 0
    entry_price = getattr(position, 'buy_avg_open_price', None)

# 6. 交易逻辑
# —— 无多头持仓：择时开多 ——
if buy_quantity == 0:
    if pd.notna(ma_1m_20_last) and pd.notna(ma_5m_10_last):
        if current_price > ma_1m_20_last and current_price > ma_5m_10_last:
            try:
                order = buy_open(context.account, symbol, 1, style=MarketOrderStyle)[0]
                if order:
                    print(f"  开多: 买入 {symbol} 1 手, 订单ID={order.order_id}")
            except Exception as e:
                print(f"  开多失败 {symbol}: {e}")
else:
    # —— 已有多头：按点数止盈/止损平仓 ——
    if entry_price is None:
        print(f"  {symbol}: 已有多头 {buy_quantity} 手，但缺少入场价，跳过平仓判断")
        return

    profit_points = current_price - entry_price
    print(
        f"  {symbol}: 当前多头={buy_quantity} 手, 开仓价={entry_price:.2f}, "
        f"浮动盈亏={profit_points:.2f} 点"
    )

    if profit_points >= context.profit_target or profit_points <= context.stop_loss:
        try:
            order = sell_close(context.account, symbol, 1, style=MarketOrderStyle)[0]
            if order:
                print(
                    f"  平多: 卖出 {symbol} 1 手, 盈亏 {profit_points:.2f} 点, "
                    f"订单ID={order.order_id}"
                )
        except Exception as e:
            print(f"  平多失败 {symbol}: {e}")

def after_trading(context):
“”“收盘后回调：输出账户与持仓情况，便于在分享中展示每日收盘状态”""
print(f"[after_trading] 当前回测交易日: {context.now}")

futures_account = context.future_account_dict.get(context.account)
if futures_account:
    print(f"  收盘后账户总权益: {futures_account.total_value}")
    print(f"  收盘后可用资金: {futures_account.cash}")
    positions = futures_account.positions
    if positions:
        for sym, pos in positions.items():
            print(
                f"  持仓 {sym}: 多头={pos.buy_quantity}, 空头={pos.sell_quantity}, "
                f"多头开仓均价={pos.buy_avg_open_price}, 总盈亏={pos.pnl}"
            )
    else:
        print("  无持仓")

方便直接复制运行。