QMT实盘策略--涨停板反包

在A股市场，“反包”是指前一日或前几日的调整后，个股再次强势拉升，并有效突破前高甚至涨停，体现主力控盘、换手充分、情绪回暖等信号。结合涨停板+反包，可以筛选出极具交易动能的个股。特征包括：前一日涨停次日调整不破支撑第三日大阳反包，并具备换手放量信号

一、策略逻辑设计

✅ 选股逻辑通过问财接口（getwencai）筛选出满足以下条件的股票：流通市值介于30亿～200亿之间排除ST、科创板、北交所、创业板当前股价低于30元（便于博弈情绪）

str1 = '流通市值大于30亿小于200亿'
str2 = '股票简称不包含st'
str3 = '上市板块不包含科创板'
str4 = '股票市场类型不包含北交所'
str5 = '上市板块不包含创业板'
str6 = '股票价格不高于30元'

✍️ 可增加指标如：换手率、涨停板次数、龙虎榜活跃度等，以增强策略筛选信度。 ✅ 买入逻辑主要逻辑位于 handlebar(C) 函数中：历史回看最近3日数据第3天的收盘价为涨停，前两天未连板，今日盘中再次大涨超9% 控制交易时段在 09:30~15:00 内每只票只买一次（避免过度加仓）

if (round(history[stk]['close'].iloc[-3],2) == history[stk]['uplimit'].iloc[-3]) \
   and (round(history[stk]['close'].iloc[-2],2) != history[stk]['uplimit'].iloc[-2]) \
   and (chg >= 9.0) \
   and A.buy_times1[stk] < 1:

💡 这是典型的 “涨停 → 调整 → 再度爆发” 的反包形态识别。 ✅ 卖出逻辑位于 zyzs(C) 函数中，分三类：止盈止损（盈利>5%或亏损>5%）最大浮盈回撤止盈最大浮盈超5%，一旦回撤2% 卖出一半，回撤5%全部清仓收盘清仓下午 14:56 后强制清仓，规避尾盘异动

二、实盘部署

#encoding:gbk
'''
何楠策略
'''
import numpy as np
import talib
import json
import requests
import pandas as pd
import logging
import os
import datetime as dt
import time
import configparser
import requests as rq
import urllib
import re
from datetime import datetime, timedelta
from decimal import Decimal, ROUND_HALF_UP


class a():
    pass

A = a()
A.REMARK = '何楠'
A.length = 10  
A.zhisun = 0.05
A.zhiying = 0.05
#盈利最高点大于过5%后在回落
A.max_profit = 0.05
#回落2%卖出一半
A.huiluo1 = 0.02
#全部卖出
A.huiluo2 = 0.05
#条件1买入数量
A.buy_nums1 = 100  # 考虑资金余额
A.max_FloatProfit = {}
#条件1运行时间
A.start_time1 = '093000'
A.medium_time1 = '145600'
A.end_time1 = '150000'
tradeStatus = {}
curr_time = dt.datetime.now().strftime('%H%M%S')

#获取文件名
def get_state_filename(C, period):
    return A.REMARK + f"{period}.json"

#保存状态，委托状态和成本价
def save_trade_status(C, trade_status, period):
    filename = get_state_filename(C, period)
    try:
        with open(filename, 'w', encoding='utf-8') as file:
            json.dump(trade_status, file, ensure_ascii=False, indent=4)
            print(f"成功保存JSON文件: {filename}")
    except Exception as e:
        print(f"保存文件时发生错误: {e}")


#读取json文件
def load_trade_status_from_file(C, period):
    filename = get_state_filename(C, period)
    try:
        with open(filename, 'r', encoding='utf-8') as file:
            data = json.load(file)
            return data
    except FileNotFoundError:
        print(f"文件未找到: {filename}")
    except json.JSONDecodeError:
        print(f"文件内容不是有效的JSON格式: {filename}")
    except Exception as e:
        print(f"读取文件时发生错误: {e}")
    return {}


def mylog(debug = False, info = False, warning = False, error = False, critical = False):
    global A
    #步骤一、生成一个日志器
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    
    #步骤二、生成处理器
    yymmdd = dt.datetime.now().strftime('%Y%m%d')
    #如果当前目录不存在，则创建一个目录
    if not os.path.exists('logs/' + str(A.acc) + '/' + A.REMARK):
        os.makedirs('logs/' + str(A.acc) + '/' + A.REMARK)
    file_name = 'logs/' + str(A.acc) + '/' + A.REMARK + '/' + yymmdd + '.log'
    if not logger.handlers:
        console_handler = logging.StreamHandler()
        file_handler = logging.FileHandler(file_name, mode='a', encoding = 'utf-8')
        console_handler.setLevel(logging.DEBUG)
        file_handler.setLevel(logging.DEBUG)
        
        #步骤三、设置输出格式
        fmt = "%(asctime)s - %(levelname)s - %(message)s"
        formatter = logging.Formatter(fmt = fmt)
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        
        #步骤四、日志器添加处理器
        logger.addHandler(file_handler)
        logger.addHandler(console_handler)
    
    #步骤五、打印日志
    if debug:
        logger.debug(debug)
    elif info:
        logger.info(info)
    elif warning:
        logger.warning(warning)
    elif error:
        logger.error(error)
    elif critical:
        logger.critical(critical)

def init(C):
    global A, tradeStatus
    try:
        A.acc = str(account)
        A.acc_type = str(accountType)
        C.do_back_test = 1
    except NameError:
        A.acc = '55012157' # 改为自己的资金账户
        A.acc_type = 'STOCK'
        C.do_back_test = 0  # 回测模式

    #all_data = get_wencai_data(question,start_date,end_date)
    #print(all_data)'每股净资产bps大于3''3日的区间换手率>9%'
    str1 = '流通市值大于30亿小于200亿'
    str2 = '股票简称不包含st'
    str3 = '上市板块不包含科创板'
    str4 = '股票市场类型不包含北交所'
    str5 = '上市板块不包含创业板'
    str6 = '股票价格不高于30元'
    strs0 = ','.join([str1, str2, str3, str4, str5 ])
#    str = ','.join([str1, str2, str3, str4, str5, str9, str10]), str8
    curr_date = dt.datetime.now().strftime('%Y%m%d')
    file_name = A.REMARK + curr_date + '.csv'
    '''
    if os.path.exists(file_name):
        pd.read_csv(os.path.join(os.getcwd(), file_name))
        xuan_data = pd.read_csv(file_name)
    else:
        xuan_data = (getwencai(strs, list=True)
        xuan_data.to_csv(file_name)
    '''
    xuan_data0 = getwencai(strs0, list=True)
    #print(list(xuan_data0['股票代码']))
    #print(list(xuan_data1['股票代码']))
    #xuan_data1 = cal_10days_limitup_count(C)
    #A.stocks = list(set(list(xuan_data0['股票代码']) + list(xuan_data0['股票代码'])))
    xuan_data =pd.DataFrame(list(set(list(xuan_data0['股票代码'])))) 
    A.stocks = list(set(list(xuan_data0['股票代码'])))
   #print(A.stocks)
    #A.stocks = list(set(list(xuan_data['股票代码'])))
    xuan_data.to_csv('第一轮.csv')
    #sector = C.get_stock_list_in_sector('自选股1')
    sector = C.get_stock_list_in_sector('沪深A股')
    intersection = [x for x in A.stocks if x in sector]
    A.stocks = intersection
    msg = f'筛选出来的股票是：{A.stocks}'
    mylog(info = msg)   
    # 创建日志器
    strToday = dt.datetime.now().strftime('%Y-%m-%d')
    # 执行收盘
    C.run_time("save", "1nDay", strToday + " 15:01:00", "SH")
    
    #获取可用资金
    A.cash = get_available_cash(C)
    msg = f'账户的可用资金为：{A.cash}'
    mylog(info = msg)
    #买入条件1
    A.buy_times1 = {}

原创 QMT实盘策略--涨停板反包

一、策略逻辑设计

二、实盘部署

sushi