策略名称：涨停板龙头股策略
策略说明：
本策略基于涨停板龙头股的短期机会，通过筛选连续涨停的股票，并结合技术指标进行交易决策。策略的核心思想是捕捉市场中的龙头股，利用其短期上涨动能获取收益。
策略核心逻辑：
- 股票池筛选：
- - 过滤ST股、停牌股、退市股、科创板股票等不符合条件的股票。
- - 筛选过去10天内连续涨停的股票，并计算其连板数。
- 风险控制：
- - 通过跌停数量判断市场风险，若跌停数量较多，则暂停买入。
- - 设置止损条件，当持仓股票亏损超过5%时，卖出止损。
- 交易决策：
- - 在开盘后，根据筛选出的龙头股进行买入操作。
- - 尾盘未涨停的股票卖出，避免隔夜风险。
- 动态仓位管理：
- - 最多持有1只股票，避免过度集中风险。
- - 防止短期内反复买入同一只股票，设置7天的冷却期。
策略特点：
- **龙头股捕捉**：通过筛选连续涨停的股票，捕捉市场中的龙头股。
- **风险控制**：通过跌停数量和止损条件，控制市场风险和个股风险。
- **动态仓位管理**：限制持仓数量，避免过度集中风险。
适用场景：
- 适用于A股市场，能够有效捕捉龙头股的短期上涨机会。
- 适合短线交易者，通过快速交易获取收益。
风险提示：
- 龙头股策略在震荡市中可能产生较多的错误信号，导致频繁交易。
- 策略依赖于历史数据的有效性，需定期优化参数以适应市场变化。
策略代码结构：
1. 初始化函数（`initialize`）：
- 设定基准指数（沪深300）。
- 配置交易手续费和滑点。
- 定义每日运行的函数（开盘前、开盘时）。
2. 开盘前运行函数（`before_market_open`）：
- 筛选过去10天内连续涨停的股票，并计算其连板数。
- 通过跌停数量判断市场风险，若跌停数量较多，则暂停买入。
3. 开盘时运行函数（`market_open`）：
- 根据筛选出的龙头股进行买入操作。
- 尾盘未涨停的股票卖出，避免隔夜风险。
- 设置止损条件，当持仓股票亏损超过5%时，卖出止损。
4. 辅助函数：
- `fit_linear`：通过线性回归判断市场趋势。
- `risk_management`：通过跌停数量判断市场风险。
策略优化建议：
1. 参数优化：
- 可对连板数的计算周期（如10天）进行优化，以适应不同的市场周期。
- 调整止损条件（如5%），以更好地控制风险。
2. 风险控制：
- 增加最大回撤限制，当回撤超过一定比例时暂停交易。
- 设置单笔交易的最大亏损比例，避免单笔交易亏损过大。
3. 多因子结合：
- 可结合其他技术指标（如成交量、MACD等）进行综合筛选，提高策略的稳定性。
总结：
本策略通过捕捉龙头股的短期上涨机会，结合风险控制和动态仓位管理，适合短线交易者在A股市场中获取收益。策略逻辑清晰，适合有一定编程基础的投资者进行进一步优化和实盘测试。
# 标题：冲天炮最高板策略迭代
# 回测分别选择 按天和按分钟的效果差别很大
# 原标题：冲天炮最高板策略，收益惊呆了我
# 标题：冲天炮最高板策略迭代
# from kuanke.user_space_api import *
from jqdata import *
import datetime as dt
from sklearn.linear_model import LinearRegression
# 2020=9-14：禁止短期内反复买入， 例如2019-1-附近，反复买卖风范股份
# 初始化程序, 整个回测只运行一次
def initialize(context):
# 设定沪深300作为基准
set_benchmark('000300.XSHG')
# 开启动态复权模式(真实价格)
set_option('use_real_price', True)
log.info('初始函数开始运行且全局只运行一次')
# 过滤掉order系列API产生的比error级别低的log
log.set_level('order', 'error')
# 最多持有的股票只数
g.max_hold_nums = 1
g.recent_bought_stocks = {} # 记录近期（7日内）买卖过的股票
# 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5),
type='stock')
run_daily(before_market_open, time='before_open', reference_security='000300.XSHG')
run_daily(market_open, time='every_bar', reference_security='000300.XSHG')
def market_open(context):
# type: (Context) -> NoReturn
curr_data = get_current_data()
hour = context.current_dt.hour
minute = context.current_dt.minute
# ------------------处理卖出-------------------
for security in context.portfolio.positions:
closeable_amount = context.portfolio.positions[security].closeable_amount
if closeable_amount > 0:
if minute > 55 and hour == 14:
if curr_data[security].last_price < curr_data[security].high_limit: # 尾盘未涨停
order_target(security, 0) # 卖出
log.info("尾盘未涨停，卖出： %s %s" % (curr_data[security].name, security))
else:
# 止损
if curr_data[security].last_price < context.portfolio.positions[security].avg_cost*0.95: order_target(security, 0) # 卖出 log.info("亏损超过5%%，止损： %s %s" % (curr_data[security].name, security)) if g.buy_list is None: return # 最多持有这么多只股票 if len(context.portfolio.positions) >= g.max_hold_nums:
return
# 涨停板数量>3
if not (fit_linear(3) > 0 and g.max_zt_days > 2):
return
if hour < 11 and len(context.portfolio.positions) < g.max_hold_nums: # 冲天炮龙头 for security in g.buy_list: if security not in context.portfolio.positions: # 防止买入已经涨停的股票 if curr_data[security].last_price >= curr_data[security].high_limit:
continue
# 计算今天还需要买入的股票数量
need_count = g.max_hold_nums - len(context.portfolio.positions)
if need_count > 0:
buy_cash = context.portfolio.available_cash / need_count
if buy_cash > (curr_data[security].last_price * 500):
# 买入这么多现金的股票
result = order_value(security, buy_cash)
if result is not None:
g.recent_bought_stocks[security] = 1 # 第一天
log.info("买入： %s %s 买入价：%.2f 涨停价: %.2f" % (
curr_data[security].name, security, result.avg_cost, curr_data[security].high_limit))
# 直线拟合
def fit_linear(count):
"""
count:拟合天数
"""
security = '000001.XSHG'#'399006.XSHE'
df = history(count=count, unit='1d', field='close', security_list=security, df=True, skip_paused=False, fq='pre')
model = LinearRegression()
x_train = np.arange(0, len(df[security])).reshape(-1, 1)
y_train = df[security].values.reshape(-1, 1)
# print(x_train,y_train)
model.fit(x_train, y_train)
# # 计算出拟合的最小二乘法方程
# # y = mx + c
# c = model.intercept_
m = model.coef_
# c1 = round(float(c), 2)
m1 = round(float(m), 2)
# print("最小二乘法方程 : y = {} + {}x".format(c1,m1))
return m1
# 跌停数量
def risk_management(context):
current_data = get_current_data()
security_list = list(get_all_securities(['stock']).index)
security_list = [stock for stock in security_list
if not current_data[stock].is_st
and not current_data[stock].paused
and 'ST' not in current_data[stock].name
and '*' not in current_data[stock].name
and '退' not in current_data[stock].name]
pre_data = get_price(security_list, end_date=context.previous_date, frequency='daily', fields=['close', 'low_limit'], skip_paused=True, fq='pre', count=1, panel=False)
pre_pre_data = get_price(security_list, end_date=context.previous_date-dt.timedelta(1), frequency='daily', fields=['close', 'low_limit'], skip_paused=True, fq='pre', count=1, panel=False)
pre_limit_down = pre_data[pre_data.close == pre_data.low_limit]
pre_pre_limit_down = pre_pre_data[pre_pre_data.close == pre_pre_data.low_limit]
pre_limit_down_num = len(pre_limit_down)
pre_pre_limit_down_num = len(pre_pre_limit_down)
print("前日跌停数量为{0},昨日跌停数量为{1}".format(pre_pre_limit_down_num, pre_limit_down_num))
if pre_limit_down_num>pre_pre_limit_down_num and pre_limit_down_num>15:
return True
else:
return False
