B站热榜视频，炒股源码来了！
大家好，我是 Jack。
视频中，承诺的量化交易教程，它来了！
这期视频播放近 70 多万，后来经过 B 站编辑老师的建议，我对视频的部分内容进行了删减。
视频中，我提到，后续我会晒实仓情况，这个行为存在政策风险。
其实，很多好心读者也都提醒过我，这样不妥，很容易造成粉丝跟盘。
所以，后面我就不公布自己的实仓情况了，我们只探讨量化交易技术本身。
希望各位理解。
同时，我自己改进的量化交易算法，里面有一些激进的选股策略，会在我人为圈定的 top20 的股池中，投票选择得分高的几只股票进行买卖。
这个也存在一个问题：
假如这篇文章，一万人阅读，10% 的人，也就是 1000 人跑了这个算法，并真投了一万元。
这也会造成极端情况下，同一时刻，一起交易一千万的情况。这样也是不好的。
所以，今天要说的这个量化交易算法，是我之前测试过的一个基础版策略，也是别人开源过的。
原理都弄懂，你也可以自己改进策略。
这个量化交易策略，8 年回测，收益 715.44%，最大回撤 28%。
OK，进入我们今天的正题，量化交易。
聚宽
我目前使用的是聚宽平台，这里也就以它为例进行讲解。
https://www.joinquant.com/
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 进行了重写。
我对每一行代码，都进行了详细的注释，并罗列了每个知识点，可以参考的文章。
直接看代码吧！
#-*- 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，我们下期见。
