# -*- coding: utf-8 -*- """ StrategyRunner — 策略调度器 - 管理多个策略实例 - 定时轮询K线，分发给各策略 - 信号收集、冲突检测、优先级排序 - 支持手动开启/关闭单个策略 """ import time import threading import logging from collections import defaultdict from .base_strategy import BaseStrategy, Signal from .signal_filter import DelayAwareFilter, FilteredSignal from .turtle import TurtleStrategy from .mean_revert import MeanReversionStrategy from .multi_factor import MultiFactorStrategy from .grid import GridStrategy from .trend import TrendFollowingStrategy logger = logging.getLogger("MyTrader") class StrategyRunner: """策略调度器""" def __init__(self, executor, instruments: dict, analyzer=None): self.executor = executor self.instruments = instruments self.analyzer = analyzer # MarketAnalyzer (for multi_factor) self.strategies: dict[str, BaseStrategy] = {} self._running = False self._thread = None self._bar_interval = 60 # 默认60秒轮询 # 延迟感知信号过滤器 self.signal_filter = DelayAwareFilter( max_signal_age=120, confirmation_period=30, max_price_slippage=0.005, pullback_pct=0.003, htf_required=True) # 有毒流检测器（由外部注入） self.toxic_detector = None # 信号统计 self.signal_log = [] # [{ts, strategy, action, score, price}] # ── 策略注册 ────────────────────────────────── def register(self, strategy: BaseStrategy): """注册策略""" self.strategies[strategy.cfg.name] = strategy logger.info(f"[Runner] 注册策略: {strategy.cfg.name} ({strategy.symbol})") def create_default_strategies(self): """创建默认5策略""" from .base_strategy import StrategyConfig # 海龟 self.register(TurtleStrategy(self.executor, StrategyConfig( name="turtle", symbol="ETH", max_position_pct=0.25))) # 均值回归 self.register(MeanReversionStrategy(self.executor, StrategyConfig( name="mean_revert", symbol="ETH", max_position_pct=0.2))) # 多因子 mf = MultiFactorStrategy(self.executor, self.analyzer, StrategyConfig( name="multi_factor", symbol="ETH", max_position_pct=0.3)) if self.analyzer: mf.set_analyzer(self.analyzer) self.register(mf) # 网格 self.register(GridStrategy(self.executor, StrategyConfig( name="grid", symbol="ETH", max_position_pct=0.4))) # 趋势 self.register(TrendFollowingStrategy(self.executor, StrategyConfig( name="trend", symbol="ETH", max_position_pct=0.3))) # ── 开关控制 ────────────────────────────────── def enable(self, name: str): if name in self.strategies: self.strategies[name].cfg.enabled = True return f"策略 {name} 已开启" return f"策略 {name} 不存在" def disable(self, name: str): if name in self.strategies: self.strategies[name].cfg.enabled = False return f"策略 {name} 已关闭" return f"策略 {name} 不存在" def list_strategies(self) -> list: """列出所有策略及状态""" return [s.get_state_summary() for s in self.strategies.values()] # ── 主循环 ──────────────────────────────────── def run_cycle(self, symbol: str = "ETH") -> list[Signal]: """执行一轮：获取K线 → 分发给各策略 → 收集信号""" executed = [] # 获取最新K线 try: cfg = self.instruments.get(symbol.upper()) if not cfg: return executed candles = self.executor.okx.candles(cfg["inst"], bar="5m", limit=3) if not candles: return executed # 取最近一根完整K线 latest = candles[0] # OKX返回 [ts, o, h, l, c, vol, volCcy] ts = int(latest[0]) o, h, l, c, v = float(latest[1]), float(latest[2]), float(latest[3]), float(latest[4]), float(latest[5]) except Exception as e: logger.warning(f"[Runner] K线获取失败: {e}") return executed # 分发给各策略 signals = {} for name, strategy in self.strategies.items(): if not strategy.enabled: continue try: signal = strategy.feed_bar(o, h, l, c, v, ts) if signal and signal.action != "WAIT": signals[name] = signal self.signal_log.append({ "ts": ts, "strategy": name, "action": signal.action, "score": signal.score, "price": c, "reason": signal.reason, }) if len(self.signal_log) > 500: self.signal_log = self.signal_log[-500:] except Exception as e: logger.error(f"[Runner] 策略 {name} 异常: {e}", exc_info=True) # 信号冲突检测：同时有LONG和SHORT → 选评分绝对值高的 long_sigs = {k: v for k, v in signals.items() if v.action == "LONG"} short_sigs = {k: v for k, v in signals.items() if v.action == "SHORT"} if long_sigs and short_sigs: best_long = max(long_sigs.items(), key=lambda x: x[1].score) best_short = max(short_sigs.items(), key=lambda x: abs(x[1].score)) if best_long[1].score >= abs(best_short[1].score): # LONG优先 sig = best_long[1] self._execute(sig, best_long[0]) executed.append(best_long[1]) logger.info(f"[Runner] 冲突仲裁: {best_long[0]} LONG > {best_short[0]} SHORT") else: sig = best_short[1] self._execute(sig, best_short[0]) executed.append(best_short[1]) logger.info(f"[Runner] 冲突仲裁: {best_short[0]} SHORT > {best_long[0]} LONG") else: for name, sig in signals.items(): self._execute(sig, name) executed.append(sig) return executed def _execute(self, sig: Signal, strategy_name: str): """执行信号（经过延迟感知过滤）""" strat = self.strategies.get(strategy_name) if strat is None: return # 标记活跃策略 (用于 trade_log 归因) if hasattr(self.executor, 'active_strategy'): self.executor.active_strategy = strategy_name # ── 延迟感知过滤 ── # 获取当前价格、HTF、有毒流状态 try: current_price = self.executor.ticker_price(sig.symbol if hasattr(sig, 'symbol') else "ETH") except Exception: current_price = sig.price t1h, t4h = 0, 0 toxic_level, toxic_bias = "normal", "neutral" # 从analyzer获取HTF if self.analyzer: try: t1h = getattr(self.analyzer, '_last_t1h', 0) or 0 t4h = getattr(self.analyzer, '_last_t4h', 0) or 0 except Exception: pass # 从有毒流检测器获取状态 if self.toxic_detector: try: summary = self.toxic_detector.analyze() toxic_level = summary.level toxic_bias = summary.bias except Exception: pass # 运行过滤器 filtered = self.signal_filter.filter( signal_action=sig.action, signal_price=sig.price, signal_ts=time.time() - 30, # 假定信号延迟30秒 current_price=current_price, t1h=t1h, t4h=t4h, toxic_level=toxic_level, toxic_bias=toxic_bias, ) # 根据过滤结果执行 if filtered.action == "VETO": logger.warning(f"[Runner] {strategy_name} {sig.action} 被否决: {filtered.reason}") return if filtered.action == "WAIT": logger.info(f"[Runner] {strategy_name} {sig.action} 等待确认: {filtered.reason}") return if filtered.entry_mode == "pullback": logger.info(f"[Runner] {strategy_name} {sig.action} 等回调@{filtered.target_price}: {filtered.reason}") # 设置一个条件单：如果价格回到目标价就入场 # 此处简化：降低仓位直接入场（实际应布设限价单） sig.size = sig.size * 0.5 # 减半仓位 sig.price = filtered.target_price try: success = strat.execute_signal(sig) if success: logger.info(f"[Runner] {strategy_name} {filtered.action} 已执行 " f"conf={filtered.confidence:.1%} mode={filtered.entry_mode} {filtered.reason[:60]}") except Exception as e: logger.error(f"[Runner] 执行失败 {strategy_name}: {e}") # ── 后台线程 ────────────────────────────────── def start_background(self, interval: int = 60): """后台线程模式""" if self._running: return self._running = True self._bar_interval = interval def _loop(): logger.info(f"[Runner] 后台策略引擎启动 interval={interval}s") while self._running: try: self.run_cycle("ETH") except Exception as e: logger.error(f"[Runner] 循环异常: {e}") time.sleep(interval) self._thread = threading.Thread(target=_loop, name="StrategyRunner", daemon=True) self._thread.start() def stop_background(self): self._running = False def is_running(self) -> bool: return self._running # ── 查询 ────────────────────────────────────── def get_recent_signals(self, n: int = 20) -> list: return self.signal_log[-n:] def get_status(self) -> dict: return { "running": self._running, "strategies": self.list_strategies(), "recent_signals": self.get_recent_signals(5), }