大家好，我是 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，我们下期见。
·················END·················

推荐阅读

•   上了B站热门榜第一，被骂惨了！•   破解隔壁老王WiFi密码，太刺激了！•   手残党抠图有救了！
继续阅读
阅读原文
关键词
数据
策略
代码
斜率
信号