2019 小土刀再启程

0%

【量化投资技术分析实战】丰富详细

发表于 更新于 分类于

量化投资和非量化模式不冲突，反而可以相互借鉴，得到更好的结果

更新历史

  • 2021.11.21：开始阅读

读后感

本书正在阅读中…

阅读笔记

序：通向量化投资之路并不平坦

量化投资没有确切的定义，它泛指通过数学分析、挖掘价格波动规律，或者通过对相关宏观经济、财务数据、量价关系、资金交易等数据进行建模，寻找数据之间的关系，以获得稳定利润为目标，持续计算生成定量化的投资信号，并通过计算机严格执行。

与传统的“主观分析+手工下单交易”的区别：

  1. 业绩稳定
  2. 概率取胜可精确回测
  3. 严谨且执行力强

量化投资和非量化模式不冲突，反而可以相互借鉴，得到更好的结果。

一 量化投资入门建议与行业概况

学习路线图

  1. 体会各类资产的择时模型，激发学习兴趣。在单一时间序列模型上，动量容易被均线类、突破类模型捕捉
  2. 克服编程语言障碍（对于程序员来说这一点可以忽略）
  3. 探索股票和期货波动特征，尝试较为复杂的股票和期货模型。因为股票和期货有足够的容量可以交易，且杠杆率可控，资金安全有保障
  4. 做足单因子分析工科，然后切换到多因子学习。观察因子值对于个股收益率的描述能力，重点分析 IC 和因子收益单调性，以便发现高质量的 Alpha 收益（超额收益）因子。在多因子阶段需要注意：做好数据处理工作，让模型更加简单有效和安全、易解释，对算法知识可以浅尝辄止，因为可供调用的算法包很多。

如何达到稳步上升的资金曲线

  1. 基于强壮稳健的投资逻辑。
    1. 不断产生假设，反复提出各种投资逻辑，并设定严格的回测条件，证实和证伪你的假设
    2. 阅读大量金融服务行业研究报告也是策略开发的思路来源
    3. 模型对应不同样本有盈利的共性，且模型参数有限、有效、有合理参数解释
  2. 基于时间的玫瑰。坚守自己的模型，保持强大的连续性和投资定力，这在一定程度上比开发更先进的模型更重要
  3. 基于多资产多策略配置。国家的货币总量是逐步递增的，这是经济正常上行的必然结果，其反应在日常生活中的表现是资产价格上涨和货币购买力缓慢降低

有保留地相信回测结果

  • 量化投资的危险之处在于，目标函数在推动你从数据里挖出黄金，也在放大噪声的影响力，所以到了某个奇怪的交易规则设置临界点下，模型由噪声主导，得到的完全都是幸存者偏差
  • 我们要相信回测结果，在一定程度上也要尝试走在市场前面，预判性地应对风险

绩效评估常见指标和方法

  1. Sharpe Ratio 夏普比率
    1. 目的是计算投资组合每承受一单位总风险，会产生多少的超额报酬
    2. 优点是不仅考虑收益，还考虑每次的波动率（回撤幅度），可以同时对策略的收益与风险进行综合考虑
    3. 公式：(策略年化收益率 - 无风险回报率) / 策略回报标准差。这里的标准差就是净值在这段时间内的标准差
  2. Alpha 阿尔法
    1. 投资中面临着系统性风险(Beta)和非系统性风险(Alpha)，Alpha 是投资者获得与市场波动无关的回报
    2. 公式：Alpha=Rp-[Rf+βp(Rm-Rf)]，含义是 策略年化收益率 - [无风险利率 + β x (基准或市场年化收益率 - 无风险利率)]，这里需要先计算 beta
  3. Beta 贝塔
    1. 表示投资的系统性风险，反映了策略对大盘变化的敏感性
    2. 计算公式：策略每日收益与基准或市场每日收益协方差 / 基准或市场每日收益方差
  4. Annualized Returns 策略年化收益率
    1. 表示投资期限为一年的预期收益率
    2. 公式：(策略最终价值 / 策略初始价值 - 1)/回测交易日数量 x 250
    3. Benchmark Annualized Returns（基准年化收益）是将策略收益换成市场收益
  5. Max Drawdown 最大回撤比率。计算公式 max(1 - 策略当日价值 / 当日之前资金最高价值）。另外还需要注意最大回撤持续时间，表示多久能重回净值最高点，一般夏普比率高，最大回撤持续时间就短
  6. Sortino 索提诺比率。
    1. 表示策略每承担一单位的下行风险，将会获得多少超额回报
    2. 公式：(策略年化收益率 - 无风险回报率) / 策略下行波动率
    3. 可以看作是夏普比率在衡量对冲基金/私募基金时的一种修正方式
    4. 下行波动率(Downside Risk) = 当（今日或本阶段策略收益率 - 平均收益率）为负时的波动率
    5. 波动率 = sqrt((收益率 - 平均收益率)^2 x 250 / (当前策略运行天数 - 1))
  7. Information Ratio(IR) 信息比率
    1. 用于衡量单位超额风险带来的超额收益，越高越好
    2. 公式：(策略年化收益率 - 基准年化收益率) / 策略与基准每日收益差值的年化标准差
    3. 更适合评估单一股票市场内的每一个交易策略

期货量化交易的重要指标

  1. 收益风险比
    1. 公式：年度收益 / 全段最大资产回撤
    2. 表示了一个策略的风险控制和收益平衡能力，越大说明同收益的前提下风险越低
  2. R 平方值
    1. 交易盈亏曲线拟合的趋势线与收益曲线之间的相关系数的平方
    2. 值趋近于 1
  3. 平均资产回撤
    1. 计算公式：资产回撤总金额 / 资产回撤计数（都是以超过最大回撤基准线以上的回撤来计算的前 N 个最大回撤）
    2. N 一般设定为 5，表示考虑亏损最严重的 5 次的平均回测
  4. 调整收益风险比
    1. 公式：年度收益 / 平均资产回撤
    2. 其可信度强于收益风险比（就是第一个）
  5. TB 系数
    1. 公式：(平均利润 x 平均利润 x 交易手数)/(平均盈利 x 平均亏损)
    2. 一个自定义绩效评价指标，有一定参考价值（考虑了交易手数）
  6. 头寸系数
    1. 公式：收益风险比 x R 平方值 x 置信度 / 最大资产回撤
    2. 收益风险比间接考虑了最大回撤
    3. R 平方值间接考虑了平均回撤
    4. 置信度间接考虑了交易次数
  7. 置信度
    1. 公式：1 - 1 / sqrt(交易次数)
    2. 在同等利润或者夏普比率下，越高的绩效置信度说明实盘阶段该绩效保持不变的可能性越高

绩效永远不仅只评估收益，而且还评估风险情况。想要走得远，就要特别关注风险控制能力。

二 快速驾驭编程语言支持

主要是介绍 Python 和聚宽平台上的函数，常用函数：

  • get_index_stocks 获取某个指数的成分股名单数据
  • get_fundamental 获取基本面财务数据
  • attribute_history 获取历史数据
  • get_current_data 获取当前时间的个股信息
  • get_security_info 获取单个标的的基本信息

三 股票期货择时交易模型

ETF 二八择时法则，跑赢基础股票指数

  • ETF 基金在股票市场中生命力持久，本质原因是它剥离了个股特质风险，通过获得市场系统性风险反而降低了个股选择的难度
  • 从时间序列量价分析的角度看，个股的时间序列噪声高、随机性强，不容易获得主趋势（信号），反而容易被次级趋势和微弱趋势（噪声）所困扰，通过择时等方法跑赢大盘不容易，也会带来过大的波动性（回撤风险）
  • 从长周期来分析，股价是围绕其基本面价值上下波动的，研究表明市场上存在单一股票价格噪声，而基本面的变化取决于经济环境特别是经济结构调整与经济周期轮动，所以个股代表了经济的各个领域，互相叠加后最终波动主趋势得以呈现
  • 选择 ETF 的持有标的，要注重差异性，比如做两个标的的轮动，一定要在基本面或者逻辑方面形成显著差异

具体代码 

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
# 导入函数库
from jqdata import *

# 初始化函数，设定基准等等
def initialize(context):
    # 输出内容到日志 log.info()
    log.info('初始函数开始运行且全局只运行一次')
    set_params() # 设置策略参数
    set_backtest() # 设置回测条件
    
    ### 股票相关设定 ###
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    # 设置滑点为交易额的千分之二
    set_slippage(PriceRelatedSlippage(0.002)) 

    ## 运行函数（reference_security为运行时间的参考标的；传入的标的只做种类区分，因此传入'000300.XSHG'或'510300.XSHG'是一样的）
      # 开盘前运行
    run_daily(before_market_open, time='before_open', reference_security='000300.XSHG')
      # 开盘时运行
    run_daily(market_open, time='open', reference_security='000300.XSHG')
      # 收盘后运行
    run_daily(after_market_close, time='after_close', reference_security='000300.XSHG')

# 设置策略参数
def set_params():
     # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    g.lag = 15 # 回溯期
    g.hour = 14 # 每天交易时间（小时）
    g.minute = 53 # 每天交易时间（分钟）
    g.sz = '000016.XSHG' # 计算标的，上证 50（超级大盘股）
    g.hs = '000300.XSHG' # 计算标的，沪深 300（价值股）
    g.zz = '000905.XSHG' # 计算标的，中证 500（成长股）
    g.ETF50 = '510050.XSHG' # 交易标的，下同
    g.ETF300 = '510300.XSHG'
    g.ETF500 = '510500.XSHG'
   

# 设置回测条件
def set_backtest():
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')

# 取得某个区间内所有收盘价
def get_stock_price(stock, interval):
    h = attribute_history(stock, interval, unit='1d', fields=('close'), skip_paused=True)
    return (h['close'].values[0], h['close'].values[-1])
    
# 计算得到操作信号
def get_signal(context):
    # yesterday 是昨日收盘价，interval 是指定周期前的收盘价
    interval50, yesterday50 = get_stock_price(g.sz, g.lag)
    interval300, yesterday300 = get_stock_price(g.hs, g.lag)
    interval500, yesterday500 = get_stock_price(g.zz, g.lag)
    # 计算前 20 日动量
    sz50increase = (yesterday50 - interval50) / interval50
    hs300increase = (yesterday300 - interval300) / interval300
    zz500increase = (yesterday500 - interval500) / interval500
    # 持仓金额
    if g.ETF50 in context.portfolio.positions:
        hold50 = context.portfolio.positions[g.ETF50].total_amount
    else: 
        hold50 = 0
    
    if g.ETF300 in context.portfolio.positions:
        hold300 = context.portfolio.positions[g.ETF300].total_amount
    else:
        hold300 = 0
    
    if g.ETF500 in context.portfolio.positions:
        hold500 = context.portfolio.positions[g.ETF500].total_amount
    else:
        hold500 = 0
        
    # 止损条件
    if (hs300increase <= 0 and hold300 > 0) \
    or (zz500increase <= 0 and hold500 > 0) \
    or (sz50increase <= 0  and hold50):
        return 'sell_the_stocks'
    # 如果 50 增长率大于 300 和 500 且幅度达到 0.01，且 50 增长率 > 0
    # 且 50,300,500 空仓，买入 50
    elif sz50increase-hs300increase>0.01 \
    and sz50increase-zz500increase>0.01 \
    and sz50increase > 0 \
    and (hold500 == 0 and hold500 == 0 and hold50 == 0):
        return 'ETF50'
    # 和上面的逻辑类似，买入 300 的条件
    elif hs300increase-sz50increase>0.01 \
    and hs300increase-zz500increase>0.01 \
    and hs300increase > 0 \
    and (hold500 == 0 and hold500 == 0 and hold50 == 0):
        return 'ETF300'
    # 和上面的逻辑类似，买入 500 的条件
    elif zz500increase-sz50increase>0.01 \
    and zz500increase-hs300increase>0.01 \
    and zz500increase > 0 \
    and (hold500 == 0 and hold500 == 0 and hold50 == 0):
        return 'ETF500'
    
# 卖出指令，清仓函数
def sell_the_stocks(context):
    for i in context.portfolio.positions.keys():
        return (log.info("卖出 %s" % i), order_target(i, 0))

# 买入指令
def buy_the_stocks(context, signal):
    return (log.info("买入 %s" % signal), order_value(eval('g.%s' % signal), context.portfolio.available_cash))

## 开盘前运行函数
def before_market_open(context):
    # 输出运行时间
    #log.info('函数运行时间(before_market_open)：'+str(context.current_dt.time()))
    # 给微信发送消息（添加模拟交易，并绑定微信生效）
    # send_message('美好的一天~')
    pass

## 开盘时运行函数
def market_open(context):
    #log.info('函数运行时间(market_open):'+str(context.current_dt.time()))
    pass
    
    # 暂时不用做啥
    
# 盘中运行
def handle_data(context, data):
    # 获得当前时间
    hour = context.current_dt.hour
    minute = context.current_dt.minute
    
    if hour == g.hour and minute == g.minute:
        signal = get_signal(context)
        
        if signal == 'sell_the_stocks':
            sell_the_stocks(context)
        elif signal == 'ETF500' or signal == 'ETF50' or signal == 'ETF300':
            buy_the_stocks(context, signal)

## 收盘后运行函数
def after_market_close(context):
    #log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time())))
    
    #得到当天所有成交记录
    trades = get_trades()
    for _trade in trades.values():
        log.info('成交记录：'+str(_trade))
    total = context.portfolio.available_cash + context.portfolio.positions_value
    cash = context.portfolio.available_cash
    hold = context.portfolio.positions_value
    log.info("一天结束，总资金 %.2f（现金 %.2f,持仓 %.2f):" % (total, cash, hold))
    log.info('############################################################')

Aberration 系统，长期活跃于期货市场

  • 是一套古老而简单的趋势类突破系统，由 Keith Fitschen 于 1986 年发明
  • 实际上是布林带系统的突破式应用，是中低频中长线持仓的一套模型
  • 基于布林带的突破模型，一定是低胜率的模型，但又可以赚钱，就一定是一个高盈亏比的模型，以此来对冲低胜率带来的反复亏本
  • Aberration 的核心构成是一条均线，加上、下两个轨道，轨道宽度是 N 倍标准差
  • 交易规则是，当价格突破上轨时做多，当价格回到中轨时平仓；反之，当价格突破下轨时做空，当价格回到中轨时平仓
  • 通过 ATR 适应波动率止损。
    • 计算 ATR 首先要计算出 TR 值，也就是单个 K 线上的真实波动量
    • TR 等于 “最高价- 最低价”和“最高价-昨收”和“昨收-最低价”，这三个值的最大值
    • 然后将 N 个周期的 TR 值做移动平均即可。如果 N 较小，ATR 反应敏锐，但也意味着噪声较高；如果 N 较大，则 ATR 反应较慢，但也更稳定

低价股+逆向双均线模型，初步探索个股特征

  • MACD 的本质是两条 EXPMA 均线通过二次平滑得到，即由快的指数移动平均线（EMA12）减去慢的指数移动平均线（EMA26）得到快线 DIF，再用 2 乘以（快线 DIF 减去 DIF 的 9 日加权移动均线 DEA）得到 MACD。可视化效果好
  • 交易逻辑为
    • 通过短期均线上穿上期均线（金叉），视为买入股票信号，全市场扫描符合该条件的股票买入
    • 在买入时做一些过滤，选择低价股买入，“低价”在 A 股本来就是一个有效因子
    • 双均线仅提供入场点，不提供出场点，出场由硬止损和追踪止损完成
  • 个股上的动量因子并不强烈，市场充满 了对于趋势交易者特别是按照均线类模型入场的交易者的绞杀
  • 散户交易者中大量存在的迷信指标（均线类动量指标）和市场实际情况发生了较为严重的偏差
  • 必须仔细检测你的交易思路是否正确
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
# 导入函数库
from jqdata import *

# 初始化函数，设定基准等等
def initialize(context):
    # 选取股票
    df = get_fundamentals(query(valuation.code))
    # 选出股票代码
    g.stocklist = list(df['code'])
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 设置滑点
    set_slippage(FixedSlippage(0))
    
    # 输出内容到日志 log.info()
    log.info('初始函数开始运行且全局只运行一次')
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    # 记录 top 值
    g.top = {}
    # 短期均线、长期均线、大盘择时动量形成期
    g.fast = 12
    g.slow = 26
    g.lag = 20

    ### 股票相关设定 ###
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')

    # 收盘后运行
    run_daily(after_market_close, time='after_close', reference_security='000300.XSHG')

# 取得某个区间内所有收盘价
def get_stock_price(stock, interval):
    h = attribute_history(stock, interval, unit='1d', fields=('close'), skip_paused=True)
    return (h['close'].values[0], h['close'].values[-1])


# 卖出指令，清仓函数
def sell_all_stocks(context):
    log.info("卖出所有股票")
    for i in context.portfolio.positions.keys():
        order_target(i, 0)

def handle_data(context, data):
    # 计算 300 和 500 指数的增长率，用于快速清仓
    interval300, yesterday300 = get_stock_price('000300.XSHG', g.lag)
    interval500, yesterday500 = get_stock_price('000905.XSHG', g.lag)
    
    hs300increase = (yesterday300 - interval300) / interval300
    zz500increase = (yesterday500 - interval500) / interval500

    if (hs300increase <= 0 and zz500increase <= 0):
        sell_all_stocks(context)
    else:
        # 卖出过程
        for security in context.portfolio.positions.keys():
            # 当前标的
            current_positions = context.portfolio.positions[security]
            # 近似初始价格
            init_cost = current_positions.avg_cost
            # 更新回撤点最高点值
            if security not in g.top:
                g.top[security] = (init_cost, 0)
            else:
                # 更新 top 值
                # 算法 1， top 值为出现过最大的峰值
                if current_positions.price > g.top[security][0]:
                    g.top[security] = (current_positions.price, 0)
                    
            # 追踪止损逻辑
            # 如果当前价格比 top 低，且处于收益的状态
            # 最高价超过 30%，且收益从最高价减少额 10%
            high = 0.30
            down = 0.15
            if ((g.top[security][0] - init_cost) / init_cost >= high 
            and (g.top[security][0] - current_positions.price) / g.top[security][0] >= down 
            and current_positions.closeable_amount > 0):
                log.info("追踪止损：卖出 %s %s 股" % (security, current_positions.closeable_amount))
                order_target(security, 0)
        
            # 硬止损：亏损 20%
            stop_loss = 0.20
            if ((current_positions.price - init_cost) / init_cost <= -stop_loss 
            and current_positions.closeable_amount > 0):
                log.info("硬追踪：卖出 %s %s 股" % (security, current_positions.closeable_amount))
                order_target(security, 0)

        # 买入过程
        cash = context.portfolio.available_cash
        if cash > 0: 
            log.info("今日现金 %s" % cash)
            # 记录出现买入信号的股票名单
            optional_list = []
            # 对应股票买入
            for security in g.stocklist:
                # 历史价格
                close_data = attribute_history(security, g.slow+2, '1d', ['close', 'volume'], df=False)
                # 昨日短线均线
                ma_fast_1 = close_data['close'][-g.fast:-1].mean()
                # 前日短线均线
                ma_fast_2 = close_data['close'][-g.fast-1:-2].mean()
                # 昨日长线均线
                ma_slow_1 = close_data['close'][-g.slow:-1].mean()
                # 前日长线均线
                ma_slow_2 = close_data['close'][-g.slow-1:-2].mean()
                
                # 如果满足买入条件
                # 短线和长线均线死叉买入
                if (ma_fast_2 > ma_slow_2 
                and ma_fast_1 <= ma_slow_1):
                    optional_list.append((security,close_data['close'][-1]))
            
            # 按照收盘价排序，优先买入便宜的股票
            optional_list.sort(key=lambda l:l[1])
            percent = 0.1
            # 遍历信号股票下单
            # 投资金额为当前可用 cash 的 10%
            use_cash = cash * percent
            for security, close in optional_list:
                current_cash = context.portfolio.available_cash
                # 没钱则结束
                if current_cash <= 0:
                    break
                # 如果所剩不足 10%
                if current_cash < use_cash:
                    use_cash = current_cash
                
                # 按股数下单
                order_value(security, use_cash)
                log.info("买入 %s，花费 %d" % (security, use_cash))


## 收盘后运行函数
def after_market_close(context):
    log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time())))
    #得到当天所有成交记录
    trades = get_trades()
    for _trade in trades.values():
        log.info('成交记录：'+str(_trade))
    log.info('一天结束')
    log.info('##############################################################')

AMA 自适应均线系统捕捉价格启动机会

  • 传统均线模型在震荡阶段价格反复向上、向下触碰均值，所以造成单均线交易法反复发出信号，亏损较为严重。
  • 双均线交易法在一定程度上有所改进，但依然在此阶段有较大亏损
  • 考夫曼自适应移动均线 KAMA 基本上是大家认为比较理想的均线系统

海龟交易法则

  • 海龟交易法则不仅从模型原理上容易理解，而且提供了一套完整的模型+风险控制+资金管理思路
  • 针对期货

四 基本面和基本面交易模型

股票模型思路形成与常见问题

  • 一个策略开发者，首先应该通过长时间在 A 股市场的交易行为，逐步感知市场参与者的风险偏好、资金流动规律。然后建立基本假设，比如假设市场偏好高业绩的股票，或者价格波动的规律偏向趋势性或反转性。最后通过撰写模型，逐一验证自己的假设。在测试模型的过程中，还要再度观察自己的假设是否合理，模型收益来源是否获得合理解释
  • 以 PDCA(Plan Do Check Action)科学管理方法建立模型
    • Plan 通过经验和观察，得到初步观察
    • Do 设计模型思路，撰写源码，验证假设
    • Check 观察交易情况，分析绩效报告
    • Action 调整逻辑，再次进入模型迭代环节
  • 可能会导致幸存者偏差的问题
    • 经济结构或市场结构过度调整：类似小市值这样的表象因子需要慎重考虑，因为这个因子的有效和失效，都伴随着市场结构和监管方式的随时变化
    • 建模数据不足，或模型错误处理数据
    • 因子超额收益不纯净：回测结果需要人工复盘观察，以股票模型为例：某些时段投资组合中的某个股票上涨原因并不是某因子能够解释的，也许是资产重组或者消息刺激等巧合导致的，这种情况要逐一排除，才能得到稳健的模型和参数

小市值二八过滤止损模型

具体代码

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
# 导入函数库
from jqdata import *

# 初始化函数，设定基准等等
def initialize(context):
    log.info('初始函数开始运行且全局只运行一次')
    set_params() # 设置策略参数
    set_backtest() # 设置回测条件
    
     # 收盘后运行
    run_daily(after_market_close, time='after_close', reference_security='000300.XSHG')


# 设置策略参数
def set_params():
    g.stockCount = 100 # 前多少支股票
    g.buyStockCount = 20 # 买入股票数量
    g.period = 5 # 调仓周期
    g.days = 0 # 计时器
    g.lag = 20 # 回溯期
    g.hour = 14
    g.minute = 53
   

# 设置回测条件
def set_backtest():
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    # 设置滑点为交易额的千分之二
    set_slippage(PriceRelatedSlippage(0.002)) 
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')

# 取得某个区间内所有收盘价
def get_stock_price(stock, interval):
    h = attribute_history(stock, interval, unit='1d', fields=('close'), skip_paused=True)
    return (h['close'].values[0], h['close'].values[-1])

# 过滤停牌
def filter_paused_stock(stock_list):
    current_data = get_current_data()
    stock_list = [stock for stock in stock_list if not current_data[stock].paused]
    return stock_list

# 过滤退市
def delisted_filter(security_list):
    current_data = get_current_data()
    security_list = [stock for stock in security_list if not '退' in current_data[stock].name]
    return security_list

# 过滤 ST
def st_filter(security_list):
    current_data = get_current_data()
    security_list = [stock for stock in security_list if not current_data[stock].is_st]
    return security_list
    
# 过滤次新股
def remove_new_stock(context, security_list):
    for stock in security_list:
        days_public = (context.current_dt.date() - get_security_info(stock).start_date).days
        if days_public < 365:
            security_list.remove(stock)
    return security_list
    
# 过滤涨停板股
def high_limit_filter(security_list):
    prices = history(1, unit='1m', field='close', security_list=security_list)
    current_data = get_current_data()
    security_list = [stock for stock in security_list if not prices[stock][-1] == current_data[stock].high_limit]
    return security_list

# 盘中运行
def handle_data(context, data):
    # 获得当前时间
    hour = context.current_dt.hour
    minute = context.current_dt.minute
    
    if hour == g.hour and minute == g.minute:
        # 获得当前总资产
        value = context.portfolio.portfolio_value
        # 计算 300 和 500 指数的增长率，用于快速清仓
        interval300, yesterday300 = get_stock_price('000300.XSHG', g.lag)
        interval500, yesterday500 = get_stock_price('000905.XSHG', g.lag)
        
        hs300increase = (yesterday300 - interval300) / interval300
        zz500increase = (yesterday500 - interval500) / interval500
        
        if (hs300increase <= 0 and zz500increase <= 0):
            sell_all_stocks(context)
        elif zz500increase > 0 or hs300increase > 0:
            buy_min_market_stocks(context)
            g.days += 1

def sell_all_stocks(context):
    log.info("卖出所有股票")
    for i in context.portfolio.positions.keys():
        order_target(i, 0)
    g.days = 0

def buy_min_market_stocks(context):
    log.info("买入股票")
    # 在每个调仓周期
    g.run_today = g.days % g.period == 0
    if not g.run_today:
        return
    log.info("买入股票且处于调仓周期")
    # 获取当前时间
    date = context.current_dt.strftime("%Y-%m-%d")
    # A 股指数和深证 A 指
    list_stock = get_index_stocks('000002.XSHG') + get_index_stocks('399107.XSHE')
    filter1 = filter_paused_stock(list_stock)
    filter2 = delisted_filter(filter1)
    filter3 = st_filter(filter2)
    filter4 = remove_new_stock(context, filter3)
    filter5 = high_limit_filter(filter4)
    
    # 按照当前市值从小到大排序
    df = get_fundamentals(query(
        valuation.code, valuation.market_cap).filter(
            valuation.code.in_(filter5)).order_by(
                valuation.market_cap.asc()).limit(
                    g.stockCount), date=date)
    g.stocks = list(df['code'][:g.buyStockCount])
    
    # 非持股的全部卖出
    for stock in context.portfolio.positions.keys():
        if stock not in g.stocks:
            order_target(stock, 0)
    
    # 应买入股票数量 valid_count
    valid_count = 0
    for stock in context.portfolio.positions.keys():
        if context.portfolio.positions[stock].total_amount > 0:
            valid_count = valid_count + 1
    
    # g.stocks 完全卖光，或 valid_count == 买入股票个数（完全满仓），停止计算
    if len(g.stocks) == 0 or valid_count == g.buyStockCount:
        return
    
    # 每支个股持有价值 = 总价值 / 目前应买入股票数量
    value = context.portfolio.available_cash / (g.buyStockCount - valid_count)
    for stock in g.stocks:
        if stock in context.portfolio.positions.keys():
            pass
        else:
            order_target_value(stock, value)


## 收盘后运行函数
def after_market_close(context):
    #得到当天所有成交记录
    trades = get_trades()
    for _trade in trades.values():
        log.info('成交记录：'+str(_trade))
    log.info('一天结束##################################################')

PEG 价值选股模型，复制彼得·林奇投资路径

  • 任何一家公司股票如果定价合理的话，市盈率就会与收益增长率相等
  • 静态 PE = 股价 / 年报每股收益(EPS)
  • 动态 PE = 股价 x 总股本 / 下一年净利润（需要预测）
  • PEG = PE/净利增长率 x 100
  • 如果 PEG > 1 则股价被高估，如果 PEG < 1 则被低估
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
# 导入函数库
from jqdata import *
import talib
import pandas as pd
import numpy as np


# 初始化函数，设定基准等等
def initialize(context):
    set_params()
    set_variables()
    set_backtest()
     # 收盘后运行
    run_daily(after_market_close, time='after_close', reference_security='000300.XSHG')

# 设置参数
def set_params():
    g.tc = 10 # 调仓天数
    g.num_stocks = 20 # 每次调仓选取的最大股票数量
    g.ATR_timeperiod = 14 # 计算 ATR 的时间周期

# 设置中间变量
def set_variables():
    g.testdays = 0 # 记录回测运行天数
    g.if_trade = False # 当天是否交易
    
# 设置回测条件
def set_backtest():
    set_option('use_real_price',True)
    log.set_level('order','error')
    set_benchmark('000300.XSHG') # 设置基准收益
    
# 开盘前要做的
def before_trading_start(context):
    log.info(str('函数运行时间(before_trading_start):'+str(context.current_dt.time())))
    if g.testdays % g.tc == 0:
        # 如果本日要进行调仓
        g.if_trade = True
        set_slip_fee(context) # 设置手续费，每天不一样
        g.stocks = get_index_stocks('000300.XSHG') # 用沪深 300 作为初始股票池
        # 得到可行股票池
        g.feasible_stocks = set_feasible_stocks(g.stocks, context)
    g.testdays += 1
    
# 设置可行股票池，过滤掉当日停牌的
def set_feasible_stocks(initial_stocks, context):
    # 判断初始股票池是否停牌
    paused_info = []
    current_data = get_current_data()
    for i in initial_stocks:
        paused_info.append(current_data[i].paused)
    df_paused_info = pd.DataFrame(data={'paused_info': paused_info},index=initial_stocks)
    unsuspended_stocks = list(df_paused_info.index[df_paused_info.paused_info == False])
    return unsuspended_stocks
    
# 根据不同时间段设置滑点与手续费
def set_slip_fee(context):
    # 将滑点设置为千分之二
    set_slippage(PriceRelatedSlippage(0.002))
    # 根据时间不同设置手续费
    dt = context.current_dt
    if dt > datetime.datetime(2013,1,1):
        set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    elif dt > datetime.datetime(2011,1,1):
        set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.001, close_commission=0.001, min_commission=5), type='stock')
    elif dt > datetime.datetime(2009,1,1):
        set_order_cost(OrderCost(open_tax=0.001, close_tax=0.002, open_commission=0.001, close_commission=0.001, min_commission=5), type='stock')
    else:
        set_order_cost(OrderCost(open_tax=0.002, close_tax=0.003, open_commission=0.001, close_commission=0.001, min_commission=5), type='stock')


# 每天交易时做的事情
def handle_data(context, data):
    if g.if_trade == True:
        sell_all = tralling_stop(context, '000300.XSHG')
        if sell_all:
            list_to_buy = []
            list_to_sell = []
            for i in context.portfolio.positions:
                list_to_sell.append(i)
        else: # 如果不需要追踪止损
            list_to_buy = stocks_to_buy(context)
            list_to_sell = stocks_to_sell(context, list_to_buy)
        # 先卖后买，腾出资金
        sell_operation(list_to_sell)
        buy_operation(context, list_to_buy)
    g.if_trade = False

# 计算股票 PEG 值
def get_PEG(context, stock_list):
    q_PE_G = query(valuation.code,valuation.pe_ratio,indicator.inc_net_profit_year_on_year).filter(valuation.code.in_(stock_list))
    df_PE_G = get_fundamentals(q_PE_G)
    # 筛选出成长股：删除市盈率或收益增长率为负值的股票
    df_Growth_PE_G = df_PE_G[(df_PE_G.pe_ratio>0)&(df_PE_G.inc_net_profit_year_on_year > 0)]
    # 去除 PE 或 G 为非数字或空值的行
    df_Growth_PE_G = df_Growth_PE_G.dropna()
    # 得到一个 Series 存放市盈率 TMM，即 PE 值
    Series_PE = df_Growth_PE_G.ix[:,'pe_ratio']
    # 得到一个 Series 存放 G 值
    Series_G = df_Growth_PE_G.ix[:,'inc_net_profit_year_on_year']
    # 计算 PEG
    Series_PEG = Series_PE / Series_G
    # 对应上股票代码
    Series_PEG.index = df_Growth_PE_G.ix[:,0]
    # 转化为 dataframe 类型
    df_PEG = pd.DataFrame(Series_PEG)
    return df_PEG
    
# 买入信号
def stocks_to_buy(context):
    list_to_buy = []
    df_PEG = get_PEG(context, g.feasible_stocks)
    # 将股票按 PEG 升序排列
    df_sort_PEG = df_PEG.sort_values(by=[0], ascending=True)
    for i in range(g.num_stocks):
        if df_sort_PEG.ix[i,0] < 0.5:
            list_to_buy.append(df_sort_PEG.index[i])
    return list_to_buy
    

# 卖出信号
def stocks_to_sell(context, list_to_buy):
    list_to_sell = []
    # 不需要的股票，全仓卖出
    for stock_sell in context.portfolio.positions:
        stock_sell_tralling_stop = tralling_stop(context, stock_sell)
        if stock_sell not in list_to_buy:
            list_to_sell.append(stock_sell)
        elif stock_sell_tralling_stop == 1:
            print('time to run')
            list_to_sell.append(stock_sell)
    return list_to_sell
    
def sell_operation(list_to_sell):
    for i in list_to_sell:
        order_target_value(i, 0)
        
def buy_operation(context, list_to_buy):
    for i in list_to_buy:
        g.capital_unit = context.portfolio.portfolio_value / len(list_to_buy)
        order_target_value(i, g.capital_unit)
        
# 根据大盘走势，做针对个股追踪止损操作
def tralling_stop(context, stock_code):
    # 获取历史数据
    Data_ATR = attribute_history(stock_code, g.ATR_timeperiod+10,'1d',['close','high','low'],df=False)
    close_ATR = Data_ATR['close']
    high_ATR = Data_ATR['high']
    low_ATR = Data_ATR['low']
    # 计算 ATR
    atr = talib.ATR(high_ATR, low_ATR, close_ATR)
    highest20 = max(close_ATR[-20:])
    if ((highest20 - close_ATR[-1]) > (2*atr[-1])):
        return 1
    else:
        return 0


## 收盘后运行函数
def after_market_close(context):
    log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time())))
    #得到当天所有成交记录
    trades = get_trades()
    for _trade in trades.values():
        log.info('成交记录：'+str(_trade))
    log.info('一天结束###################################################')

技术指标测试平台

  • 技术指标分析法(Technical Indicator)是证券投资中一种重要的操作方法，从传统交易方法来看，这是技术分析法的一个重要分支，但这些技术指标本质都是价量关系的数学表达，只不过公式不同，侧重点不同
  • 技术分析法的重要理论奠基——道氏理论(Dow Theory)认为市场行为包容一切，即所有的基础事件、经济事件、社会实践等最终都会反映到价格和成交量变化中来。这种分析法不建议再针对基本面做任何建模，因为除了量化分析之外的工作都是无效的
  • 使用 TA-Lib 进行相关计算
  • 关键之处在于交易规则，可以用金叉、死叉方法买入。但为了得到指标值对于股票涨跌力度的解释，更倾向于得到指标值之后，直接以此对股票排序，然后买卖股票
  • 一些重要的技术指标：MACD、布林带、RSI、CCI、MTM、KDJ、CMO、OBV、CORREL 等
  • 指标必须和止损、大盘判断、个股特指等因子完整地结合，才能具备一定的生命力

动量效应和反转效应

  • 在较为成熟的市场上，反转效应较强，或者说换手率过高的、投资标的选择自由的市场（特别是 A 股市场），反转效应显得更有优势
  • 虽然市场存在信息不对称，但是信息传导的速度过快，经常导致套利空间被压缩，在动量能够被观测到的时候，已经变成了反转
  • 大部分 A 股投资者不关心公司的基本面，其核心原因是 A 股定价和基本面关系较弱，而和资金推动力关系较强，且行业未形成垄断格局
相关文章