自行拿去研究。# 聚宽研究环境直接运行import pandas as pdfrom datetime import datetime, timedeltadef get_trade_days_simple(end_date=None, count=15):if end_date is None:end_date = context.current_dt.date() if 'context' in globals() else datetime.now().date()try:return get_trade_days(end_date=end_date, count=count)except:days, cur = [], end_datewhile len(days) < count:if cur.weekday() < 5: days.insert(0, cur)cur -= timedelta(days=1)return daysdef select_dq_dq_pattern(end_date=None, lookback=20, amp_thr=4.0):"""形态:跌停→撬开>amp_thr%→收盘仍跌停返回:[(代码,名称,日期,拉起幅度), ...] 日期倒序"""end_date = end_date or (context.current_dt.date() if 'context' in globals() else datetime.now().date())tdays = get_trade_days_simple(end_date=end_date, count=lookback)# 股票池stocks = []for idx in ['000001.XSHG','399001.XSHE','399006.XSHE']:stocks.extend(get_index_stocks(idx)[:200])stocks = list(set(stocks))hits = []for stock in stocks:try:df = get_price(stock, start_date=tdays[0], end_date=end_date,frequency='daily', fields=['open','close','high','low','low_limit'])if df is None or len(df)==0: continuefor _,r in df.iterrows():dl = r['low_limit']if dl<=0: continue# ①最低≈跌停 ②收盘=跌停(0.5%容差)③拉起幅度if r['low']<=dl*1.005 and abs(r['close']-dl)/dl<=0.005:pull = (r['high']-dl)/dl*100if pull >= amp_thr:info = get_security_info(stock)hits.append((stock,info.display_name if info else "未知名称",r.name,round(pull,2)))break # 只记最近一次except Exception: continuehits.sort(key=lambda x: x[2], reverse=True)return hits# 运行入口def run():print("=== 跌停→撬开→仍跌停 形态选股 ===")res = select_dq_dq_pattern()print(f"共 {len(res)} 只(日期倒序)")if res:print("日期\t\t股票代码\t\t股票名称\t\t拉起幅度")for r in res:print(f"{r[2].strftime('%Y-%m-%d')}\t{r[0]}\t{r[1]:<15}\t{r[3]}%")else:print("未找到符合形态的股票")return resif __name__ == "__main__":run()
运行结果:

Day1 跌停→被撬开→振幅>4%→最终仍封回跌停(收盘=跌停价)
Day2 收盘 高于 Day1 的跌停价(次日脱离或反包)
运行结果:
代码:

import pandas as pdfrom datetime import datetime, timedeltadef get_trade_days_simple(end_date=None, count=25):if end_date is None:end_date = context.current_dt.date() if 'context' in globals() else datetime.now().date()try:return get_trade_days(end_date=end_date, count=count)except:days, cur = [], end_datewhile len(days) < count:if cur.weekday() < 5: days.insert(0, cur)cur -= timedelta(days=1)return daysdef select_dq_lock_rebound(end_date=None, lookback=20, amp_thr=4.0):"""形态:Day1 最低≈跌停且拉起>amp_thr% 且 收盘=跌停(≤0.5%误差)Day2 收盘 > Day1跌停价返回:[(代码,名称,Day1日期,Day1拉起%,Day2溢价%), ...] 日期倒序"""end_date = end_date or (context.current_dt.date() if 'context' in globals() else datetime.now().date())tdays = get_trade_days_simple(end_date=end_date, count=lookback)stocks = []for idx in ['000001.XSHG','399001.XSHE','399006.XSHE']:stocks.extend(get_index_stocks(idx)[:200])stocks = list(set(stocks))hits = []for stock in stocks:try:df = get_price(stock, start_date=tdays[0], end_date=end_date,frequency='daily', fields=['open','close','high','low','low_limit'])if df is None or len(df)<2: continuefor i in range(1, len(df)):day1, day2 = df.iloc[i-1], df.iloc[i]dl1 = day1['low_limit']if dl1<=0: continue# Day1 条件day1_low, day1_high, day1_close = day1['low'], day1['high'], day1['close']if day1_low<=dl1*1.005 and abs(day1_close-dl1)/dl1<=0.005:pull = (day1_high-dl1)/dl1*100if pull >= amp_thr:# Day2 条件day2_close = day2['close']premium = (day2_close-dl1)/dl1*100if premium>0:info = get_security_info(stock)hits.append((stock,info.display_name if info else "未知名称",day1.name,round(pull,2),round(premium,2)))break # 只记最近一次except Exception: continuehits.sort(key=lambda x: x[2], reverse=True)return hits# 运行入口def run():print("=== Day1跌停撬开>4%且收盘仍跌停,Day2收盘>Day1跌停价 形态选股 ===")res = select_dq_lock_rebound()print(f"共 {len(res)} 只(Day1日期倒序)")if res:print("Day1日期\t\t股票代码\t\t股票名称\t\tDay1拉起%\tDay2溢价%")for r in res:print(f"{r[2].strftime('%Y-%m-%d')}\t{r[0]}\t{r[1]:<15}\t{r[3]}%\t\t{r[4]}%")else:print("未找到符合形态的股票")return resif __name__ == "__main__":run()
Day1 跌停→被撬开→振幅>4%(无论是否封回跌停)
Day2 收盘 高于 Day1 的跌停价(确认次日反包或脱离跌停区域)
运行结果:

代码:
import pandas as pdfrom datetime import datetime, timedeltadef get_trade_days_simple(end_date=None, count=25):if end_date is None:end_date = context.current_dt.date() if 'context' in globals() else datetime.now().date()try:return get_trade_days(end_date=end_date, count=count)except:days, cur = [], end_datewhile len(days) < count:if cur.weekday() < 5: days.insert(0, cur)cur -= timedelta(days=1)return daysdef select_dq_rebound_pattern(end_date=None, lookback=20, amp_thr=4.0):"""形态:Day1 最低≈跌停且拉起>amp_thr%Day2 收盘 > Day1跌停价返回:[(代码,名称,Day1日期,Day1拉起%,Day2收盘相对跌停溢价%), ...] 日期倒序"""end_date = end_date or (context.current_dt.date() if 'context' in globals() else datetime.now().date())tdays = get_trade_days_simple(end_date=end_date, count=lookback)# 股票池stocks = []for idx in ['000001.XSHG','399001.XSHE','399006.XSHE']:stocks.extend(get_index_stocks(idx)[:200])stocks = list(set(stocks))hits = []for stock in stocks:try:df = get_price(stock, start_date=tdays[0], end_date=end_date,frequency='daily', fields=['open','close','high','low','low_limit'])if df is None or len(df)<2: continuefor i in range(1, len(df)):day1, day2 = df.iloc[i-1], df.iloc[i]dl1 = day1['low_limit']if dl1<=0: continue# Day1 条件day1_low, day1_high = day1['low'], day1['high']if day1_low<=dl1*1.005:pull = (day1_high-dl1)/dl1*100if pull >= amp_thr:# Day2 条件day2_close = day2['close']premium = (day2_close-dl1)/dl1*100if premium>0:info = get_security_info(stock)hits.append((stock,info.display_name if info else "未知名称",day1.name,round(pull,2),round(premium,2)))break # 只记最近一次except Exception: continuehits.sort(key=lambda x: x[2], reverse=True)return hits# 运行入口def run():print("=== Day1跌停撬开>4% 且 Day2收盘>Day1跌停价 形态选股 ===")res = select_dq_rebound_pattern()print(f"共 {len(res)} 只(Day1日期倒序)")if res:print("Day1日期\t\t股票代码\t\t股票名称\t\tDay1拉起%\tDay2溢价%")for r in res:print(f"{r[2].strftime('%Y-%m-%d')}\t{r[0]}\t{r[1]:<15}\t{r[3]}%\t\t{r[4]}%")else:print("未找到符合形态的股票")return resif __name__ == "__main__":run()
需要自取,仅研究,不作为买卖依据,如果造成亏损,需要自行负责。
如果想进行聚宽实盘,但是不知道怎么链接的可以看一下这个教程,只需要一个支持Mini的QMT:
一文带你了解量化交易使用最多的两款量化软件:QMT与PTrade,超详细剖析
如果您对某个想法或者策略感兴趣,想获得更详细的落地方案,欢迎留言或私信我。我们可以进行一对一的免费初步交流,期待为您提供真正有价值的帮助。
