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 初始化") 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启动 {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 self.last_price = tick.last_price # 过滤无效买卖一价(开盘首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线记一次过程数据,便于复盘还原止损线演变 if self.stage == "holding" and (self.long_pos > 0 or self.short_pos > 0): n, stage = self._current_n() if self.long_pos > 0: prof_pts = self.last_price - self.entry_price else: prof_pts = 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线不检查止损,等下一根K线收线后(计数增加)才开始保护 if self.bar_count <= self.entry_bar_count: return if self.long_pos > 0: self.peak = max(self.peak, self.last_price) # HH 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 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: self.peak = min(self.peak, self.last_price) # LL 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 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("本次交易完成,策略进入已完成状态,不再开新仓(如需下一单请手动重启)")