Ptrade量化实盘部署策略

Ptrade 量化实盘部署：从策略到自动交易

量化交易的最后一步

策略回测再好，不上实盘就没有意义。Ptrade（国金证券）是国内主流的券商量化平台，支持：

  • Python 策略编写——标准的 Python 语法，支持 pandas/numpy
  • 回测引擎——内置回测系统，支持多周期
  • 模拟交易——实盘环境仿真，验证策略
  • 实盘交易——直接对接交易所，自动化执行
  • 多策略管理——一个账户同时运行多个策略

Ptrade 基础架构

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  策略研究    │────▶│  回测验证    │────▶│  模拟交易    │
│  (本地开发)  │     │  (Ptrade)    │     │  (Ptrade)    │
└──────────────┘     └──────────────┘     └──────────────┘
                                              │
                                              ▼
                                         ┌──────────────┐
                                         │  实盘交易    │
                                         │  (Ptrade)    │
                                         └──────────────┘

四个阶段逐步推进，每个阶段验证通过再进入下一阶段。

环境准备

开通 Ptrade

  1. 联系国金证券客户经理开通 Ptrade 权限
  2. 下载安装 Ptrade 客户端（Windows）
  3. 登录后进入”量化交易”模块

Ptrade 知道这些 API

# 获取行情
get_history(security, count, unit='1d', fields=['close'])
get_current_data()  # 实时行情

# 交易
order(security, amount, price=None, style=None)
order_target_value(security, target_value)
cancel_order(order_id)

# 账户
get_portfolios()   # 总资产
get_positions()    # 持仓

# 回测
get_industry(security)  # 行业分类
get_fundamentals(query_object)  # 基本面数据

⚠️ 关键坑：get_history 返回的是 long-format，需 pivot() 转 wide-format

策略架构：多策略统一管理

当你有多个策略（ETF轮动、白马选股、聚宽因子）时，需要统一管理。下面是一个三合一策略架构：

combined_strategy.py

# combined_strategy.py
import pandas as pd

def initialize(context):
    """初始化：注册各子策略"""
    context.strategies = {
        'etf_rotation': {
            'active': True,
            'capital_ratio': 0.2,  # 20%资金
            'run_freq': 'daily',
            'func': etf_rotation_logic
        },
        'white_horse': {
            'active': True,
            'capital_ratio': 0.6,  # 60%资金
            'run_freq': 'monthly',
            'func': white_horse_logic
        },
        'jq_factor': {
            'active': True,
            'capital_ratio': 0.2,  # 20%资金
            'run_freq': 'monthly',
            'func': jq_factor_logic
        }
    }
    
    # 全局风控
    context.max_position_ratio = 0.95  # 最大仓位
    context.stop_loss = -0.05          # 单日止损 -5%
    context.max_concentration = 0.3    # 单票上限 30%
    
    # 运行计数器
    context.run_days = 0

def handle_data(context, data):
    """每日运行"""
    context.run_days += 1
    today = str(get_datetime())[:10]
    
    # 风控检查
    if not risk_check(context):
        return
    
    # 运行各策略
    for name, cfg in context.strategies.items():
        if not cfg['active']:
            continue
        
        # 判断是否该运行
        if not should_run(cfg, context.run_days):
            continue
        
        # 执行策略逻辑
        try:
            signals = cfg['func'](context)
            execute_signals(signals, context)
            log.info(f"[{today}] 策略 {name} 执行完成")
        except Exception as e:
            log.error(f"[{today}] 策略 {name} 执行失败: {e}")
    
    # 盘后记录
    record(context)

def etf_rotation_logic(context):
    """ETF轮动策略：基于20日动量选前3"""
    etfs = get_history('all_etf', 20, '1d', ['close'])
    etfs = etfs.pivot_table(index='date', columns='security', values='close')
    
    momentum = etfs.iloc[-1] / etfs.iloc[0] - 1
    top3 = momentum.nlargest(3).index.tolist()
    
    return {
        'buy': [{'security': e, 'ratio': 1/3} for e in top3],
        'sell': 'all_except_top3'
    }

def white_horse_logic(context):
    """白马股策略：基本面选股"""
    q = query(
        valuation.market_cap,
        indicator.roa,
        indicator.roe
    ).filter(
        indicator.roa > 0.05,
        indicator.roe > 0.1
    ).order_by(valuation.market_cap.asc()).limit(10)
    
    stocks = get_fundamentals(q)
    # ... 策略逻辑
    return signals

def jq_factor_logic(context):
    """聚宽因子选股"""
    # ... 因子逻辑
    return signals

风控模块

def risk_check(context):
    """风控检查"""
    portfolio = get_portfolios()
    
    # 1. 总资产检查
    available = context.portfolio.available_cash
    if available < 100:
        log.warn("现金不足，暂停交易")
        return False
    
    # 2. 当日回撤检查
    day_pnl = portfolio.total_assets / portfolio.total_assets_yesterday - 1
    if day_pnl < context.stop_loss:
        log.warn(f"当日回撤{day_pnl:.2%}，触发止损")
        return False
    
    # 3. 集中度检查
    for pos in get_positions():
        ratio = pos.amount * pos.price / portfolio.total_assets
        if ratio > context.max_concentration:
            log.warn(f"集中度{ratio:.1%}超限")
            # 减仓
    
    return True

部署流程

阶段一：本地开发

# 在本地写好策略
~/quant/ptrade/
├── combined_strategy.py   # 主策略
├── config.py              # 配置文件
├── utils.py               # 工具函数
└── README.md              # 说明文档

阶段二：Ptrade 回测

  1. 打开 Ptrade → 量化交易 → 新建策略
  2. 粘贴代码
  3. 设置回测参数（起始日期、初始资金、手续费等）
  4. 点击回测

阶段三：模拟交易

回测通过后，切换到”模拟交易”：

  1. 选择策略 → 启动模拟
  2. 观察运行情况（至少运行 1-2 周）
  3. 检查信号准确性，对比回测表现

阶段四：实盘上线

模拟交易稳定后：

  1. 创建实盘策略实例
  2. 设置资金规模（建议先用 10% 资金）
  3. 启动实盘
  4. 前两周每天检查执行情况

资金配置方案

以 10 万账户为例：

策略资金占比金额调仓频率
白马选股 v2.360%6 万每月
ETF轮动20%2 万每日
聚宽因子选股20%2 万每月

监控和维护

每日检查

def after_trading_end(context, data):
    """收盘后检查"""
    # 检查成交
    trades = get_trades()
    for t in trades.values():
        if t.status == 'unfilled':
            log.warn(f"未成交订单: {t.security} {t.amount}股")
    
    # 检查持仓
    for pos in get_positions():
        pnl = pos.amount * (pos.price - pos.avg_cost)
        log.info(f"{pos.security}: 盈亏 {pnl:.2f}")

常见问题处理

问题原因解决方法
订单未成交价格偏离市场改为市价单或放宽限价范围
策略没跑权限/时间问题检查策略状态，确认运行时间
信号异常数据缺失检查 get_history 返回值
收益偏差滑点/手续费调低预期，加缓冲

Ptrade 已知坑

# ❌ get_history 直接使用
data = get_history(['000001.SZ'], 10)  
# 返回的是 long format（列是 date, security, close）

# ✅ 正确用法
data = get_history(['000001.SZ'], 10)  
data = data.pivot_table(index='date', columns='security', values='close')

# ❌ 获取可用资金
cash = context.portfolio.cash  # 包含冻结资金

# ✅ 正确
cash = context.portfolio.available_cash

# ❌ after_trading_end 没有写参数
def after_trading_end(context):  # 需要2个参数

# ✅ 正确
def after_trading_end(context, data):
    pass

# ⚠️ 手续费
# 最低佣金 1.0 元（即使按万0.5算不到1块也要收1块）

小结

从策略到实盘的四步路线图：

策略研究 → 回测验证 → 模拟交易 → 实盘上线
  ↓            ↓           ↓           ↓
 本地Python   Ptrade      Ptrade      Ptrade
  + Agent     回测        模拟        实盘

每一步验证通过后再推进，不要急于实盘。量化交易的核心不是策略多厉害，而是系统稳定、风控到位。

本文是量化学习路径的第 5 步，也是最后一步。

💬 评论