我更深的学习编代码,短期怕是有点难了,只能用手工半自动的笨办法啦,先用个股测试这个简的代码运行,分别测试自己心中的标的,回撒小的加进来,够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天一换的

@努力jjian 先用简版试票,固定后轮动,相当于半自动
2021-11-30
@飞越极限 那还得再做一步,如果选的是较差的票,是否也能相对跌的少呢,才能体现这个优势
2021-11-30
@努力jjian 自己对比可以用差的来比较,反正简版跑得差的票我是不加进去的
2021-11-30
@飞越极限 问题就是未来不知道哪些会表现差额,现在回测,当然可以预见
2021-11-30
@努力jjian 你不会把测试时间缩到近期
关键标的质量要好,历史波动大的
2021-11-30
@拯救兄弟 你自己先用简版自己选股后自建一个呀,文中说得很清楚了
2021-11-30