策略名称：涨停板打板策略
策略说明：
本策略基于涨停板打板逻辑，通过筛选上午触及涨停的股票，并在特定时间点进行买入和卖出操作。策略的核心思想是捕捉涨停板的短期机会，结合主力吸筹信号和价格波动进行交易决策。
策略核心逻辑：
1. **股票池筛选**：
- 过滤科创板、ST股、停牌股、退市股、新股、1元股和99元以上股。
- 筛选前3天未涨停的股票。
- 筛选前3天至少有一天主力吸筹（均线涨幅大于3%）的股票。
2. **触及涨停筛选**：
- 在上午交易时段（11:30前）筛选出触及涨停的股票，并按触及涨停的时间排序。
3. **交易决策**：
- **卖出前买入**：在卖出时间点前触及涨停的股票，按触及涨停的时间顺序买入，最多买入（最大持仓数 - 开盘持仓数）只股票。
- **卖出后买入**：在卖出时间点后触及涨停的股票，按触及涨停的时间顺序买入，最多买入至最大持仓数。
4. **卖出规则**：
- 每天10:30判断一次卖出，未涨停的股票卖出，涨停的股票留到下一个交易日。
策略特点：
- **涨停板捕捉**：通过筛选触及涨停的股票，捕捉短期涨停机会。
- **主力吸筹信号**：结合主力吸筹信号，提高选股的准确性。
- **动态仓位管理**：根据触及涨停的时间和持仓数量动态调整买入顺序，控制风险。
适用场景：
- 适用于A股市场，能够有效捕捉涨停板的短期机会。
- 适合短线交易者，通过快速交易获取收益。
风险提示：
- 涨停板策略在震荡市中可能产生较多的错误信号，导致频繁交易。
- 策略依赖于历史数据的有效性，需定期优化参数以适应市场变化。
策略代码结构：
1. **初始化函数（`initialize`）**：
- 设定基准指数（上证指数）。
- 配置交易手续费和滑点。
- 定义每日运行的函数（开盘前、上午筛选、卖出）。
2. **开盘前运行函数（`before_open`）**：
- 获取开盘前的持仓数量和时间。
3. **上午筛选函数（`Screening_stock_purchases`）**：
- 筛选上午触及涨停的股票，并按触及涨停的时间排序。
- 根据卖出时间点分为两部分，分别进行买入操作。
4. **买入函数（`buy`）**：
- 根据触及涨停的时间顺序买入股票，并记录买入信息。
5. **卖出函数（`sell`）**：
- 每天10:30判断一次卖出，未涨停的股票卖出，涨停的股票留到下一个交易日。
6. **辅助函数**：
- **`before_selling`**：筛选卖出时间点前触及涨停的股票。
- **`after_selling`**：筛选卖出时间点后触及涨停的股票。
- **`attracting_funds`**：筛选前3天至少有一天主力吸筹的股票。
- **`No_limit_for_three_days`**：筛选前3天未涨停的股票。
- **`zt`**：筛选上午触及涨停的股票。
- **`filter_specials`**：过滤不符合条件的股票。
---
策略优化建议：
1. **参数优化**：
- 可对主力吸筹的条件（如均线涨幅）进行优化，以提高选股的准确性。
- 调整最大持仓数量和卖出时间点，以适应不同的市场环境。
2. **风险控制**：
- 增加止损机制，避免在极端行情中遭受过大损失。
- 设置最大回撤限制，当回撤超过一定比例时暂停交易。
3. **多因子结合**：
- 可结合其他技术指标（如成交量、MACD等）进行综合筛选，提高策略的稳定性。
总结：
本策略通过涨停板打板逻辑捕捉短期涨停机会，结合主力吸筹信号和动态仓位管理，适合短线交易者在A股市场中获取收益。策略逻辑清晰，适合有一定编程基础的投资者进行进一步优化和实盘测试。
# 标题：最近2年年化100%以上回撤12%的打板策略-验证策略
# 作者：大思维者
# enable_profile()
# 作者：大思维者
# 导入函数库
# from __future__ import division #用于/相除的时候,保留真实结果.小数，python2专用
from jqdata import *
import json #Python2 输出中文列表使用
# # 触及涨停打板----验证策略，首板策略思路
# 1）股票池：前3天不涨停，前三天至少一天主力吸筹（均线涨幅大于3%）
# 2),11::30筛选上午股票池中触及涨停的股票，按触及涨停时间排序
# 3）把11:30选出的股票列表，按卖出时间分成两部分,记录股票，离开盘涨停分钟数组成列表，before_selling_list，after_selling_list
# 卖出之前：计算可以买的数量（最大持仓数-开盘持仓数），按顺序买入before_selling_list中的股票
# 卖出之后：按持仓数量g.number_of_positions，按顺序买after_selling_list中的股票
# 4）每天十点判断一次卖出，没涨停的卖出，涨停的留到下一天
# 初始化函数，设定基准等等
def initialize(context):
set_benchmark('000001.XSHG')# 设定上证作为基准
set_option('use_real_price', True)# 开启动态复权模式(真实价格)
log.info('初始函数开始运行且全局只运行一次')# 输出内容到日志 log.info()
log.set_level('order', 'error') # 过滤掉order系列API产生的比error级别低的log
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')# 手续费是：买入万三，卖出万三加千分之一印花税, 最低扣5块钱
set_option("match_by_signal", True) # 强制撮合，仅支持限价单。
set_option("avoid_future_data", True) # 避免未来数据
g.buy_list=[]
g.minute=60 # 卖出里开盘分钟数
g.buy_stock_count=10 # 最大持仓数量
g.limit_lists=[] # 筛序出的触及涨停股票列表
g.position_count=[] #卖出以后持仓数
g.number_of_positions=0 # 初始持仓数量0
# 替换代码
def after_code_changed(context):
unschedule_all()
# 开盘前运行
run_daily(before_open, '9:30')
run_daily(Screening_stock_purchases, '11:30')
# run_daily(buy, '15:00')
run_daily(sell, '10:30')
# 开盘前数据获取
def before_open(context):
g.open_positions=len(context.portfolio.positions) #开盘前持仓数量
g.open_time=context.current_dt #开盘时间
log.info('===================================================')
log.info('开盘持仓数：',g.open_positions)
# 筛选股票买入
def Screening_stock_purchases(context):
QA=list(get_all_securities().index)# 全A股列表
QA= filter_specials(context,QA) #过滤科创板，ST，停牌，退市，上市半年内的新股,1元股,99元以上股
QA=No_limit_for_three_days(context,QA) # 前3天不涨停
QA=attracting_funds(context,QA) #主力吸筹
log.info('')
log.info('今日筛选股票池数量：',len(QA))
QA=zt(context,QA) # 上午触及涨停的
log.info('上午触及涨停数量：',len(QA))
before_selling_list=before_selling(context,QA,g.minute,g.limit_lists)#卖出之前触及涨停的
log.info('卖出时间之前：数量：%s'%len(before_selling_list),before_selling_list)
before_selling_list=before_selling_list[:g.buy_stock_count-g.open_positions] #卖出之前触及涨停的，取（最大持仓数-开盘持仓）个买入
after_selling_list=after_selling(context,QA,g.minute,g.limit_lists) #卖出之后触及涨停的
log.info('卖出时间之后：数量：%s'%len(after_selling_list),after_selling_list)
for stock_and_time in before_selling_list:
if g.number_of_positions<10:
buy(context,stock_and_time)
g.number_of_positions+=1
log.info('当前持仓数量：',g.number_of_positions,stock_and_time)
for stock_and_time in after_selling_list:
if g.number_of_positions<10:
buy(context,stock_and_time)
g.number_of_positions+=1
log.info('持仓数量：',g.number_of_positions,stock_and_time)
def before_selling(context,stock_list,minute,limit_lists): # 卖出前可买列表及触及涨停时间离开盘分钟数
lists=[]
for stock_and_time in limit_lists: # 触及涨停股票和触及涨停离开盘分钟数
if stock_and_time[0] in stock_list:
if stock_and_time[-1]=minute:
lists.append(stock_and_time) # 卖出时间前涨停的
stock_list=lists
stock_list=sorted(stock_list, key=lambda y:y[-1], reverse = False) # 涨停时间顺序，从小到大
return stock_list
# 股票代码列表=》股票名称列表
def stock_name(lists):
a=[]
for i in lists:
a.append(get_security_info(i).display_name)
lists =json.dumps(a,ensure_ascii=False)
return lists
# 买入股票
def buy(context,stock_and_time):
# log.info(stock_and_time)
g.zzc=round(context.portfolio.total_value,2) # 获取总资产
high_limit=get_current_data()[stock_and_time[0]].high_limit
limit_time=g.open_time+datetime.timedelta(minutes=stock_and_time[-1])
order_value(stock_and_time[0], g.zzc/g.buy_stock_count,LimitOrderStyle(high_limit)) # 实盘，按股数下单
log.info(limit_time,"买入股票：%s "%get_security_info(stock_and_time[0]).display_name,stock_and_time[0],"下单后，剩余资金：",round(context.portfolio.cash,2),"第%s分钟检测到触及涨停"%stock_and_time[1])
# 卖出股票,不涨停卖出
def sell(context):
if len(context.portfolio.positions)!=0:
# log.info(context.current_dt,"可用资金：",round(context.portfolio.cash,2))
for position in list(context.portfolio.long_positions.values()):
mc=get_security_info(position.security).display_name# 股票，名称
cccb = round(context.subportfolios[0].long_positions[position.security].acc_avg_cost,2)#持仓成本
day_open=get_current_data()[position.security].day_open
last_price=get_current_data()[position.security].last_price
low_limit=get_current_data()[position.security].low_limit
high_limit=get_current_data()[position.security].high_limit
sl=context.portfolio.positions[position.security].total_amount# 持仓数量
if position.closeable_amount>0 and high_limit != last_price:# 可卖数量大于0，不涨停卖出
order_target_value(position.security, 0,LimitOrderStyle(last_price))
g.number_of_positions+=-1
log.info("卖出股票：%s "%get_security_info(position.security).display_name,position.security,'最新价格：%s'%last_price,'个股盈亏：%s'%round((((last_price-cccb)*100/cccb)-0.15),2)+'%')
log.info('当前持仓数量：',g.number_of_positions)
else:
log.info("涨停！继续持有==》》%s "%get_security_info(position.security).display_name,position.security,'最新价格：%s'%last_price,'个股盈亏：%s'%round((((last_price-cccb)*100/cccb)-0.15),2)+'%')
log.info('当前持仓数量：',g.number_of_positions)
g.position_count=len(context.portfolio.positions)
log.info('卖出以后持仓数量：',g.position_count)
# 3天中主力必须吸筹1天以上。判断主力吸条件：均线大于昨天收盘价的1.03
def attracting_funds(context,stock_list):
stock_avg=history(3, unit='1d', field='avg', security_list=stock_list, df=False, skip_paused=True, fq='pre')
stock_close=history(4, unit='1d', field='close', security_list=stock_list, df=False, skip_paused=True, fq='pre')
stock_list = [stock for stock in stock_list if
stock_avg[stock][-1]>=stock_close[stock][-2]*1.03
or stock_avg[stock][-2]>=stock_close[stock][-3]*1.03
or stock_avg[stock][-3]>=stock_close[stock][-4]*1.03
]
return stock_list
# 前三天不涨停
def No_limit_for_three_days(context,stock_list):
stock_close=history(3, unit='1d', field='close', security_list=stock_list, df=False, skip_paused=True, fq='pre')
stock_high_limit=history(3, unit='1d', field='high_limit', security_list=stock_list, df=False, skip_paused=True, fq='pre')
stock_list = [stock for stock in stock_list if stock_close[stock][-1] != stock_high_limit[stock][-1]]
stock_list = [stock for stock in stock_list if stock_close[stock][-2] != stock_high_limit[stock][-2]]
stock_list = [stock for stock in stock_list if stock_close[stock][-3] != stock_high_limit[stock][-3]]
return stock_list
# 过滤器，过滤停牌，ST，科创，新股，1元股，一字板（可选择过滤）
def filter_specials(context,stock_list):
curr_data = get_current_data()
stock_list=[stock for stock in stock_list if stock[0:3] != '688'] #过滤科创板'688'
# stock_list=[stock for stock in stock_list if stock[0:3] != '300'] #过滤创业板'300'
stock_list = [stock for stock in stock_list if not curr_data[stock].is_st]
stock_list = [stock for stock in stock_list if not curr_data[stock].paused]
stock_list = [stock for stock in stock_list if 'ST' not in curr_data[stock].name]
stock_list = [stock for stock in stock_list if '*' not in curr_data[stock].name]
stock_list = [stock for stock in stock_list if '退' not in curr_data[stock].name]
stock_list = [stock for stock in stock_list if curr_data[stock].day_open>1] # 过滤一元股
stock_list = [stock for stock in stock_list if curr_data[stock].high_limit<99] # 过滤99以上元股
stock_list = [stock for stock in stock_list if curr_data[stock].day_open!= curr_data[stock].high_limit] # 过滤一字板
return stock_list
解锁查看函数
# 筛序盘中触及涨停
def zt(context,stock_list):
