小果量化交易Ptrade全能助手

小果量化交易Ptrade全能助手

作者：小果（微信：xg_quant）
版本：2.1.0
适用平台：Ptrade 量化交易终端（恒生电子）
适用策略：趋势跟踪、双均线、金叉死叉、多因子、网格、动量轮动等

---

一、功能概述

本技能帮助用户在 Ptrade 量化交易终端 上高效完成从策略设计、代码编写、回测验证到实盘上线的全流程工作。
核心覆盖以下场景：

• ✅ 策略脚本生成：基于用户需求自动生成 Ptrade 可运行的 Python 策略代码
• ✅ 回测参数配置：引导设置标的、周期、手续费、滑点、资金管理等关键参数
• ✅ 回测结果分析：解读收益率、夏普比率、最大回撤、胜率、盈亏比等核心指标
• ✅ 交易日志分析：逐笔分析成交记录，定位亏损原因，优化交易逻辑
• ✅ 实盘交易指导：提供仓位管理建议、止盈止损设置、风险预警提示
• ✅ 策略优化建议：基于回测数据给出参数调优、过滤条件增强等专业建议

---

二、双均线策略（金叉死叉）快速上手

2.1 策略逻辑

• 金叉买入：短期均线（如 MA5）上穿长期均线（如 MA20），产生买入信号
• 死叉卖出：短期均线下穿长期均线，产生卖出信号
• 可选增强：可叠加成交量过滤、RSI 过滤、大盘择时等条件

2.2 Ptrade 策略代码模板（Python）

# ============================================================
# 小果量化 - Ptrade 双均线策略模板
# 作者：小果（微信：xg_quant）
# 版本：2.0
# 适用周期：日线 / 30分钟 / 60分钟
# ============================================================

def initialize(context):
    # 设置标的
    g.stock = '510300.SS'  # 沪深300ETF
    g.short_ma = 5         # 短期均线周期
    g.long_ma = 20         # 长期均线周期
    g.position_pct = 0.95  # 仓位比例
    g.stop_loss_pct = 0.05 # 止损比例（5%）
    g.take_profit_pct = 0.10 # 止盈比例（10%）
    
    # 定时任务：每日开盘后执行
    run_daily(main_strategy, time='09:35')
    # 每日收盘前检查止损止盈
    run_daily(check_stop, time='14:55')

def main_strategy(context):
    # 获取历史收盘价
    close_prices = history_bars(g.stock, g.long_ma + 5, '1d', 'close')
    if len(close_prices) < g.long_ma + 1:
        return
    
    # 计算均线
    ma_short = close_prices[-g.short_ma:].mean()
    ma_long = close_prices[-g.long_ma:].mean()
    
    # 当前持仓
    current_position = context.portfolio.positions[g.stock].quantity
    current_price = get_current_data()[g.stock].last_price
    
    # ===== 金叉买入（无持仓时） =====
    if ma_short > ma_long and current_position == 0:
        cash = context.portfolio.cash * g.position_pct
        order_value(g.stock, cash)
        log.info(f'金叉买入 {g.stock}，金额：{cash:.2f}，价格：{current_price:.3f}')
        # 记录买入价格用于止损止盈
        g.entry_price = current_price
    
    # ===== 死叉卖出（有持仓时） =====
    elif ma_short < ma_long and current_position > 0:
        order_target(g.stock, 0)
        log.info(f'死叉卖出 {g.stock}，价格：{current_price:.3f}')

def check_stop(context):
    """止损止盈检查"""
    current_position = context.portfolio.positions[g.stock].quantity
    if current_position == 0:
        return
    
    current_price = get_current_data()[g.stock].last_price
    if not hasattr(g, 'entry_price'):
        return
    
    # 止损
    if current_price < g.entry_price * (1 - g.stop_loss_pct):
        order_target(g.stock, 0)
        log.warning(f'止损触发，卖出 {g.stock}，价格：{current_price:.3f}')
    
    # 止盈
    elif current_price > g.entry_price * (1 + g.take_profit_pct):
        order_target(g.stock, 0)
        log.info(f'止盈触发，卖出 {g.stock}，价格：{current_price:.3f}')
## 二、使用步骤（四步走）

### 步骤1：新建策略
1. 登录 Ptrade 终端，点击左上角 **“策略”** 图标
2. 点击 **“添加策略”**，选择业务类型为 **“股票”**
3. 填写策略名称，例如：`小果双均线策略`
4. 点击 **“确定”**，系统会自动生成一个默认策略模板

### 步骤2：编写策略代码
1. 在策略编辑器中，**删除**默认模板的所有代码
2. 复制 `scripts/dual_ma_strategy.py` 中的完整代码粘贴进去
3. 点击 **“保存”**（快捷键：Ctrl+S）

### 步骤3：配置回测参数
1. 点击 **“新建回测”** 按钮
2. 填写以下参数：
   - **开始时间**：例如 `2023-01-01`
   - **结束时间**：例如 `2023-12-31`
   - **回测资金**：`1000000`（100万元）
   - **回测基准**：`000300.SS`（沪深300指数）
   - **回测频率**：选择 **“日线”**（分钟级别可选）
3. 点击 **“保存”** 按钮
4. 点击 **“回测”** 按钮，系统开始运行回测

### 步骤4：分析回测结果
回测完成后，查看以下面板：
- **策略评价指标**：策略收益、基准收益、Alpha比率、Beta比率、夏普比率、索提诺比率
- **收益曲线**：对比策略收益与基准收益的走势
- **交易日志**：查看每次买卖的详细信息（时间、价格、数量）
- **每日盈亏**：查看每日的盈亏明细

## 三、策略逻辑详解

### 核心逻辑
买入条件：五日均线（MA5）上穿十日均线（MA10）
卖出条件：五日均线（MA5）下穿十日均线（MA10）

### 交易标的
- **股票代码**：`60.SS`（电子）
- **股票池设置**：`set_universe(g.security)`

### 资金管理
- 买入时：**全仓买入**（使用全部可用现金）
- 卖出时：**全部卖出**（清空该股票持仓）

### 代码结构说明
| 函数 | 是否必选 | 执行时机 | 作用 |
|------|----------|----------|------|
| `initialize(context)` | 必选 | 策略启动时（仅1次） | 初始化股票池、全局变量、参数 |
| `handle_data(context, data)` | 必选 | 每分钟/每天执行 | 交易逻辑判断、下单操作 |

## 四、注意事项

### 回测注意事项
1. **股票池设置**：`set_universe()` 必须在 `initialize()` 中调用，否则 `get_history()` 无法获取数据
2. **停牌处理**：停牌日成交量为0，复盘后价格可能跳空，策略需增加过滤条件
3. **滑点设置**：建议设置 `set_slippage(0.001)` 使回测更贴近实盘
4. **佣金设置**：建议设置 `set_commission(0.0003, 5.0)` 模拟真实交易成本
5. **最小交易单位**：股票最小下单100股，可转债最小10张
6. **回测频率**：日线级别在 `15:00` 执行，分钟级别在 `9:31~15:00` 每分钟执行

### 实盘注意事项
1. **交易时间**：股票交易时间为 `9:30~15:00`，建议使用 `run_daily()` 定时执行
2. **涨跌停限制**：涨停无法买入，跌停无法卖出，下单前需判断涨停价和跌停价
3. **资金管理**：不建议全仓买入，建议分批建仓（如每次买入总资金的20%）
4. **日志记录**：使用 `log.info()` 记录关键操作，便于排查问题
5. **行情快照**：`get_snapshot(security)` 仅支持交易模块，回测中不可用

## 五、常见问题解答

### Q1：为什么回测结果显示没有交易？
**可能原因**：
- 股票池设置错误（`set_universe()` 未调用或参数错误）
- 均线周期过大，回测时间太短，没有产生金叉信号
- 股票停牌导致无法交易

**解决方法**：
- 检查 `g.security` 是否正确，确保股票代码格式为 `600570.SS`
- 缩小均线周期（如 MA5/MA10 → MA3/MA5）
- 延长回测时间（至少1年以上）

### Q2：为什么日志报错“股票不在股票池中”？
**原因**：`order()` 函数中指定的股票不在 `set_universe()` 设置的股票池中

**解决方法**：确保 `order(security, amount)` 中的 `security` 与 `set_universe()` 中的股票代码一致

### Q3：如何切换成其他股票？
**修改方法**：在 `initialize()` 函数中修改 `g.security` 的值为其他股票代码
```python
g.security = '0001.SZ'  # 改为银行
## 一、策略必备函数（必写）

```python
initialize(context)                          # 初始化（必写）
handle_data(context, data)                   # 行情/交易主函数（必写）
before_trading_start(context, data)          # 盘前运行
after_trading_end(context, data)             # 盘后运行
tick_data(context, data)                     # Tick 级别处理
on_order_response(context, order_list)       # 委托回调
on_trade_response(context, trade_list)       # 成交回调

handle_data 的第二个参数 data 为当根 K 线数据，可不使用但建议保留签名。

---

二、生成策略工作流（模型必须遵循）

1. 澄清需求：确认策略类型（择时 / 选股 / 套利 / 事件驱动）、标的范围、周期、风控要求。
2. 锁定 API：只使用本文件列出的 PTRADE 全局函数；严禁 security.xxx / trade.xxx / get_json() 等 QMT 写法。
3. 选择 / 套用模板：优先使用下方「策略模板」，或在其基础上修改。
4. 代码生成：输出完整的 initialize + handle_data，参数用模块级常量定义，便于在 PTRADE 界面调整。
5. 自检：生成的代码必须通过 scripts/validate_strategy.py 检查（无 QMT API、含 initialize/handle_data）。
6. 运行说明：附 PTRADE 平台导入 / 回测 / 实盘步骤。

---

三、PTRADE 关键约定（必读，否则策略无法运行）

• 全局函数：get_price、get_history、get_positions、get_position、get_index_stocks、get_fundamentals、order 等均为全局函数，直接调用，不要当成对象方法。
• g 全局对象：用 g.xxx = yyy 在 initialize 中保存自定义状态，在 handle_data 中读取。g 在 PTRADE 中已预定义。
• context 对象：context.portfolio.cash、context.portfolio.total_value、context.portfolio.positions、context.blotter.current_dt。
• 下单 order(security, amount, limit_price=None)：amount > 0 买入，amount < 0 卖出；limit_price=None 表示市价。每笔数量须为 100 的整数倍（1 手 = 100 股）。
• T+1：A 股当日买入不可当日卖出，卖出请用 position.enable_amount（可用持仓）。
• 行情返回：get_price(..., fields=[...], count=N) 返回 pandas DataFrame，列名为字段名；取数组用 df["close"].values。
• 回测撮合：默认以开盘价成交；使用最新价需订阅实时行情。

---

四、设置类 API

set_universe(security_list)                      # 设置股票池
set_benchmark(sids)                              # 设置基准指数
set_commission(commission_ratio, min_commission, type)  # 设置佣金
set_fixed_slippage(fixedslippage)                # 设置固定滑点
set_slippage(slippage)                           # 设置滑点比例
set_volume_ratio(volume_ratio)                   # 设置成交量限制比例
set_limit_mode(limit_mode)                       # 设置涨跌停模式
set_yesterday_position(poslist)                  # 设置昨日持仓
set_parameters(**kwargs)                         # 设置策略参数
set_email_info(email_address, smtp_code, email_subject)  # 设置邮件

---

五、定时或周期 API

run_daily(context, func, time='9:31')     # 每日定时执行
run_interval(context, func, seconds=10)   # 间隔执行（秒）

---

六、获取信息类 API

6.1 交易日

get_trading_day(day)                          # 获取第N个交易日
get_all_trades_days(date=None)                # 获取所有交易日
get_trade_days(start_date=None, end_date=None, count=None)  # 获取日期范围交易日
get_trading_day_by_date(query_date, day=0)    # 按日期查询交易日

6.2 市场信息

get_market_list()                             # 获取市场列表
get_market_detail(finance_mic)                # 获取市场详情

6.3 行情信息

# 获取历史K线（核心API）
get_history(count, frequency, field, security_list, fq=None, include=False, fill='nan', is_dict=False)
get_price(security, start_date=None, end_date=None, frequency='1d', fields=None, fq=None, count=None, is_dict=False)

# 个股委托/成交明细
get_individual_entrust(stocks=None, data_count=50, start_pos=0, search_direction=1, is_dict=False)
get_individual_transaction(stocks=None, data_count=50, start_pos=0, search_direction=1, is_dict=False)

# Tick数据
get_tick_direction(symbols=None, query_date=0, start_pos=0, search_direction=1, data_count=50, is_dict=False)

# 板块排序
get_sort_msg(sort_type_grp=None, sort_field_name=None, sort_type=1, data_count=100)

# 五档行情
get_gear_price(sids)

# 快照行情
get_snapshot(security)

# 涨跌停股票
get_trend_data(date=None, stocks=None, market=None)

6.4 证券信息

get_stock_name(stocks)                        # 获取股票名称
get_stock_info(stocks, field=None)            # 获取股票基本信息
get_stock_status(stocks, query_type='ST', query_date=None)  # 获取股票状态
get_underlying_code(symbols)                  # 获取正股代码
get_stock_exrights(stock_code, date=None)     # 获取复权因子
get_stock_blocks(stock_code)                  # 获取股票所属板块
get_index_stocks(index_code, date)            # 获取指数成分股
get_industry_stocks(industry_code)            # 获取行业成分股
get_fundamentals(security, table, fields=None, date=None, start_year=None, end_year=None, report_types=None, merge_type=None, is_dataframe=False)  # 财务数据
get_Ashares(date=None)                        # 获取全部A股
get_etf_list()                                # 获取ETF列表
get_etf_info(etf_code)                        # 获取ETF信息
get_etf_stock_list(etf_code)                  # 获取ETF成分股
get_etf_stock_info(etf_code, security)        # 获取ETF成分股信息
get_ipo_stocks()                              # 获取今日可申购新股
get_cb_list()                                 # 获取可转债列表
get_cb_info()                                 # 获取可转债信息
get_reits_list(date=None)                     # 获取REITs列表

---

七、交易 API

7.1 股票交易

order(security, amount, limit_price=None)                  # 限价/市价委托（正买负卖）
order_target(security, amount, limit_price=None)           # 目标仓位
order_value(security, value, limit_price=None)             # 目标市值
order_target_value(security, value, limit_price=None)      # 目标市值
order_market(security, amount, market_type, limit_price=None)  # 市价委托
ipo_stocks_order(submarket_type=None, black_stocks=None)   # 新股申购
after_trading_order(security, amount, entrust_price)       # 盘后委托
after_trading_cancel_order(order_param)                    # 盘后撤单
etf_basket_order(etf_code, amount, price_style=None, position=True, info=None)  # ETF篮子委托
etf_purchase_redemption(etf_code, amount, limit_price=None) # ETF申购赎回

7.2 公共交易

order_tick(sid, amount, priceGear='1', limit_price=None)   # 逐笔委托
cancel_order(order_param)                                  # 撤单
cancel_order_ex(order_param)                               # 增强撤单
debt_to_stock_order(security, amount)                      # 转债转股

7.3 融资融券

margin_trade(security, amount, limit_price=None, market_type=None)       # 融资融券
margincash_open(security, amount, limit_price=None, market_type=None, cash_group=None)   # 融资开仓
margincash_close(security, amount, limit_price=None, market_type=None, cash_group=None)  # 融资平仓
margincash_direct_refund(value, cash_group=None)                       # 融资直接还款
marginsec_open(security, amount, limit_price=None, cash_group=None)    # 融券开仓
marginsec_close(security, amount, limit_price=None, market_type=None)  # 融券平仓
marginsec_direct_refund(security, amount, cash_group=None)            # 融券直接还券

7.4 期货

buy_open(contract, amount, limit_price=None)        # 期货开多
sell_close(contract, amount, limit_price=None, close_today=False)  # 期货平多
sell_open(contract, amount, limit_price=None)       # 期货开空
buy_close(contract, amount, limit_price=None, close_today=False)    # 期货平空

---

八、查询持仓/订单/成交

get_position(security)              # 获取单只股票持仓
get_positions()                     # 获取所有持仓
get_open_orders(security=None)      # 获取未成交订单
get_order(order_id)                 # 获取订单详情
get_orders(security=None)           # 获取订单列表
get_all_orders(security=None)       # 获取所有订单
get_trades()                        # 获取成交记录
get_deliver(start_date, end_date)   # 获取交割单
get_fundjour(start_date, end_date)  # 获取资金流水
get_lucky_info(start_date, end_date) # 获取中签信息

---

九、计算/指标 API

get_MACD(close, short=12, long=26, m=9)
get_KDJ(high, low, close, n=9, m1=3, m2=3)
get_RSI(close, n=6)
get_CCI(high, low, close, n=14)

---

十、其他工具 API

log.debug/info/warning/error/critical()   # 日志输出
is_trade()                                # 是否交易时间
check_limit(security, query_date=None)    # 检查涨跌停
send_email(...)                           # 发送邮件
send_qywx(...)                            # 发送企业微信
permission_test(account=None, end_date=None)  # 权限测试
create_dir(user_path)                     # 创建目录
get_frequency()                           # 获取K线周期
get_business_type()                       # 获取业务类型
get_current_kline_count()                 # 获取当前K线根数
filter_stock_by_status(stocks, filter_type=["ST","HALT","DELISTING"], query_date=None)  # 按状态过滤
check_strategy(strategy_content=None, strategy_path=None)  # 检查策略
fund_transfer(trans_direction, occur_balance, exchange_type="1")  # 资金划转
market_fund_transfer(exchange_type, occur_balance)            # 市场间资金划转
get_research_path()                       # 获取研究模块路径
get_trade_name()                          # 获取交易账户名称
get_user_name(login_account=True)         # 获取用户名称

---

十一、数据结构（对象）

# g 全局对象（在 initialize 中使用 g.xxx = yyy 赋值）
g = {}  # 自定义全局变量（PTRADE 已预定义，可直接 g.xxx = yyy）

# Context 上下文
context.portfolio            # 资产对象
context.portfolio.cash      # 可用现金
context.portfolio.total_value  # 总资产
context.portfolio.positions # 持仓dict
context.blotter.current_dt  # 当前时间

# BarData K线数据（DataFrame 列）
# open, close, high, low, volume, price, preclose, high_limit, low_limit

# Portfolio 资产对象
context.portfolio.cash           # 可用现金
context.portfolio.total_value     # 总资金
context.portfolio.positions       # 持仓字典 {股票代码: Position}

# Position 持仓对象
position.security           # 证券代码
position.total_amount        # 总持仓数量
position.enable_amount       # 可用持仓（T+1）
position.cost_basis          # 成本
position.avg_cost            # 平均成本

# Order 订单对象
order.id                     # 订单ID
order.dt                     # 委托时间
order.limit_price            # 委托价格
order.amount                 # 委托数量
order.filled                 # 已成交数量
order.status                 # 订单状态

---

十二、数据字典（状态常量）

# 委托状态(order.status)
0   # 未报
1   # 待报
2   # 已报
3   # 已报待撤
4   # 部成待撤
5   # 部撤
6   # 已撤
7   # 部成
8   # 已成
9   # 废单

# 交易状态(trade_status)
START     # 开盘
PRETR     # 盘前
OCALL     # 集合竞价
TRADE     # 交易
HALT      # 停牌
SUSP      # 暂停
BREAK     # 熔断
POSTR     # 盘后
ENDTR     # 收盘
STOPT     # 停牌
DELISTED  # 退市

# 买卖方向
entrust_bs: 1  # 买
entrust_bs: 2  # 卖

business_direction: 0  # 买
business_direction: 1  # 卖

# frequency 参数频率
"1d"   # 日线
"1m"   # 1分钟
"5m"   # 5分钟
"15m"  # 15分钟
"30m"  # 30分钟
"60m"  # 60分钟

---

十三、策略模板

模板 1：日线双均线策略（金叉买入 / 死叉卖出，含止损止盈）

import numpy as np

SECURITY    = "000001.XSHG"
SHORT_WIN   = 5
LONG_WIN    = 20
TRADE_UNIT  = 100          # 1手=100股
PERIOD      = "1d"
STOP_LOSS   = 0.05         # 止损 5%
TAKE_PROFIT = 0.15         # 止盈 15%

def initialize(context):
    g.security = SECURITY
    g.short_win = SHORT_WIN
    g.long_win = LONG_WIN
    g.buy_price = None
    set_benchmark("000300.XSHG")
    set_universe([SECURITY])
    set_commission(0.00025, 5.0)
    set_slippage(0.002)
    log.info("双均线策略启动 | 标的=%s" % SECURITY)

def handle_data(context, data):
    sym = g.security
    df = get_price(sym, frequency=PERIOD, fields=["close"], count=g.long_win + 5)
    if df is None or len(df) < g.long_win:
        return
    close = df["close"].values.astype(float)
    short_ma = float(np.mean(close[-g.short_win:]))
    long_ma  = float(np.mean(close[-g.long_win:]))
    prev_short = float(np.mean(close[-g.short_win-1:-1]))
    prev_long  = float(np.mean(close[-g.long_win-1:-1]))
    price = float(close[-1])

    pos = get_position(sym)
    holding = int(pos.total_amount) if pos else 0

    # 止损止盈（优先）
    if holding > 0 and g.buy_price:
        pnl = (price - g.buy_price) / g.buy_price
        if pnl <= -STOP_LOSS or pnl >= TAKE_PROFIT:
            order(sym, -holding)
            g.buy_price = None
            return

    gold = (prev_short <= prev_long) and (short_ma > long_ma)
    dead = (prev_short >= prev_long) and (short_ma < long_ma)

    if gold and holding == 0:
        order(sym, TRADE_UNIT)
        g.buy_price = price
        log.info("金叉买入 %s @ %.2f" % (sym, price))
    elif dead and holding > 0:
        order(sym, -holding)
        g.buy_price = None
        log.info("死叉卖出 %s @ %.2f" % (sym, price))

完整可运行文件见 templates/dual_ma_strategy.py。

模板 2：布林带策略

import numpy as np

SECURITY = "000001.XSHG"
BOLL_PERIOD = 20
BOLL_STD = 2.0

def initialize(context):
    set_benchmark("000300.XSHG")
    set_universe([SECURITY])

def handle_data(context, data):
    sym = SECURITY
    df = get_price(sym, frequency="1d", fields=["close"], count=BOLL_PERIOD + 5)
    close = df["close"].values.astype(float)
    current = float(close[-1])
    period = close[-BOLL_PERIOD:]
    boll_mid = np.mean(period)
    boll_std = np.std(period, ddof=1)
    upper = boll_mid + BOLL_STD * boll_std
    lower = boll_mid - BOLL_STD * boll_std
    if current <= lower:
        order(sym, 100)
    elif current >= upper:
        pos = get_position(sym)
        holding = int(pos.total_amount) if pos else 0
        if holding > 0:
            order(sym, -holding)

模板 3：止损止盈策略（修复版，注意 iloc[-1] 取最新价）

STOP_LOSS = 0.05    # 止损5%
TAKE_PROFIT = 0.15  # 止盈15%

def handle_data(context, data):
    sym = "000001.XSHG"
    pos = get_position(sym)
    if pos and pos.total_amount > 0:
        df = get_price(sym, frequency="1d", fields=["close"], count=1)
        current = float(df["close"].iloc[-1])   # 取最新一根，勿用 iloc[-0]
        cost = float(pos.cost_basis)
        profit = (current - cost) / cost
        if profit <= -STOP_LOSS:
            order(sym, -int(pos.enable_amount))  # 止损（用可用持仓）
        elif profit >= TAKE_PROFIT:
            order(sym, -int(pos.enable_amount))  # 止盈

模板 4（新增）：网格交易策略

SECURITY = "000001.XSHG"
GRID_PCT = 0.02        # 每格间距 2%
UNIT     = 100

def initialize(context):
    g.security = SECURITY
    g.base = None
    set_universe([SECURITY])
    set_benchmark("000300.XSHG")

def handle_data(context, data):
    sym = g.security
    df = get_price(sym, frequency="1d", fields=["close"], count=2)
    price = float(df["close"].values[-1])
    if g.base is None:
        g.base = price
        return
    step = g.base * GRID_PCT
    if price <= g.base - step:
        order(sym, UNIT); g.base = price
        log.info("网格买入 %s @ %.2f" % (sym, price))
    elif price >= g.base + step:
        pos = get_position(sym)
        holding = int(pos.total_amount) if pos else 0
        if holding >= UNIT:
            order(sym, -UNIT); g.base = price
            log.info("网格卖出 %s @ %.2f" % (sym, price))

模板 5（新增）：二八轮动（沪深300 / 中证500 动量）

IDX_A = "000300.XSHG"   # 沪深300
IDX_B = "000905.XSHG"   # 中证500
LOOKBACK = 20

def initialize(context):
    g.idx = [IDX_A, IDX_B]
    set_universe(g.idx)

def _ret(sym):
    df = get_price(sym, frequency="1d", fields=["close"], count=LOOKBACK)
    c = df["close"].values.astype(float)
    return (c[-1] / c[0]) - 1

def handle_data(context, data):
    ra, rb = _ret(IDX_A), _ret(IDX_B)
    winner = IDX_A if ra >= rb else IDX_B
    for pos in get_positions():
        if pos.security != winner and pos.total_amount > 0:
            order(pos.security, -int(pos.total_amount))
    pos = get_position(winner)
    if (pos is None) or (pos.total_amount == 0):
        order(winner, 100)
        log.info("轮动买入 %s (300收益=%.2f%%, 500收益=%.2f%%)" % (winner, ra*100, rb*100))

模板 6（新增）：ETF 定投策略

ETF = "510300.XSHG"    # 沪深300ETF
PERIOD_DAYS = 5        # 每5个交易日定投一次
UNIT = 100

def initialize(context):
    g.etf = ETF
    g.cnt = 0
    set_universe([ETF])

def handle_data(context, data):
    g.cnt += 1
    if g.cnt % PERIOD_DAYS != 0:
        return
    order(g.etf, UNIT)
    log.info("定投买入 %s %d 股" % (g.etf, UNIT))

模板 7（新增）：多标的等权组合（因子选股）

完整文件见 templates/factor_strategy.py。核心：每月从指数成分股按估值因子选股、等权配置。

import numpy as np

INDEX = "000300.XSHG"
N_HOLD = 10
FACTOR = "PE"

def initialize(context):
    g.index = INDEX
    g.n = N_HOLD
    g.factor = FACTOR
    g.day = 0
    set_benchmark(INDEX)
    set_universe(get_index_stocks(INDEX))

def handle_data(context, data):
    g.day += 1
    if g.day % 20 != 0:
        return
    stocks = get_index_stocks(g.index)
    scores = {}
    for s in stocks:
        try:
            fd = get_fundamentals(s, "valuation", fields=[g.factor])
            if hasattr(fd, "empty") and not fd.empty:
                v = float(fd[g.factor].iloc[-1])
                if v > 0:
                    scores[s] = v
        except Exception as e:
            log.warning("财务数据获取失败 %s: %s" % (s, e))
    target = [s for s, _ in sorted(scores.items(), key=lambda x: x[1])[:g.n]]
    for pos in get_positions():
        if pos.security not in target and pos.total_amount > 0:
            order(pos.security, -int(pos.total_amount))
    for s in target:
        pos = get_position(s)
        if (pos is None) or (pos.total_amount == 0):
            order(s, 100)

---

十四、风控与最佳实践

• 必须设置股票池：set_universe(...) 决定 handle_data 能拿到的数据范围。
• 数量取整到 100：order 的 amount 必须是 100 的整数倍，否则委托会被拒。
• T+1 卖出约束：卖出数量不要超过 position.enable_amount。
• ST / 退市过滤：上线前用 filter_stock_by_status(stocks, filter_type=["ST","HALT","DELISTING"]) 剔除风险标的。
• 先回测再实盘：用 check_strategy(strategy_path=...) 做静态检查；在 PTRADE 回测通过后再开启实盘。
• 日志而非打印：用 log.info(...) 而非 print(...)，便于在 PTRADE 日志面板查看。
• 全局函数陷阱：再一次强调 —— PTRADE 没有 security.get_bars / trade.open_long / get_json，误用会直接运行失败。

---

十五、在 PTRADE 平台运行步骤

1. 打开 PTRADE → 策略研究 → 新建策略
2. 粘贴生成的代码 → 保存（Ctrl+S）
3. 回测：设置时间范围与初始资金，查看收益 / 回撤
4. 参数优化：调整模块级常量（如 SHORT_WIN / LONG_WIN）
5. 实盘：订阅实时行情，开启自动执行

---

十六、离线校验（新增）

仓库附带 scripts/validate_strategy.py，可在本地（非 PTRADE 环境）快速检查生成的策略是否误用了 QMT API：

python scripts/validate_strategy.py your_strategy.py

校验项：是否包含 initialize / handle_data、是否出现 security.get_bars / trade.open_long / get_json 等禁用写法、是否使用 PTRADE 全局函数。

交易带们参考

""" 策略名称： 三因子日线交易策略 运行周期: 日线 策略流程： 盘前将中小板成分股中st、停牌、退市的股票过滤得到股票池 盘中： 1、获取市场风险溢价、市值因子、账面市值比因子三因子数据， 2、分组差值做线性回归处理，最终得到得分，选择得分高的标的调仓买入 3、每15天换仓一次 注意事项： 策略中调用的order_target_value接口的使用有场景限制，回测可以正常使用，交易谨慎使用。 回测场景下撮合是引擎计算的，因此成交之后持仓信息的更新是瞬时的，但交易场景下信息的更新依赖于柜台数据 的返回，无法做到瞬时同步，可能造成重复下单。详细原因请看帮助文档。 """

导入函数库

import pandas as pd import numpy as np import statsmodels.api as sm from statsmodels import regression from decimal import Decimal

初始化此策略

def initialize(context): g.factor_params_info = { 'total_shareholder_equity': ['balance_statement', 'total_shareholder_equity'], 'roe': ['profit_ability', 'roe'] } set_params() # 设置策参数 set_variables() # 设置中间变量 if not is_trade(): set_backtest() # 设置回测条件

设置策参数

def set_params(): g.tc = 15 # 调仓频率 g.yb = 63 # 样本长度 g.N = 10 # 持仓数目 g.NoF = 3 # 三因子模型

设置中间变量

def set_variables(): g.t = 0 # 记录连续回测天数 g.rf = 0.04 # 无风险利率 g.if_trade = False # 当天是否交易

设置回测条件

def set_backtest(): set_limit_mode('UNLIMITED')

每天盘前处理

def before_trading_start(context, data): g.current_date = context.blotter.current_dt.strftime("%Y%m%d") # 2005-06-01前回测由于数据不足，不执行。 if g.current_date < '20050601': g.trade_flag = False else: g.trade_flag = True

g.rf = 0.04
g.all_stocks = get_index_stocks('000300.XBHS', g.current_date)

if g.t % g.tc == 0:
    # 每隔g.tc天，交易一次
    g.if_trade = True
    # 将ST、停牌、退市三种状态的股票剔除当日的股票池
    g.all_stocks = filter_stock_by_status(g.all_stocks, filter_type=["ST", "HALT", "DELISTING"], query_date=None)
g.t += 1

每天交易时要做的事情

def handle_data(context, data): if not g.trade_flag: return

if g.if_trade:
    df_scores = get_scores(g.all_stocks, str(get_trading_day(-63)), str(get_trading_day(-1)), g.rf)
    # 为每个持仓股票分配资金
    # 依打分排序，当前需要持仓的股票
    if df_scores.empty:
        stock_sort = list()
    else:
        stock_sort = df_scores.sort_values('score')['code'].tolist()
    # 把涨停状态的股票剔除
    up_limit_stock = get_limit_stock(stock_sort)['up_limit']
    # stock_sort = list(set(stock_sort)-set(up_limit_stock))
    stock_sort = [stock for stock in stock_sort if stock not in up_limit_stock]
    position_list = get_position_list(context)
    # 持仓中跌停的股票不做卖出
    limit_info = get_limit_stock(position_list)
    hold_down_limit_stock = limit_info['down_limit']
    log.info('持仓跌停股：%s' % hold_down_limit_stock)
    position_list = get_position_list(context)
    # 持仓中除了不处于前g.N且跌停不能卖的股票进行卖出
    sell_stocks = list(set(position_list) - set(stock_sort[:g.N]) - set(hold_down_limit_stock))
    # 对不在换仓列表中且飞跌停股的股票进行卖出操作
    order_stock_sell(sell_stocks)
    # 获取仍在持仓中的股票
    position_list = get_position_list(context)
    # 获取调仓买入的股票
    buy_stocks = [stock for stock in stock_sort if stock not in position_list][:(g.N - len(position_list))]
    # 仓位动态平衡的股票
    balance_stocks = list(set(buy_stocks + position_list) - set(hold_down_limit_stock))
    every_stock = context.portfolio.portfolio_value / g.N
    order_stock_balance(balance_stocks, every_stock)
    order_stock_balance(balance_stocks, every_stock)
g.if_trade = False

不在换仓目标中且没有跌停的股票进行清仓操作

def order_stock_sell(sell_stocks): # 对于不需要持仓的股票，全仓卖出 for stock in sell_stocks: order_target_value(stock, 0)

非跌停的换仓目标股进行仓位再平衡

def order_stock_balance(balance_stocks, every_stock): for stock in balance_stocks: order_target_value(stock, every_stock)

获取综合得分

def get_scores(stocks, begin, end, rf): try: length = len(stocks) market_cap_df = get_fundamentals(stocks, 'valuation', fields='total_value', date=begin) market_cap_df.dropna(inplace=True) if market_cap_df.empty: print('获取市值数据失败，股票因子评分失败') return pd.DataFrame() total_shareholder_equity_df = get_factor_values(stocks, 'total_shareholder_equity', begin, g.factor_params_info) total_shareholder_equity_df.dropna(inplace=True) if total_shareholder_equity_df.empty: print('获取total_shareholder_equity财务数据失败，股票因子评分失败') return pd.DataFrame() roe_df = get_factor_values(stocks, 'roe', begin, g.factor_params_info) roe_df.dropna(inplace=True) if roe_df.empty: print('获取roe财务数据失败，股票因子评分失败') return pd.DataFrame() df_all = pd.concat([market_cap_df, total_shareholder_equity_df, roe_df], axis=1) df_all.dropna(inplace=True) df_all['BTM'] = df_all['total_shareholder_equity'] / df_all['total_value'] df_all = df_all.reset_index() S = df_all.sort_values('total_value')['index'][:int(length / 3)] B = df_all.sort_values('total_value')['index'][length - int(length / 3):] L = df_all.sort_values('BTM')['index'][:int(length / 3)] H = df_all.sort_values('BTM')['index'][length - int(length / 3):] W = df_all.sort_values('roe')['index'][:int(length / 3)] R = df_all.sort_values('roe')['index'][length - int(length / 3):]

    close_data = get_price(stocks, begin, end, fields='close', frequency='1d', is_dict=True)

    close_df = pd.DataFrame()
    for stock_code, stock_data in close_data.items():
        date_info = pd.to_datetime(stock_data['datetime'], format='%Y%m%d')
        close_info = stock_data['close']
        close_df[stock_code] = pd.Series(close_info, index=date_info)
    close_df.sort_index(inplace=True)
    df = np.diff(np.log(close_df), axis=0) + 0 * close_df[1:]
    SMB = df[S].T.sum() / len(S) - df[B].T.sum() / len(B)
    HML = df[H].T.sum() / len(H) - df[L].T.sum() / len(L)
    RMW = df[R].T.sum() / len(R) - df[W].T.sum() / len(W)
    dp = get_price('000300.XSHG', begin, end, '1d')['close']
    if len(dp)-len(df)>1:
        log.info('历史行情数据缺失，股票因子评分失败')
        return pd.DataFrame()
    RM = np.diff(np.log(dp)) - rf / 252
    X = pd.DataFrame({"RM": RM, "SMB": SMB, "HML": HML, "RMW": RMW})
    factor_flag = ["RM", "SMB", "HML", "RMW"][:g.NoF]
    X = X[factor_flag]
    t_scores = [0.0] * length
    for i in range(length):
        t_stock = stocks[i]
        t_r = linreg(X, df[t_stock] - rf / 252, len(factor_flag))
        t_scores[i] = t_r[0]
    scores = pd.DataFrame({'code': stocks, 'score': t_scores})
    df_scores = scores.sort_values(by='score')
    return df_scores
except:
    print('股票因子评分失败，请检查数据')
    return pd.DataFrame()

获取因子值

def get_factor_values(stock_list, factor, date, factor_params_info): """ 获取因子值方法 入参： 1、股票池：stock_list 2、因子名称：factor 3、计算日期：date 4、因子数据获取需要维护的信息（因子名称、表名、字段名） """ data = get_fundamentals(stock_list, table=factor_params_info[factor][0], fields=factor_params_info[factor][1], date=date, is_dataframe=True) factor_info = {} for stock in stock_list.copy(): if stock not in data.index: continue factor_info[stock] = data.loc[stock, factor_params_info[factor][1]] if factor_info == {}: return pd.DataFrame() factor_df = pd.DataFrame.from_dict(factor_info, orient='index') factor_df.columns = [factor_params_info[factor][1]] return factor_df

线性回归

def linreg(x, y, columns=3): x = sm.add_constant(np.array(x)) y = np.array(y) if len(y) > 1: results = regression.linear_model.OLS(y, x).fit() return results.params else: return [float("nan")] * (columns + 1)

保留小数点两位

def replace(x): x = Decimal(x) x = float(str(round(x, 2))) return x

生成昨日持仓股票列表

def get_position_list(context): return [ position.sid for position in context.portfolio.positions.values() if position.amount != 0 ]

日级别回测获取持仓中不能卖出的股票(涨停就不卖出)

def get_limit_stock(stock_list): out_info = {'up_limit': [], 'down_limit': []} for stock in stock_list: limit_status = check_limit(stock)[stock] if limit_status == 1: out_info['up_limit'].append(stock) elif limit_status == -1: out_info['down_limit'].append(stock) return out_info ################################ """ 策略名称： 指数增强日线交易策略 策略流程： 盘前： 1、将沪深300成分股中st、停牌、退市的股票过滤得到股票池 2、示例用roe作为单因子选出排名第一档的股票作为目标股票池 盘中： 1、财报调仓日或者固定间隔调仓日通过线性规划的方法进行调仓，以图实现增强效果 注意事项： 策略中调用的order_target、order_target_value接口的使用有场景限制，回测可以正常使用，交易谨慎使用。 回测场景下撮合是引擎计算的，因此成交之后持仓信息的更新是瞬时的，但交易场景下信息的更新依赖于柜台数据 的返回，无法做到瞬时同步，可能造成重复下单。详细原因请看帮助文档。 """ import pandas as pd import numpy as np import statsmodels.api as sm import math from decimal import Decimal import datetime from datetime import date as justdate from scipy.optimize import minimize

初始化

def initialize(context): g.factor = 'roe' g.factor_params_info = {'roe': ['profit_ability', 'roe', False], # 净资产收益率,最后布尔值为排序方式 'operating_revenue_grow': ['growth_ability', 'operating_revenue_grow_rate', False], # 营收增速 'net_profit_grow': ['growth_ability', 'np_parent_company_cut_yoy', False], # 扣非净利润增速 } set_params() # 设置策参数 set_variables() # 设置中间变量 is_trade_flag = is_trade() if is_trade_flag: pass else: set_backtest() # 设置回测条件

设置策参数

def set_params(): g.percent = 0.10 g.est_interval = 80 # 记录优化区间，使用二次规划根据这个区间最优化权重 g.lamda = 0 g.hold_days = 60 g.max_hold_num = 20 # 最大持仓的股票 g.run_days = 0 g.benchmark = '000300.SS' # 财报季度调仓所依据的指定日期 g.finance_update_date_list = ['0401', '0801', '1001']

设置中间变量

def set_variables(): g.init_screen = True g.is_update_stocks = False

设置回测条件

def set_backtest(): set_limit_mode('UNLIMITED') # 回测撮合不限制成交量

盘前处理

def before_trading_start(context, data): g.current_date = context.blotter.current_dt.strftime('%Y%m%d') # 2008-01-01前回测由于数据不足，不执行。 if g.current_date < '20080101': g.trade_flag = False else: g.trade_flag = True if not g.trade_flag: return g.everyStock = 0 if is_pub_date(g.current_date): # 财报调仓日 g.stocks = create_stocks() g.is_update_stocks = True elif g.init_screen: '''初始化一个组合，这一小段代码只会用一次''' g.stocks = create_stocks() g.is_update_stocks = True g.init_screen = False # 将Flag置为False，保证下次不再运行

每天交易时要做的事情

def handle_data(context, data): if not g.trade_flag: return # 如果到公告日更新了调仓 if g.is_update_stocks: stock_sort = g.stocks log.info('初始日或调仓日股票池') log.info(stock_sort) if not stock_sort: return previous_date = get_trading_day(-1) # 通过二次规划确定权重 weight = get_weights(stock_sort, previous_date) stock_weight = dict(zip(stock_sort, weight)) stocks = stock_sort current_hold_set = set(context.portfolio.positions.keys()) if set(stocks) != current_hold_set: need_buy = set(stocks).difference(current_hold_set) need_sell = current_hold_set.difference(stocks) current_stocks = set(stocks).difference(need_buy) try: for stock in need_sell: order_target(stock, 0) for stock in need_buy: order_value(stock, context.portfolio.portfolio_value * stock_weight[stock]) for stock in current_stocks: order_target_value(stock, context.portfolio.portfolio_value * stock_weight[stock]) except: pass g.is_update_stocks = False g.run_days = 0

elif g.run_days % g.hold_days == 0:
    stocks = g.stocks
    log.info('非调仓日股票池')
    log.info(stocks)
    if not stocks:
        return
    '''这里的权重通过二次规划确定'''
    weight = get_weights(stocks, context.previous_date)
    stock_weight = dict(zip(stocks, weight))
    try:
        for stock in stocks:
            order_target_value(stock, context.portfolio.portfolio_value * stock_weight[stock])
    except:
        pass
if context.portfolio.cash > 0:
    # 如果可用资金大于0，说明没有全仓，就是撮合单的时候出问题，所以需要重新买入，这时候重新全仓买入几个ETF
    log.info('尝试把剩余资金用完，买入ETF')
    cash = context.portfolio.cash
    order_value('510300.SS', cash / 10 * 4)
    order_value('510330.SS', cash / 10 * 3)
    order_value('510500.SS', cash / 10 * 3)
g.run_days += 1

建立股票池

def create_stocks(): g.all_stocks = get_index_stocks('000300.XBHS', g.current_date) for stock in g.all_stocks.copy(): if stock[:3] == '688': g.all_stocks.remove(stock) # 将ST、停牌、退市三种状态的股票剔除当日的股票池 g.all_stocks = filter_stock_by_status(g.all_stocks, filter_type=["ST", "HALT", "DELISTING"], query_date=None) return get_stocks(g.all_stocks, str(get_trading_day(-1)), g.factor)

获取拟持仓股票池

def get_stocks(stocks, date, factor): sort_type = g.factor_params_info[factor][-1] df = get_factor_values(stocks, factor, date, g.factor_params_info) df.dropna(inplace=True) if df.empty: print('%s数据获取失败，选股失败' % factor) return list() # 3倍标准差去极值 df = winsorize(df, factor, std=3, have_negative=True) # z标准化 df = standardize(df, factor, ty=2) # 市值中性化 market_cap_df = get_fundamentals(stocks, 'valuation', fields='total_value', date=date) market_cap_df = market_cap_df[['total_value']] market_cap_df.dropna(inplace=True) if market_cap_df.empty: print('市值数据获取失败，选股失败') return list() df = neutralization(df, factor, market_cap_df) df = df.sort_values(by=factor, ascending=sort_type) return list(df.head(int(len(df) * g.percent)).index)

获取因子值

def get_factor_values(stock_list, factor, date, factor_params_info): """ 获取因子值方法 入参： 1、股票池：stock_list 2、因子名称：factor 3、计算日期：date 4、因子数据获取需要维护的信息（因子名称、表名、字段名） """ data = get_fundamentals(stock_list, table=factor_params_info[factor][0], fields=factor_params_info[factor][1], date=date, is_dataframe=True) factor_info = {} for stock in stock_list.copy(): if stock not in data.index: continue factor_info[stock] = data.loc[stock, factor_params_info[factor][1]] if factor_info == {}: return pd.DataFrame() factor_df = pd.DataFrame.from_dict(factor_info, orient='index') factor_df.columns = [factor_params_info[factor][1]] return factor_df

使用二次规划确定权重

def get_weights(stocks, date): date = date.strftime('%Y-%m-%d') start_date = get_trading_day(-(g.est_interval + 1)).strftime('%Y-%m-%d') price_data = get_price(stocks, start_date=start_date, end_date=date, frequency='daily', fields=['close'], is_dict=True)

close_df = pd.DataFrame()
for stock_code, stock_data in price_data.items():
    date_info = pd.to_datetime(stock_data['datetime'], format='%Y%m%d')
    close_info = stock_data['close']
    close_df[stock_code] = pd.Series(close_info, index=date_info)
close_df.sort_index(inplace=True)    

code_list = list(close_df.columns)
df_list = []
for stock in code_list:
    df = close_df[[stock]]
    df['change'] = 0 
    df['change'] = df[stock] / df[stock].shift(1) - 1
    df[stock] = df['change']
    df = df[[stock]]
    df.fillna(0, inplace=True)
    df_list.append(df)

result = pd.concat(df_list, axis=1)
index_price = get_price(g.benchmark, start_date=start_date, end_date=date, frequency='daily',
                        fields=['close'], is_dict=False)
index_r = index_price.pct_change()
index_r.fillna(0, inplace=True)
weight = calculate_weight(np.array(result), np.array(index_r))
return weight

def calculate_weight(train_returns, target_returns): length = len(train_returns.T)

# 定义二次线性规划目标函数
def objective(weights):
    return np.sum((np.dot(train_returns, weights) - target_returns) ** 2)

# 定义约束条件
constraints = [{'type': 'eq', 'fun': lambda weights: np.sum(weights) - 1},
               {'type': 'ineq', 'fun': lambda weights: 0.2 - np.max(weights)}
               ]
# 定义权重的取值范围（可以设置最小权重和最大权重区间）
min_weight = (1 / length) * 0.2  # 最小权重
max_weight = (1 / length) * 5  # 最大权重
bounds = [(min_weight, max_weight)] * train_returns.shape[1]
# 初始化权重
initial_weights = np.ones(train_returns.shape[1]) / train_returns.shape[1]
# 最小化目标函数，求解权重
result = minimize(objective, initial_weights, constraints=constraints, bounds=bounds)
# 输出结果
test_weights = result.x
# print("测试集投资权重：", test_weights)
return test_weights

保留小数点两位

def replace(x): x = Decimal(x) x = float(str(round(x, 2))) return x

去极值函数（3倍标准差去极值）

def winsorize(factor_data, factor, std=3, have_negative=True): """ 去极值函数 factor:以股票code为index，因子值为value的Series std为几倍的标准差，have_negative 为布尔值，是否包括负值 输出Series """ r = factor_data[factor] if not have_negative: r = r[r >= 0] # 取极值 edge_up = r.mean() + std * r.std() edge_low = r.mean() - std * r.std() r[r > edge_up] = edge_up r[r < edge_low] = edge_low r = pd.DataFrame(r) return r

z－score标准化函数：

def standardize(factor_data, factor, ty=2): """ s为Series数据 ty为标准化类型:1 MinMax,2 Standard,3 maxabs """ temp = factor_data[factor] re = 0 if int(ty) == 1: re = (temp - temp.min()) / (temp.max() - temp.min()) elif ty == 2: re = (temp - temp.m