翘板选股策略代码:跌停→撬开→仍跌停→第二日收盘价>跌停价

# 聚宽研究环境直接运行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_date        while 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: continue            for _,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*100                    if 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: continue    hits.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()

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_date        while 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: continue            for 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*100                    if pull >= amp_thr:                        # Day2 条件                        day2_close = day2['close']                        premium = (day2_close-dl1)/dl1*100                        if 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: continue    hits.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()

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_date        while 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: continue            for 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*100                    if pull >= amp_thr:                        # Day2 条件                        day2_close = day2['close']                        premium = (day2_close-dl1)/dl1*100                        if 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: continue    hits.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()