回测日期:2014-12-5 至 2025-05-31
选股策略:
选取中证1000指数中的1000只股票
当日涨跌幅 3-5
当日换手率 3-10
当日均线 MA5 > MA10
止损止盈条件 收益 > 5 或者 收益-3卖出
# 中证1000选股策略
# 作者 但盼风雨来
# 选股策略:
# 当日涨跌幅 3-5
# 当日换手率 3-10
# 当日均线 MA5 > MA10
# 止损止盈条件 收益 > 5 或者 收益-3卖出
import pandas as pd
# 初始化函数,全局只运行一次
def init(context):
# 设置基准收益:沪深300指数
set_benchmark('000300.SH')
# 打印日志
log.info('策略开始运行,初始化函数全局只运行一次')
# 设置股票每笔交易的手续费为万分之二(手续费在买卖成交后扣除,不包括税费,税费在卖出成交后扣除)
set_commission(PerShare(type='stock',cost=0.0000854,min_trade_cost=0))
# 设置股票交易双边滑点0.5%,表示买入价为实际价格乘1.0025,卖出价为实际价格乘0.9975
set_slippage(PriceSlippage(0.005))
# 设置日级最大成交比例25%,分钟级最大成交比例50%
# 日频运行时,下单数量超过当天真实成交量25%,则全部不成交
# 分钟频运行时,下单数量超过当前分钟真实成交量50%,则全部不成交
set_volume_limit(0.25,0.5)
# 设置要操作的股票:同花顺
context.security = ['300033.SZ']
# 回测区间、初始资金、运行频率请在右上方设置
context.stock_list = ['300033.SZ','000718.SZ']
run_daily(function)
#每日开盘前9:00被调用一次,用于储存自定义参数、全局变量,执行盘前选股等
def before_trading(context):
# 获取日期
date = get_datetime().strftime('%Y-%m-%d %H:%M:%S')
# 打印日期
log.info('{} 盘前运行'.format(date))
## 开盘时运行函数
def handle_bar(context, bar_dict):
# 获取时间
time = get_datetime().strftime('%Y-%m-%d %H:%M:%S')
log.info(bar_dict)
# 买卖策略
positions = context.stock_account.positions
to_sell = []
for position in positions:
shouyi = positions[position].profit_rate * 100
# log.info(f"{positions[position].symbol}的收益率为{shouyi}%")
stock = positions[position].symbol
if shouyi > 6 or shouyi <-3:
to_sell.append(stock)
for stock in to_sell:
order_target(stock,0)
log.warn(f"已清仓了{stock}")
# 打印时间
log.info('{} 盘中运行'.format(time))
stock_list = context.stock_list
log.info("下单的股票为",str(stock_list))
for stock in stock_list:
price = get_current(stock)
current_price = price[stock].open
# log.info(f"股票代码{stock},价格{current_price}")
order(stock,1000,price = current_price)
date = get_datetime()
# log.info('{} 运行'.format(date))
log.info(f"已购买{stock},价格为{current_price}")
## 收盘后运行函数,用于储存自定义参数、全局变量,执行盘后选股等
def after_trading(context):
# 获取时间
time = get_datetime().strftime('%Y-%m-%d %H:%M:%S')
# 打印时间
log.info('{} 盘后运行'.format(time))
log.info('一天结束')
def function(context,*args):
get_all_stock(context)
get_rate_stock(context)
change_hand(context)
avg_stock(context)
# 获取所有股票
def get_all_stock(context):
# 获取时间
time = get_datetime()
# 获取市场所有股票
# all_stock = get_all_securities(ty='stock', date=time)
# df = all_stock
# df = df[~df['display_name'].str.contains('ST', na=False)]
# df = df[~df['order_book_id'].str.contains('BJ')]
# stock_list = df['order_book_id'].tolist()
# 获取中证1000
stock_list= get_index_stocks('000852.SH',time)
log.info("已获取到所有股票代码")
context.stock_list = stock_list
log.info(f"中证1000股票数量为{len(stock_list)}")
log.info("已获取中证1000股票")
# 筛选符合涨幅条件的股票
def get_rate_stock(context):
time = get_datetime().strftime('%Y-%m-%d')
stock_list = context.stock_list
null_list = []
fields = ['quote_rate','open']
current = get_price(
stock_list,
start_date=None,
end_date=time,
fre_step='1d',
fields=fields,
skip_paused=False,
fq='pre',
bar_count=1,
is_panel=False,
)
for stock in stock_list:
if (current[stock].quote_rate[0] > 3) & (current[stock].quote_rate[0] < 5):
null_list.append(stock)
context.stock_list = null_list
log.info(f"符合涨跌幅限制的股票数量为{len(null_list)}")
log.info("执行筛选符合涨幅条件的股票")
# 筛选符合换手率的股票
def change_hand(context):
current_stock = context.stock_list
hand_list = []
time = get_datetime().strftime('%Y%m%d')
fields = ['turnover_rate']
change_hand = get_price(
current_stock,
start_date=None,
end_date=time,
fre_step='1d',
fields=fields,
skip_paused=False,
fq='pre',
bar_count=1,
is_panel=False,
)
for stock in current_stock:
if (float(change_hand[stock]['turnover_rate'])>3)&(float(change_hand[stock]['turnover_rate'])<10):
hand_list.append(stock)
context.stock_list = hand_list
log.info(f"符合换手率的股票数量为{len(hand_list)}")
log.info("执行筛选符合换手率的股票")
# 筛选符合均线的股票
def avg_stock(context):
time = get_datetime().strftime('%Y%m%d')
stock_list = context.stock_list
avg_list = []
price = get_price(
stock_list,
start_date=None,
end_date=time,
fre_step='1d',
fields=['close'],
skip_paused=False,
fq='pre',
bar_count=20,
is_panel=False,
)
for stock in stock_list:
MA5 = price[stock]['close'].iloc[-5:].mean()
MA10 = price[stock]['close'].iloc[-10:].mean()
if MA5 > MA10 :
avg_list.append(stock)
context.stock_list = avg_list
log.info(f"符合均线策略的股票数量为{len(avg_list)}")
log.info("执行筛选符合均线的股票")