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="SHFE", title="交易所代码") instrument_id: str = Field(default="ao2510", title="合约代码") kline_style: KLineStyleType = Field(default="M5", title="K线周期") direction: int = Field(default=1, title="本次方向") expma_fast: int = Field(default=21, title="EXPMA快线") expma_slow: int = Field(default=42, title="EXPMA慢线") atr_period: int = Field(default=20, title="ATR窗口") k: float = Field(default=2.5, title="吊灯ATR倍数") max_loss: float = Field(default=15, title="最大止损点数") lots: int = Field(default=1, title="开仓手数") over_price: int = Field(default=1, title="委托超价") class State(BaseState): """状态映射模型（界面显示）""" status: str = Field(default="等待数据", title="运行状态") stage: str = Field(default="待进场", title="阶段") expma_fast: float = Field(default=0.0, title="EXPMA快线") expma_slow: float = Field(default=0.0, title="EXPMA慢线") atr: float = Field(default=0.0, title="ATR") entry_price: float = Field(default=0.0, title="进场价") stop_line: float = Field(default=0.0, title="当前止损线") class expma_chandelier_pg(BaseStrategy): """ 手动启动 · 一次性执行工具（不预测方向，方向由你启动前指定） 逻辑： 1. 启动前设定“本次方向”(1做多/-1做空)及各参数。 2. 启动后等待【指定方向】的 EXPMA 交叉：做多等金叉、做空等死叉； K线收线确认（[-1]与[-2]），下一跳对手价进场。 3. 进场后逐 tick 用【吊灯线】与【固定止损上限】中更紧的一条保护： - 多：止损线 = max(最高价since进场 - k*ATR, 进场价 - max_loss) - 空：止损线 = min(最低价since进场 + k*ATR, 进场价 + max_loss) 价格触线即平。反向交叉也平（兜底）。 4. 平仓后进入“已完成”，不再开新仓（要下一单请手动重启策略）。 仅做模拟盘验证执行纪律，不赌信号能预测行情。 """ 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.pending_action = "" # "" / "open" / "close" 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]: return { "EXPMA快": self.expma_fast_val, "EXPMA慢": self.expma_slow_val, "止损线": self.cur_stop, } # ---------------- 生命周期 ---------------- def on_init(self) -> None: super().on_init() self.output("EXPMA吊灯一次单 初始化") 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() if p.expma_fast >= p.expma_slow: self.output("错误: EXPMA快线周期须小于慢线周期") 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} EXPMA{p.expma_fast}/{p.expma_slow} " f"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.is_trading_time(): self._update_position() # 持仓中：逐tick盘中检查吊灯/止损 if self.stage == "holding" and (self.long_pos > 0 or self.short_pos > 0): self._check_exit_intrabar() if not self._has_pending_order(): self._execute_pending_action() 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() if self.trading and self.is_trading_time(): self._evaluate_bar(kline) 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) + 3 if len(bars.close) < need: return fast_arr = bars.ema(timeperiod=p.expma_fast, array=True) slow_arr = bars.ema(timeperiod=p.expma_slow, array=True) self.expma_fast_val = float(fast_arr[-1]) self.expma_slow_val = float(slow_arr[-1]) self.state_map.expma_fast = round(self.expma_fast_val, 4) self.state_map.expma_slow = round(self.expma_slow_val, 4) 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: atr_arr = bars.atr(timeperiod=n, array=True) v = float(atr_arr[-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): tr = max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])) s += tr return s / n except Exception: return self.atr_val def _get_confirmed_cross(self) -> dict | None: """收线后用刚走完的[-1]与再前一根[-2]判金叉/死叉""" if self.kline_generator is None: return None p = self.params_map bars = self.kline_generator.producer if len(bars.close) < max(p.expma_fast, p.expma_slow) + 2: return None fast_arr = bars.ema(timeperiod=p.expma_fast, array=True) slow_arr = bars.ema(timeperiod=p.expma_slow, array=True) cf, cs = float(fast_arr[-1]), float(slow_arr[-1]) pf, ps = float(fast_arr[-2]), float(slow_arr[-2]) if min(cf, cs, pf, ps) <= 0: return None return { "golden": pf <= ps and cf > cs, "death": pf >= ps and cf < cs, } def _evaluate_bar(self, kline: KLineData) -> None: sig = self._get_confirmed_cross() if sig is None: return p = self.params_map dt_str = kline.datetime.strftime("%Y-%m-%d %H:%M:%S") # 待进场：只接受“指定方向”的交叉 if self.stage == "waiting" and self.long_pos == 0 and self.short_pos == 0: if p.direction == 1 and sig["golden"]: self.pending_action = "open" self.output(f"{dt_str} 金叉，等待下一跳开多") elif p.direction == -1 and sig["death"]: self.pending_action = "open" self.output(f"{dt_str} 死叉，等待下一跳开空") return # 持仓：反向交叉平仓（兜底出场） if self.stage == "holding": if self.long_pos > 0 and sig["death"]: self.pending_action = "close" self.output(f"{dt_str} 持多遇死叉，等待下一跳平多") elif self.short_pos > 0 and sig["golden"]: self.pending_action = "close" self.output(f"{dt_str} 持空遇金叉，等待下一跳平空") def _check_exit_intrabar(self) -> None: """逐tick：吊灯线与固定止损取更紧者，触线则平""" p = self.params_map if self.last_price <= 0 or self.atr_val <= 0: 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 self.cur_stop = max(chand, fixed) self.state_map.stop_line = round(self.cur_stop, 4) if self.last_price <= self.cur_stop: self.pending_action = "close" self.output(f"触发出场：多单 现价{self.last_price:.2f} <= 止损线{self.cur_stop:.2f}") 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 self.cur_stop = min(chand, fixed) self.state_map.stop_line = round(self.cur_stop, 4) if self.last_price >= self.cur_stop: self.pending_action = "close" self.output(f"触发出场：空单 现价{self.last_price:.2f} >= 止损线{self.cur_stop:.2f}") def _execute_pending_action(self) -> None: action = self.pending_action if not action: return p = self.params_map if action == "open" and self.stage == "waiting" and self.long_pos == 0 and self.short_pos == 0: if p.direction == 1: self._open("buy", "金叉开多") else: self._open("sell", "死叉开空") elif action == "close" and self.stage == "holding": if self.long_pos > 0: self._close("sell", "平多") elif self.short_pos > 0: self._close("buy", "平空") # ---------------- 下单 ---------------- def _open(self, side: str, reason: str) -> None: if self.open_orderid is not None: return price = self._opp_price_buy() if side == "buy" else self._opp_price_sell() if price is None: self.output(f"开仓跳过：{reason} 对手价无效") return self.open_orderid = self.send_order( exchange=self.params_map.exchange, instrument_id=self.params_map.instrument_id, volume=self.params_map.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"{reason} 对手价{price:.2f} {self.params_map.lots}手") def _close(self, side: str, reason: str) -> None: if self.close_orderid is not None: return 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: self.output(f"平仓跳过：{reason} 对手价无效") return self.close_orderid = self.auto_close_position( exchange=self.params_map.exchange, instrument_id=self.params_map.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"{reason} 对手价{price:.2f} {vol}手") def _opp_price_buy(self) -> float | None: 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) -> float | None: 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 _has_pending_order(self) -> bool: return self.open_orderid is not None or self.close_orderid is not None 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 not self._has_pending_order(): 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 self.pending_action = "" elif order.order_id == self.close_orderid: self.close_orderid = None # 平仓被撤，保留close意图，下一tick重挂 self.pending_action = "close" if self.stage == "holding" else "" 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.pending_action = "" 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.pending_action = "" self.state_map.stage = "已完成" self.state_map.status = "已完成(平仓后停)" self.output("本次交易完成，策略进入已完成状态，不再开新仓（如需下一单请手动重启）") def is_trading_time(self) -> bool: """交易时段判断。默认按氧化铝(SHFE)时段：日盘+夜盘21:00-23:00。 若换品种，请按该品种夜盘收盘时间调整下面的时段。""" now = datetime.now().time() periods = [ (time(9, 0), time(10, 15)), (time(10, 30), time(11, 30)), (time(13, 30), time(15, 0)), (time(21, 0), time(23, 0)), # 氧化铝夜盘；其他品种按需改 ] for start, end in periods: if start <= now < end: return True return False