前天介绍了完整的拐点交易算法的分析，今天详细的介绍一下里面的子模块，挂点卖出，先看简单的原理图，这个是固定的，更改起点会变成连续动态的分析算法研究--日内拐点交易算法原理研究
通过基准价与涨幅比例计算涨幅阈值价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()))
原理图
流程图
