PTrade 入门指南

finance
, ,

首先介绍的是最为出名的一个券商能提供的可交易的量化软件 PTrade, 底层应该是套壳二开的经典开源软件 (ziplin?)

让我们一起来看看量化软件到底是啥样的

0x00. 环境

由于券商不同于技术习惯 Unix 类环境, 很多还是 Windows 环境强绑定的, 这点 MacOS 的同学要注意, 有些类似 QMT 对虚拟机下支持不好

  • OS: Windows10
  • PTrade-Client: V1.0-202204-GS定制版本 + 可读写外网(非指海外)
  • 最大并行数: 5个 (策略)

整体框架

首先建议参考一下简单的 PTrade 介绍文, 大概知道它的核心运行逻辑, 显然它是围绕交易时间的盘前/中/后来设计的, 那么这些函数自然都有

注: 个人理解 PTrade 就像一个 10 人共享 1CPU 的主机, 回测速度非常慢, 所以建议不要

基本函数

PTrade 内置了一些必要的函数, 这里简单列一下, 其他的查它的 api 文档:

  1. response:
    • on_trade_response(context, tradeList):
    • on_order_response(context, orderList):
  2. order
    • order_value(security, value, limit_prive=None) : 按指定价下单, 常见
    • over_night_order(): 隔夜单

MACD

Moving Average Convergence/Divergence 是全称, 我们尝试不看其他内容, 从单词本身去理解一下

前两个单词是最基础的 MA (均线), 然后converge 代表汇聚/收敛, diverge 代表发散, 是一对明显的反义词, 所以我们就可以理解英文里 / 的含义了, 本来它就可以是 MAC 或 MAD, 也就是 “收敛/发散均线”, 那么啥叫 “收敛均线”呢? 先尝试看一下实际的 MACD 图: (有两条线)

  • 通常白色的 DIF ()线
  • 通常黄色的 DEA ()线

顶/底背离

股市/金融里有许多听起来很奇怪的词汇, 因为频次很高常用的也不会太多, 强烈建议大家不要自己猜测/跳过, 而是尽早理解一下

经常会听到”日线顶背离“的说法, 代表股市/股价很可能走下行趋势, 那么什么叫顶背离呢? (底背离则相反)

简单说, 股价上涨, MACD 指标却出现下跌趋势就可理解为顶背离 (当然其他指标 RSI 等也是类似), 所以基础指标还是必须得了解含义的

priceUpParamDown00

0x02. 小市值回测

小市值策略作为最经典好懂的高收益策略之一, 先来看看它的模板代码和效果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
'''
策略名称：小市值日线交易策略
运行周期: 日线
策略流程：
盘前将中小板综成分股中st、停牌、退市的股票过滤得到股票池
盘中换仓，始终持有当日流通市值最小的股票（涨停标的不换仓）。
'''
# 导入函数库
import pandas as pd
import numpy as np
import time
from decimal import Decimal

#初始化
def initialize(context):

    set_benchmark('000300.XSHG')  
    #股票池
    g.index = '399101.XBHS' #中小板综
    #持有股票数量
    g.buy_stock_count = 3
    #筛选股票数量
    g.screen_stock_count = 15
    is_trade_flag = is_trade()
    if is_trade_flag:
        pass
    else:
        set_backtest()      #设置回测条件    

#设置回测条件
def set_backtest():
    set_limit_mode('UNLIMITED')

#盘前处理
def before_trading_start(context, data):
    
    g.position_last_map = []  
    g.stock_list = get_index_stocks(g.index)
    #获取股票的状态ST、停牌、退市
    st_status = get_stock_status(g.stock_list, 'ST')
    halt_status = get_stock_status(g.stock_list, 'HALT')
    delisting_status = get_stock_status(g.stock_list, 'DELISTING')
    
    #将三种状态的股票剔除当日的股票池
    for stock in g.stock_list.copy():
        if st_status[stock] or halt_status[stock] or delisting_status[stock]:
            g.stock_list.remove(stock)

    # g.current_date = context.blotter.current_dt.strftime("%Y%m%d")
    g.current_date = str(get_trading_day(-1))

    count = 0
    flag = False
    #获取财务数据尝试5次
    while count < 5:
        if get_df(context):
            log.info('本次获取财务数据成功')
            flag = True
            break
        else:
            count +=1
            time.sleep(60)
    if not flag:
        g.handle_data_flag = False
        log.info('本次获取财务数据不成功，请检查数据')
    else:
        g.handle_data_flag = True
        
   
#盘中处理         
def handle_data(context, data):
    if not g.handle_data_flag:
        return

    position_last_close_init(context)
    # g.data = data
    buy_stocks = get_trade_stocks(context, data)
    log.info('buy_stocks:%s'%buy_stocks)
    trade(context, buy_stocks)      


#交易函数
def trade(context, buy_stocks):
    #卖出
    for stock in context.portfolio.positions:
        if stock not in buy_stocks:
            order_target_value(stock, 0)
            log.info('sell:%s'%stock)
    #买入
    #持有股票数量
    position_last_map = [
        position.sid
        for position in context.portfolio.positions.values()
        if position.amount != 0]
    position_count = len(position_last_map)
    log.info('position_count%s'%position_count)
    # position_count = len(context.portfolio.positions)
    if g.buy_stock_count > position_count:
        value = context.portfolio.cash / (g.buy_stock_count - position_count)        
        for stock in buy_stocks:
            if stock not in context.portfolio.positions:
                order_target_value(stock, value)


#获取市值数据函数
def get_df(context):
    try:
        df = get_fundamentals(g.stock_list, 'valuation', fields=['total_value','a_floats'], date=g.current_date)
        g.df2 = df.sort_values(by='a_floats')
        return True
    except:
        return False


#获取买入股票池（涨停股不参与换仓）
def get_trade_stocks(context, data):
    g.df2['curr_float_value'] = 0
    stocks = list(g.df2.index)
    for stock in stocks:
        df = g.df2[g.df2.index == stock]
        if not df.empty:
            g.df2['curr_float_value'][stock] = g.df2['a_floats'][stock]*data[stock].close
        else:
            g.df2['curr_float_value'][stock] = 0
    g.df2 = g.df2[g.df2['curr_float_value']!=0]
    df3 = g.df2.sort_values(by='curr_float_value')
    stocks = list(df3.head(g.screen_stock_count).index)  
    up_limit_stock = get_limit_stock(context, stocks)['up_limit']
    stocks = list(set(stocks)-set(up_limit_stock))
    hold_up_limit_stock = get_limit_stock(context, g.position_last_map)['up_limit']
    log.info('持仓涨停股：%s'%hold_up_limit_stock)    
    count = g.buy_stock_count - len(hold_up_limit_stock)
    check_out_lists = stocks[:count]
    check_out_lists = hold_up_limit_stock + check_out_lists
    return check_out_lists


#生成昨日持仓股票列表
def position_last_close_init(context):
    g.position_last_map = [
        position.sid
        for position in context.portfolio.positions.values()
        if position.amount != 0
    ]


#保留小数点两位
def replace(x):
    y = Decimal(x)
    y = float(str(round(x, 2)))  
    return y       


#日级别回测获取涨停和跌停状态的股票
def get_limit_stock(context, stock_list):
    today_date = context.blotter.current_dt.strftime("%Y%m%d")
    st_status = get_stock_status(stock_list, 'ST')        
    out_info = {'up_limit':[], 'down_limit':[]}
    history = get_history(5, '1d', ['close','volume'], stock_list, fq='dypre', include=True)
    history = history.swapaxes("minor_axis", "items") 
    def get_limit_rate(stock, ST_flag=True, date=None):
        rate = 0.1
        # 科创板为20%
        if stock[:2] == '68':
            rate = 0.2
        # 创业板2020年8月24日之前为10%，之后为20%
        elif stock[0] == '3' and date>='20200824':
            rate = 0.2
        # 非科创和非创业板的股票在st状态下为5%
        elif stock[0] != '3' and stock[:2] != '68' and ST_flag:
            rate = 0.05
        return rate
    for stock in stock_list:
        df = history[stock] 
        df = df[df['volume']>0] 
        if len(df.index)<2:   
            continue          
        last_close = df['close'].values[:][-2]
        curr_price = df['close'].values[:][-1]
        ST_flag = st_status[stock]
        rate = get_limit_rate(stock, ST_flag, today_date)
        up_limit_price = last_close*(1+rate)
        up_limit_price = replace(up_limit_price)
        if curr_price >= up_limit_price:
            out_info['up_limit'].append(stock)
        down_limit_price = last_close*(1-rate)
        down_limit_price = replace(down_limit_price)
        if curr_price <= down_limit_price:
            out_info['down_limit'].append(stock)                    
    return out_info

以最近一年为例, 随便一跑回测 150% 的受益, 直觉告诉我们这是不科学的, 那问题可能出在哪几个地方呢:

  1. 没有设置滑点, 滑点默认值是 0.1 代表什么呢?
  2. 交易费率/成本过低, 这里是否有考虑印花税, 还是不需要考虑

然后看看为啥回测这么慢, 带上性能分析之后, 看看几个大头:

  1. set_benchmark: 75%
  2. get_trade_stock()

0x0n. 备注

上传策略以后如果重启，要先双击策略再点重启才能正常运行，不然看着是运行的，但是策略进程没起来

交易开启的时间

交易在任何时间都可以开启，开启后会立刻运行initialize和before_trading_start（如果策略中定义的话），要注意的是：开盘前开启交易，before_trading_start肯定会先于handle_data开启；但开盘期间开启交易，before_trading_start和handle_data可能会同时运行，策略逻辑防止混乱。

策略中模块的运行顺序

交易中before_trading_start、handle_data、tick_data、run_interval都是独立的线程，当交易开启之后，每个线程都会启动，理论上没有先后顺序关系。因此建议在策略中设置强制顺序的控制系统，比如运行完before_trading_start后打开handle_data的运行开关，运行完handle_data后打开tick_data或者run_interval的运行开关。

重启功能

  1. 实盘日线策略的handle_data 时间 2：50
  2. 盘前初始化时用到昨天交易数据
  3. 可以在第一天的after_trading_end里把当天的get_orders（）保存成文件，然后第二天读出来
  4. 集合竞价：un_daily可以定义某个时间执行的函数，定义后order就可以
  5. order_tick 或order发出下单指令，到实际委托，最快要多久的
  6. for循环报300笔委托的情况下测试 ，300笔委托2秒钟内全部报到柜台
  7. run_daily 周末不执行
  8. Portfolio是间隔6s更新+成交主推触发Portfolio更新 若是有委托，会根据主推更新的 若没委托，就是间隔6s更新
  9. 委托主推和成交主推用entrust_no可以关联到同一个order_id上
  10. 保护源码的措施。可以在回测中下载策略，然后再上传，就无法编辑和查看策略了

0x0n. Alpha-T

简单说类似聚宽这中量化平台联合券商提供了一种可以自动做T的量化权限, 基于你已有的股票, 在一定规则下进行自动的做T, 以获得超额的收益, 这个在聚宽中被称作 Alpha-T, 其实说人话就是一个量化版的”网格条件单”, 我们可以从中学习一下一些策略来, 先看看它的基本要求:

  1. 日内波动(最好)较大 (基本做T要素)
  2. 成交量比较活跃 (避免滑点)
  3. 持仓周期比较长 (>1个月)

算法特点:

  1. 短持仓时间
  2. 多笔交易分摊风险:
  3. 严格控制止损: 如果发现做T不当出现反向操作, 会尽快平仓止损,

缺点: 觉得最主要是收的手续费太高 – 万分之3.5(包括交易费率), 叠加千分之一的印花税,也会导致收益率大幅下滑, 但是我们能去掉这个手续费自己模拟出类似的自动做T效果, 还是不错的.

参考资料:

  1. PTrade 入门介绍 - zhihu
  2. PTrade 官网 20 课
  3. Alpha-T-ZhiHu 简介
  4. 网格黄金分割回调线、反弹线的思考 - wx)
  5. 量化交易平台入门避坑指南 - zhihu.com