B站熱榜視頻，炒股源碼來(lái)了！

大家好，我是 Jack。

視頻中，承諾的量化交易教程，它來(lái)了！

這期視頻播放近 70 多萬(wàn)，后來(lái)經(jīng)過(guò) B 站編輯老師的建議，我對(duì)視頻的部分內(nèi)容進(jìn)行了刪減。

視頻中，我提到，后續(xù)我會(huì)曬實(shí)倉(cāng)情況，這個(gè)行為存在政策風(fēng)險(xiǎn)。

其實(shí)，很多好心讀者也都提醒過(guò)我，這樣不妥，很容易造成粉絲跟盤。

所以，后面我就不公布自己的實(shí)倉(cāng)情況了，我們只探討量化交易技術(shù)本身。

希望各位理解。

同時(shí)，我自己改進(jìn)的量化交易算法，里面有一些激進(jìn)的選股策略，會(huì)在我人為圈定的 top20 的股池中，投票選擇得分高的幾只股票進(jìn)行買賣。

這個(gè)也存在一個(gè)問(wèn)題：

假如這篇文章，一萬(wàn)人閱讀，10% 的人，也就是 1000 人跑了這個(gè)算法，并真投了一萬(wàn)元。

這也會(huì)造成極端情況下，同一時(shí)刻，一起交易一千萬(wàn)的情況。這樣也是不好的。

所以，今天要說(shuō)的這個(gè)量化交易算法，是我之前測(cè)試過(guò)的一個(gè)基礎(chǔ)版策略，也是別人開(kāi)源過(guò)的。

原理都弄懂，你也可以自己改進(jìn)策略。

這個(gè)量化交易策略，8 年回測(cè)，收益 715.44%，最大回撤 28%。

OK，進(jìn)入我們今天的正題，量化交易。

聚寬

我目前使用的是聚寬平臺(tái)，這里也就以它為例進(jìn)行講解。

https://www.joinquant.com/

PS：有聚寬工作的朋友嗎？廣告費(fèi)記得結(jié)一下。

聚寬是一個(gè)量化交易平臺(tái)，在這個(gè)平臺(tái)有很多開(kāi)源的量化交易策略，社區(qū)不錯(cuò)。

同時(shí)，使用這個(gè)平臺(tái)，還可以回測(cè)我們實(shí)現(xiàn)的策略。

左邊寫好代碼，選擇時(shí)間和金額，就可以使用歷史數(shù)據(jù)進(jìn)行回測(cè)。

因?yàn)樯婕暗骄帉懘a，所以你必須具備 Python 編程基礎(chǔ)。

沒(méi)有 Python 基礎(chǔ)的小伙伴，先看我的 Python 入門視頻吧：

https://www.bilibili.com/video/BV1Sh411a76E/

一定要先好好學(xué) Python，無(wú)論你是不是程序員，都很有用。

屬于，好學(xué)又實(shí)用的編程語(yǔ)言。

聚寬平臺(tái)，有兩個(gè) api，可以使用。

一個(gè)是在聚寬平臺(tái)使用的 api：

https://www.joinquant.com/help/api/help#api:API%E6%96%87%E6%A1%A3

如果你是在網(wǎng)頁(yè)，進(jìn)行回測(cè)，那就需要使用這個(gè) api。

另一個(gè)，就是本地化數(shù)據(jù) JQData：

https://www.joinquant.com/help/api/help#JQData:JQData

這個(gè) api 是我平時(shí)使用的本地化服務(wù)接口，只需要 pip 安裝一下，就可以本地環(huán)境調(diào)用接口，獲取數(shù)據(jù)了。

如果你有 Python 基礎(chǔ)，那我想這兩份 api 使用起來(lái)，應(yīng)該很簡(jiǎn)單。

ETF 動(dòng)量輪動(dòng)

今天要講的這個(gè)量化交易策略，就是在聚寬社區(qū)，其他人開(kāi)源的量化交易算法，起了個(gè)名字，叫 ETF 動(dòng)量輪動(dòng)。

其實(shí)，就是一種長(zhǎng)期定投 ETF 的策略，定投大法好。

策略核心有兩塊，選哪個(gè) ETF，以及何時(shí)買賣。

我將這個(gè)策略進(jìn)行了重構(gòu)，用本地化數(shù)據(jù) JQData 的 api 進(jìn)行了重寫。

我對(duì)每一行代碼，都進(jìn)行了詳細(xì)的注釋，并羅列了每個(gè)知識(shí)點(diǎn)，可以參考的文章。

直接看代碼吧！

#-*- 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 為你自己的帳號(hào)， bb 為你自己的密碼
jq.auth('aa','bb')

# http://fund.eastmoney.com/ETFN_jzzzl.html
stock_pool = [
    '159915.XSHE', # 易方達(dá)創(chuàng)業(yè)板ETF
    '510300.XSHG', # 華泰柏瑞滬深300ETF
    '510500.XSHG', # 南方中證500ETF
]

# 動(dòng)量輪動(dòng)參數(shù)
stock_num = 1           # 買入評(píng)分最高的前 stock_num 只股票
momentum_day = 29       # 最新動(dòng)量參考最近 momentum_day 的

ref_stock = '000300.XSHG' #用 ref_stock 做擇時(shí)計(jì)算的基礎(chǔ)數(shù)據(jù)
N = 18 # 計(jì)算最新斜率 slope，擬合度 r2 參考最近 N 天
M = 600 # 計(jì)算最新標(biāo)準(zhǔn)分 zscore，rsrs_score 參考最近 M 天
score_threshold = 0.7 # rsrs 標(biāo)準(zhǔn)分指標(biāo)閾值
# ma 擇時(shí)參數(shù)
mean_day = 20 # 計(jì)算結(jié)束 ma 收盤價(jià)，參考最近 mean_day
mean_diff_day = 3 # 計(jì)算初始 ma 收盤價(jià)，參考(mean_day + mean_diff_day)天前，窗口為 mean_diff_day 的一段時(shí)間

day = 1

# 1-1 選股模塊-動(dòng)量因子輪動(dòng) 
# 基于股票年化收益和判定系數(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'])
        # 收盤價(jià)
        y = data['log'] = np.log(data.close)
        # 分析的數(shù)據(jù)個(gè)數(shù)（天）
        x = data['num'] = np.arange(data.log.size)
        # 擬合 1 次多項(xiàng)式
        # 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 擇時(shí)模塊-計(jì)算線性回歸統(tǒng)計(jì)值
# 對(duì)輸入的自變量每日最低價(jià) x(series) 和因變量每日最高價(jià) y(series) 建立 OLS 回歸模型,返回元組(截距,斜率,擬合度)
# R2 統(tǒng)計(jì)學(xué)線性回歸決定系數(shù)，也叫判定系數(shù)，擬合優(yōu)度。
# R2 范圍 0 ~ 1，擬合優(yōu)度越大，自變量對(duì)因變量的解釋程度越高，越接近 1 越好。
# 公式說(shuō)明：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í)模塊-設(shè)定初始斜率序列
# 通過(guò)前 M 日最高最低價(jià)的線性回歸計(jì)算初始的斜率,返回斜率的列表
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 擇時(shí)模塊-計(jì)算標(biāo)準(zhǔn)分
# 通過(guò)斜率列表計(jì)算并返回截至回測(cè)結(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 擇時(shí)模塊-計(jì)算綜合信號(hào)
# 1.獲得 rsrs 與 MA 信號(hào),rsrs 信號(hào)算法參考優(yōu)化說(shuō)明，MA 信號(hào)為一段時(shí)間兩個(gè)端點(diǎn)的 MA 數(shù)值比較大小
# 2.信號(hào)同時(shí)為 True 時(shí)返回買入信號(hào)，同為 False 時(shí)返回賣出信號(hào)，其余情況返回持倉(cāng)不變信號(hào)
# 解釋：
#       MA 信號(hào)：MA 指標(biāo)是英文(Moving average)的簡(jiǎn)寫，叫移動(dòng)平均線指標(biāo)。
#       RSRS 擇時(shí)信號(hào)：
#               https://www.joinquant.com/view/community/detail/32b60d05f16c7d719d7fb836687504d6?type=1
def get_timing_signal(stock):
    # 計(jì)算 MA 信號(hào)
    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()
    # 計(jì)算 rsrs 信號(hào)
    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
    # 綜合判斷所有信號(hào)
    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] # 除去回測(cè)第一天的 slope ，避免運(yùn)行時(shí)重復(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))
        #獲取綜合擇時(shí)信號(hào)
        timing_signal = get_timing_signal(ref_stock)
        print('今日擇時(shí)信號(hào):{}'.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))
    #獲取綜合擇時(shí)信號(hào)
    timing_signal = get_timing_signal(ref_stock)
    print('今日擇時(shí)信號(hào):{}'.format(timing_signal))
    print('*' * 100)

策略很短，不到 200 行。

需要注意的是，這個(gè)本地化的 api，需要通過(guò)官網(wǎng)申請(qǐng)后，才能使用。

申請(qǐng)地址：

https://www.joinquant.com/default/index/sdk

對(duì)應(yīng)的，可以直接在聚寬平臺(tái)運(yùn)行的代碼，在這里：

https://github.com/Jack-Cherish/quantitative/blob/main/lesson1/quantitive-etf-jq.py

輸入代碼，就可以直接運(yùn)行，回測(cè)效果了。

時(shí)間有限，這里先寫這么多。

這個(gè)策略，只用了寬基，輪動(dòng)選擇。

后續(xù)我會(huì)繼續(xù)講解，怎樣將這個(gè)策略部署到我們的服務(wù)器上，并定時(shí)給我們的手機(jī)發(fā)送郵件，進(jìn)行交易提醒。

股市有風(fēng)險(xiǎn)，入市需謹(jǐn)慎，請(qǐng)謹(jǐn)慎使用~

有什么問(wèn)題，歡迎在評(píng)論區(qū)里留言。

我是 Jack，我們下期見(jiàn)。