# ============================================================ # AdaptiveChandelierV2 (自适应吊灯 半自动版) # 版本: V2.6.3 # 定位: 金叉/死叉自动进场(或手动进场) + 四阶段吊灯出场,做一单停 # # 更新说明: # V2.0 金叉/死叉自动进场(启动后等信号确认,金叉多/死叉空) # V2.1 进场当根K线不止损,下一根才开始保护 # V2.5 吊灯增加阶段4(浮盈≥8ATR → N=1.0,超大浮盈锁死) # V2.6 快速锁利(浮盈到 lock_trigger 点 → 锁死 进场±lock_protect 点,可调) # 状态页增加: 当前浮盈点数 / 锁利状态 / 止损线离进场 # V2.6.1 增加平仓诊断日志 + 持仓过程逐根日志(便于复盘验证) # V2.6.2 进场当根改为"跟踪峰值+显示止损线,但不触发平仓"(修止损线显示1e18) # V2.6.3 峰值改用K线高/低价更新(方案A,抗脏tick),平仓止损仍逐tick实时 # 进场模式开关: auto_entry(1自动/0手动) + manual_dir(手动方向) # ============================================================ from datetime import datetime from pythongo.base import BaseParams, BaseState, Field from pythongo.classdef import KLineData, OrderData, TickData, TradeData from pythongo.core import KLineStyleType from pythongo.ui import BaseStrategy from pythongo.utils import KLineGenerator class Params(BaseParams): """参数（启动前设置）""" exchange: str = Field(default="CFFEX", title="交易所代码") instrument_id: str = Field(default="IC2609", title="合约代码") kline_style: KLineStyleType = Field(default="M5", title="K线周期") auto_entry: int = Field(default=1, title="自动进场") # 1=自动等金叉死叉 0=手动启动即进场 manual_dir: int = Field(default=1, title="手动方向") # 手动模式下:1做多 -1做空(自动模式忽略) atr_period: int = Field(default=20, title="ATR窗口") stage2_trigger: float = Field(default=1.5, title="进阶段2浮盈ATR") # 浮盈达到几倍ATR进阶段2 stage3_trigger: float = Field(default=3.0, title="进阶段3浮盈ATR") stage4_trigger: float = Field(default=8.0, title="进阶段4浮盈ATR") k_stage1: float = Field(default=3.5, title="阶段1_ATR倍数") # 前松 k_stage2: float = Field(default=3.0, title="阶段2_ATR倍数") # 中稳 k_stage3: float = Field(default=2.5, title="阶段3_ATR倍数") # 后紧 k_stage4: float = Field(default=1.0, title="阶段4_ATR倍数") # 超大浮盈锁死 max_loss: float = Field(default=15, title="最大止损点数") # 0=关闭只用吊灯 lock_trigger: float = Field(default=10, title="锁利触发点数") # 浮盈到这么多点触发锁利(0=关闭) lock_protect: float = Field(default=7, title="锁利保护点数") # 锁死进场上方这么多点 lots: int = Field(default=1, title="开仓手数") over_price: int = Field(default=1, title="委托超价") expma_fast: int = Field(default=21, title="EXPMA快线") # 仅画图参考 expma_slow: int = Field(default=42, title="EXPMA慢线") # 仅画图参考 class State(BaseState): """状态（界面显示）""" status: str = Field(default="等待数据", title="运行状态") stage: str = Field(default="待进场", title="阶段") entry_price: float = Field(default=0.0, title="进场价") atr: float = Field(default=0.0, title="ATR") profit_atr: float = Field(default=0.0, title="浮盈ATR倍数") profit_pts: float = Field(default=0.0, title="当前浮盈点数") lock_status: str = Field(default="-", title="锁利状态") stop_offset: float = Field(default=0.0, title="止损线离进场") stop_line: float = Field(default=0.0, title="止损线") class AdaptiveChandelierV2(BaseStrategy): """ 启动即进场 · 三阶段自适应吊灯止损 · 一次性执行工具（无限易 pythongo） 进场:启动前设 direction(1做多/-1做空);策略一启动收到行情立即按方向进场(不验时段)。 出场:三阶段自适应吊灯(按浮盈以ATR计,前松-中稳-后紧调整N倍数) + 棘轮(只朝锁利方向移动) + 固定止损上限,取更紧者,现价触及即平。平仓后不再进场(一次性)。 阶段(用最高价HH算浮盈,达到过即进更紧阶段,不回退): 浮盈 N=k_stage1 (前松,给呼吸空间) stage2_trigger~stage3 -> N=k_stage2 (中稳,推向保本) >=stage3_trigger -> N=k_stage3 (后紧,保护利润) EXPMA 快慢线仅画图参考。仅供模拟盘验证执行纪律。 """ def __init__(self) -> None: super().__init__() self.params_map = Params() self.state_map = State() self.kline_generator: KLineGenerator | None = None self.last_price = 0.0 self.bid_price1 = 0.0 self.ask_price1 = 0.0 self.price_tick = 1.0 self.long_pos = 0 self.short_pos = 0 self.expma_fast_val = 0.0 self.expma_slow_val = 0.0 self.atr_val = 0.0 # 状态机: waiting -> holding -> done self.stage = "waiting" self.entry_price = 0.0 self.peak = 0.0 # 多:进场后最高价HH / 空:进场后最低价LL self.max_profit_atr = 0.0 # 达到过的最大浮盈(ATR计),单调不减,用于分阶段 self.cur_stop = 0.0 # 棘轮后的实际止损线 self.open_orderid: int | None = None self.close_orderid: int | None = None self.signal_price = 0.0 self._opened_by_self = False self._first_tick_logged = False self.pending_dir = 0 # 待执行方向: 0=还没等到信号, 1=金叉做多, -1=死叉做空 self.bar_count = 0 # 已收线K线计数 self.entry_bar_count = -1 # 进场时的K线计数(进场当根不止损,下根才开始) @property def main_indicator_data(self) -> dict[str, float]: if self.stage == "holding" and abs(self.cur_stop) < 1e17: stop_display = self.cur_stop else: stop_display = float("nan") return { "EXPMA快": self.expma_fast_val, "EXPMA慢": self.expma_slow_val, "止损线": stop_display, } # ---------------- 生命周期 ---------------- def on_init(self) -> None: super().on_init() self.output("AdaptiveChandelierV2 初始化 版本 V2.6.3 (自动/手动进场+四阶段吊灯+快速锁利)") def on_start(self) -> None: p = self.params_map self.kline_generator = KLineGenerator( callback=self.on_bar, real_time_callback=self.on_bar_realtime, exchange=p.exchange, instrument_id=p.instrument_id, style=p.kline_style, ) self.kline_generator.push_history_data() try: inst = self.get_instrument_data(p.exchange, p.instrument_id) if inst: self.price_tick = inst.price_tick except Exception as e: self.output(f"[警告] 获取合约信息失败: {e}") super().on_start() self.stage = "waiting" # 手动模式:启动即按手动方向进场(pending_dir直接设好,不等信号);自动模式:等金叉/死叉 if p.auto_entry == 0: self.pending_dir = 1 if p.manual_dir == 1 else -1 mode_txt = f"手动进场(启动即{'做多' if self.pending_dir == 1 else '做空'})" else: self.pending_dir = 0 mode_txt = "自动进场(等金叉做多/死叉做空)" self.state_map.stage = "待进场" if p.auto_entry == 0 else "待信号" self.output( f"V2.6.3启动 {p.exchange}.{p.instrument_id} {p.kline_style} {mode_txt} " f"EXPMA{p.expma_fast}/{p.expma_slow} ATR{p.atr_period} 四阶段N={p.k_stage1}/{p.k_stage2}/{p.k_stage3}/{p.k_stage4} " f"阈值={p.stage2_trigger}/{p.stage3_trigger}/{p.stage4_trigger}ATR 最大止损={p.max_loss}点 锁利{p.lock_trigger}→{p.lock_protect} —— 做一单停" ) def on_stop(self) -> None: super().on_stop() self.output("策略停止") # ---------------- Tick(实时驱动,不验时段) ---------------- def on_tick(self, tick: TickData) -> None: super().on_tick(tick) if tick.instrument_id != self.params_map.instrument_id: return # 过滤无效最新价(e+308 之类占位极值) lp = tick.last_price if not (0 < lp < 1e10): return self.last_price = lp # 过滤无效买卖一价(开盘首tick常出现 1.79e308 占位极值),无效则置0 bid = tick.bid_price1 ask = tick.ask_price1 self.bid_price1 = bid if (0 < bid < 1e10) else 0.0 self.ask_price1 = ask if (0 < ask < 1e10) else 0.0 if not self._first_tick_logged: self._first_tick_logged = True self.output(f"[诊断] 首个tick: 现价{self.last_price} 买一{self.bid_price1} " f"卖一{self.ask_price1} 阶段={self.stage}") self._update_position() # 待进场:等到金叉/死叉信号(pending_dir)、空仓、无挂单才进场 if (self.stage == "waiting" and self.pending_dir != 0 and self.long_pos == 0 and self.short_pos == 0 and self.open_orderid is None): self._open() elif (self.stage == "holding" and self.open_orderid is None and (self.long_pos > 0 or self.short_pos > 0)): self._check_exit() if self.kline_generator: self.kline_generator.tick_to_kline(tick) self.update_status_bar() # ---------------- Bar(算指标) ---------------- def on_bar(self, kline: KLineData) -> None: self.bar_count += 1 # 每根K线收线计数+1 self._calc_indicators() self._check_signal() # 持仓期间:用【已收线K线的高/低价】更新峰值(方案A,抗脏tick),不再用逐tick最新价 if self.stage == "holding" and (self.long_pos > 0 or self.short_pos > 0): if self.long_pos > 0: self.peak = max(self.peak, kline.high) # 多单:峰值取K线最高 prof = (self.peak - self.entry_price) / self.atr_val if self.atr_val > 0 else 0 else: self.peak = min(self.peak, kline.low) # 空单:峰值取K线最低 prof = (self.entry_price - self.peak) / self.atr_val if self.atr_val > 0 else 0 self.max_profit_atr = max(self.max_profit_atr, prof) # 达到过就不回退 n, stage = self._current_n() prof_pts = (self.last_price - self.entry_price) if self.long_pos > 0 else (self.entry_price - self.last_price) self.output(f"[过程] {kline.datetime} 收{kline.close} 高{kline.high} 低{kline.low} " f"峰值{self.peak:.1f} 浮盈{prof_pts:.1f}点/{self.max_profit_atr:.2f}ATR " f"阶段{stage}·N={n} ATR{self.atr_val:.2f} 止损线{self.cur_stop:.1f}") self._update_chart(kline) def on_bar_realtime(self, kline: KLineData) -> None: self._calc_indicators() self._update_chart(kline) def _check_signal(self) -> None: """收线确认金叉/死叉,只在【自动模式】等待进场阶段设置 pending_dir(用[-1]/[-2],防闪烁)""" if self.params_map.auto_entry == 0: # 手动模式不等信号 return if self.stage != "waiting" or self.pending_dir != 0: return if self.kline_generator is None: return p = self.params_map bars = self.kline_generator.producer if len(bars.close) < max(p.expma_fast, p.expma_slow) + 3: return try: ef = bars.ema(timeperiod=p.expma_fast, array=True) es = bars.ema(timeperiod=p.expma_slow, array=True) except Exception: return f1, s1 = float(ef[-1]), float(es[-1]) # 刚收线的这根 f2, s2 = float(ef[-2]), float(es[-2]) # 前一根 if f2 <= s2 and f1 > s1: self.pending_dir = 1 self.output("金叉确认,等待下一跳做多进场") elif f2 >= s2 and f1 < s1: self.pending_dir = -1 self.output("死叉确认,等待下一跳做空进场") def _calc_indicators(self) -> None: if self.kline_generator is None: return p = self.params_map bars = self.kline_generator.producer need = max(p.expma_fast, p.expma_slow, p.atr_period) + 2 if len(bars.close) < need: return try: self.expma_fast_val = float(bars.ema(timeperiod=p.expma_fast, array=True)[-1]) self.expma_slow_val = float(bars.ema(timeperiod=p.expma_slow, array=True)[-1]) except Exception: pass self.atr_val = self._calc_atr(bars, p.atr_period) self.state_map.atr = round(self.atr_val, 4) def _calc_atr(self, bars, n: int) -> float: try: v = float(bars.atr(timeperiod=n, array=True)[-1]) if v > 0: return v except Exception: pass try: H, L, C = bars.high, bars.low, bars.close m = len(C) if m < n + 1: return self.atr_val s = 0.0 for i in range(m - n, m): s += max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])) return s / n except Exception: return self.atr_val # ---------------- 三阶段 N ---------------- def _current_n(self): p = self.params_map if self.max_profit_atr >= p.stage4_trigger: return p.k_stage4, 4 if self.max_profit_atr >= p.stage3_trigger: return p.k_stage3, 3 if self.max_profit_atr >= p.stage2_trigger: return p.k_stage2, 2 return p.k_stage1, 1 # ---------------- 进场 / 出场 ---------------- def _open(self) -> None: p = self.params_map if self.pending_dir == 1: price = self._opp_price_buy() or self.last_price side = "buy" elif self.pending_dir == -1: price = self._opp_price_sell() or self.last_price side = "sell" else: return if not price or price <= 0: self.output("[诊断] 价格无效,暂不进场") return self.open_orderid = self.send_order( exchange=p.exchange, instrument_id=p.instrument_id, volume=p.lots, price=price, order_direction=side, ) if self.open_orderid is not None: self._opened_by_self = True self.signal_price = price if side == "buy" else -price self.state_map.status = "进场委托中" self.output(f"{'开多' if side == 'buy' else '开空'} 委托价{price:.2f} {p.lots}手 orderid={self.open_orderid}") else: self.output("[诊断] send_order 返回空,下单失败") def _check_exit(self) -> None: """逐tick:三阶段自适应吊灯 + 棘轮 + 固定止损上限,取更紧者,现价触及即平""" p = self.params_map if self.last_price <= 0 or self.atr_val <= 0 or self.close_orderid is not None: return # 进场当根:仍然跟踪峰值、计算并显示止损线,但【不触发平仓】(下一根K线起才允许平) allow_exit = self.bar_count > self.entry_bar_count if self.long_pos > 0: # 峰值改由 on_bar 用K线高低更新(方案A),这里只用当前峰值计算,不再用tick更新峰值 prof = (self.peak - self.entry_price) / self.atr_val self.max_profit_atr = max(self.max_profit_atr, prof) # 达到过就不回退 n, stage = self._current_n() chand = self.peak - n * self.atr_val fixed = self.entry_price - p.max_loss if p.max_loss > 0 else -1e18 # 快速锁利:浮盈曾达到 lock_trigger 点 -> 锁死 进场+lock_protect locked = p.lock_trigger > 0 and (self.peak - self.entry_price) >= p.lock_trigger lock_line = (self.entry_price + p.lock_protect) if locked else -1e18 raw = max(chand, fixed, lock_line) self.cur_stop = max(self.cur_stop, raw) # 棘轮:只升不降 self.state_map.stop_line = round(self.cur_stop, 4) self.state_map.profit_atr = round(self.max_profit_atr, 2) self.state_map.profit_pts = round(self.last_price - self.entry_price, 1) self.state_map.stop_offset = round(self.cur_stop - self.entry_price, 1) self.state_map.lock_status = f"已锁利(+{p.lock_protect:g})" if locked else f"未锁(触发{p.lock_trigger:g})" self.state_map.stage = f"持多·阶段{stage}·N={n}" + ("·已锁利" if locked else "") if allow_exit and self.last_price <= self.cur_stop: which = "锁利线" if (locked and abs(self.cur_stop-lock_line)<0.01) else ("吊灯线" if abs(self.cur_stop-chand)<0.01 else "固定止损") self.output(f"[平仓诊断] 触发平多: 现价{self.last_price} <= 止损线{self.cur_stop:.1f}({which}) " f"峰值{self.peak:.1f} 阶段{stage}·N={n} ATR{self.atr_val:.2f} " f"吊灯{chand:.1f} 锁利{lock_line if locked else 0:.1f} 固定{fixed:.1f}") self._close("sell") elif self.short_pos > 0: # 峰值改由 on_bar 用K线高低更新(方案A),这里只用当前峰值计算 prof = (self.entry_price - self.peak) / self.atr_val self.max_profit_atr = max(self.max_profit_atr, prof) n, stage = self._current_n() chand = self.peak + n * self.atr_val fixed = self.entry_price + p.max_loss if p.max_loss > 0 else 1e18 # 快速锁利:浮盈曾达到 lock_trigger 点 -> 锁死 进场-lock_protect locked = p.lock_trigger > 0 and (self.entry_price - self.peak) >= p.lock_trigger lock_line = (self.entry_price - p.lock_protect) if locked else 1e18 raw = min(chand, fixed, lock_line) self.cur_stop = min(self.cur_stop, raw) # 棘轮:只降不升 self.state_map.stop_line = round(self.cur_stop, 4) self.state_map.profit_atr = round(self.max_profit_atr, 2) self.state_map.profit_pts = round(self.entry_price - self.last_price, 1) self.state_map.stop_offset = round(self.entry_price - self.cur_stop, 1) self.state_map.lock_status = f"已锁利(+{p.lock_protect:g})" if locked else f"未锁(触发{p.lock_trigger:g})" self.state_map.stage = f"持空·阶段{stage}·N={n}" + ("·已锁利" if locked else "") if allow_exit and self.last_price >= self.cur_stop: which = "锁利线" if (locked and abs(self.cur_stop-lock_line)<0.01) else ("吊灯线" if abs(self.cur_stop-chand)<0.01 else "固定止损") self.output(f"[平仓诊断] 触发平空: 现价{self.last_price} >= 止损线{self.cur_stop:.1f}({which}) " f"峰值{self.peak:.1f} 阶段{stage}·N={n} ATR{self.atr_val:.2f} " f"吊灯{chand:.1f} 锁利{lock_line if locked else 0:.1f} 固定{fixed:.1f}") self._close("buy") def _close(self, side: str) -> None: if self.close_orderid is not None: return p = self.params_map vol = self.long_pos if side == "sell" else self.short_pos if vol <= 0: return price = (self._opp_price_sell() or self.last_price) if side == "sell" else (self._opp_price_buy() or self.last_price) if not price or price <= 0: return self.close_orderid = self.auto_close_position( exchange=p.exchange, instrument_id=p.instrument_id, volume=vol, price=price, order_direction=side, ) if self.close_orderid is not None: self.signal_price = -price if side == "sell" else price self.state_map.status = "平仓委托中" self.output(f"{'平多' if side == 'sell' else '平空'} 对手价{price:.2f} {vol}手") def _opp_price_buy(self): if self.ask_price1 <= 0: return None return self._round_price(self.ask_price1 + self.params_map.over_price * self.price_tick) def _opp_price_sell(self): if self.bid_price1 <= 0: return None return self._round_price(self.bid_price1 - self.params_map.over_price * self.price_tick) def _round_price(self, price: float) -> float: tick = self.price_tick if self.price_tick > 0 else 1.0 return round(round(price / tick) * tick, 10) def _update_position(self) -> None: position = self.get_position(instrument_id=self.params_map.instrument_id) self.long_pos = position.long.close_available self.short_pos = position.short.close_available # 自愈:仅当本次运行程序从没开过仓、却发现已有持仓(真正的遗留仓)才接管 if (self.stage == "waiting" and not self._opened_by_self and self.open_orderid is None and (self.long_pos > 0 or self.short_pos > 0)): self.stage = "holding" try: avg = position.long.open_avg_price if self.long_pos > 0 else position.short.open_avg_price except Exception: avg = 0.0 if not avg or avg <= 0: avg = self.last_price self.entry_price = avg self.peak = avg self.max_profit_atr = 0.0 self.cur_stop = -1e18 if self.long_pos > 0 else 1e18 self.state_map.stage = "持仓(接管)" self.state_map.entry_price = round(avg, 4) # 在图上补画一个接管标记,便于看到程序在此接管了遗留仓 self.signal_price = avg if self.long_pos > 0 else -avg self.output(f"[自愈] 检测到遗留持仓,接管并启动吊灯保护 进场价≈{avg:.2f}") if self.long_pos > 0: self.state_map.status = f"持多{self.long_pos}手" elif self.short_pos > 0: self.state_map.status = f"持空{self.short_pos}手" elif self.stage == "done": self.state_map.status = "已完成(平仓后停)" elif self.open_orderid is None and self.close_orderid is None: self.state_map.status = "待进场" def _update_chart(self, kline: KLineData) -> None: if not self.widget: return self.widget.recv_kline({ "kline": kline, "signal_price": self.signal_price, **self.main_indicator_data, }) self.signal_price = 0.0 # ---------------- 回调 ---------------- def on_order_cancel(self, order: OrderData) -> None: super().on_order_cancel(order) if order.order_id == self.open_orderid: self.open_orderid = None elif order.order_id == self.close_orderid: self.close_orderid = None def on_order(self, order: OrderData) -> None: super().on_order(order) if order.status in ("全部成交", "已撤销", "部分撤销"): if order.order_id == self.open_orderid: self.open_orderid = None elif order.order_id == self.close_orderid: self.close_orderid = None def on_trade(self, trade: TradeData) -> None: super().on_trade(trade) direction_text = "买入" if trade.direction == "0" else "卖出" offset_text = "开仓" if trade.offset == "0" else ("平今" if trade.offset == "3" else "平仓") self.output(f"成交:{direction_text}{offset_text} {trade.volume}手 价格{trade.price}") if trade.offset == "0": # 分批成交只在【第一笔开仓】初始化状态,后续分批成交不再覆盖进场价/峰值/止损线 if self.stage != "holding": self.stage = "holding" self.entry_price = trade.price self.peak = trade.price self.entry_bar_count = self.bar_count # 记进场当根,当根不止损 self.max_profit_atr = 0.0 # 用实际成交方向初始化棘轮(买入=多→-1e18,卖出=空→+1e18) self.cur_stop = -1e18 if trade.direction == "0" else 1e18 self.state_map.stage = "持仓" self.state_map.entry_price = round(trade.price, 4) self.output(f"已进场 @{trade.price},三阶段吊灯+固定止损开始保护") else: self.output(f"分批成交追加 @{trade.price}(进场价保持 {self.entry_price})") elif trade.offset in ("1", "3"): self.stage = "done" self.cur_stop = 0.0 self.peak = 0.0 self.max_profit_atr = 0.0 self.pending_dir = 0 self.entry_bar_count = -1 self.state_map.stop_line = 0.0 self.state_map.profit_pts = 0.0 self.state_map.stop_offset = 0.0 self.state_map.lock_status = "-" self.state_map.stage = "已完成" self.state_map.status = "已完成(平仓后停)" self.output("本次交易完成,策略进入已完成状态,不再开新仓(如需下一单请手动重启)")