[策略模板] 一个新手适用的交易策略的模板（分钟回测，过滤涨跌停、停牌……）

一个新手适用的交易模型的模板

新鲜出品： https://www.ricequant.com/community/topic/1150/?utm_source=v2ex

关键词：策略模板、策略、策略交易、新人、模板、模块……

引言： 在 rice 里混了大半年了，学习了不少大牛的有用知识，也编写了一大堆的有的没的策略，但是每次都面临大量的重复劳动，费时费力，于是这里就总结了一个适合新人的交易策略的模板分享给大家。

原理: 看了大家的策略，和查阅了一些资料，也总结了和归纳了一些，大概分为，选股、进场时机、持仓平衡、现金管理、出场时机、风险管理，一些工具组件~ 不废话了，直接上 demo 代码简单写~

模板代码： 说明：分钟回测，组合初始 100 万现金，交易手续默认的，无卖空， benchmark 为默认，进场策略输出了股票列表，出场策略也是返回要卖出的股票列表……

1 、传奇的小市值策略（市值最小的 100 只股票做为每天的备选列表），这个因子表现的最好，为避免模板的 demo 曲线表现太差，所以用了这个吸引眼球的选股因子，高手勿喷，勿笑；

2 、进场以大家熟悉的 5 日均线上传 10 日均线，并保持 20 日线为进场条件；（感觉自己写复杂了，反正是模板）

3 、出场为低于 20 日线的 99%为强制出场（ 20 日线，在近两年基本作为了行业标配了，反正有点用）

4 、持仓策略：在市值不便的情况下平均持仓，每日临近收盘进行再平衡，理想情况保持每只持仓占比相等（

5 、现金管理：本来想再收盘前现金买进“银华日利”，但由于默认市价交易滑点太大，就省略了（要限价交易才可能有利润，但是这里也发现尾盘表现非常不好，就注释掉了）

6 、风险管理：略了，流传一个大盘跌过 3%的强制止损风险策略，小市值也可以增加二八轮动的择时，没有加上，有兴趣可以自己弄着玩，加这个模块里；

7 、交易方式：为了避免过大的成家量超过 25%的 error ，这里都下的限价单，但是后续模块化吧，另控制了单只持仓不超过 10%~（模块写起来比较复杂，还要新建 dict 进行撤单再下单等计算，后续成熟了，再拿出来分享）

8 、工具：

trans 、历史数据强制转化成真正的 DataFrame(效率问题，做了.T 的来回变换)，问 licco 说，其实 2016 年 5 月 2X 后就不需要了

n 日内随机交易的收益率概率（例子中未用到）

多个 list 里取交集，懒得每次都写了，干脆写了个小函数

标的上市自然日的函数，避免次新股对收益干扰太大，要真像炒次新股，要好好研究一下，个人做过尝试发现，高风险高利润，大盘择时比较关键，干脆这里做了过滤比如一定要上市超过 60 个自然日

是否涨跌停区间，一定要可交易，人家都封版了，交易量有，关咱策略毛事，所以也进行了过滤

因为分钟回测，所以选择了 14 ： 50 作为买入时间点，而出场选择每 15 分钟采样计算一次（性能压力和必要性的问题）， 14 ： 59 （后来发现深圳市场应该 14 点 56 ，要不 57 开始深圳会集合竞价了，但是例子里没有调整，所以还是 error 一大堆，见笑了）

最后，个人也是新学小白用户，以新人最常见的 5 日金叉做例子，结合小市值选股做了这个模板，期望可以减少新人的重复无价值劳动~另还请大牛们给与指正~

import pandas as pd import numpy as np import time import math import itertools # 数据准备 def init_variables (context): context.init = 0 context.days = 0 context.barcount = 0 context.choosenum = 300 context.obv = 50 context.tj1 = 5 # 5 日均线 context.tj2 = 10 # 10 日均线 context.tj3 = 20 # 20 日均线 context.his = pd.DataFrame() return '''第 1 部、选择标的''' def choose_target(context): # 最小市值的 100 只标的 df = get_fundamentals( query(fundamentals.eod_derivative_indicator.market_cap) .order_by(fundamentals.eod_derivative_indicator.market_cap.asc()) .limit(context.choosenum) ) context.stocks = [stock for stock in df][:100] return context.stocks '''第 2 部、入场策略''' #2.1 大盘环境问题 #可增加外部数据 #2.2 个股选择问题，最后还要过滤非跌停、上市天数、非停牌的标的（ st 未过滤） def for_buy(context, bar_dict, his): #2.2.1 备选中标的站上 5 日线 def tj1(context, bar_dict, his): ma_n = pd.rolling_mean(his, context.tj1) temp = his - ma_n temp_s = list(temp[temp>0].iloc[-1,:].dropna().index) return temp_s #2.2.2 备选中标的站上 10 日线 def tj2(context, bar_dict, his): ma_n = pd.rolling_mean(his, context.tj2) temp = his - ma_n temp_s = list(temp[temp>0].iloc[-1,:].dropna().index) return temp_s #2.2.2 所谓金叉，今天短均线大于长均线，上一个 bar 反之 def tj3(context, bar_dict, his): mas = pd.rolling_mean(his, context.tj1) mal = pd.rolling_mean(his, context.tj2) temp = mas - mal temp_jc = list(temp[temp>0].iloc[-1,:].dropna().index) temp_r = list(temp[temp>0].iloc[-2,:].dropna().index) temp = [] for stock in temp_jc: if stock not in temp_r: temp.append(stock) return temp #整合各个子条件的交集 l1 = tj1(context, bar_dict, his) l2 = tj2(context, bar_dict, his) l3 = tj3(context, bar_dict, his) l_tar = jj_list([l1,l2,l3]) to_buy = [] #过滤上市时间、是否涨停、是否停牌等条件 if l_tar: for stock in l_tar: con1 = ipo_days(stock,context.now)>60 con2 = zdt_trade(stock,context,bar_dict) con3 = bar_dict[stock].is_trading if con1 & con2 & con3: to_buy.append(stock) return to_buy '''第 3 部、持仓组合的微调策略''' # 平均市值做微调 def for_balance(context, bar_dict): #mvalues = context.portfolio.market_value #avalues = context.portfolio.portfolio_value #per = mvalues / avalues hlist = [] for stock in context.portfolio.positions: hlist.append([stock,bar_dict[stock].last * context.portfolio.positions[stock].quantity]) if hlist: hlist = sorted(hlist,key=lambda x:x[1], reverse=True) temp = 0 for li in hlist: temp += li[1] for li in hlist: if bar_dict[li[0]].is_trading: order_target_value(li[0], temp/len(hlist)) return '''第 4 部、出场策略''' # 小于 20 日均线，并且可交易，没跌停 def for_sell(context, bar_dict): to_sell = [] for stock in context.portfolio.positions: con1 = bar_dict[stock].last < 0.99 * bar_dict[stock].mavg(20, frequency='day') con2 = bar_dict[stock].is_trading con3 = zdt_trade(stock,context,bar_dict) if con1 & con2 & con3: to_sell.append(stock) return to_sell '''第 5 部、闲置资金效率最大化''' def for_cash(context, bar_dict): cash = context.portfolio.cash #order_target_value('511880.XSHG',cash) 注释掉因为滑点太大，可以买一个货基，或者逆回购 return '''第 6 部、风险控制''' def alert_risk(context,bar_dict): #这里如果给出策略，要强制执行，注意在 handle 优先级高于所有 pass '''第 7 部、备用组件''' #7.1 将 his 的非标 DF 进行转换， licco 说现在不用转换了，我还是保留了：） def trans(df): temp = pd.DataFrame() for col in df.index: temp[col] = df.T[col] return temp.T #7.2 计算 n 日概率随机交易的概率收益率 def rts_sj(df,n,m): dfp_pct = df.pct_change() def the_list(df,n,m): temp = [] for i in range(n,n+m): temp.append(df.iloc[-i,:] + 1) return temp def from_list(self,num): result = [] for i in range(1,num+1): result.extend(list(itertools.combinations(self,i))) return result def rts_n(tu): sum0 = [] for i in tu: temp = 1 for z in i: temp = temp * z temp = temp**(1/len(i)) sum0.append(temp) sum1 = 0 for i in sum0: sum1 = sum1 + i - 1 return sum1/len(sum0) return rts_n(from_list(the_list(dfp_pct,n,m),m)) #7.3 多 list 获得并集 def jj_list(tar_list): temp = tar_list[0] for i in tar_list: temp = list(set(temp).intersection(set(i))) return temp #7.4 标的上市时间距离参照时间的自然日数量 def ipo_days(stock, today): ssrq = instruments(stock).listed_date.replace(tzinfo=None) today = today.replace(tzinfo=None) return (today - ssrq).days #7.5 判断当前标在可交易区间内（非涨跌停） def zdt_trade(stock, context, bar_dict): yesterday = history(2,'1d', 'close')[stock].values[-1] zt = round(1.10 * yesterday,2) dt = round(0.90 * yesterday,2) return dt < bar_dict[stock].last < zt '''--------------华丽的分割线----------------''' def init(context): init_variables(context) choose_target(context) # before_trading 此函数会在每天交易开始前被调用，当天只会被调用一次 def before_trading(context, bar_dict): choose_target(context) update_universe(context.stocks) context.his = trans(history(context.obv,'1d','close')) context.barcount = 0 context.init = 1 pass # 你选择的证券的数据更新将会触发此段逻辑，例如日或分钟历史数据切片或者是实时数据切片更新 def handle_bar(context, bar_dict): context.barcount += 1 alert_risk(context,bar_dict) #模拟交易第一次开始，如果是交易时间可能运行不了 before_trading,所以这里做了个参数来控制这种出错的特例 if context.init == 0: update_universe(context.stocks) context.his = trans(history(context.obv, '1d', 'close')) context.barcount = 0 context.init = 1 else: pass if context.barcount % 15 == 0: to_sell = for_sell(context, bar_dict) if to_sell: for oid in get_open_orders().keys(): cancel_order(oid) for stock in to_sell: order_target_value(stock, 0, style=LimitOrder(bar_dict[stock].last*0.995)) if context.barcount == 230: his = trans(history(2,'1m','close')) his = context.his.append(his.iloc[-1,:],ignore_index=True) to_buy = for_buy(context, bar_dict, his) if to_buy: print (to_buy) hnum = len(list(set(to_buy).union(set(context.portfolio.positions.keys())))) for stock in to_buy: if hnum <10: print ('buy', stock, bar_dict[stock].high * 1.005) order_target_percent(stock, 0.99/10, style=LimitOrder(bar_dict[stock].high * 1.005)) else: order_target_percent(stock, 0.99/hnum, style=LimitOrder(bar_dict[stock].high * 1.005)) if context.barcount == 236: his = trans(history(2,'1m','close')) his = context.his.append(his.iloc[-1,:],ignore_index=True) for_balance(context, bar_dict) for_cash(context, bar_dict) 

新鲜出品： https://www.ricequant.com/community/topic/1150/?utm_source=v2ex