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线周期") direction: 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") 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倍数") # 后紧 max_loss: float = Field(default=15, title="最大止损点数") # 0=关闭只用吊灯 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倍数") stop_line: float = Field(default=0.0, title="止损线") class AdaptiveChandelier(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 @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("AdaptiveChandelier 初始化") 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" dir_txt = "做多" if p.direction == 1 else "做空" self.state_map.stage = "待进场" self.output( f"策略启动 {p.exchange}.{p.instrument_id} {p.kline_style} 方向={dir_txt} " f"ATR{p.atr_period} 三阶段N={p.k_stage1}/{p.k_stage2}/{p.k_stage3} " f"阈值={p.stage2_trigger}/{p.stage3_trigger}ATR 最大止损={p.max_loss}点 —— 收到行情即进场" ) 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 self.bid_price1 = tick.bid_price1 self.ask_price1 = tick.ask_price1 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() if (self.stage == "waiting" and self.long_pos == 0 and self.short_pos == 0 and self.open_orderid is None): self._open() elif self.stage == "holding" 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._calc_indicators() self._update_chart(kline) def on_bar_realtime(self, kline: KLineData) -> None: self._calc_indicators() self._update_chart(kline) 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.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 p.direction == 1: price = self._opp_price_buy() or self.last_price side = "buy" elif p.direction == -1: price = self._opp_price_sell() or self.last_price side = "sell" else: self.output("[诊断] direction 非 1/-1,不进场") 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 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 raw = max(chand, fixed) 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.stage = f"持多·阶段{stage}·N={n}" if self.last_price <= self.cur_stop: 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 raw = min(chand, fixed) 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.stage = f"持空·阶段{stage}·N={n}" if self.last_price >= self.cur_stop: 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.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": self.stage = "holding" self.entry_price = trade.price self.peak = trade.price self.max_profit_atr = 0.0 self.cur_stop = -1e18 if self.params_map.direction == 1 else 1e18 self.state_map.stage = "持仓" self.state_map.entry_price = round(trade.price, 4) self.output(f"已进场 @{trade.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.state_map.stop_line = 0.0 self.state_map.stage = "已完成" self.state_map.status = "已完成(平仓后停)" self.output("本次交易完成,策略进入已完成状态,不再开新仓(如需下一单请手动重启)")