from datetime import datetime, time 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窗口") k: float = Field(default=2.5, title="吊灯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") stop_line: float = Field(default=0.0, title="止损线") class ChandelierExec(BaseStrategy): """ 启动即进场 · 吊灯止盈止损 · 一次性执行工具（无限易 pythongo） 方向与时机由你判断:启动前设 direction(1做多/-1做空),策略一启动进入交易时段 立即按方向市价进场;进场后逐 tick 用【吊灯线】(棘轮,只朝锁利方向移动)与 【固定止损】(进场∓max_loss)中更紧者保护,现价触及即平;平仓后不再进场(一次性)。 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 # 多:进场后最高价 / 空:进场后最低价 self.cur_stop = 0.0 # 棘轮后的实际止损线 self.open_orderid: int | None = None self.close_orderid: int | None = None self.signal_price = 0.0 @property def main_indicator_data(self) -> dict[str, float]: # 止损线只在持仓、且已算出有效值时才画;否则用 nan(不显示),避免残留上一单的值 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("ChandelierExec 初始化") 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} " f"本次方向={dir_txt} ATR{p.atr_period} k={p.k} 最大止损={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 self.trading and self.is_trading_time(): 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() # 持仓:逐tick盯吊灯+固定止损(棘轮) 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) if self.trading: 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: """优先内置 atr,不可用则用 高/低/昨收 手工算""" 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 # ---------------- 进场 / 出场 ---------------- def _open(self) -> None: p = self.params_map if p.direction == 1: price = self._opp_price_buy() side = "buy" elif p.direction == -1: price = self._opp_price_sell() side = "sell" else: return if price is None: 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.signal_price = price if side == "buy" else -price self.state_map.status = "进场委托中" self.output(f"{'开多' if side == 'buy' else '开空'} 对手价{price:.2f} {p.lots}手") 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_atr = p.k * self.atr_val if self.long_pos > 0: self.peak = max(self.peak, self.last_price) chand = self.peak - k_atr 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) if self.last_price <= self.cur_stop: self._close("sell") elif self.short_pos > 0: self.peak = min(self.peak, self.last_price) chand = self.peak + k_atr 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) 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() if side == "sell" else self._opp_price_buy() if price is None: 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.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": # 开仓成交 -> 进入持仓,用成交价作进场价与初始峰值;棘轮初始设极值,首次检查即被raw赋值 self.stage = "holding" self.entry_price = trade.price self.peak = trade.price 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.state_map.stop_line = 0.0 self.state_map.stage = "已完成" self.state_map.status = "已完成(平仓后停)" self.output("本次交易完成,策略进入已完成状态,不再开新仓(如需下一单请手动重启)") def is_trading_time(self) -> bool: """交易时段。默认按 IC(中金所股指)日盘;换商品期货请改成对应日盘+夜盘时段。""" now = datetime.now().time() periods = [ (time(9, 30), time(11, 30)), (time(13, 0), time(15, 0)), ] for start, end in periods: if start <= now < end: return True return False