DeepSeek 4.2.4利用整理形态中的上升旗形买入,当股价跌破5日均线时卖出,编写一个能在JoinQuant量化交易平台上运行的交易策略,要求编写Python代码。
·
以下是一个基于上升旗形买入、5日均线卖出的量化策略,适用于JoinQuant平台。策略每日开盘时判断信号,以开盘价交易,避免未来函数。
# 导入函数库
import numpy as np
import pandas as pd
def initialize(context):
"""
初始化函数,设定要操作的股票、参数等
"""
# 股票池:沪深300成分股(可自行修改)
g.stocks = get_index_stocks('000300.XSHG')
# 最大持仓数量
g.max_hold = 5
# 均线周期
g.ma_days = 5
# 运行每日策略(开盘后立即执行)
run_daily(handle, time='open')
def handle(context):
"""
每日开盘执行:先卖出,后买入
"""
# 获取当前持仓
current_positions = context.portfolio.positions
# 获取所有可用股票(剔除ST、停牌等)
stock_list = filter_valid_stocks(g.stocks, context)
# ========== 卖出部分:跌破5日均线 ==========
stocks_to_sell = []
for stock in current_positions:
# 判断是否跌破5日均线(基于昨日数据)
if sell_signal(stock, context):
stocks_to_sell.append(stock)
for stock in stocks_to_sell:
# 清空该股票持仓
order_target_percent(stock, 0)
log.info(f"卖出 {stock}:跌破5日均线")
# ========== 买入部分:上升旗形突破 ==========
# 获取现金占总资产比例
cash_ratio = context.portfolio.available_cash / context.portfolio.total_value
if cash_ratio <= 0.01: # 现金不足1%则不买入
return
# 当前持仓股票数
hold_count = len(current_positions)
# 还可买入的股票数量
can_buy_count = g.max_hold - hold_count
if can_buy_count <= 0:
return
# 选出未持仓且出现买入信号的股票
buy_candidates = []
for stock in stock_list:
if stock in current_positions:
continue
if buy_signal(stock, context):
buy_candidates.append(stock)
# 如果候选过多,按信号强度排序(此处简单取前N只)
if len(buy_candidates) > can_buy_count:
# 可自行添加排序逻辑,如成交量放大比例等
buy_candidates = buy_candidates[:can_buy_count]
# 买入:等比例分配剩余资金
if buy_candidates:
cash = context.portfolio.available_cash
percent_per_stock = 0.95 / g.max_hold # 留5%现金防止误差
for stock in buy_candidates:
order_target_percent(stock, percent_per_stock)
log.info(f"买入 {stock}:上升旗形突破")
def filter_valid_stocks(stock_list, context):
"""
过滤ST、停牌、上市不满60天的股票
"""
current_date = context.current_dt.strftime('%Y-%m-%d')
# 获取ST状态
st_data = get_extras('is_st', stock_list, start_date=current_date, end_date=current_date, df=True)
# 获取上市日期
listed_date = get_fundamentals(query(valuation.code, valuation.market_cap, income.total_operation_income,
indicator.adjust_factor).filter(valuation.code.in_(stock_list)))
# 这里简化处理,实际可详细过滤
valid = []
for stock in stock_list:
if st_data.loc[current_date, stock]: # ST剔除
continue
# 简单用attribute_history检查是否有足够数据(例如60天)
df = attribute_history(stock, 60, '1d', fields=['close'], skip_paused=True)
if df is None or len(df) < 60:
continue
valid.append(stock)
return valid
def sell_signal(stock, context):
"""
卖出信号:昨日收盘价 < 昨日5日均线
"""
df = attribute_history(stock, g.ma_days + 1, '1d', fields=['close'], skip_paused=True)
if df is None or len(df) < g.ma_days + 1:
return False
# 昨日收盘价
last_close = df['close'][-1]
# 昨日5日均线(需要前5日收盘价,包括昨日)
ma5 = df['close'][-g.ma_days:].mean()
return last_close < ma5
def buy_signal(stock, context):
"""
买入信号:上升旗形整理后突破
简化条件:
1. 最近20日涨幅 > 5%
2. 最近10日最高价、最低价均呈下降趋势(线性回归斜率<0)
3. 通道宽度(最高-最低)稳定,变异系数 < 0.1
4. 最近5日均量 < 最近10日均量 * 0.9(缩量整理)
5. 昨日收盘价突破过去10日(不含昨日)最高点
6. 昨日成交量 > 最近5日均量 * 1.1(放量突破)
"""
# 获取近30日数据(需包含足够天数)
df = attribute_history(stock, 40, '1d', fields=['close','high','low','volume'], skip_paused=True)
if df is None or len(df) < 25:
return False
# 1. 最近20日涨幅
ret_20 = df['close'][-1] / df['close'][-20] - 1
if ret_20 < 0.05:
return False
# 提取最近10日作为整理期
df10 = df.iloc[-10:] # 最近10个交易日
if len(df10) < 10:
return False
highs = df10['high'].values
lows = df10['low'].values
vols = df10['volume'].values
# 2. 线性回归判断趋势
x = np.arange(len(highs))
# 最高价斜率
coeff_high = np.polyfit(x, highs, 1)[0]
# 最低价斜率
coeff_low = np.polyfit(x, lows, 1)[0]
if coeff_high >= 0 or coeff_low >= 0:
return False
# 3. 通道宽度稳定性
widths = highs - lows
mean_w = np.mean(widths)
std_w = np.std(widths)
if mean_w == 0 or std_w / mean_w > 0.1:
return False
# 4. 成交量萎缩
vol_ma5 = df['volume'].tail(5).mean()
vol_ma10 = df['volume'].tail(10).mean()
if vol_ma5 > 0.9 * vol_ma10: # 缩量不足
return False
# 5. 突破过去10日(不含昨日)最高点
# 过去9日最高价最大值(不含昨日)
high_max_prev = df.iloc[-11:-1]['high'].max() # 倒数第11到倒数第2
last_close = df['close'][-1]
if last_close <= high_max_prev:
return False
# 6. 放量突破
last_vol = df['volume'][-1]
if last_vol <= vol_ma5 * 1.1:
return False
# 附加:可增加平行性检验,此处简化
return True
策略说明
· 买入逻辑:识别上升旗形整理后的放量突破。判断条件包括:前期上涨、整理期高点低点同步下移、通道宽度稳定、成交量先缩后放、价格突破整理区高点。
· 卖出逻辑:股价跌破5日均线时清仓。
· 交易时间:每日开盘时根据昨日数据产生信号,以开盘价执行。
· 股票池:沪深300成分股(可自行修改),并自动过滤ST、停牌、次新股。
· 资金管理:最多同时持有5只股票,每只分配等比例资金。
注意事项
· 策略中的形态参数(如涨幅阈值、变异系数、放量倍数)可根据历史回测优化调整。
· 由于简化处理,上升旗形的识别并非100%准确,实盘前建议充分测试。
· 可结合更多过滤器(如大盘环境、行业)提高胜率。
更多推荐

所有评论(0)