算法研究--日内拐点卖出算法原理研究

前天介绍了完整的拐点交易算法的分析，今天详细的介绍一下里面的子模块，挂点卖出，先看简单的原理图，这个是固定的，更改起点会变成连续动态的分析算法研究--日内拐点交易算法原理研究

通过基准价与涨幅比例计算涨幅阈值价p1，当最新价突破p1后，以突破后的最高价p2与回调比例计算回调阈值价p3，当最新价突破p3后，如果P3>P1则以（价格策略—价格偏移*最小价差）挂单卖出

最高价说明：当涨幅比例阈值p1满足后，开始记录最高价p2，此时的最高价p2会根据行情的更新而随时变化，p2取到的是从p1满足起开始到当前时间内的最高价

一、原理：捕捉“冲高回落”的动能衰竭点

该算法的核心原理是：识别一个上涨趋势在经历“加速赶顶”后的“动能衰竭”信号，并以此作为卖点。它认为，一个健康的上涨趋势转为危险的下跌趋势，通常会经历两个阶段：

加速上涨阶段：价格快速拉升，突破某个关键位置（P1），这证明多头力量达到顶峰。

高位回落阶段：价格在创出新高后，无法维持，开始显著回落。这个“回落”动作本身就是多头力量衰竭、空头开始反扑的直接证据。策略捕捉的正是这个“从最高点回落”的拐点。

二、核心交易思路：一个带有“门控”的两阶段确认模型

这个算法的思路非常严谨，可以概括为一个 “门控式”的两阶段确认模型。

第一阶段：上涨趋势的“入场券”——突破P1，作用：P1是基准价上方的一个“门槛”。价格必须向上突破这个门槛，才被纳入监控范围。思路：这起到了一个趋势强度过滤器的作用。如果价格连P1都无法突破，说明上涨动能不足，不值得启动“高位回落”监控。这避免了在横盘震荡行情中频繁产生无效信号。

第二阶段：动能衰竭的“确认键”——跌破P3

作用：一旦价格突破P1，策略便开始寻找这个上涨波段中的最高点（P2），并计算出从P2回落的“警戒线”P3。当价格从高点回落到P3以下时，触发卖出。思路：这个阶段是本策略的精髓。它不预测顶部，而是等待顶部出现（P2）并从顶部显著回落（跌破P3）后才行动，是一种典型的“右侧交易”思维。它牺牲了顶部的一小部分利润，来换取趋势可能反转的更高确定性。

最关键的约束条件：P3 > P1

代码体现：if p_3>p_1:思路：这是整个策略逻辑中最精妙、最关键的一环。它要求“回调警戒线P3”必须高于“上涨启动线P1”。这保证了触发卖出时，股价仍然高于最初确认上涨趋势的位置。

深层含义：如果 P3 <= P1，意味着股价从高点P2回落的幅度过大，不仅吃掉了所有涨幅，还跌回了“上涨趋势确认线”下方。这种情况虽然也是亏损，但它可能属于“假突破”或趋势的彻底反转，与策略定义的“冲高回落的动能衰竭”模式不符。策略主动放弃处理这种情况，以保持其信号的纯粹性。

三、设计理念：有取舍的模式

这个算法的设计体现了清晰的取舍：它要赚什么钱？ 它旨在捕捉一个已确认的上升趋势中，最后一段加速上涨后，趋势发生温和反转的那一段利润。它要放弃什么行情？放弃窄幅震荡：价格无法突破P1，永不触发。

放弃深度V型反转：股价突破P1后快速拉升，然后掉头向下，直接跌破P1（此时 P3 <= P1）。这种行情它不参与，也因此避开了追高买入后立刻被套在山顶的风险。放弃预测最高点：它在最高点（P2）出现并回落之后才行动，属于确认易，而非预测易。

策略内部计算的核心变量

不懂的问我就可以，加我备注入群可以加入量化群

量化开户需要找我就可以，专业量化技术支持服务

代码的研究学习参考不做交易参考

def run_sell_trade(c):
    '''
    卖出拐点运行函数
    "上涨比例":,
    "回调比例":,
    '''
    path=c.text['记录文件路径']
    up_ratio=c.text['上涨比例']
    down_ratio=c.text['回调比例']
    trader_type=c.text['交易模式']
    value=c.text['下单值']
    position=get_position(c,c.account,c.account_type)
    account=get_account(c,c.account,c.account_type)
    if position.shape[0]>0:
        position=position[position['持仓量']>=10]
        if position.shape[0]>0:
            hold_list=position['证券代码'].tolist()
            av_amount_dict=dict(zip(position['证券代码'],position['可用数量']))
        else:
            hold_list=[]
            av_amount_dict={}
    else:
        hold_list=[]
        av_amount_dict={}
    if check_is_trader_date_1(c):
        df=get_trader_stock(c)
        if df.shape[0]>0:
            for stock,name in zip(df['证券代码'],df['名称']):
                try:
                    now_date=int(''.join(str(datetime.now())[:10].split('-')))
                    date_time=int(''.join(str(datetime.now()).split('.')[0][-8:].split(':')))
                    log=check_trader_log(c)
                    price=get_price(c,stock)
                    price=round(price,3)
                    if log.shape[0]>0:
                        log=log[['证券代码','名称',"拐点类型",'交易模式',
                            '交易方向','交易值','交易价格','交易时间','交易日']]
                        log=log.sort_values(by='交易时间',ascending=True)
                        log['交易日']=log['交易日'].astype(int)
                        log=log[log['交易日']==now_date]
                    else:
                        log=pd.DataFrame()
                    #获取开始的基础价格
                    if log.shape[0]>0:
                        log_stock=log[log['证券代码']==stock]
                        if log_stock.shape[0]>0:
                            base_price=log_stock['交易价格'].tolist()[-1]
                            last_time=log_stock['交易时间'].tolist()[-1]
                            stats=True
                        else:
                            base_price=get_base_price(c,stock)
                            last_time=93000
                            stats=False
                    else:
                        base_price=get_base_price(c,stock)
                        last_time=93000
                        stats=False
                    if stats==True:
                        tick=get_stock_lx_tick(c,stock,date=str(now_date))
                        tick=tick[tick['time']>=last_time]
                    else:
                        tick=get_stock_lx_tick(c,stock,date=str(now_date))
                    if tick.shape[0]>0:
                        lastClose_list=tick['lastPrice'].tolist()
                        #最新值
                        last_price=lastClose_list[-1]
                        #最大值
                        max_value=max(lastClose_list)
                        #p1值
                        p_1=base_price*(1+up_ratio/100)
                        p_1=round(p_1,3)
                        #目前涨跌幅
                        last_zdf=((last_price-base_price)/base_price)*100
                        last_zdf=round(last_zdf,2)
                        #最高点涨跌幅
                        last_max_zdf=((max_value-base_price)/base_price)*100
                        last_max_zdf=round(last_max_zdf,2)
                        #p2最大价
                        p_2=max_value
                        p_2=round(p_2,3)
                        #回调比例计算p3
                        p_3=p_2*(1-down_ratio/100)
                        p_3=round(p_3,3)
                        #目前回调比例
                        last_ht_ratio=((last_price-p_2)/p_2)*100
                        last_ht_ratio=round(last_ht_ratio,2)
                        #最新价大于P1,开始监控
                        if last_price>=p_1:
                            #回调没有跌破p1
                            if p_3>p_1:
                                #最新价跌破回调值
                                if last_price<=p_3:
                                    print('触发拐点卖出,时间:{},股票:{},名称:{},基准价:{},p1值:{},p2值:{},p3:值{},最新价:{},目前涨跌幅:{},最高涨跌幅:{},目前回调比例:{} '.format(str(datetime.now())[:-7:],
                                        stock,name,base_price,p_1,p_2,p_3,last_price,
                                        last_zdf,last_max_zdf,last_ht_ratio))
                                    if trader_type=='数量':
                                        amount=value
                                    elif trader_type=='金额':
                                        amount=value/price
                                        amount=adjust_amount(stock,amount)
                                    elif trader_type=='百分比':
                                        total=account['总资产']
                                        value=total*value
                                        amount=value/price
                                        amount=adjust_amount(stock,amount)
                                    else:
                                        amount=0
                                    av_amount=av_amount_dict.get(stock,0)
                                    if av_amount>=amount:
                                        amount=amount
                                    else:
                                        amount=av_amount
                                    if check_is_sell(c,c.account,c.account_type,stock=stock,amount=amount):
                                        maker='{},拐点卖出,{},{},{},{}'.format(c.st_name,stock,'sell',amount,price)
                                        passorder(c.sell_code, 1101,c.account, stock, c.sell_price_code, 0, amount, maker,1,maker,c)
                                        trader_log=pd.DataFrame()
                                        trader_log['证券代码']=[stock]
                                        trader_log['名称']=[name]
                                        trader_log['拐点类型']=['拐点卖出']
                                        trader_log['交易模式']=[trader_type]
                                        trader_log['交易方向']=['sell']
                                        trader_log['交易值']=[value]
                                        trader_log['交易价格']=[price]
                                        trader_log['交易时间']=[int(date_time)]
                                        trader_log['交易日']=[int(now_date)]
                                        #第一次循环运行卖出不记录
                                        if a.sell_**==0:
                                            print(stock,'第一次循环运行卖出不记录********')
                                            a.sell_**=1
                                        else:
                                            log=pd.concat([log,trader_log],ignore_index=True)
                                            if log.shape[0]>0:
                                                log['证券代码']=log['证券代码'].apply(lambda x: '0'*(6-len(str(x)))+str(x))
                                            else:
                                                log=log
                                            log.to_excel(r'{}'.format(path))
                                            print(maker)
                                    else:
                                        print(stock,'卖出失败可能没有持股')
                                else:
                                    print('没有触发拐点卖出,会调没有跌破P3,时间:{},股票:{},名称:{},基准价:{},p1值:{},p2值:{},p3:值{},最新价:{},目前涨跌幅:{},最高涨跌幅:{},目前回调比例:{} '.format(str(datetime.now())[:-7:],
                                        stock,name,base_price,p_1,p_2,p_3,last_price,
                                        last_zdf,last_max_zdf,last_ht_ratio))
                            else:
                                print('没有触发拐点卖出,P3小于P1,时间:{},股票:{},名称:{},基准价:{},p1值:{},p2值:{},p3:值{},最新价:{},目前涨跌幅:{},最高涨跌幅:{},目前回调比例:{} '.format(str(datetime.now())[:-7:],
                                    stock,name,base_price,p_1,p_2,p_3,last_price,
                                    last_zdf,last_max_zdf,last_ht_ratio))
                        else:
                            print('没有触发拐点卖出,最新加没有突破P1,时间:{},股票:{},名称:{},基准价:{},p1值:{},p2值:{},p3:值{},最新价:{},目前涨跌幅:{},最高涨跌幅:{},目前回调比例:{} '.format(str(datetime.now())[:-7:],
                                stock,name,base_price,p_1,p_2,p_3,last_price,
                                last_zdf,last_max_zdf,last_ht_ratio))
                    else:
                        print(str(datetime.now())[:-7],stock,'拐点卖出，拐点触发后的时间没有数据,拐点触发交易时间大于15:00')
                except Exception as e:
                    print(e,stock,'拐点卖出有问题*********')
        else:
            print('拐点买入没有自定义交易股票池*********')
    else:
        print('{}拐点卖出目前不是交易时间*********'.format(datetime.now()))

原理图

流程图

不懂的问我就可以作者微信xg_quant