一、项目目标：从“能跑”到“敢上实盘”

很多时候策略逻辑本身不复杂，真正难的是让它在真实环境里稳定跑起来。这次的目标是：

用一个简单的 RB 分钟级策略做载体；

把从 日志设计、数据调用、持仓读取、风险控制 到 自动下单 的工程流程走一圈；

最终形成一套“能直接迁移到实盘/仿真”的代码骨架。

当前策略特点：

标的：RB2605.SHF；

频率：以分钟为驱动事件；

核心逻辑：

1 分钟 + 5 分钟双周期均线过滤；

固定 30 点止盈止损；

引入完整日志体系 SRLogger 追踪每一笔决策与数据。

二、系统架构与模块拆分

1. 策略模块

策略核心代码由四个事件函数组成：

initialize(context)：初始化策略，记录当前时间、环境信息；

before_trading(context)：开盘前检查；

handle_data(context, data)：每分钟执行，包含数据拉取、信号生成、下单；

after_trading(context)：收盘后做总结与回顾。

关键点：

使用 SYMBOLS = [“RB2605.SHF”] 管理合约列表；

用当前分钟 now.minute % len(SYMBOLS) 做索引，保留了扩展到多合约轮询的能力；

所有关键步骤都通过 SRLogger.info 输出，方便复盘和排查。

2. 数据模块

采用 future_api_quotation 作为分钟行情来源：

通过 panda_quant.get_previous_trading_date 获取上一交易日，构造：

start_date = 上一交易日

end_date = 今日

分别拉取：

period=‘1m’ 的 1 分钟数据

period=‘5m’ 的 5 分钟数据

为避免初始阶段均线数据不足，策略中加入数据量检查：

if len(quotation_df_1m) < 20 or len(quotation_df_5m) < 10:
SRLogger.info(f"合约 {symbol} 的数据不足以计算均线")
return
这样可以避免开盘前几分钟出现 NaN 均线导致的错误信号。

3. 交易与持仓模块

交易接口：

开多：buy_open(account, symbol, 1)

平多：sell_close(account, symbol, 1)

持仓读取：

通过期货账户字典获取：context.future_account_dict[‘1826’]

从中取出指定合约的持仓对象：position = …positions[symbol]

关键字段：

position.buy_quantity 当前多头手数；

position.buy_avg_open_price 多头开仓均价。

这样可以保证所有交易动作都建立在真实持仓信息上，不会出现“没有仓位却平仓”的错误。

三、策略逻辑：简单但可解释

这里的策略逻辑刻意保持简单，更重要的是“可解释性”和“易调试”。

1. 信号生成逻辑

1）均线计算

2）买入条件（只做多）：

当前无多头持仓：buy_quantity == 0

当前 1 分钟收盘价满足：

current_price > ma_5m_10[-1]

current_price > ma_1m_20[-1]

理解：短周期价格站上短期趋势（1m 20 均）和中期趋势（5m 10 均），视作一个“短线顺势突破”信号。

3）平仓条件（止盈止损）：

当前有多头持仓；

浮动盈亏点数：profit_points = current_price - entry_price

当：

profit_points >= 30（盈利 ≥ 30 点），或者

profit_points <= -30（亏损 ≤ -30 点）
即触发平多。

这形成一个非常明确、可验证的规则集，便于后续逐个环节排查逻辑。

四、日志与调试：如何排查每一个决策

这一部分是整个项目的“灵魂”：要让策略出了问题，能迅速从日志找到原因。

1. 关键日志点设计

策略在以下节点输出详细日志：

1）策略生命周期

SRLogger.info(“initialize”)
SRLogger.info(“before_trading”)
SRLogger.info(“after_trading”)

2）每分钟运行信息

SRLogger.info(“handle_data”)
SRLogger.info(f"当前分钟: {now.minute}, 选择合约: {symbol}")
SRLogger.info(f"调用参数: symbol_list={SYMBOLS}, start_date={start_date}, end_date={end_date}, period=‘1m’")
SRLogger.info(quotation_df_1m)
SRLogger.info(quotation_df_5m)

3）风控与信号决策

数据不足时：明确说明是哪一个合约缺数据；

买入开仓时：记录合约、方向和数量；

平仓时：记录当前浮动盈亏点数：

SRLogger.info(f"平仓 {symbol} - 当前收益: {profit_points} 点")

2. 日志驱动的调参与迭代

通过观察一段时间的日志，你可以很清晰地回答：

某一分钟为什么没交易？（数据不足 / 持仓不为空 / 条件未触发）

某一笔平仓为什么在那一刻执行？（恰好达到 +/-30 点）

均线值是否如预期般跟随行情变化？

后续优化时，可以在日志中额外记录：

每次买入时的 MA 值、ATR 值；

每次平仓后，是否后面又大幅朝有利方向发展（评估止盈止损是否过紧）。

五、实盘视角下的风险与改进

从工程视角看，这个策略已经具备“跑起来”的基本条件，但距离真正的实盘，还有不少坑需要注意：

1. 交易频率与滑点

分钟级策略在高波动时段可能频繁触发开平仓；

实盘滑点、手续费会大幅侵蚀策略的理论收益；

可以在日志中统计：

日交易次数

单笔平均持仓时长

理论收益与考虑手续费后的收益差距。

2. 合约与主力切换

目前合约列表中只有 RB2605.SHF，在跨月份或主力切换时，会出现：

成交量下降

价差跳变

后续可以：

使用主力连续合约做回测；

实盘中通过 panda_data.get_future_dominant 或交易所数据动态切主力；

在日志标记每次“换月动作”。

3. 风险控制

目前仅有单笔 ±30 点止盈止损，还可以加入：

单日最大亏损限制（如当日亏损超过 X 点退出交易）；

连续亏损 N 笔后暂停交易；

单品种最大持仓手数限制等。

这些都可以在 before_trading 和 after_trading 中管理和记录。

六、后续计划与开放问题

未来我准备在这个工程骨架上继续做几件事：

参数化与优化

把 MA 窗口和止盈止损点数从写死变成参数；

做一轮简单的参数扫描，看看在 RB 历史周期中，

哪一组参数组合在样本内/样本外相对稳健；

策略逻辑扩展

增加做空逻辑（均线死叉 +价格跌破）；

增加趋势过滤：例如只在日线趋势向上时做多，向下时做空。

监控与报警

接入简单的监控：

策略挂了是否能收到报警；

某个小时无成交是否触发提醒；

七、总结

这篇分享并不是一个“多么厉害的策略”，而更像是一个完整工程化过程的 demo：

从一个非常朴素的 RB 分钟级均线策略出发，在 panda_backtest / panda_trading 环境下，完整走了一遍数据获取、信号生成、持仓读取、自动下单和日志调试的流程。逻辑简单，但结构清晰、可解释、易维护，为后续更复杂的模型（多因子、机器学习等）提供了一个可以复用的工程框架。

欢迎大家在这个骨架基础上改造、扩展，如果你有更好的止盈止损方式、主力切换逻辑或者风控方案，也可以在评论区一起讨论。
完整代码如下需要自取：

具体代码如下：
from datetime import datetime
import panda_quant
from panda_backtest.api.future_api import future_api_quotation
from panda_trading.trading_common.api.api import buy_open, sell_close
from panda_trading import SRLogger

全部合约列表

SYMBOLS = [
“RB2605.SHF”
]

def initialize(context):
SRLogger.info(“initialize”)
SRLogger.info('context.now: ’ + str(context.now))
# 获取当前分钟对应的下单合约索引
now = datetime.now()
print(‘时间’, now)

def before_trading(context):
SRLogger.info(“before_trading”)

def handle_data(context, data):
SRLogger.info(“handle_data”)
account = context.run_info.future_account

# 获取当前分钟对应的下单合约索引
now = datetime.now()
print('时间', now)
idx = now.minute % len(SYMBOLS)
symbol = SYMBOLS[idx]

SRLogger.info(f"当前分钟: {now.minute}, 选择合约: {symbol}")

# 使用上一交易日作为开始日期
start_date = panda_quant.get_previous_trading_date(datetime.now().strftime('%Y%m%d'))
end_date = datetime.now().strftime('%Y%m%d')

# 打印入参
SRLogger.info(f"调用参数: symbol_list={SYMBOLS}, start_date={start_date}, end_date={end_date}, period='1m'")

quotation_df_1m = future_api_quotation(symbol_list=[symbol], start_date=start_date, end_date=end_date, period='1m')
quotation_df_5m = future_api_quotation(symbol_list=[symbol], start_date=start_date, end_date=end_date, period='5m')
SRLogger.info(quotation_df_1m)
SRLogger.info(quotation_df_5m)

if len(quotation_df_1m) < 20 or len(quotation_df_5m) < 10:
    SRLogger.info(f"合约 {symbol} 的数据不足以计算均线")
    return

close_prices_1m = quotation_df_1m['close']
close_prices_5m = quotation_df_5m['close']

# 输出1分钟和5分钟的收盘价数据
print(f"1分钟收盘价数据({symbol}): {close_prices_1m}")
print(f"5分钟收盘价数据({symbol}): {close_prices_5m}")

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

# 打印均线值
print(f"5分钟周期10日均线({symbol}): {ma_5m_10.iloc[-1]}")
print(f"1分钟周期20日均线({symbol}): {ma_1m_20.iloc[-1]}")

# 获取当前持仓量
position = context.future_account_dict['1826'].positions[symbol]
buy_quantity = position.buy_quantity

# 根据持仓情况判断交易条件
if buy_quantity == 0:
    # 没有持仓时判断是否符合买入条件
    current_price = close_prices_1m.iloc[-1]
    if current_price > ma_5m_10.iloc[-1] and current_price > ma_1m_20.iloc[-1]:
        # 当前价格大于5分钟周期10日均线且大于1分钟周期20日均线
        buy_open(account, symbol, 1)
        SRLogger.info(f"买入开仓 {symbol}")
else:
    # 有持仓时检查是否满足平仓条件
    entry_price = position.buy_avg_open_price
    current_price = close_prices_1m.iloc[-1]
    profit_points = current_price - entry_price
    
    if profit_points >= 30 or profit_points <= -30:
        # 收益或亏损达到30个点时进行平仓
        sell_close(account, symbol, 1)
        SRLogger.info(f"平仓 {symbol} - 当前收益: {profit_points} 点")

def after_trading(context):
SRLogger.info(“after_trading”)