大家好，我是 Jack。
视频中，承诺的量化交易教程，它来了！
这期视频播放近 70 多万，后来经过 B 站编辑老师的建议，我对视频的部分内容进行了删减。
视频中，我提到，后续我会晒实仓情况，这个行为存在政策风险。
其实，很多好心读者也都提醒过我，这样不妥，很容易造成粉丝跟盘。
所以，后面我就不公布自己的实仓情况了，我们只探讨量化交易技术本身。
希望各位理解。
同时，我自己改进的量化交易算法，里面有一些激进的选股策略，会在我人为圈定的 top20 的股池中，投票选择得分高的几只股票进行买卖。
这个也存在一个问题：
假如这篇文章，一万人阅读，10% 的人，也就是 1000 人跑了这个算法，并真投了一万元。
这也会造成极端情况下，同一时刻，一起交易一千万的情况。这样也是不好的。
所以，今天要说的这个量化交易算法，是我之前测试过的一个基础版策略，也是别人开源过的。
原理都弄懂，你也可以自己改进策略。
这个量化交易策略，8 年回测，收益 715.44%，最大回撤 28%。
OK，进入我们今天的正题，量化交易。
聚宽
我目前使用的是聚宽平台，这里也就以它为例进行讲解。
PS：有聚宽工作的朋友吗？广告费记得结一下。
聚宽是一个量化交易平台，在这个平台有很多开源的量化交易策略，社区不错。
同时，使用这个平台，还可以回测我们实现的策略。
左边写好代码，选择时间和金额，就可以使用历史数据进行回测。
因为涉及到编写代码，所以你必须具备 Python 编程基础。
没有 Python 基础的小伙伴，先看我的 Python 入门视频吧：
https://www.bilibili.com/video/BV1Sh411a76E/
一定要先好好学 Python，无论你是不是程序员，都很有用。
属于，好学又实用的编程语言。
聚宽平台，有两个 api，可以使用。
一个是在聚宽平台使用的 api：
https://www.joinquant.com/help/api/help#api:API%E6%96%87%E6%A1%A3
如果你是在网页，进行回测，那就需要使用这个 api。
另一个，就是本地化数据 JQData：
https://www.joinquant.com/help/api/help#JQData:JQData
这个 api 是我平时使用的本地化服务接口，只需要 pip 安装一下，就可以本地环境调用接口，获取数据了。
如果你有 Python 基础，那我想这两份 api 使用起来，应该很简单。
ETF 动量轮动
今天要讲的这个量化交易策略，就是在聚宽社区，其他人开源的量化交易算法，起了个名字，叫 ETF 动量轮动。
其实，就是一种长期定投 ETF 的策略，定投大法好。
策略核心有两块，选哪个 ETF，以及何时买卖。
我将这个策略进行了重构，用本地化数据 JQData 的 api 进行了重写。
我对每一行代码，都进行了详细的注释，并罗列了每个知识点，可以参考的文章。
直接看代码吧！
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 | #-*- codig:utf-8 -*- import jqdatasdk as jq from datetime import datetime, timedelta import time import numpy as np import math # https://www.joinquant.com/help/api/help#api:API%E6%96%87%E6%A1%A3 # https://www.joinquant.com/help/api/help#JQData:JQData # aa 为你自己的帐号， bb 为你自己的密码 jq.auth('aa','bb') # http://fund.eastmoney.com/ETFN_jzzzl.html stock_pool = [ '159915.XSHE', # 易方达创业板ETF '510300.XSHG', # 华泰柏瑞沪深300ETF '510500.XSHG', # 南方中证500ETF ] # 动量轮动参数 stock_num = 1 # 买入评分最高的前 stock_num 只股票 momentum_day = 29 # 最新动量参考最近 momentum_day 的 ref_stock = '000300.XSHG' #用 ref_stock 做择时计算的基础数据 N = 18 # 计算最新斜率 slope，拟合度 r2 参考最近 N 天 M = 600 # 计算最新标准分 zscore，rsrs_score 参考最近 M 天 score_threshold = 0.7 # rsrs 标准分指标阈值 # ma 择时参数 mean_day = 20 # 计算结束 ma 收盘价，参考最近 mean_day mean_diff_day = 3 # 计算初始 ma 收盘价，参考(mean_day + mean_diff_day)天前，窗口为 mean_diff_day 的一段时间 day = 1 # 1-1 选股模块-动量因子轮动 # 基于股票年化收益和判定系数打分,并按照分数从大到小排名 def get_rank(stock_pool): score_list = [] for stock in stock_pool: current_dt = time.strftime("%Y-%m-%d", time.localtime()) current_dt = datetime.strptime(current_dt, '%Y-%m-%d') previous_date = current_dt - timedelta(days = day) data = jq.get_price(stock, end_date = previous_date, count = momentum_day, frequency='daily', fields=['close']) # 收盘价 y = data['log'] = np.log(data.close) # 分析的数据个数（天） x = data['num'] = np.arange(data.log.size) # 拟合 1 次多项式 # y = kx + b, slope 为斜率 k，intercept 为截距 b slope, intercept = np.polyfit(x, y, 1) # (e ^ slope) ^ 250 - 1 annualized_returns = math.pow(math.exp(slope), 250) - 1 r_squared = 1 - (sum((y - (slope * x + intercept))**2) / ((len(y) - 1) * np.var(y, ddof=1))) score = annualized_returns * r_squared score_list.append(score) stock_dict = dict(zip(stock_pool, score_list)) sort_list = sorted(stock_dict.items(), key = lambda item:item[1], reverse = True) print("#" * 30 + "候选" + "#" * 30) for stock in sort_list: stock_code = stock[0] stock_score = stock[1] security_info = jq.get_security_info(stock_code) stock_name = security_info.display_name print('{}({}):{}'.format(stock_name, stock_code, stock_score)) print('#' * 64) code_list = [] for i in range((len(stock_pool))): code_list.append(sort_list[i][0]) rank_stock = code_list[0:stock_num] return rank_stock # 2-1 择时模块-计算线性回归统计值 # 对输入的自变量每日最低价 x(series) 和因变量每日最高价 y(series) 建立 OLS 回归模型,返回元组(截距,斜率,拟合度) # R2 统计学线性回归决定系数，也叫判定系数，拟合优度。 # R2 范围 0 ~ 1，拟合优度越大，自变量对因变量的解释程度越高，越接近 1 越好。 # 公式说明：https://blog.csdn.net/snowdroptulip/article/details/79022532 # https://www.cnblogs.com/aviator999/p/10049646.html def get_ols(x, y): slope, intercept = np.polyfit(x, y, 1) r2 = 1 - (sum((y - (slope * x + intercept))**2) / ((len(y) - 1) * np.var(y, ddof=1))) return (intercept, slope, r2) # 2-2 择时模块-设定初始斜率序列 # 通过前 M 日最高最低价的线性回归计算初始的斜率,返回斜率的列表 def initial_slope_series(): current_dt = time.strftime("%Y-%m-%d", time.localtime()) current_dt = datetime.strptime(current_dt, '%Y-%m-%d') previous_date = current_dt - timedelta(days = day) data = jq.get_price(ref_stock, end_date = previous_date, count = N + M, frequency='daily', fields=['high', 'low']) return [get_ols(data.low[i:i+N], data.high[i:i+N])[1] for i in range(M)] # 2-3 择时模块-计算标准分 # 通过斜率列表计算并返回截至回测结束日的最新标准分 def get_zscore(slope_series): mean = np.mean(slope_series) std = np.std(slope_series) return (slope_series[-1] - mean) / std # 2-4 择时模块-计算综合信号 # 1.获得 rsrs 与 MA 信号,rsrs 信号算法参考优化说明，MA 信号为一段时间两个端点的 MA 数值比较大小 # 2.信号同时为 True 时返回买入信号，同为 False 时返回卖出信号，其余情况返回持仓不变信号 # 解释： # MA 信号：MA 指标是英文(Moving average)的简写，叫移动平均线指标。 # RSRS 择时信号： # https://www.joinquant.com/view/community/detail/32b60d05f16c7d719d7fb836687504d6?type=1 def get_timing_signal(stock): # 计算 MA 信号 current_dt = time.strftime("%Y-%m-%d", time.localtime()) current_dt = datetime.strptime(current_dt, '%Y-%m-%d') previous_date = current_dt - timedelta(days = day) close_data = jq.get_price(ref_stock, end_date = previous_date, count = mean_day + mean_diff_day, frequency = 'daily', fields = ['close']) # 0 0 0 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1，23 天，要后 20 天 today_MA = close_data.close[mean_diff_day:].mean() # 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 0 0 0，23 天，要前 20 天 before_MA = close_data.close[:-mean_diff_day].mean() # 计算 rsrs 信号 high_low_data = jq.get_price(ref_stock, end_date = previous_date, count = N, frequency='daily', fields = ['high', 'low']) intercept, slope, r2 = get_ols(high_low_data.low, high_low_data.high) slope_series.append(slope) rsrs_score = get_zscore(slope_series[-M:]) * r2 # 综合判断所有信号 if rsrs_score > score_threshold and today_MA > before_MA: return "BUY" elif rsrs_score < -score_threshold and today_MA < before_MA: return "SELL" else: return "KEEP" slope_series = initial_slope_series()[:-1] # 除去回测第一天的 slope ，避免运行时重复加入 def get_test(): for each_day in range(1, 100)[::-1]: current_dt = time.strftime("%Y-%m-%d", time.localtime()) current_dt = datetime.strptime(current_dt, '%Y-%m-%d') previous_date = current_dt - timedelta(days = each_day - 1) day = each_day print(each_day, previous_date) check_out_list = get_rank(stock_pool) for each_check_out in check_out_list: security_info = jq.get_security_info(each_check_out) stock_name = security_info.display_name stock_code = each_check_out print('今日自选股:{}({})'.format(stock_name, stock_code)) #获取综合择时信号 timing_signal = get_timing_signal(ref_stock) print('今日择时信号:{}'.format(timing_signal)) print('*' * 100) if __name__ == "__main__": check_out_list = get_rank(stock_pool) for each_check_out in check_out_list: security_info = jq.get_security_info(each_check_out) stock_name = security_info.display_name stock_code = each_check_out print('今日自选股:{}({})'.format(stock_name, stock_code)) #获取综合择时信号 timing_signal = get_timing_signal(ref_stock) print('今日择时信号:{}'.format(timing_signal)) print('*' * 100) |
策略很短，不到 200 行。
需要注意的是，这个本地化的 api，需要通过官网申请后，才能使用。
申请地址：
https://www.joinquant.com/default/index/sdk
对应的，可以直接在聚宽平台运行的代码，在这里：
https://github.com/Jack-Cherish/quantitative/blob/main/lesson1/quantitive-etf-jq.py
输入代码，就可以直接运行，回测效果了。
时间有限，这里先写这么多。
这个策略，只用了宽基，轮动选择。
后续我会继续讲解，怎样将这个策略部署到我们的服务器上，并定时给我们的手机发送邮件，进行交易提醒。
股市有风险，入市需谨慎，请谨慎使用~
有什么问题，欢迎在评论区里留言。
我是 Jack，我们下期见。
