B站熱榜視頻，炒股源碼來了！
大家好，我是 Jack。
視頻中，承諾的量化交易教程，它來了！
這期視頻播放近 70 多萬，后來經(jīng)過 B 站編輯老師的建議，我對視頻的部分內(nèi)容進行了刪減。
視頻中，我提到，后續(xù)我會曬實倉情況，這個行為存在政策風(fēng)險。
其實，很多好心讀者也都提醒過我，這樣不妥，很容易造成粉絲跟盤。
所以，后面我就不公布自己的實倉情況了，我們只探討量化交易技術(shù)本身。
希望各位理解。
同時，我自己改進的量化交易算法，里面有一些激進的選股策略，會在我人為圈定的 top20 的股池中，投票選擇得分高的幾只股票進行買賣。
這個也存在一個問題：
假如這篇文章，一萬人閱讀，10% 的人，也就是 1000 人跑了這個算法，并真投了一萬元。
這也會造成極端情況下，同一時刻，一起交易一千萬的情況。這樣也是不好的。
所以，今天要說的這個量化交易算法，是我之前測試過的一個基礎(chǔ)版策略，也是別人開源過的。
原理都弄懂，你也可以自己改進策略。
這個量化交易策略，8 年回測，收益 715.44%，最大回撤 28%。
OK，進入我們今天的正題，量化交易。
聚寬
我目前使用的是聚寬平臺，這里也就以它為例進行講解。
https://www.joinquant.com/
PS：有聚寬工作的朋友嗎？廣告費記得結(jié)一下。
聚寬是一個量化交易平臺，在這個平臺有很多開源的量化交易策略，社區(qū)不錯。
同時，使用這個平臺，還可以回測我們實現(xiàn)的策略。
左邊寫好代碼，選擇時間和金額，就可以使用歷史數(shù)據(jù)進行回測。
因為涉及到編寫代碼，所以你必須具備 Python 編程基礎(chǔ)。
沒有 Python 基礎(chǔ)的小伙伴，先看我的 Python 入門視頻吧：
https://www.bilibili.com/video/BV1Sh411a76E/
一定要先好好學(xué) Python，無論你是不是程序員，都很有用。
屬于，好學(xué)又實用的編程語言。
聚寬平臺，有兩個 api，可以使用。
一個是在聚寬平臺使用的 api：
https://www.joinquant.com/help/api/help#api:API%E6%96%87%E6%A1%A3
如果你是在網(wǎng)頁，進行回測，那就需要使用這個 api。
另一個，就是本地化數(shù)據(jù) JQData：
https://www.joinquant.com/help/api/help#JQData:JQData
這個 api 是我平時使用的本地化服務(wù)接口，只需要 pip 安裝一下，就可以本地環(huán)境調(diào)用接口，獲取數(shù)據(jù)了。
如果你有 Python 基礎(chǔ)，那我想這兩份 api 使用起來，應(yīng)該很簡單。
ETF 動量輪動
今天要講的這個量化交易策略，就是在聚寬社區(qū)，其他人開源的量化交易算法，起了個名字，叫 ETF 動量輪動。
其實，就是一種長期定投 ETF 的策略，定投大法好。
策略核心有兩塊，選哪個 ETF，以及何時買賣。
我將這個策略進行了重構(gòu)，用本地化數(shù)據(jù) 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', # 易方達創(chuàng)業(yè)板ETF
'510300.XSHG', # 華泰柏瑞滬深300ETF
'510500.XSHG', # 南方中證500ETF
]
# 動量輪動參數(shù)
stock_num = 1 # 買入評分最高的前 stock_num 只股票
momentum_day = 29 # 最新動量參考最近 momentum_day 的
ref_stock = '000300.XSHG' #用 ref_stock 做擇時計算的基礎(chǔ)數(shù)據(jù)
N = 18 # 計算最新斜率 slope，擬合度 r2 參考最近 N 天
M = 600 # 計算最新標(biāo)準(zhǔn)分 zscore，rsrs_score 參考最近 M 天
score_threshold = 0.7 # rsrs 標(biāo)準(zhǔn)分指標(biāo)閾值
# ma 擇時參數(shù)
mean_day = 20 # 計算結(jié)束 ma 收盤價，參考最近 mean_day
mean_diff_day = 3 # 計算初始 ma 收盤價，參考(mean_day + mean_diff_day)天前，窗口為 mean_diff_day 的一段時間
day = 1
# 1-1 選股模塊-動量因子輪動
# 基于股票年化收益和判定系數(shù)打分,并按照分?jǐn)?shù)從大到小排名
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)
# 分析的數(shù)據(jù)個數(shù)（天）
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 擇時模塊-計算線性回歸統(tǒng)計值
# 對輸入的自變量每日最低價 x(series) 和因變量每日最高價 y(series) 建立 OLS 回歸模型,返回元組(截距,斜率,擬合度)
# R2 統(tǒng)計學(xué)線性回歸決定系數(shù)，也叫判定系數(shù)，擬合優(yōu)度。
# R2 范圍 0 ~ 1，擬合優(yōu)度越大，自變量對因變量的解釋程度越高，越接近 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 擇時模塊-設(shè)定初始斜率序列
# 通過前 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 擇時模塊-計算標(biāo)準(zhǔn)分
# 通過斜率列表計算并返回截至回測結(jié)束日的最新標(biāo)準(zhǔn)分
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 信號算法參考優(yōu)化說明，MA 信號為一段時間兩個端點的 MA 數(shù)值比較大小
# 2.信號同時為 True 時返回買入信號，同為 False 時返回賣出信號，其余情況返回持倉不變信號
# 解釋：
# MA 信號：MA 指標(biāo)是英文(Moving average)的簡寫，叫移動平均線指標(biāo)。
# 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 ，避免運行時重復(fù)加入
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，需要通過官網(wǎng)申請后，才能使用。
申請地址：
https://www.joinquant.com/default/index/sdk
對應(yīng)的，可以直接在聚寬平臺運行的代碼，在這里：
https://github.com/Jack-Cherish/quantitative/blob/main/lesson1/quantitive-etf-jq.py
輸入代碼，就可以直接運行，回測效果了。
時間有限，這里先寫這么多。
這個策略，只用了寬基，輪動選擇。
后續(xù)我會繼續(xù)講解，怎樣將這個策略部署到我們的服務(wù)器上，并定時給我們的手機發(fā)送郵件，進行交易提醒。
股市有風(fēng)險，入市需謹(jǐn)慎，請謹(jǐn)慎使用~
有什么問題，歡迎在評論區(qū)里留言。
我是 Jack，我們下期見。
