上漲趨勢策略
看了聚寬上大神寫的上漲趨勢策略，感覺不錯。感興趣的可以轉(zhuǎn)至苦咖啡《上漲趨勢策略修改版》，地址為：
https://www.joinquant.com/view/community/detail/15c783ea414ddac3abb8fd55f39df2cc?type=1
隨著中美對抗的持續(xù)進行，原有很多以美國為出口主要對象的企業(yè)必然走向內(nèi)卷，大企業(yè)之間的競爭也將進一步加大，只會強者恒強，進一步走向壟斷。A股全面注冊制也將到來，小盤股時代終將結(jié)束，跟著機構(gòu)抱團，找大型的，具有壟斷特征的股票才是出路。
作者在掘金平臺上對該策略進行了實現(xiàn)，對2019到2020年進行了回測，表現(xiàn)不錯。
策略源碼如下，覺得有幫助的，還望點贊關(guān)注！對量化投資感興趣的小伙伴可以添加如下微信群，一起用技術(shù)改變世界，實現(xiàn)財務自由。
掘金實盤可參考掘金量化實盤指導
# coding=utf-8from gm.api import *# 克隆自聚寬文章：https://www.joinquant.com/post/31232# 標題：2020年 267%年化 13%回撤 上漲趨勢策略修改版# 作者：苦咖啡# 克隆自聚寬文章：https://www.joinquant.com/post/31232# 標題：2020年 267%年化 13%回撤 上漲趨勢策略修改版# 作者：苦咖啡# 克隆自聚寬文章：https://www.joinquant.com/post/30863# 標題：上漲股策略2020年收益123%# 作者：scottchenrui# 2021-02-06# 張連重 掘金實現(xiàn)import numpy as npfrom scipy.stats import linregressfrom gm.api import *from datetime import datetimeimport pandas as pdfrom jqdatasdk import *#?jqdata?login??需替換為個人聚寬賬號auth('15520768082','ZXXXXXX')# 初始化函數(shù)，設定要操作的股票、基準等等def init(context):# 最大建倉數(shù)量context.max_hold_stock_nums = 2# 選出來的股票context.target_lists = []schedule(schedule_func=check_stocks, date_rule='1w', time_rule='09:32:00')schedule(schedule_func=sell, date_rule='1w', time_rule='09:34:00')schedule(schedule_func=buy, date_rule='1w', time_rule='09:35:00')# 股票篩選def check_stocks(context):today = context.now.strftime("%Y-%m-%d")last_date = get_trade_days(end_date=today, count=2)[0]current = context.now.strftime("%Y-%m-%d %H:%M:%S")last_date = get_trade_days(end_date=today, count=2)[0]# 滬深300成分股check_out_lists = get_index_stocks("000300.XSHG", date=last_date)check_out_lists = st_check(context,check_out_lists)check_out_lists = science_check(context,check_out_lists)stocks = filter_stks_listed_n_days(check_out_lists,last_date,100)df = get_price(stocks,end_date=current,frequency='1m',fields=['open','close','high_limit','low_limit','paused'] ,count=1,panel=False)df = df[(df['close'] < df['high_limit']) & (df['close'] > df['low_limit']) & (df['paused'] == 0)]checked_stocks = df['code'].tolist()# 昨收盤價不高于260元/股s_close_1 = get_price(checked_stocks,end_date=last_date,frequency='1d',fields=['close'] ,count=1,panel=False)s_close_1.drop(['time'], axis=1,inplace=True)s_close_1.set_index(["code"], inplace=True)check_out_lists = list(s_close_1[s_close_1['close'] <= 260].index)# 近30個交易日的最高價 / 昨收盤價 <=1.1, 即HHV(HIGH,30)/C[-1] <= 1.1high_30 = get_price(check_out_lists,end_date=last_date,frequency='1d',fields=['high'] ,count=30,panel=False)high_max_30 = high_30.pivot(index='time',columns='code',values='high').max()dict_high_max_30 = {'code':high_max_30.index,'high':high_max_30.values}df_high_max_30 = pd.DataFrame(dict_high_max_30)df_high_max_30.set_index('code',drop=True,inplace=True)s_fall = df_high_max_30['high'] / s_close_1['close']check_out_lists = list(s_fall[s_fall <= 1.1].index)# 近7個交易日的交易量均值 與 近180給交易日的成交量均值 相比，放大不超過1.5倍 MA(VOL,7)/MA(VOL,180) <=1.5volume_180 = get_price(check_out_lists,end_date=last_date,frequency='1d',fields=['volume'] ,count=180,panel=False)volume_mean_7 = volume_180.pivot(index='time',columns='code',values='volume').iloc[-7:].mean()volume_mean_180 = volume_180.pivot(index='time',columns='code',values='volume').mean()s_vol_ratio = volume_mean_7 / volume_mean_180check_out_lists = list(s_vol_ratio[s_vol_ratio <= 1.5].index)# 對近120個交易日的股價進行線性回歸：入選條件 slope / intercept > 0.005 and r_value**2 > 0.8target_dict = {}x = np.arange(120)for stock in check_out_lists:y = get_price(stock,end_date=last_date,frequency='1d',fields=['close'] ,count=120,panel=False)['close']slope, intercept, r_value, p_value, std_err = linregress(x, y)if slope / intercept > 0.005 and r_value ** 2 > 0.8:target_dict[stock] = r_value ** 2# 入選股票按照R Square 降序排序, 取前N名context.target_lists = []if target_dict:df_score = pd.DataFrame.from_dict(target_dict, orient='index', columns=['score', ]).sort_values(by='score', ascending=False)#context.target_lists = list(df_score.index[:context.max_hold_stock_nums])# 去除ST股票def st_check(context,security_list):currentDate = context.now.strftime("%Y-%m-%d")current_data = get_extras('is_st', security_list, end_date=currentDate, df=True, count=1)security_list = [stock for stock in security_list if not current_data[stock][-1]]# 返回結(jié)果return security_list#def filter_stks_listed_n_days(source_stk_list, end_date, n=100):# type: (list, Union[str, datetime.date, datetime.datetime], int) -> list"""過濾掉source_stk_list中尚未上市的，或者上市天數(shù)不足n天的股票"""# 1. 取得n個交易日之前的交易日期trd_datetrd_date = get_trade_days(end_date=end_date, count=n)[0]# 2. 取trd_date日就已經(jīng)上市的所有股票all_stks = get_all_securities(date=trd_date)# 3. 過濾source_stk_list，剔除掉不在all_stks中的valid_stk_list = list(all_stks[all_stks.index.isin(source_stk_list)].index)return valid_stk_list# 去除科創(chuàng)板股票def science_check(context,security_list):stock_list = []for stock in security_list:if not stock.startswith('688'):stock_list.append(stock)# 返回結(jié)果return stock_list# 交易函數(shù) - 入場def buy(context):buy_lists = context.target_listsif buy_lists:amount = context.account().cash['nav'] / len(buy_lists)for stock in buy_lists:stock_code_jj = transStockCode(stock,'JJ')if stock_code_jj in get_position_list(context):account_position = context.account().position(symbol=stock_code_jj,side = PositionSide_Long)if abs(amount - account_position['amount']) > account_position['price'] * 100 + 5:_order = order_target_value(symbol=stock, value=amount, position_side=PositionSide_Long, order_type=OrderType_Market)if _order is not None:print('調(diào)倉: %s (%s)' % (get_security_info(stock).display_name, stock))for stock in buy_lists:stock_code_jj = transStockCode(stock,'JJ')if stock_code_jj not in get_position_list(context):_order = order_target_value(symbol=stock_code_jj, value=amount, position_side=PositionSide_Long, order_type=OrderType_Market)if _order is not None:print('買入: %s (%s)' % (get_security_info(stock).display_name, stock))def sell(context):"""賣出不在buy_lists中的股票"""buy_lists = context.target_listsfor stock in get_position_list(context):stock_code_jk = transStockCode(stock,'JK')if stock_code_jk not in buy_lists:_order = order_target_value(symbol=stock, value=0, position_side=PositionSide_Long, order_type=OrderType_Market)if _order is not None:print('賣出: %s (%s)' % (get_security_info(stock_code_jk).display_name, stock))##def transStockListCode(stock_list_src,target_type):stock_list_target = []if target_type=='JJ':for stock in stock_list_src:target = 'SHSE' if stock[7:]=='XSHG' else 'SZSE'stock_list_target.append(target +'.'+ stock[0:6])elif target_type=='JK':for stock in stock_list_src:target = 'XSHG' if stock[0:4]=='SHSE' else 'XSHE'stock_list_target.append(stock[5:] +'.'+ target)return stock_list_target## 單股名稱轉(zhuǎn)換def transStockCode(stock_src,target_type):if target_type=='JJ':target = 'SHSE' if stock_src[7:]=='XSHG' else 'SZSE'return target +'.'+ stock_src[0:6]elif target_type=='JK':target = 'XSHG' if stock_src[0:4]=='SHSE' else 'XSHE'return stock_src[5:] +'.'+ target## 獲取持倉股票列表def get_position_list(context):in_stock = []Account_positions = context.account().positions()for position in Account_positions:in_stock.append(position['symbol'])return in_stockif __name__ == '__main__':'''strategy_id策略ID, 由系統(tǒng)生成filename文件名, 請與本文件名保持一致mode運行模式, 實時模式:MODE_LIVE回測模式:MODE_BACKTESTtoken綁定計算機的ID, 可在系統(tǒng)設置-密鑰管理中生成backtest_start_time回測開始時間backtest_end_time回測結(jié)束時間backtest_adjust股票復權(quán)方式, 不復權(quán):ADJUST_NONE前復權(quán):ADJUST_PREV后復權(quán):ADJUST_POSTbacktest_initial_cash回測初始資金backtest_commission_ratio回測傭金比例backtest_slippage_ratio回測滑點比例'''run(strategy_id='b056605f-687a-11eb-8a4e-5254009e6375',filename='main.py',mode=MODE_BACKTEST,????????token='f8e34a0dacdbbbba2f2cab37b9c01772f80c53',backtest_start_time='2019-01-01 08:00:00',backtest_end_time='2021-02-10 16:00:00',backtest_adjust=ADJUST_PREV,backtest_initial_cash=50000,backtest_commission_ratio=0.0001,backtest_slippage_ratio=0.0001)
評論
圖片
表情