我更深的学习编代码，短期怕是有点难了，只能用手工半自动的笨办法啦，先用个股测试这个简的代码运行，分别测试自己心中的标的，回撒小的加进来，够10多只轮转着就行，快接近目了，附件就不传上来了，大家自己用简单的代码这个运行测试，试好一个加进去一个就可，都是传入昨天时间，避免未来函数。 附上代码，大家直接拷贝，不用克隆。 一个票的： # 导入函数库 import jqdata from jqlib.technical_analysis import * # 初始化函数，设定基准等等 def initialize(context): # 设定沪深300作为基准 set_benchmark('000300.XSHG') # 开启动态复权模式(真实价格) set_option('use_real_price', True) #设置开启避免未来数据模式 set_option("avoid_future_data", True) # 输出内容到日志 log.info() log.info('初始函数开始运行且全局只运行一次') ### 股票相关设定 ### # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱 set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock') ## 运行函数（reference_security为运行时间的参考标的；传入的标的只做种类区分，因此传入'000300.XSHG'或'510300.XSHG'是一样的） # 开盘前运行 run_daily(before_market_open, time='before_open', reference_security='000300.XSHG') # 开盘时运行 run_daily(market_open, time='open', reference_security='000300.XSHG') # 收盘后运行 run_daily(after_market_close, time='after_close', reference_security='000300.XSHG') def before_market_open(context): # 输出运行时间 log.info('函数运行时间(before_market_open)：'+str(context.current_dt.time())) # 给微信发送消息 send_message('美好的一天，祝您交易顺利！') # 要操作的股票：云南白药（g.为全局变量） g.security = '510300.XSHG' def market_open(context): log.info('函数运行时间(market_open):'+str(context.current_dt.time())) security = g.security #print(type(security)) now =context.previous_date #前一交易日 today_now = context.current_dt #调用BIAS函数，获取该函数的K值和D值 BIAS1,BIAS2,BIAS3 = BIAS(security,check_date= now, N1=10, N2=25, N3=73) #print(BIAS3) # 取得当前的现金 cash = context.portfolio.available_cash # 如果BIAS3在低位向上交叉BIAS2时, 则全仓买 if BIAS3[security] < -11: # 记录这次买入 log.info('买入股票 %s' % (security)) # 用所有cash买入股票 order_value(security, cash) # BIAS3处于高位，并且目前有头寸,有可卖出股票，则全仓卖出 elif BIAS3[security] >= 14 and context.portfolio.positions[security].closeable_amount > 0: # 记录这次卖出 log.info('卖出股票 %s' % (security)) # 卖出所有股票，使这只股票的最终持有量为0 order_target(security, 0) def after_market_close(context): log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time()))) #得到当天所有成交记录 trades = get_trades() for _trade in list(trades.values()): log.info('成交记录：'+str(_trade)) log.info('一天的交易结束，祝你心情愉快！') 轮换的： from jqdata import * from jqlib.technical_analysis import * import numpy as np import pandas as pd import seaborn as sns from six import BytesIO ## 初始化函数，设定要操作的股票、基准等等 def initialize(context): # 设定沪深300作为基准 set_benchmark('000300.XSHG') # True为开启动态复权模式，使用真实价格交易 set_option('use_real_price', True) # 设定成交量比例 set_option('order_volume_ratio', 1) # 股票类交易手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱 set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock') # 持仓数量 g.stocknum = 3 # 交易日计时器 g.days = 0 # 调仓频率 g.refresh_count = 1 # 运行函数（reference_security为运行时间的参考标的；传入的标的只做种类区分，因此传入'000300.XSHG'或'510300.XSHG'是一样的） # 开盘前运行 run_daily(before_market_open, time='before_open', reference_security='000300.XSHG') # 开盘时运行 run_daily(market_open, time='open', reference_security='000300.XSHG') run_daily(check_holding,time='open', reference_security='000300.XSHG') run_daily(trade,time='open', reference_security='000300.XSHG') run_daily(check_holding,time='10:57', reference_security='000300.XSHG') run_daily(check_holding,time='13:57', reference_security='000300.XSHG') run_daily(check_holding,time='14:29', reference_security='000300.XSHG') # 收盘后运行 run_daily(after_market_close, time='after_close', reference_security='000300.XSHG') def before_market_open(context): # 输出运行时间 log.info('函数运行时间(before_market_open)：'+str(context.current_dt.time())) # 给微信发送消息 #send_message('美好的一天，祝您交易顺利！') def market_open(context): pass ## 选出符合条件的股票 def check_stocks(context): buylist = [] #填你研究环境的csv名称 a = pd.read_csv(BytesIO(read_file('bias.csv'))) #读取第1列数据 b = a.iloc[ :,0] security_list = b # 设置沪深300为初始股票池 # 输出 security_list 的 BIAS 值 now =context.previous_date #前一交易日 today_now = context.current_dt BIAS1,BIAS2,BIAS3 = BIAS(security_list,check_date= now, N1=10, N2=25, N3=73) #print (BIAS3) for security in security_list: if BIAS3[security] < -11: buylist.append(security) if len(buylist) == g.stocknum: break return buylist # 过滤停牌股票 buylist = filter_paused_stock(buylist) # 过滤st,*st,退市 buylist = filter_st_stock(buylist) #过滤涨停板和跌停板 buylist= filter_limitup_stock(context,buylist) return buylist[:g.stocknum] #每天检查持仓bias如果达到卖出条件加入卖出列表 def check_holding(context): sell_list = [] holding_list = list(context.portfolio.positions.keys()) now =context.previous_date #前一交易日 today_now = context.current_dt BIAS1,BIAS2,BIAS3 = BIAS(holding_list,check_date = now, N1=10, N2=25, N3=73) for security in holding_list: if BIAS3[security] >= 40 or BIAS3[security] < BIAS2[security]: sell_list.append(security) return sell_list # 如果有需要调仓的股票，卖出 if len(sell) > 0 : for stock in sell: order_target_value(stock, 0) ## 交易函数 def trade(context): if g.days%g.refresh_count == 0: #这里的意思是g.days可以被g.refresh_count整除（days是count的倍数） #换仓日，调整持仓 #先清算持仓 check_holding(context) ## 获取持仓列表和买入列表，删除其中的重复项，即不需要卖出的股票 now_list = list(context.portfolio.positions.keys()) stock_list = check_stocks(context) sell = [] buy=[] for stock in now_list: if stock not in stock_list: sell.append(stock) for stock in stock_list: if stock not in now_list: buy.append(stock) # 如果有需要调仓的股票，卖出 if len(sell) > 0 : for stock in sell: order_target_value(stock, 0) #分钱 if len(buy) > 0: Cash = context.portfolio.cash/len(buy) else: Cash = 0 ## 买入股票 for stock in buy: order_value(stock, Cash) # 天计数加一 g.days = 1 else: g.days += 1 #非换仓日，只卖出跌破长线的股票，不买入 sell_list = check_holding(context) #找到短线跌破长线的股票并卖出 if len(sell_list) > 0 : for stock in sell_list: order_target_value(stock, 0) # 过滤停牌股票 def filter_paused_stock(stock_list): current_data = get_current_data() return [stock for stock in stock_list if not current_data[stock].paused] # 过滤ST及其他具有退市标签的股票 def filter_st_stock(stock_list): current_data = get_current_data() return [stock for stock in stock_list if not current_data[stock].is_st and 'ST' not in current_data[stock].name and '*' not in current_data[stock].name and '退' not in current_data[stock].name] #过滤涨停板和跌停板 #输入选股列表，返回剔除未持有且已涨停股票后的列表 def filter_limitup_stock(context,stock_list): last_prices = history(1, unit='1m', field='close', security_list=stock_list) current_data = get_current_data() # 已存在于持仓的股票即使涨停也不过滤，避免此股票再次可买，但因被过滤而导致选择别的股票 return [stock for stock in stock_list if stock in context.portfolio.positions.keys() or last_prices[stock][-1] < current_data[stock].high_limit] #过滤模块-过滤跌停的股票 #输入股票列表，返回剔除已跌停股票后的列表 def filter_limitdown_stock(context,stock_list): last_prices = history(1, unit='1m', field='close', security_list=stock_list) current_data = get_current_data() return [stock for stock in stock_list if stock in context.portfolio.positions.keys() or last_prices[stock][-1] > current_data[stock].low_limit] def after_market_close(context): log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time()))) #得到当天所有成交记录 trades = get_trades() for _trade in list(trades.values()): log.info('成交记录：'+str(_trade)) log.info('一天的交易结束，祝你心情愉快！') 如果每天交易，交易计时器可以删了。 分别试了1、3、5、8、13、21天，只有1天的效果最好。下图是3天一换的