PandaAI工作流-策略帮助文档
PandaAI官方 2025年06月23日
框架基本方法
基础方法说明
该策略为事件驱动性策略，需要实现框架中约定的事件回调方法，实现后回测、仿真、实盘通用。
策略头部需要默认引用内置API，运行代码为：from panda_backtest.api.api import *，后文不再重复赘述。
期货策略双模式架构（MODE模式）
策略通过 MODE 变量兼容性能模式和通用模式两种场景，默认 MODE = 'backtest'。
MODE 说明：
MODE = 'backtest'：性能模式。回测速度提升数倍。仅支持回测。MODE = 'live'：通用模式。兼容回测、仿真、实盘回测速度稍慢。
期货策略标准代码模板
默认 MODE = 'backtest'（性能模式）。切换到通用模式时只需改 MODE = 'live'，其余代码不动。
以下是一个完整的实战策略示例——双动量策略（Dual Momentum），覆盖15个期货品种，包含动量计算、ATR波动率过滤、风险仓位管理等完整逻辑。每一部分都有详细注释说明。
# =====================================================================
# 第一部分：导入和全局常量
# =====================================================================
from panda_backtest.api.api import * # 必须引入，包含下单函数、MarketOrderStyle等
import panda_data # 数据查询API，用于查主力合约和合约乘数
import numpy as np # 数值计算
# 运行模式开关（整个文件只需改这一行即可切换模式）
# 'backtest' : 性能模式
# 'live' : 通用模式
MODE = 'backtest'
# 交易品种列表（全局常量，方便统一管理）
PRODUCTS = ['AU', 'AG', 'CU', 'AL', 'ZN', 'RB', 'HC', 'I', 'BU', 'TA', 'PP', 'MA', 'M', 'P', 'SR']
# =====================================================================
# 第二部分：工具函数
# =====================================================================
def _safe_date(d):
"""将各种日期格式统一转为 YYYYMMDD 字符串
context.now 可能返回 '20250101' 或 '2025-01-01' 等格式，此函数确保统一
"""
if d is None:
return None
s = str(d).replace('-', '')
digits = ''.join(c for c in s if c.isdigit())
return digits[:8] if len(digits) >= 8 else None
# =====================================================================
# 第三部分：数据预加载（仅 backtest 模式使用）
# =====================================================================
def _preload_all_data(context):
start_date = _safe_date(context.run_info.start_date) # 回测起始日期
end_date = _safe_date(context.run_info.end_date) # 回测结束日期
context._dominant_map = {}
try:
dom_df = panda_data.get_future_dominant(
underlying_symbol=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():
d = _safe_date(row.get('date'))
sym = row.get('symbol')
if d and sym:
context._dominant_map[(row['underlying_symbol'], d)] = sym
except Exception:
pass
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:
pass
print(f"[预加载完成] 主力合约映射: {len(context._dominant_map)}天, "
f"合约乘数: {len(context._mul_map)}个合约")
# =====================================================================
# 第四部分：策略初始化（只在策略启动时执行一次）
# =====================================================================
def initialize(context):
"""策略初始化：设置账户、参数、缓存变量，backtest 模式下预加载数据"""
context.account = '5588' # 期货账号（回测固定5588，仿真会自动替换）
context.mode = MODE # 将全局 MODE 赋值到 context，供 before_trading 判断
# ===== 策略参数（根据策略修改） =====
context.mom_days = 10 # 动量计算周期（天数）
context.atr_period = 14 # ATR 计算周期
context.atr_expand_ratio = 1.1 # ATR 放大阈值（当前ATR > 均值ATR × 此比率 → 波动放大）
context.mom_threshold = 0.015 # 动量阈值（涨跌幅超过1.5%才产生信号）
context.risk_ratio = 0.01 # 每次下单风险比例（账户总权益的1%）
context.margin_ratio = 0.10 # 保证金比例
context.capital_use_ratio = 0.6 # 资金使用率上限
context.max_len = 60 # 历史数据最大保留长度
# ===== 以下变量名固定不可改（MODE 模式必须） =====
context.today_dominant = {} # 当日主力合约 {品种: symbol}，before_trading 每天刷新
context.contract_mul = {} # 当日合约乘数 {symbol: 乘数}，before_trading 每天刷新
# ===== 策略运行状态（用于在 handle_data 中跨 bar 维护历史数据） =====
context.close_hist = {} # 收盘价历史 {symbol: [float]}
context.high_hist = {} # 最高价历史 {symbol: [float]}
context.low_hist = {} # 最低价历史 {symbol: [float]}
# ===== 预加载缓存（backtest 模式下由 _preload_all_data 填充） =====
context._dominant_map = {} # {(品种, 日期): symbol}
context._mul_map = {} # {symbol: 乘数}
# backtest 模式：启动时一次性预加载全区间数据
if context.mode == 'backtest':
_preload_all_data(context)
# =====================================================================
# 第五部分：开盘前数据准备（每个交易日执行一次）
# =====================================================================
def _before_trading_backtest(context, today):
"""【backtest 模式】从内存字典查表，零网络请求
将预加载的全区间数据中「今天」的部分写入 today_dominant 和 contract_mul
"""
for product in PRODUCTS:
symbol = context._dominant_map.get((product, today))
if symbol:
context.today_dominant[product] = symbol
context.contract_mul[symbol] = context._mul_map.get(symbol, 1.0)
def _before_trading_live(context, today):
"""【live 模式】每天查询 panda_data 获取当日主力合约和乘数
兼容回测、仿真、实盘三种环境
"""
try:
dom_df = panda_data.get_future_dominant(
underlying_symbol=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):
"""开盘前运行：按 MODE 分发到不同的数据获取逻辑，然后订阅合约"""
today = _safe_date(getattr(context, 'now', None))
if not today:
return
# 每天重置，因为主力合约可能换月
context.today_dominant = {}
context.contract_mul = {}
# 按模式分发
if context.mode == 'backtest':
_before_trading_backtest(context, today)
else:
_before_trading_live(context, today)
# 订阅当日主力合约行情（换月后需订阅新合约）
symbols = list(context.today_dominant.values())
if symbols:
try:
sub_future_symbol(symbols)
except Exception:
pass
# =====================================================================
# 第六部分：策略核心逻辑（每根 bar 执行，两种模式完全相同）
# =====================================================================
def handle_data(context, data):
"""策略核心：每根 bar 执行
"""
# 遍历当日所有主力合约（由 before_trading 每天更新）
for product, symbol in context.today_dominant.items():
# 获取当前 bar 行情
try:
bar = data[symbol]
except Exception:
continue
if not bar or bar.close <= 0:
continue
close, high, low = float(bar.close), float(bar.high), float(bar.low)
mul = context.contract_mul.get(symbol, 1.0) # 从缓存读取合约乘数
# ----- 维护历史价格数据（跨 bar 累积） -----
if symbol not in context.close_hist:
context.close_hist[symbol] = []
context.high_hist[symbol] = []
context.low_hist[symbol] = []
context.close_hist[symbol].append(close)
context.high_hist[symbol].append(high)
context.low_hist[symbol].append(low)
# 限制历史长度，避免内存无限增长
if len(context.close_hist[symbol]) > context.max_len:
context.close_hist[symbol].pop(0)
context.high_hist[symbol].pop(0)
context.low_hist[symbol].pop(0)
closes = context.close_hist[symbol]
highs = context.high_hist[symbol]
lows = context.low_hist[symbol]
# 数据不足时跳过（需要 mom_days + atr_period + 缓冲）
if len(closes) < context.mom_days + context.atr_period + 5:
continue
# ----- 计算动量 -----
prev_close = closes[-context.mom_days - 1]
mom = (close - prev_close) / prev_close if prev_close > 0 else 0
# ----- 计算 ATR（平均真实波幅） -----
n = len(closes)
trs = []
for i in range(n - context.atr_period, n):
h, l_ = highs[i], lows[i]
c1 = closes[i - 1] if i > 0 else closes[i]
trs.append(max(h - l_, abs(h - c1), abs(l_ - c1)))
atr_val = float(np.mean(trs)) if trs else 0.0
# ----- 计算 ATR 均值（判断波动是否放大） -----
if len(closes) > context.atr_period + 20:
atr_hist = []
for j in range(context.atr_period, len(closes)):
t = []
for i in range(j - context.atr_period, j):
h, l_ = highs[i], lows[i]
c1 = closes[i - 1] if i > 0 else closes[i]
t.append(max(h - l_, abs(h - c1), abs(l_ - c1)))
atr_hist.append(float(np.mean(t)))
atr_ma = float(np.mean(atr_hist[-20:]))
else:
atr_ma = atr_val
# 波动放大信号：当前 ATR > 历史 ATR 均值 × 放大比率
vol_expand = atr_val > atr_ma * context.atr_expand_ratio if atr_ma > 0 else False
if atr_val <= 0:
continue
# ----- 计算下单手数（基于风险管理） -----
fut_acct = context.future_account_dict.get(context.account)
if not fut_acct:
continue
total_value = fut_acct.total_value
risk_capital = total_value * context.risk_ratio # 单次风险金额
stop_distance = max(atr_val * mul * 2, close * mul * 0.01) # 止损距离
risk_lots = risk_capital / stop_distance # 基于风险的手数
capital_per_symbol = total_value / len(PRODUCTS) # 每品种分配资金
margin_per_lot = close * mul * context.margin_ratio # 每手保证金
max_lots = int((capital_per_symbol * context.capital_use_ratio) / margin_per_lot) if margin_per_lot > 0 else 1
lots = max(1, int(min(risk_lots, max_lots))) # 取较小值，至少1手
# ----- 获取当前持仓 -----
pos = fut_acct.positions[symbol]
long_qty = pos.buy_quantity
short_qty = pos.sell_quantity
# ----- 交易信号判断与下单 -----
# 已有多头 → 动量反转时平仓
if long_qty > 0:
if mom < -context.mom_threshold:
try:
sell_close(context.account, symbol, pos.closable_buy_quantity, style=MarketOrderStyle)
except Exception:
pass
continue
# 已有空头 → 动量反转时平仓
if short_qty > 0:
if mom > context.mom_threshold:
try:
buy_close(context.account, symbol, pos.closable_sell_quantity, style=MarketOrderStyle)
except Exception:
pass
continue
# 无持仓 → 波动放大 + 动量突破时开仓
if vol_expand and mom > context.mom_threshold:
try:
buy_open(context.account, symbol, lots, style=MarketOrderStyle)
except Exception:
pass
elif vol_expand and mom < -context.mom_threshold:
try:
sell_open(context.account, symbol, lots, style=MarketOrderStyle)
except Exception:
pass
# =====================================================================
# 第七部分：收盘后汇总（每天只执行一次，适合打印日志）
# =====================================================================
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}个")
策略结构总结
上述代码分为七个部分，每个期货策略都遵循这个结构：
| 部分 | 内容 | 说明 |
|---|---|---|
| 第一部分 | 导入和全局常量 | MODE 开关、PRODUCTS 品种列表 |
| 第二部分 | 工具函数 | _safe_date 等辅助函数 |
| 第三部分 | _preload_all_data | backtest 模式预加载全区间数据 |
| 第四部分 | initialize | 账户、参数、缓存变量初始化 |
| 第五部分 | before_trading 系列 | 按 MODE 分发，更新当日主力和乘数 |
| 第六部分 | handle_data | 策略核心逻辑，两种模式完全一样 |
| 第七部分 | after_trading | 盘后日志汇总 |
如何基于此模板编写自己的策略：
- 修改
PRODUCTS列表为你关注的品种 - 修改
initialize中的策略参数 - 修改
handle_data中的信号计算和下单逻辑 - 其他部分（
_preload_all_data、before_trading、after_trading）保持不变
变量命名规范
| 变量 | 含义 | 赋值位置 |
|---|---|---|
| products | 品种列表 [‘AU’] | initialize |
| today_dominant | 当日主力 {品种: symbol} | before_trading 每天更新 |
| contract_mul | 当日乘数 {symbol: float} | before_trading 每天更新 |
| _dominant_map | 预加载主力 {(品种,日期): symbol} | _preload_all_data |
| _mul_map | 预加载乘数 {symbol: float} | _preload_all_data |
策略生命周期函数
策略初始化（必选）
函数： initialize
描述： 策略初始化，主要用于初始化策略上下文中的变量，只在策略启动时运行一次。backtest 模式下同时调用 _preload_all_data 预加载全区间数据。
代码：
def initialize(context):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| context | Context对象 | 策略上下文对象 |
例子：
def initialize(context):
# 期货账号 回测时写5588 仿真盘会自动替换为真实账号
context.account = '5588'
context.mode = MODE
# 关注的期货品种
context.products = ['RB']
# 策略参数
context.short_window = 5
context.long_window = 20
# 用于保存历史价格数据
context.historical_prices = []
# 用于保存订单ID
context.order_ids = []
# 固定变量（MODE 模式必须）
context.today_dominant = {}
context.contract_mul = {}
context._dominant_map = {}
context._mul_map = {}
if context.mode == 'backtest':
_preload_all_data(context)
策略bar运行（必选）
函数： handle_data
描述： 策略bar触发运行函数，日线回测为每日调用一次，分钟则为每个交易分钟时间调用一次。注意：handle_data 中禁止调用 panda_data 等网络请求，禁止高频 print。
运行时间说明：
当有基金交易时，分为普通交易和所有时间交易，分钟运行时间参考下表：
| 策略类型 | 运行时间 |
|---|---|
| 股票 | 9:30 ~ 15:00 |
| 期货 | 9:00 ~ 15:00 |
代码：
def handle_data(context, data):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| context | Context对象 | 策略上下文对象 |
| data | Bar对象 | bar行情字典对象 |
例子：
def handle_data(context, data):
# 获取期货账户信息
futures_account = context.future_account_dict.get(context.account)
if not futures_account:
return
# 遍历当日主力合约（MODE 模式标准写法）
for product, symbol in context.today_dominant.items():
try:
bar = data[symbol]
except Exception:
continue
if not bar or bar.close <= 0:
continue
close_price = bar.close
open_price = bar.open
high_price = bar.high
low_price = bar.low
volume = bar.volume
# 从缓存获取合约乘数（禁止调用 panda_data）
mul = context.contract_mul.get(symbol, 1.0)
# 获取持仓信息
positions = futures_account.positions
if symbol in list(positions.keys()):
position = positions[symbol]
buy_quantity = position.buy_quantity # 多头持仓
sell_quantity = position.sell_quantity # 空头持仓
# 策略核心逻辑（计算指标、判断信号、下单等）
# buy_open(context.account, symbol, 1, style=MarketOrderStyle)
策略开盘前（可选）
函数： before_trading
描述： 策略开盘前运行函数。使用 MODE 模式时，此函数按 context.mode 分发到 _before_trading_backtest（内存查表）或 _before_trading_live（网络查询）。
注意：该函数只在交易日调用，分钟回测调用时间参考下表
运行时间：
| 策略类型 | 运行时间 |
|---|---|
| 股票 | 8:30 |
| 期货 | 20:30 |
代码：
def before_trading(context):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| context | Context对象 | 策略上下文对象 |
例子：
def before_trading(context):
today = str(context.now)
context.today_dominant = {}
context.contract_mul = {}
if context.mode == 'backtest':
# backtest 模式：从预加载的内存字典查表，零网络请求
for product in context.products:
symbol = context._dominant_map.get((product, today))
if symbol:
context.today_dominant[product] = symbol
context.contract_mul[symbol] = context._mul_map.get(symbol, 1.0)
else:
# live 模式：每天查 panda_data
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
# 订阅当日主力合约
symbols = list(context.today_dominant.values())
if symbols:
sub_future_symbol(symbols)
策略收盘后（可选）
函数： after_trading
描述： 策略收盘后运行函数。日志只在此函数中打印，禁止在 handle_data 中高频打印。
注意：该函数只在交易日调用，调用时间为15:30
代码：
def after_trading(context):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| context | Context对象 | 策略上下文对象 |
例子：
def after_trading(context):
# 盘后汇总（每天只执行一次，适合打印日志）
futures_account = context.future_account_dict.get(context.account)
if futures_account:
pos_count = sum(1 for p in futures_account.positions.values()
if p.buy_quantity > 0 or p.sell_quantity > 0)
print(f"[{context.now}] 权益={futures_account.total_value:.0f} 持仓={pos_count}个 盈亏={futures_account.holding_pnl:.0f}")
回调函数
股票报单回报（可选）
函数： on_stock_trade_rtn
描述： 当有报单委托成交后触发
代码：
def on_stock_trade_rtn(context, order):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| context | Context对象 | 策略上下文对象 |
| order | Order对象 | 订单信息 |
例子：
def on_stock_trade_rtn(context, order):
print("【股票报单回报】")
print(f"订单ID: {order.order_id}")
print(f"合约: {order.order_book_id}")
print(f"买卖方向: {order.side} (1:买, 2:卖)")
print(f"订单价格: {order.price}")
print(f"下单数量: {order.quantity}")
print(f"已成交数量: {order.filled_quantity}")
print(f"剩余数量: {order.unfilled_quantity}")
print(f"订单状态: {order.status} (1:未成交, 2:已成交, 3:已撤, -1:拒单)")
# 如果订单已成交，更新持仓信息
if order.status == 2:
stock_account = context.stock_account_dict.get(context.account)
if stock_account and order.order_book_id in stock_account.positions:
position = stock_account.positions[order.order_book_id]
print(f"当前持仓数量: {position.quantity}")
print(f"持仓市值: {position.market_value}")
股票撤单回报（可选）
函数： stock_order_cancel
描述： 当有报单委托被撤单后触发
代码：
def stock_order_cancel(context, order):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| context | Context对象 | 策略上下文对象 |
| order | Order对象 | 订单信息 |
例子：
def stock_order_cancel(context, order):
print("【股票撤单回报】")
print(f"订单ID: {order.order_id}")
print(f"合约: {order.order_book_id}")
print(f"订单状态: {order.status} (3:已撤)")
# 从订单列表中移除
if hasattr(context, 'order_ids') and order.order_id in context.order_ids:
context.order_ids.remove(order.order_id)
print(f"订单 {order.order_id} 已从列表中移除")
期货报单回报（可选）
函数： on_future_trade_rtn
描述： 当有报单委托成交后触发
代码：
def on_future_trade_rtn(context, order):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| context | Context对象 | 策略上下文对象 |
| order | Order对象 | 订单信息 |
例子：
def on_future_trade_rtn(context, order):
print("【期货报单回报】")
print(f"订单ID: {order.order_id}")
print(f"合约: {order.order_book_id}")
print(f"买卖方向: {order.side} (1:买, 2:卖)")
print(f"开平方向: {order.effect} (0:开, 1:平)")
print(f"订单价格: {order.price}")
print(f"下单数量: {order.quantity}")
print(f"已成交数量: {order.filled_quantity}")
print(f"剩余数量: {order.unfilled_quantity}")
print(f"订单状态: {order.status} (1:未成交, 2:已成交, 3:已撤, -1:拒单)")
print(f"订单信息: {order.message}")
# 如果订单已成交，更新持仓信息
if order.status == 2:
futures_account = context.future_account_dict.get(context.account)
if futures_account and order.order_book_id in futures_account.positions:
position = futures_account.positions[order.order_book_id]
print(f"当前多头持仓: {position.buy_quantity}")
print(f"当前空头持仓: {position.sell_quantity}")
print(f"多头盈亏: {position.buy_pnl}")
print(f"空头盈亏: {position.sell_pnl}")
# 从订单ID列表中移除已成交订单
if hasattr(context, 'order_ids') and order.order_id in context.order_ids:
context.order_ids.remove(order.order_id)
print(f"订单 {order.order_id} 已成交，从订单列表中移除")
期货撤单回报（可选）
函数： future_order_cancel
描述： 当有报单委托被撤单后触发
代码：
def future_order_cancel(context, order):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| context | Context对象 | 策略上下文对象 |
| order | Order对象 | 订单信息 |
例子：
def future_order_cancel(context, order):
print("【期货撤单回报】")
print(f"订单ID: {order.order_id}")
print(f"合约: {order.order_book_id}")
print(f"买卖方向: {order.side} (1:买, 2:卖)")
print(f"开平方向: {order.effect} (0:开, 1:平)")
print(f"订单状态: {order.status} (3:已撤)")
# 从订单ID列表中移除
if hasattr(context, 'order_ids') and order.order_id in context.order_ids:
context.order_ids.remove(order.order_id)
print(f"订单 {order.order_id} 已撤单，从订单列表中移除")
事件拓展
系统支持自定义事件，详细参考开源代码 src/panda_backtest/backtest_common/system/event/event.py
基础对象说明
context对象
对象： context
描述： 全局上下文对象，可在基础函数中传递，同时会内置数据对象
代码：
context.变量
内置变量：
| 字段 | 类型 | 描述 |
|---|---|---|
| now | str | 当前日期（yyyMMdd） |
| trade_date | str | 交易日期 |
| trade_time | str | 交易时间 |
| portfolio_dict | dict | 收益信息字典（key为account，value为Portfolio对象） |
| stock_account_dict | dict | 股票账户信息字典（key为account，value为StockAccount对象） |
| future_account_dict | dict | 期货账户信息字典（key为account，value为FutureAccount对象） |
| sub_future_symbol_list | set | 订阅的期货合约列表 |
| sub_stock_symbol_list | set | 订阅的股票合约列表 |
| sub_strategy_future_symbol_list | set | 策略订阅的期货合约列表 |
| sub_strategy_stock_symbol_list | set | 策略订阅的股票合约列表 |
例子：
# 获取当前日期
current_date = context.now
# 获取股票账户
stock_account = context.stock_account_dict['15032863']
# 获取期货账户
future_account = context.future_account_dict['5588']
# 获取订阅的期货合约列表
subscribed_futures = context.sub_future_symbol_list
# 遍历所有期货账户
for account_id, future_account in context.future_account_dict.items():
print(f"账户 {account_id} 总权益: {future_account.total_value}")
Bar对象
对象： Bar
描述： 当前bar行情对象
代码：
data[合约]
# 或
bar_dict[合约]
内置变量：
| 字段 | 类型 | 描述 |
|---|---|---|
| symbol | str | 合约 |
| open | double | 开盘价 |
| high | double | 最高价 |
| low | double | 最低价 |
| close | double | 收盘价 |
| settle | double | 结算价（期货） |
| last | double | 最新价 |
| volume | long | 成交量 |
| oi | long | 持仓量（期货） |
| turnover | double | 成交金额 |
例子：
def handle_data(context, data):
# 获取平安银行当前bar收盘价
stock_bar = data['000001.SZ']
close_price = stock_bar.close
open_price = stock_bar.open
high_price = stock_bar.high
low_price = stock_bar.low
volume = stock_bar.volume
# 获取黄金2002合约当前bar信息
future_bar = data['AU2002.SHF']
close_price = future_bar.close
settle_price = future_bar.settle # 结算价
last_price = future_bar.last # 最新价
volume = future_bar.volume # 成交量
oi = future_bar.oi # 持仓量
turnover = future_bar.turnover # 成交金额
# 遍历所有订阅的合约
if hasattr(data, 'future_real_time_bar_map'):
future_map = data.future_real_time_bar_map
for symbol in future_map.keys():
bar = data[symbol]
print(f"合约 {symbol}: 收盘价={bar.close}, 成交量={bar.volume}")
Order对象
对象： Order
描述： 下单返回订单对象
代码：
order = order_shares('15032863','000001.SZ', 100)
order = buy_open('5588','AU2002.SHF', 1)
内置变量：
| 字段 | 类型 | 描述 |
|---|---|---|
| order_id | str | 订单唯一标识 |
| order_book_id | str | 下单合约 |
| side | int | 买卖方向（1：买 2：卖） |
| effect | int | 开平方向（0：开，1：平，仅期货） |
| price | double | 订单价格，只有在订单类型为’限价单’的时候才有意义 |
| quantity | int | 下单数量 |
| filled_quantity | int | 订单已成交数量 |
| unfilled_quantity | int | 订单剩余数量 |
| status | int | 订单状态（1：未成交，2：已成交，3：已撤，-1：拒单） |
| message | str | 订单信息 |
例子：
def handle_data(context, data):
# 股票下单
order = order_shares('15032863','000001.SZ', 100, style=MarketOrderStyle)
if order:
print(f"订单ID: {order.order_id}")
print(f"订单状态: {order.status}")
# 保存订单ID用于后续撤单
context.order_ids.append(order.order_id)
# 期货下单
order = buy_open('5588','AU2002.SHF', 1, style=MarketOrderStyle)
if order:
print(f"订单ID: {order.order_id}")
print(f"买卖方向: {order.side} (1:买, 2:卖)")
print(f"开平方向: {order.effect} (0:开, 1:平)")
print(f"已成交数量: {order.filled_quantity}")
# 检查订单是否完全成交
if order.status == 2 and order.filled_quantity == order.quantity:
print("订单已完全成交")
StockAccount对象
对象： StockAccount
描述： 股票账号信息实体对象
代码：
context.stock_account_dict['15032863']
内置变量：
| 字段 | 类型 | 描述 |
|---|---|---|
| total_value | double | 总权益 |
| cash | double | 可用资金 |
| frozen_cash | double | 冻结资金 |
| market_value | double | 持仓市值 |
| positions | dict | 持仓字典（key为合约，value为StockPositions对象） |
例子：
def handle_data(context, data):
# 获取股票账号
stock_account = context.stock_account_dict.get('15032863')
if stock_account:
# 获取账户基本信息
total_value = stock_account.total_value # 总权益
cash = stock_account.cash # 可用资金
frozen_cash = stock_account.frozen_cash # 冻结资金
market_value = stock_account.market_value # 持仓市值
print(f"账户总权益: {total_value}")
print(f"可用资金: {cash}")
print(f"冻结资金: {frozen_cash}")
print(f"持仓市值: {market_value}")
# 遍历所有持仓
positions = stock_account.positions
if isinstance(positions, dict):
print(f"持仓数量: {len(positions)}")
for symbol, position in positions.items():
print(f"\n持仓合约: {symbol}")
print(f" 持仓数量: {position.quantity}")
print(f" 可卖数量: {position.sellable}")
print(f" 持仓市值: {position.market_value}")
print(f" 持仓均价: {position.avg_price}")
print(f" 持仓盈亏: {position.pnl}")
# 检查特定股票的持仓
if '000001.SZ' in stock_account.positions:
position = stock_account.positions['000001.SZ']
if position.quantity > 0:
print(f"平安银行持仓: {position.quantity} 股")
print(f"持仓盈亏: {position.pnl}")
StockPosition对象
对象： StockPositions
描述： 股票持仓对象
代码：
context.stock_account_dict['15032863'].positions['000001.SZ']
内置变量：
| 字段 | 类型 | 描述 |
|---|---|---|
| order_book_id | str | 合约 |
| quantity | int | 持仓数量 |
| sellable | int | 可卖数量 |
| market_value | double | 持仓市值 |
| avg_price | double | 持仓均价 |
| pnl | double | 持仓盈亏 |
例子：
def handle_data(context, data):
stock_account = context.stock_account_dict.get('15032863')
if stock_account and '000001.SZ' in stock_account.positions:
position = stock_account.positions['000001.SZ']
# 获取持仓详细信息
symbol = position.order_book_id # 合约代码
quantity = position.quantity # 持仓数量
sellable = position.sellable # 可卖数量
market_value = position.market_value # 持仓市值
avg_price = position.avg_price # 持仓均价
pnl = position.pnl # 持仓盈亏
print(f"合约: {symbol}")
print(f"持仓数量: {quantity}")
print(f"可卖数量: {sellable}")
print(f"持仓市值: {market_value}")
print(f"持仓均价: {avg_price}")
print(f"持仓盈亏: {pnl}")
# 根据持仓情况决定是否卖出
if quantity > 0 and sellable > 0:
# 如果盈利超过10%，卖出50%
if pnl > 0 and (pnl / market_value) > 0.1:
sell_quantity = int(sellable * 0.5)
order_shares('15032863', '000001.SZ', -sell_quantity, style=MarketOrderStyle)
FutureAccount对象
对象： FutureAccount
描述： 期货账号信息实体对象
代码：
context.future_account_dict['5588']
内置变量：
| 字段 | 类型 | 描述 |
|---|---|---|
| total_value | double | 总权益 |
| cash | double | 可用资金 |
| frozen_cash | double | 冻结资金 |
| holding_pnl | double | 持仓盈亏 |
| realized_pnl | double | 平仓盈亏 |
| margin | double | 保证金 |
| transaction_cost | double | 手续费 |
| positions | dict | 持仓字典（key为合约，value为FuturePositions对象） |
例子：
def handle_data(context, data):
# 获取期货账户
futures_account = context.future_account_dict.get('5588')
if futures_account:
# 获取账户基本信息
total_value = futures_account.total_value # 总权益
cash = futures_account.cash # 可用资金
frozen_cash = futures_account.frozen_cash # 冻结资金
holding_pnl = futures_account.holding_pnl # 持仓盈亏
realized_pnl = futures_account.realized_pnl # 平仓盈亏
margin = futures_account.margin # 保证金
transaction_cost = futures_account.transaction_cost # 手续费
print(f"账户总权益: {total_value}")
print(f"可用资金: {cash}")
print(f"冻结资金: {frozen_cash}")
print(f"持仓盈亏: {holding_pnl}")
print(f"平仓盈亏: {realized_pnl}")
print(f"保证金: {margin}")
print(f"手续费: {transaction_cost}")
# 计算账户使用率
if total_value > 0:
margin_ratio = margin / total_value
print(f"保证金使用率: {margin_ratio:.2%}")
# 遍历所有持仓
positions = futures_account.positions
if isinstance(positions, dict):
print(f"持仓合约数量: {len(positions)}")
for symbol, position in positions.items():
print(f"\n持仓合约: {symbol}")
print(f" 多头持仓: {position.buy_quantity}")
print(f" 空头持仓: {position.sell_quantity}")
# 检查特定合约的持仓
if 'AU2002.SHF' in futures_account.positions:
position = futures_account.positions['AU2002.SHF']
if position.buy_quantity > 0:
print(f"黄金2002多头持仓: {position.buy_quantity} 手")
print(f"多头盈亏: {position.buy_pnl}")
FuturePosition对象
对象： FuturePositions
描述： 期货持仓对象
代码：
context.future_account_dict['5588'].positions['AU2601.SHF']
内置变量：
| 字段 | 类型 | 描述 |
|---|---|---|
| order_book_id | str | 合约 |
| buy_quantity | int | 多头持仓 |
| buy_today_quantity | int | 多头今日持仓 |
| closable_buy_quantity | int | 多头可平持仓 |
| buy_margin | double | 多头保证金 |
| buy_pnl | double | 多头累计收益 |
| buy_avg_open_price | double | 多头开仓均价 |
| buy_avg_holding_price | double | 多头持仓均价 |
| buy_transaction_cost | double | 多头手续费 |
| sell_quantity | int | 空头持仓 |
| sell_today_quantity | int | 空头今日持仓 |
| closable_sell_quantity | int | 空头可平持仓 |
| sell_margin | double | 空头保证金 |
| sell_pnl | double | 空头累计收益 |
| sell_avg_open_price | double | 空头开仓均价 |
| sell_avg_holding_price | double | 空头持仓均价 |
| sell_transaction_cost | double | 空头手续费 |
| pnl | double | 总盈亏 |
| daily_pnl | double | 当日盈亏 |
| holding_pnl | double | 持仓盈亏 |
| realized_pnl | double | 已实现盈亏 |
| transaction_cost | double | 总手续费 |
| margin | double | 总保证金 |
| market_value | double | 持仓市值 |
例子：
def handle_data(context, data):
futures_account = context.future_account_dict.get('5588')
if futures_account and 'AU2002.SHF' in futures_account.positions:
position = futures_account.positions['AU2002.SHF']
# 获取合约信息
symbol = position.order_book_id
# 多头信息
buy_quantity = position.buy_quantity # 多头持仓
buy_today_quantity = position.buy_today_quantity # 多头今日持仓
closable_buy_quantity = position.closable_buy_quantity # 多头可平持仓
buy_margin = position.buy_margin # 多头保证金
buy_pnl = position.buy_pnl # 多头累计收益
buy_avg_open_price = position.buy_avg_open_price # 多头开仓均价
buy_avg_holding_price = position.buy_avg_holding_price # 多头持仓均价
buy_transaction_cost = position.buy_transaction_cost # 多头手续费
# 空头信息
sell_quantity = position.sell_quantity # 空头持仓
sell_today_quantity = position.sell_today_quantity # 空头今日持仓
closable_sell_quantity = position.closable_sell_quantity # 空头可平持仓
sell_margin = position.sell_margin # 空头保证金
sell_pnl = position.sell_pnl # 空头累计收益
sell_avg_open_price = position.sell_avg_open_price # 空头开仓均价
sell_avg_holding_price = position.sell_avg_holding_price # 空头持仓均价
sell_transaction_cost = position.sell_transaction_cost # 空头手续费
# 总体信息
total_pnl = position.pnl # 总盈亏
daily_pnl = position.daily_pnl # 当日盈亏
holding_pnl = position.holding_pnl # 持仓盈亏
realized_pnl = position.realized_pnl # 已实现盈亏
total_transaction_cost = position.transaction_cost # 总手续费
total_margin = position.margin # 总保证金
market_value = position.market_value # 持仓市值
print(f"合约: {symbol}")
print(f"\n多头信息:")
print(f" 多头持仓: {buy_quantity}")
print(f" 多头可平: {closable_buy_quantity}")
print(f" 多头盈亏: {buy_pnl}")
print(f" 多头开仓均价: {buy_avg_open_price}")
print(f"\n空头信息:")
print(f" 空头持仓: {sell_quantity}")
print(f" 空头可平: {closable_sell_quantity}")
print(f" 空头盈亏: {sell_pnl}")
print(f" 空头开仓均价: {sell_avg_open_price}")
print(f"\n总体信息:")
print(f" 总盈亏: {total_pnl}")
print(f" 持仓盈亏: {holding_pnl}")
print(f" 总保证金: {total_margin}")
# 根据持仓情况决定是否平仓
if buy_quantity > 0 and closable_buy_quantity > 0:
# 如果多头盈利超过5%，平掉一半
if buy_pnl > 0 and (buy_pnl / (buy_avg_open_price * buy_quantity)) > 0.05:
close_quantity = int(closable_buy_quantity * 0.5)
sell_close('5588', symbol, close_quantity, style=MarketOrderStyle)
if sell_quantity > 0 and closable_sell_quantity > 0:
# 如果空头亏损超过3%，止损平仓
if sell_pnl < 0 and abs(sell_pnl / (sell_avg_open_price * sell_quantity)) > 0.03:
buy_close('5588', symbol, closable_sell_quantity, style=MarketOrderStyle)
交易函数
股票交易
指定股数下单
函数： order_shares
描述： 指定股数进行股票交易
代码：
def order_shares(account, symbol, amount, style):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 股票账号 |
| symbol | str | 股票合约 |
| amount | int | 股数（正数代表买入，负数代表卖出） |
| style | enum | 订单类型, MarketOrderStyle=市价单, LimitOrderStyle=限价单 |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| order | Order对象 | 订单对象 |
例子：
def handle_data(context, data):
# 按照市价最新价买入100股平安银行
order = order_shares('15032863','000001.SZ', 100, style=MarketOrderStyle)
if order:
print(f"买入订单ID: {order.order_id}")
context.order_ids.append(order.order_id)
# 按照市价最新价卖出100股平安银行
order = order_shares('15032863','000001.SZ', -100, style=MarketOrderStyle)
if order:
print(f"卖出订单ID: {order.order_id}")
# 按照12.89价格买入100股平安银行（限价单）
order = order_shares('15032863','000001.SZ', 100, style=LimitOrderStyle(12.89))
if order:
print(f"限价买入订单ID: {order.order_id}")
# 限价单可以撤单
if order.status == 1: # 未成交
cancel_order('15032863', order.order_id)
按照目标持仓下单
函数： target_stock_group_order
描述： 按照目标持仓下单，在1分钟内，以最小代价，将当前持仓改为目标持仓
代码：
def target_stock_group_order(account, symbol_dict):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 股票账号 |
| symbol_dict | dict | 股票合约和股数 （{“000001.SZ”:100}） |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| order | Order对象 | 订单对象 |
例子：
def handle_data(context, data):
# 平掉当前持仓，买入中国平安100股
target_dict = {'000001.SZ': 100}
order = target_stock_group_order('15032863', target_dict)
if order:
print(f"目标持仓下单成功，订单ID: {order.order_id}")
# 多只股票目标持仓
target_dict = {
'000001.SZ': 1000, # 平安银行1000股
'600000.SH': 500, # 浦发银行500股
'000002.SZ': 200 # 万科A 200股
}
order = target_stock_group_order('15032863', target_dict)
撤单
函数： cancel_order
描述： 股票撤单，一般只用于限价单挂单，市价单为即成即撤无法撤单
代码：
def cancel_order(account, order_id):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 股票账号 |
| order_id | str | 订单id |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| result | bool | 是否撤单成功 |
例子：
def handle_data(context, data):
# 下限价单
order = order_shares('15032863','000001.SZ', 100, style=LimitOrderStyle(12.50))
if order:
context.order_ids.append(order.order_id)
# 对订单进行撤单
if context.order_ids:
for order_id in context.order_ids[:]: # 使用切片复制列表
result = cancel_order('15032863', order_id)
if result:
print(f"撤单成功，订单ID: {order_id}")
context.order_ids.remove(order_id)
else:
print(f"撤单失败，订单ID: {order_id}")
期货交易
买入开仓
函数： buy_open
描述： 期货买入开仓
代码：
def buy_open(account, symbol, amount, style):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 期货账号 |
| symbol | str | 期货合约 |
| amount | int | 手数 |
| style | enum | 订单类型, MarketOrderStyle=市价单, LimitOrderStyle=限价单 |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| order | Order对象 | 订单对象 |
例子：
def handle_data(context, data):
# 按照市价最新价开仓买入1手AG2002
order = buy_open('5588','AG2002.SHF', 1, style=MarketOrderStyle)
if order:
print(f"买入开仓订单ID: {order.order_id}")
print(f"订单状态: {order.status}")
context.order_ids.append(order.order_id)
# 按照4280价格开仓买入1手AG2002（限价单）
current_price = data['AG2002.SHF'].close
limit_price = current_price * 0.99 # 低于当前价1%
order = buy_open('5588','AG2002.SHF', 1, style=LimitOrderStyle(limit_price))
if order:
print(f"限价买入开仓订单ID: {order.order_id}")
print(f"限价: {limit_price}")
卖出开仓
函数： sell_open
描述： 期货卖出开仓
代码：
def sell_open(account, symbol, amount, style):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 期货账号 |
| symbol | symbol | 期货合约 |
| amount | int | 手数 |
| style | enum | 订单类型, MarketOrderStyle=市价单, LimitOrderStyle=限价单 |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| order | Order对象 | 订单对象 |
例子：
def handle_data(context, data):
# 按照市价最新价开仓卖出1手AG2002
order = sell_open('5588','AG2002.SHF', 1, style=MarketOrderStyle)
if order:
print(f"卖出开仓订单ID: {order.order_id}")
context.order_ids.append(order.order_id)
# 按照4280价格开仓卖出1手AG2002（限价单）
current_price = data['AG2002.SHF'].close
limit_price = current_price * 1.01 # 高于当前价1%
order = sell_open('5588','AG2002.SHF', 1, style=LimitOrderStyle(limit_price))
if order:
print(f"限价卖出开仓订单ID: {order.order_id}")
买入平仓
函数： buy_close
描述： 期货买入平仓，即平空头仓位
代码：
def buy_close(account, symbol, amount, style):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 期货账号 |
| symbol | str | 期货合约 |
| amount | int | 手数 |
| style | enum | 订单类型, MarketOrderStyle=市价单, LimitOrderStyle=限价单 |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| order | Order对象 | 订单对象 |
例子：
def handle_data(context, data):
# 获取账户和持仓信息
futures_account = context.future_account_dict.get('5588')
if futures_account and 'AG2002.SHF' in futures_account.positions:
position = futures_account.positions['AG2002.SHF']
# 如果有空头持仓，平掉空头
if position.sell_quantity > 0:
closable_quantity = position.closable_sell_quantity
if closable_quantity > 0:
# 按照市价最新价平仓空头
order = buy_close('5588','AG2002.SHF', closable_quantity, style=MarketOrderStyle)
if order:
print(f"买入平仓订单ID: {order.order_id}")
print(f"平仓数量: {closable_quantity} 手")
# 限价平仓示例
current_price = data['AG2002.SHF'].close
limit_price = current_price * 0.99 # 低于当前价1%平仓
order = buy_close('5588','AG2002.SHF', 1, style=LimitOrderStyle(limit_price))
if order:
print(f"限价买入平仓订单ID: {order.order_id}")
卖出平仓
函数： sell_close
描述： 期货卖出平仓，即平多头仓位
代码：
def sell_close(account, symbol, amount, style):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 期货账号 |
| symbol | str | 期货合约 |
| amount | int | 手数 |
| style | enum | 订单类型, MarketOrderStyle=市价单, LimitOrderStyle=限价单 |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| order | Order对象 | 订单对象 |
例子：
def handle_data(context, data):
# 获取账户和持仓信息
futures_account = context.future_account_dict.get('5588')
if futures_account and 'AG2002.SHF' in futures_account.positions:
position = futures_account.positions['AG2002.SHF']
# 如果有多头持仓，平掉多头
if position.buy_quantity > 0:
closable_quantity = position.closable_buy_quantity
if closable_quantity > 0:
# 按照市价最新价平仓多头
order = sell_close('5588','AG2002.SHF', closable_quantity, style=MarketOrderStyle)
if order:
print(f"卖出平仓订单ID: {order.order_id}")
print(f"平仓数量: {closable_quantity} 手")
# 限价平仓示例
current_price = data['AG2002.SHF'].close
limit_price = current_price * 1.01 # 高于当前价1%平仓
order = sell_close('5588','AG2002.SHF', 1, style=LimitOrderStyle(limit_price))
if order:
print(f"限价卖出平仓订单ID: {order.order_id}")
按照目标持仓下单
函数： target_future_group_order
描述： 按照目标持仓下单，在1分钟内，以最小代价，将当前持仓改为目标持仓
代码：
def target_future_group_order(account, long_symbol_dict, short_symbol_dict):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 期货账号 |
| long_symbol_dict | dict | 多头期货合约和手数 （{“AG2509.SHF”:1}） |
| short_symbol_dict | dict | 空头期货合约和手数 （{“AG2509.SHF”:1}） |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| order | Order对象 | 订单对象 |
例子：
def handle_data(context, data):
# 平掉当前持仓，建立多头AG2505.SHF 1手，空头A2505.DCE 1手
long_dict = {"AG2505.SHF": 1}
short_dict = {"A2505.DCE": 1}
order = target_future_group_order('5588', long_dict, short_dict)
if order:
print(f"目标持仓下单成功，订单ID: {order.order_id}")
# 多合约目标持仓
long_dict = {
"AG2505.SHF": 2, # 黄金多头2手
"AU2506.SHF": 1 # 白银多头1手
}
short_dict = {
"A2505.DCE": 1, # 豆粕空头1手
"RB2505.SHF": 1 # 螺纹钢空头1手
}
order = target_future_group_order('5588', long_dict, short_dict)
# 只建立多头，平掉所有空头
long_dict = {"AG2505.SHF": 1}
short_dict = {} # 空字典表示平掉所有空头
order = target_future_group_order('5588', long_dict, short_dict)
# 只建立空头，平掉所有多头
long_dict = {} # 空字典表示平掉所有多头
short_dict = {"A2505.DCE": 1}
order = target_future_group_order('5588', long_dict, short_dict)
期货撤单
函数： cancel_future_order
描述： 期货撤单，一般只用于限价单挂单，市价单为即成即撤无法撤单
代码：
def cancel_future_order(account, order_id):
参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| account | str | 期货账号 |
| order_id | str | 订单id |
输出参数：
| 字段 | 类型 | 描述 |
|---|---|---|
| result | bool | 是否撤单成功 |
例子：
def handle_data(context, data):
# 下限价单
current_price = data['AG2002.SHF'].close
limit_price = current_price * 0.99
order = buy_open('5588','AG2002.SHF', 1, style=LimitOrderStyle(limit_price))
if order:
context.order_ids.append(order.order_id)
# 对订单进行撤单
if context.order_ids:
for order_id in context.order_ids[:]: # 使用切片复制列表
result = cancel_future_order('5588', order_id)
if result:
print(f"撤单成功，订单ID: {order_id}")
context.order_ids.remove(order_id)
else:
print(f"撤单失败，订单ID: {order_id}")
完整案例
案例1：双均线策略（期货 - MODE模式）
描述： 基于双均线的期货交易策略，使用 MODE 双模式架构，backtest 模式下 before_trading 零网络请求
from panda_backtest.api.api import *
import panda_data
import numpy as np
MODE = 'backtest'
def _preload_all_data(context):
start_date = str(context.run_info.start_date)
end_date = str(context.run_info.end_date)
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:
pass
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:
pass
def initialize(context):
context.account = '5588'
context.mode = MODE
context.products = ['RB']
context.short_window = 5
context.long_window = 20
context.historical_prices = {}
context.today_dominant = {}
context.contract_mul = {}
context._dominant_map = {}
context._mul_map = {}
if context.mode == 'backtest':
_preload_all_data(context)
def _before_trading_backtest(context, today):
for product in context.products:
symbol = context._dominant_map.get((product, today))
if symbol:
context.today_dominant[product] = symbol
context.contract_mul[symbol] = context._mul_map.get(symbol, 1.0)
def _before_trading_live(context, today):
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):
today = str(context.now)
context.today_dominant = {}
context.contract_mul = {}
if context.mode == 'backtest':
_before_trading_backtest(context, today)
else:
_before_trading_live(context, today)
symbols = list(context.today_dominant.values())
if symbols:
sub_future_symbol(symbols)
def handle_data(context, data):
futures_account = context.future_account_dict.get(context.account)
if not futures_account:
return
for product, symbol in context.today_dominant.items():
try:
bar = data[symbol]
except Exception:
continue
if not bar or bar.close <= 0:
continue
mul = context.contract_mul.get(symbol, 1.0)
if symbol not in context.historical_prices:
context.historical_prices[symbol] = []
prices = context.historical_prices[symbol]
prices.append(bar.close)
if len(prices) < context.long_window:
continue
if len(prices) > context.long_window:
prices.pop(0)
short_ma = np.mean(prices[-context.short_window:])
long_ma = np.mean(prices)
positions = futures_account.positions
has_long = (symbol in list(positions.keys()) and positions[symbol].buy_quantity > 0)
has_short = (symbol in list(positions.keys()) and positions[symbol].sell_quantity > 0)
if not has_long and not has_short:
hands = int(futures_account.total_value * 0.05 / (bar.close * mul)) if mul > 0 else 0
if hands > 0:
if short_ma > long_ma:
try:
buy_open(context.account, symbol, hands, style=MarketOrderStyle)
except Exception:
pass
elif short_ma < long_ma:
try:
sell_open(context.account, symbol, hands, style=MarketOrderStyle)
except Exception:
pass
else:
position = positions[symbol]
if short_ma < long_ma and position.buy_quantity > 0:
try:
sell_close(context.account, symbol, position.closable_buy_quantity, style=MarketOrderStyle)
except Exception:
pass
elif short_ma > long_ma and position.sell_quantity > 0:
try:
buy_close(context.account, symbol, position.closable_sell_quantity, style=MarketOrderStyle)
except Exception:
pass
def after_trading(context):
futures_account = context.future_account_dict.get(context.account)
if futures_account:
pos_count = sum(1 for p in futures_account.positions.values()
if p.buy_quantity > 0 or p.sell_quantity > 0)
print(f"[{context.now}] 权益={futures_account.total_value:.0f} 持仓={pos_count}个")
案例2：盘后汇总账户和持仓信息
描述： 在 after_trading 中汇总打印账户和持仓信息（每天只执行一次，不影响回测性能）
注意：不要在 handle_data 中打印账户/持仓信息，分钟回测中会产生几十万条日志
from panda_backtest.api.api import *
def handle_data(context, data):
pass
def after_trading(context):
if not hasattr(context, 'future_account_dict') or not context.future_account_dict:
return
for account_id, futures_account in context.future_account_dict.items():
print(f"[{context.now}] 账户{account_id}: "
f"权益={futures_account.total_value:.0f}, "
f"可用={futures_account.cash:.0f}, "
f"保证金={futures_account.margin:.0f}, "
f"持仓盈亏={futures_account.holding_pnl:.0f}")
positions = futures_account.positions
for symbol, pos in positions.items():
if pos.buy_quantity > 0 or pos.sell_quantity > 0:
print(f" {symbol}: 多头={pos.buy_quantity}手 空头={pos.sell_quantity}手 "
f"盈亏={pos.pnl:.0f}")
注意事项
-
引用路径：
- 回测环境：
from panda_backtest.api.api import * - 仿真/实盘环境：
from panda_trading.trading_common.api.api import *
- 回测环境：
-
订单类型：
MarketOrderStyle：市价单，立即成交LimitOrderStyle(price)：限价单，需要指定价格
-
持仓对象：
- positions是FuturePositions或StockPositions对象（不是dict）
- 访问时使用
list(positions.keys())获取持仓合约列表 - 遍历时直接使用
positions.items()
-
日期格式：
- 所有日期参数格式为：yyyyMMdd（如：20250101）
-
错误处理：
- 建议对所有API调用和对象访问进行异常处理
- 使用try-except捕获可能的异常
-
MODE模式使用：
- 期货策略推荐使用
MODE = 'backtest'极致性能模式 - 切换到仿真/实盘只需改
MODE = 'live'，其余代码不动 - handle_data 中禁止调用 panda_data 等网络请求，禁止高频 print
- 日志只在 after_trading 中打印
- 期货策略推荐使用
-
股票、期货账户：
- 回测股票账户使用
"15032863" - 回测期货账户使用
"5588"
- 回测股票账户使用
常见问题
Q1: 如何获取所有账户的持仓信息？
# 建议放在 after_trading 中打印，不要在 handle_data 中逐 bar 打印
for account_id, futures_account in context.future_account_dict.items():
positions = futures_account.positions
for symbol, position in positions.items():
print(f"账户{account_id} 合约{symbol}: 多头={position.buy_quantity}, 空头={position.sell_quantity}")
Q2: 如何判断是否有持仓？
futures_account = context.future_account_dict.get('5588')
if futures_account:
positions = futures_account.positions
if 'RB2601.SHF' in list(positions.keys()):
position = positions['RB2601.SHF']
has_position = position.buy_quantity > 0 or position.sell_quantity > 0
Q3: 如何计算持仓市值？
# 在 before_trading 中查合约乘数（推荐 MODE 模式，已自动缓存到 context.contract_mul）
# 在 handle_data 中使用缓存
def handle_data(context, data):
for product, symbol in context.today_dominant.items():
mul = context.contract_mul.get(symbol, 1.0)
bar = data[symbol]
futures_account = context.future_account_dict.get(context.account)
if futures_account and symbol in list(futures_account.positions.keys()):
position = futures_account.positions[symbol]
long_value = position.buy_quantity * bar.close * mul
short_value = position.sell_quantity * bar.close * mul
Q4: 如何获取主力合约？
import panda_data
# 推荐方式：在 before_trading 中查当日主力合约
def before_trading(context):
today = str(context.now)
dom_df = panda_data.get_future_dominant(
underlying_symbol=["AG"],
start_date=today,
end_date=today
)
if dom_df is not None and not dom_df.empty:
context.today_dominant = {
row['underlying_symbol']: row['symbol']
for _, row in dom_df.iterrows()
}
Q5: 如何使用 MODE 模式？
# 1. 文件顶部设置 MODE = 'backtest'（回测极致性能）或 MODE = 'live'（仿真/实盘）
# 2. initialize 中设置 context.mode = MODE，末尾调用 _preload_all_data
# 3. before_trading 中按 context.mode 分发到不同的数据获取逻辑
# 4. handle_data 两种模式完全相同，只读 context.today_dominant 和 context.contract_mul
# 5. 切换模式只需改 MODE 变量值，其余代码不动
