# -*- coding: utf-8 -*- """ TradeManager — 仓位管理与交易执行 - load_state / save_state : 持久化峰值/当日统计 - get_active_positions : 查询所有活跃持仓 - check_drawdown : 回撤检测与重置 - calc_position_size : 信号强度 → 张数/杠杆 - _find_wall : 订单簿挂单墙定位 - calc_tp_sl : 自适应止盈止损 - manage_position : 决策树动态管理 - should_trade : 冷却期检查 - execute_trade : 开仓执行（带风控） - run_cycle : 主循环一轮 """ import os import json import math import time from datetime import datetime from config import ( INSTRUMENTS, MAX_POSITION_PCT, MAX_TOTAL_PCT, DEFAULT_LEVERAGE, MAX_CONCURRENT, MAX_DRAWDOWN_PCT, STOP_LOSS_PCT, SIGNAL_COOLDOWN, SCORE_THRESHOLD, SCORE_ADD_THRESHOLD, STATE_FILE, MANUAL_TRADE_COOLDOWN, TWO_STRIKE_COOLDOWN, MAX_MANUAL_SESSION, MANUAL_IDLE_TIMEOUT, US_SESSION_START_UTC, US_SESSION_END_UTC, DB_ENABLED, logger, ) from notify.telegram import tg_send from strategy.position_journal import PositionJournal class TradeManager: def __init__(self, okx, analyzer): self.okx = okx self.analyzer = analyzer self.ws_book = None # 由 my_trader.py 注入 OKXBookWS 实例 self.last_signal = {} # {symbol_direction: timestamp} self.last_loss_time = 0 # 上次止损时间 self.consecutive_losses = 0 # 连续亏损次数 self.peak_equity = 0 self.trades_today = 0 self.pnl_today = 0.0 self.trading_paused = False # 回撤硬止损标志 self.start_equity = 0 self.bot_positions = {} # {(inst_id, pos_side): {"entry": price, "ts": time}} self._active_journals = {} # {inst_id: PositionJournal} # 猎杀系统：增强风险控制 self.coin_losses = {} # {symbol: {"count": int, "last_ts": float}} 每币连续止损 self.coin_cooldown_until = {} # {symbol: float} 每币冷却到期时间戳 self.manual_session_start = 0 # 手动操作开始时间戳 self.manual_last_close_ts = 0 # 最后平仓时间戳 self.is_manual_session = False # 是否在手动操作会话中 self.load_state() # ────────────────────────────────────── # 状态持久化 # ────────────────────────────────────── def load_state(self): state_path = STATE_FILE # Docker volume mount: if the file doesn't exist on host, Docker creates a directory if os.path.isdir(state_path): state_path = os.path.join(state_path, "state.json") if os.path.exists(state_path) and os.path.isfile(state_path): try: with open(state_path, 'r') as f: s = json.load(f) self.peak_equity = s.get("peak_equity", 0) self.trades_today = s.get("trades_today", 0) self.pnl_today = s.get("pnl_today", 0) self.trading_paused = s.get("trading_paused", False) self.start_equity = s.get("start_equity", 0) self.bot_positions = { tuple(k): v for k, v in s.get("bot_positions", {}).items() if isinstance(k, (list, tuple)) and len(k) == 2 } self.coin_losses = s.get("coin_losses", {}) self.coin_cooldown_until = s.get("coin_cooldown_until", {}) self.manual_session_start = s.get("manual_session_start", 0) self.is_manual_session = s.get("is_manual_session", False) # 兼容旧格式 set → dict if isinstance(s.get("bot_positions"), list): for p in s["bot_positions"]: if isinstance(p, (list, tuple)) and len(p) == 2: self.bot_positions[tuple(p)] = {"entry": 0, "ts": 0} except: pass def save_state(self): state_path = STATE_FILE if os.path.isdir(state_path): state_path = os.path.join(state_path, "state.json") try: with open(state_path, 'w') as f: json.dump({ "peak_equity": self.peak_equity, "trades_today": self.trades_today, "pnl_today": self.pnl_today, "start_equity": self.start_equity, "trading_paused": self.trading_paused, "bot_positions": {str(list(k)): v for k, v in self.bot_positions.items()}, "coin_losses": self.coin_losses, "coin_cooldown_until": self.coin_cooldown_until, "manual_session_start": self.manual_session_start, "is_manual_session": self.is_manual_session, "updated": datetime.now().isoformat(), }, f) except: pass # ────────────────────────────────────── # 持仓查询 # ────────────────────────────────────── def get_active_positions(self): """获取所有活跃持仓，并标记是否为机器人开仓（is_bot）""" active = [] for sym, cfg in INSTRUMENTS.items(): positions = self.okx.positions(cfg["inst"]) for p in positions: qty = float(p.get("pos", 0)) if qty != 0: inst_id = cfg["inst"] pos_side = p.get("posSide") entry = float(p.get("avgPx", 0)) key = (inst_id, pos_side) rec = self.bot_positions.get(key) # 匹配：有记录 + 入场价相差<0.5%（防误判） is_bot = False if rec: ref_entry = rec.get("entry", 0) if ref_entry > 0: is_bot = abs(entry - ref_entry) / ref_entry < 0.005 else: is_bot = True # 旧格式兼容（无entry字段） active.append({ "symbol": sym, "inst_id": inst_id, "side": pos_side, "qty": abs(qty), "entry": entry, "upl": float(p.get("upl", 0)), "lever": p.get("lever", "0"), "margin": float(p.get("margin", 0)), "last": float(p.get("last", 0)), "is_bot": is_bot, }) return active # ────────────────────────────────────── # 风控 # ────────────────────────────────────── def check_drawdown(self, equity): """ 检查回撤，硬止损：回撤超限后暂停交易，需手动 /reset 解除。 不再仅重置峰值，避免慢性失血。 """ if equity > self.peak_equity: self.peak_equity = equity self.trading_paused = False # 创新高时自动解除暂停 if self.peak_equity > 0: dd = (self.peak_equity - equity) / self.peak_equity if dd > MAX_DRAWDOWN_PCT: if not self.trading_paused: self.trading_paused = True logger.critical( f"🚨 回撤 {dd:.1%} 超限，硬暂停交易！" f"峰值${self.peak_equity:.2f} → 当前${equity:.2f}" ) tg_send( f"🚨 回撤 {dd:.1%} > {MAX_DRAWDOWN_PCT:.0%}，交易已暂停！\n" f"峰值: ${self.peak_equity:.2f} 当前: ${equity:.2f}\n" f"发送 /reset 手动恢复" ) return True, dd return False, 0 # ────────────────────────────────────── # 仓位计算 # ────────────────────────────────────── def calc_position_size(self, symbol, price, equity, score): """按余额比例开仓 — 信号越强占比越大""" cfg = INSTRUMENTS[symbol] ct_val = cfg["ct_val"] ct_usd = ct_val * price # 1张面值(USD) lot_sz = 0.01 # 信号强度 → 仓位占余额比例 # 0.35(阈值) → 5%, 0.5 → 12%, 0.7 → 20%, 1.0 → 25% abs_score = abs(score) position_pct = 0.05 + (abs_score - SCORE_THRESHOLD) * 0.35 position_pct = max(0.05, min(MAX_POSITION_PCT, position_pct)) lever = DEFAULT_LEVERAGE target_usd = equity * position_pct * lever # 杠杆后的名义价值 target_sz = target_usd / ct_usd sz = math.floor(target_sz / lot_sz) * lot_sz sz = round(sz, 2) if sz < lot_sz: logger.warning(f"[仓位] {symbol} 余额不足开仓") return 0, 0 actual_margin = sz * ct_usd / lever margin_pct = actual_margin / equity * 100 logger.info( f"[仓位] {symbol}: score={abs_score:.2f} → 占比{position_pct:.0%} " f"{sz}张 {lever}x 保证金${actual_margin:.2f}({margin_pct:.1f}%)" ) return sz, lever # ────────────────────────────────────── # 订单簿挂单墙 # ────────────────────────────────────── def _find_wall(self, inst_id, price, direction): """找订单簿挂单密集区 — 止损藏在密集区后方""" ob = self.okx.orderbook(inst_id, 400) bucket_pct = 0.001 # 0.1%一个桶 if direction == "LONG": orders = [(float(b[0]), float(b[1])) for b in ob.get("bids", []) if price * 0.97 < float(b[0]) < price * 0.998] else: orders = [(float(a[0]), float(a[1])) for a in ob.get("asks", []) if price * 1.002 < float(a[0]) < price * 1.03] if not orders: return None buckets = {} for px, sz in orders: bucket = round(px / (price * bucket_pct)) * (price * bucket_pct) buckets[bucket] = buckets.get(bucket, 0) + sz if not buckets: return None wall_px = max(buckets, key=buckets.get) wall_sz = buckets[wall_px] avg_sz = sum(buckets.values()) / len(buckets) if wall_sz < avg_sz * 2: # 墙要比平均大2倍才算 return None sl = wall_px * 0.999 if direction == "LONG" else wall_px * 1.001 logger.info( f"[挂单墙] {'bid' if direction=='LONG' else 'ask'}密集区 " f"${wall_px:.1f} qty:{wall_sz:.0f} (avg:{avg_sz:.0f}) → SL=${sl:.2f}" ) return sl # ────────────────────────────────────── # 止盈止损 # ────────────────────────────────────── def calc_tp_sl(self, symbol, price, direction, fr, details=None): """自适应止盈止损：以清算密集区为首选止盈锚，ATR为兜底，压缩盈亏比""" cfg = INSTRUMENTS[symbol] inst_id = cfg["inst"] candles = self.okx.candles(inst_id, bar="5m", limit=12) # ATR计算（用近6根5m K线） if len(candles) >= 6: atr = sum(float(c[2]) - float(c[3]) for c in candles[:6]) / 6 atr_pct = atr / price else: atr_pct = STOP_LOSS_PCT # 判断趋势强度 trend_strength = 0 if details: scores = details.get("scores", {}) raw = details.get("raw", {}) t1h = raw.get("t1h", 0) t4h = raw.get("t4h", 0) trend = scores.get("trend", 0) btc = scores.get("btc_corr", 0) vd = scores.get("vol_delta", 0) total = abs(details.get("total", 0)) if t1h == t4h and t1h != 0: trend_strength += 1 if btc == trend and trend != 0: trend_strength += 1 if vd == trend and trend != 0: trend_strength += 1 if total >= 0.4: trend_strength += 1 trend_strength = min(trend_strength, 2) # ── 止损计算 ── atr_sl_pct = max(STOP_LOSS_PCT, atr_pct * 1.0) # ATR倍数1.0（原1.2） atr_sl_pct = min(atr_sl_pct, 0.012) # 上限1.2%（原2%） wall_sl = self._find_wall(inst_id, price, direction) if wall_sl: wall_pct = abs(price - wall_sl) / price sl_pct = wall_pct if 0.001 <= wall_pct <= 0.012 else atr_sl_pct else: sl_pct = atr_sl_pct # ── 止盈：优先用清算密集区锚点 ── tp_pct = None liq_detail = details.get("raw", {}).get("liq_detail", {}) if details else {} if liq_detail and isinstance(liq_detail, dict): zones_str = liq_detail.get("zones", "") try: import re if direction == "LONG": m = re.search(r'上方\$[\d.]+M\(([\d.]+)%\)', zones_str) else: m = re.search(r'下方\$[\d.]+M\(([\d.]+)%\)', zones_str) if m: zone_dist = float(m.group(1)) / 100 if 0.003 <= zone_dist <= 0.03: tp_cand = zone_dist * 0.85 # 清算锚止盈必须满足盈亏比 ≥ 1.5，否则放弃 if tp_cand >= sl_pct * 1.5: tp_pct = tp_cand except Exception: pass # 清算区锚点不可用/盈亏比不足时，用趋势盈亏比兜底（≥1.5） if tp_pct is None: if trend_strength == 0: tp_pct = sl_pct * 1.5 mode = "震荡" elif trend_strength == 1: tp_pct = sl_pct * 2.0 mode = "弱趋势" else: tp_pct = sl_pct * 2.5 mode = "强趋势" else: mode = "清算锚" sl_pct = min(sl_pct, 0.012) tp_pct = min(tp_pct, 0.04) if direction == "LONG": tp = price * (1 + tp_pct) sl = price * (1 - sl_pct) else: tp = price * (1 - tp_pct) sl = price * (1 + sl_pct) logger.info( f"[TP/SL] {mode}模式(强度{trend_strength}) " f"止盈:{tp_pct:.2%} 止损:{sl_pct:.2%} 盈亏比:{tp_pct/sl_pct:.1f}:1" ) return round(tp, 2), round(sl, 2) # ────────────────────────────────────── # 决策树持仓管理 # ────────────────────────────────────── def manage_position(self, pos, details): """决策树动态管理持仓：平仓/加仓/减仓/调TP-SL/调杠杆""" symbol = pos["symbol"] side = pos["side"] entry = pos["entry"] qty = pos.get("qty", 0) price = details["price"] lever = int(pos.get("lever", DEFAULT_LEVERAGE)) pnl_pct = (price - entry) / entry if side == "long" else (entry - price) / entry scores = details.get("scores", {}) raw = details.get("raw", {}) inst_id = INSTRUMENTS[symbol]["inst"] total_score = details.get("total", 0) # 持仓管理用评分（忽略EX衰竭，EX只影响开仓） hold_score = details.get("total_hold", total_score) # ── 手动单保护：持仓未被 is_bot 标记的，跳过所有程序干预 ── if not pos.get("is_bot", False): logger.info(f"[手动单] {symbol} {side} 非机器人开仓，跳过管理") return False t1h = raw.get("t1h", 0) t4h = raw.get("t4h", 0) trend = scores.get("trend", 0) direction_sign = 1 if side == "long" else -1 trend_against = (trend == -direction_sign) # 修复v2.3: against_count只用核心动量因子（去掉OI/VD，趋势盘常反指） # exhaust因子专用于开仓决策，不参与持仓管理判断 core_check_keys = ["taker", "btc_corr", "gamma", "smart_money"] trend_check_keys = ["taker", "vol_delta", "btc_corr", "oi", "gamma", "smart_money"] aligned_count = sum(1 for k in core_check_keys if scores.get(k, 0) * direction_sign > 0.3) against_count = sum(1 for k in core_check_keys if scores.get(k, 0) * direction_sign < -0.3) # 强趋势（1H+4H同向）时，进一步提高平仓门槛（需要更多核心因子反向） is_trend_aligned = (t1h == t4h and t1h == direction_sign) trend_strength = 0 if t1h == t4h and t1h == direction_sign: trend_strength += 1 if scores.get("btc_corr", 0) * direction_sign > 0.3: trend_strength += 1 if scores.get("vol_delta", 0) * direction_sign > 0.3: trend_strength += 1 avail, equity = self.okx.balance() margin = qty * INSTRUMENTS[symbol]["ct_val"] * price / lever margin_pct = margin / equity if equity > 0 else 1 # ── 0. 强制平仓：综合评分已反转超过开仓阈值（无论盈亏）── if hold_score * direction_sign < -SCORE_THRESHOLD: return self._close_pos( inst_id, side, f"综合评分反转({hold_score*direction_sign:+.3f}<-{SCORE_THRESHOLD})，" f"{'锁利' if pnl_pct >= 0 else '止损'}{pnl_pct:.2%}" ) # ── 1. 强制平仓：趋势反转（允许小亏损时也平，避免扩大损失）── REVERSAL_PNL_FLOOR = -0.005 # 最多允许亏0.5%时仍触发反转平仓 if is_trend_aligned: if t1h == -direction_sign and t4h == -direction_sign and pnl_pct > REVERSAL_PNL_FLOOR: return self._close_pos(inst_id, side, f"趋势双反转(1H:{t1h:+d}/4H:{t4h:+d})，{pnl_pct:.2%}") else: if trend_against and t1h == -direction_sign and pnl_pct > REVERSAL_PNL_FLOOR: return self._close_pos(inst_id, side, f"趋势反转(1H:{t1h:+d})，{pnl_pct:.2%}") # ── 2. 强制平仓：核心因子多数反向（允许小亏损时也平）── close_threshold = 4 if is_trend_aligned else 3 if against_count >= close_threshold and pnl_pct > REVERSAL_PNL_FLOOR: return self._close_pos(inst_id, side, f"{against_count}核心因子反向，{pnl_pct:.2%}") # ── 3. 减仓：核心因子2+反向 + 盈利中（趋势盘不减仓）── if not is_trend_aligned and against_count >= 2 and pnl_pct > 0.005 and qty > 0.02: reduce_sz = round(qty * 0.5, 2) if reduce_sz >= 0.01: close_side = "sell" if side == "long" else "buy" r = self.okx.post("/api/v5/trade/order", { "instId": inst_id, "tdMode": "isolated", "side": close_side, "posSide": side, "ordType": "market", "sz": str(reduce_sz), "reduceOnly": True }) if r.get("code") == "0": logger.info(f"[决策树] 减仓{reduce_sz}张 ({against_count}因子反向，盈利{pnl_pct:.2%})") tg_send(f"📉 减仓{reduce_sz}张 (剩{qty-reduce_sz}张)\n{against_count}因子反向，锁部分利润{pnl_pct:.2%}") # ── 4. 加仓：信号加强 + 盈利 + 仓位不满（用hold_score排除EX衰竭干扰）── score_aligned = (hold_score * direction_sign > SCORE_ADD_THRESHOLD) if score_aligned and pnl_pct > 0.003 and margin_pct < 0.20: add_margin = min(equity * 0.10, margin * 0.5) ct_usd = INSTRUMENTS[symbol]["ct_val"] * price add_sz = math.floor(add_margin * lever / ct_usd / 0.01) * 0.01 if add_sz >= 0.01: open_side = "buy" if side == "long" else "sell" r = self.okx.place_order(inst_id, open_side, side, add_sz) if r.get("code") == "0": logger.info(f"[决策树] 加仓{add_sz}张 (hold_score:{hold_score*direction_sign:+.3f}>{SCORE_ADD_THRESHOLD} 盈利{pnl_pct:.2%})") tg_send(f"📈 加仓{add_sz}张 (共{qty+add_sz}张)\n信号{hold_score*direction_sign:+.3f}加强，盈利{pnl_pct:.2%}") # ── 5. 动态调整TP/SL ── try: algos = self.okx.get_algo_orders(inst_id) for algo in algos: algo_id = algo.get("algoId") old_sl = float(algo.get("slTriggerPx", 0)) old_tp = float(algo.get("tpTriggerPx", 0)) if not algo_id: continue new_sl = None new_tp = None # 移动止损（盈利越多止损越紧） if pnl_pct > 0.02 and old_sl: target_sl = entry * (1 + 0.01 * direction_sign) if side == "long" else entry * (1 - 0.01 * direction_sign) should_move = (target_sl > old_sl) if side == "long" else (target_sl < old_sl) if should_move: new_sl = target_sl elif pnl_pct > 0.01 and old_sl: target_sl = entry * (1 + 0.003 * direction_sign) if side == "long" else entry * (1 - 0.003 * direction_sign) should_move = (target_sl > old_sl) if side == "long" else (target_sl < old_sl) if should_move: new_sl = target_sl elif pnl_pct > 0.005 and old_sl: target_sl = entry * (1 + 0.001 * direction_sign) if side == "long" else entry * (1 - 0.001 * direction_sign) should_move = (target_sl > old_sl) if side == "long" else (target_sl < old_sl) if should_move: new_sl = target_sl # 趋势加强时放大TP if trend_strength >= 2 and old_tp: target_tp = old_tp * (1 + 0.005) if side == "long" else old_tp * (1 - 0.005) should_expand = (target_tp > old_tp) if side == "long" else (target_tp < old_tp) if should_expand: new_tp = target_tp # 执行修改 if new_sl or new_tp: r = self.okx.amend_algo(algo_id, inst_id, new_tp=new_tp, new_sl=new_sl) if r.get("code") == "0": parts = [] if new_sl: parts.append(f"SL:{old_sl:.2f}→{new_sl:.2f}") if new_tp: parts.append(f"TP:{old_tp:.2f}→{new_tp:.2f}") msg = " ".join(parts) logger.info(f"[决策树] {msg} (盈利{pnl_pct:.2%})") tg_send(f"🔄 {msg}") else: # amend失败，cancel+重下 self.okx.cancel_algo(algo_id, inst_id) time.sleep(0.5) final_tp = new_tp or (old_tp if old_tp > 0 else None) final_sl = new_sl or (old_sl if old_sl > 0 else None) self.okx.place_algo_tpsl(inst_id, side, qty, tp=final_tp, sl=final_sl) logger.info(f"[决策树] 重下TP/SL tp={final_tp} sl={final_sl}") except Exception as e: logger.warning(f"[决策树] TP/SL调整失败: {e}") return False def _close_pos(self, inst_id, side, reason): """执行平仓""" logger.info(f"[决策树] 平仓 {side}: {reason}") r = self.okx.close_position(inst_id, side) if r.get("code") == "0": self.bot_positions.pop((inst_id, side), None) self.save_state() # 写入 MySQL if DB_ENABLED: try: from storage.mysql_client import close_trade, insert_event tid = getattr(self, '_last_trade_db_id', None) if tid: close_trade(tid, 0, 0, 0, reason, self._last_equity) insert_event('INFO', 'trader', 'trade_close', f'{inst_id} {side}: {reason}') except Exception as e: logger.warning(f'[DB] Trade close error: {e}') # 关闭持仓日志 journal = self._active_journals.pop(inst_id, None) if journal: try: tick = self.okx.ticker(inst_id) exit_px = float(tick.get("last", 0)) # 简单估算PnL（止损/止盈/手动平仓时） entry = journal.entry_price if journal.direction == "SHORT": pnl = (entry - exit_px) / entry * journal.size * journal.leverage else: pnl = (exit_px - entry) / entry * journal.size * journal.leverage journal.close(exit_px, pnl, pnl / (journal.size * journal.entry_price / journal.leverage) * 100 if journal.entry_price > 0 else 0, reason) except Exception as e: logger.warning(f"[Journal] Close error: {e}") tg_send(f"🔄 动态平仓 {side}\n原因: {reason}") else: logger.warning(f"[决策树] 平仓失败: {r}") return True # ────────────────────────────────────── # 猎杀系统：两振出局（per-coin） # ────────────────────────────────────── def _record_coin_loss(self, symbol): """记录币种止损，连续两次触发16小时冷却""" now = time.time() if symbol not in self.coin_losses: self.coin_losses[symbol] = {"count": 0, "last_ts": 0} c = self.coin_losses[symbol] # 如果上次止损超过24小时，重置计数 if now - c["last_ts"] > 86400: c["count"] = 1 else: c["count"] += 1 c["last_ts"] = now if c["count"] >= 2: self.coin_cooldown_until[symbol] = now + TWO_STRIKE_COOLDOWN hours = TWO_STRIKE_COOLDOWN / 3600 logger.warning( f"[两振] {symbol} 连续{c['count']}次止损，冷却{hours:.0f}小时" f" 至 {datetime.fromtimestamp(self.coin_cooldown_until[symbol]).strftime('%m-%d %H:%M')}" ) tg_send( f"⛔ 两振出局 — {symbol}\n" f"连续{c['count']}次止损，该币种冷却{hours:.0f}小时\n" f"恢复时间: {datetime.fromtimestamp(self.coin_cooldown_until[symbol]).strftime('%m-%d %H:%M')}" ) self.save_state() def _record_coin_win(self, symbol): """止盈重置币种止损计数""" if symbol in self.coin_losses: self.coin_losses[symbol]["count"] = 0 if symbol in self.coin_cooldown_until: del self.coin_cooldown_until[symbol] self.save_state() def _is_coin_cooldown(self, symbol): """检查币种是否在两振冷却期""" if symbol in self.coin_cooldown_until: if time.time() < self.coin_cooldown_until[symbol]: remaining = self.coin_cooldown_until[symbol] - time.time() return True, f"两振冷却 {remaining/3600:.1f}h" else: del self.coin_cooldown_until[symbol] return False, "" # ────────────────────────────────────── # 猎杀系统：手动操作会话追踪 # ────────────────────────────────────── def _start_manual_session(self): """标记手动操作会话开始""" now = time.time() if not self.is_manual_session: self.is_manual_session = True self.manual_session_start = now logger.info(f"[手动会话] 开始于 {datetime.fromtimestamp(now).strftime('%H:%M:%S')}") self.save_state() def _end_manual_session(self): """手动操作结束后检查是否需要冷却""" if self.is_manual_session: now = time.time() duration = now - self.manual_session_start if duration >= MAX_MANUAL_SESSION: logger.warning(f"[手动会话] 超时{duration/3600:.1f}h，触发16h冷却") tg_send(f"⚠️ 手动操作超时{duration/3600:.1f}h，请停止交易，休息16小时") self.is_manual_session = False self.manual_session_start = 0 self.save_state() def _check_manual_cooldown(self) -> tuple: """检查手动交易冷却状态""" now = time.time() # 如果上次手动会话超过3h，需要冷却16h if self.manual_session_start > 0 and not self.is_manual_session: elapsed = now - self.manual_session_start if elapsed < MANUAL_TRADE_COOLDOWN: remaining = MANUAL_TRADE_COOLDOWN - elapsed return False, f"手动冷却 {remaining/3600:.1f}h (16h冷却中)" # 正在手动会话中，检查是否超时 if self.is_manual_session: elapsed = now - self.manual_session_start if elapsed >= MAX_MANUAL_SESSION: return False, f"手动超时 {elapsed/3600:.1f}h，请休息后再操作" return True, "" # ────────────────────────────────────── # 美股时段检测 # ────────────────────────────────────── def _is_us_session(self) -> tuple: """检测当前是否在美股交易时段，返回 (in_session: bool, phase: str)""" now = datetime.utcnow() hour = now.hour + now.minute / 60 weekday = now.weekday() # 0=Monday if weekday >= 5: # 周末 return False, "周末休市" # 美股正常交易时段 (UTC 14:00-21:00) if US_SESSION_START_UTC <= hour < US_SESSION_END_UTC: return True, "美股盘中" # 盘前1小时 (UTC 13:00-14:00) if US_SESSION_START_UTC - 1 <= hour < US_SESSION_START_UTC: return True, "美股盘前" # 盘后1小时 (UTC 21:00-22:00) if US_SESSION_END_UTC <= hour < US_SESSION_END_UTC + 1: return True, "美股盘后" return False, "非美股时段" # ────────────────────────────────────── # 冷却检查（增强版） # ────────────────────────────────────── def should_trade(self, symbol, direction, score=0): """检查是否应该交易（冷却期 + 止损冷却 + 两振出局 + 连亏阈值提升）""" key = f"{symbol}_{direction}" # ── 币种两振冷却检查 ── in_cooldown, cooldown_reason = self._is_coin_cooldown(symbol) if in_cooldown: return False, cooldown_reason if key in self.last_signal: elapsed = time.time() - self.last_signal[key] if elapsed < SIGNAL_COOLDOWN: return False, f"方向冷却 {int(SIGNAL_COOLDOWN-elapsed)}s" if self.last_loss_time > 0: loss_cooldown = SIGNAL_COOLDOWN * (1 + self.consecutive_losses) elapsed = time.time() - self.last_loss_time if elapsed < loss_cooldown: return False, f"止损冷却 {int(loss_cooldown-elapsed)}s (连亏{self.consecutive_losses})" # 连亏时提高开仓阈值：连亏2次要求总分≥0.45，连亏3+要求≥0.55 if self.consecutive_losses >= 3 and abs(score) < 0.55: return False, f"连亏{self.consecutive_losses}次，要求|总分|≥0.55 当前{score:+.3f}" if self.consecutive_losses >= 2 and abs(score) < 0.45: return False, f"连亏{self.consecutive_losses}次，要求|总分|≥0.45 当前{score:+.3f}" return True, "" # ────────────────────────────────────── # 开仓执行 # ────────────────────────────────────── def execute_trade(self, symbol, direction, details): """执行交易""" cfg = INSTRUMENTS[symbol] inst_id = cfg["inst"] price = details["price"] fr = details["funding_rate"] score = details.get("total", 0) avail, equity = self.okx.balance() # 风控检查 drawdown, dd_pct = self.check_drawdown(equity) if drawdown: logger.warning(f"回撤锁定! {dd_pct:.1%} > {MAX_DRAWDOWN_PCT:.0%}") return {"action": "BLOCKED", "reason": f"drawdown {dd_pct:.1%}"} active = self.get_active_positions() if len(active) >= MAX_CONCURRENT: return {"action": "BLOCKED", "reason": f"max concurrent {len(active)}/{MAX_CONCURRENT}"} total_margin = sum(p["margin"] for p in active) if total_margin / equity > MAX_TOTAL_PCT: return {"action": "BLOCKED", "reason": f"total margin {total_margin/equity:.0%} > {MAX_TOTAL_PCT:.0%}"} can_trade, reason = self.should_trade(symbol, direction, score=score) if not can_trade: return {"action": "COOLDOWN", "reason": reason} sz, lever = self.calc_position_size(symbol, price, equity, score) if sz == 0: return {"action": "BLOCKED", "reason": "insufficient balance for min size"} tp, sl = self.calc_tp_sl(symbol, price, direction, fr, details) side = "buy" if direction == "LONG" else "sell" pos_side = "long" if direction == "LONG" else "short" self.okx.set_leverage(inst_id, lever, pos_side) time.sleep(0.3) result = self.okx.place_order(inst_id, side, pos_side, sz, tp=tp, sl=sl) self.last_signal[f"{symbol}_{direction}"] = time.time() self.trades_today += 1 self.save_state() code = result.get("code", "?") success = (code == "0") if success: self._open_equity = equity self.bot_positions[(inst_id, pos_side)] = {"entry": price, "ts": time.time()} self.save_state() # 写入 MySQL if DB_ENABLED: try: from storage.mysql_client import insert_trade, insert_event tid = insert_trade(symbol, direction, price, sz, lever, tp, sl, score, equity, is_bot=True) setattr(self, '_last_trade_db_id', tid) insert_event('INFO', 'trader', 'trade_open', f'{symbol} {direction} {sz}张 @${price:.2f} TP${tp} SL${sl}', {'score': score, 'leverage': lever}) except Exception as e: logger.warning(f'[DB] Trade insert error: {e}') # 启动持仓日志 self._active_journals[inst_id] = PositionJournal( symbol, inst_id, direction, price, sz, lever) logger.info(f"[Journal] Started for {symbol} {direction}") ord_id = "" if success and result.get("data"): ord_id = result["data"][0].get("ordId", "") return { "action": direction, "symbol": symbol, "size": sz, "price": price, "tp": tp, "sl": sl, "leverage": lever, "success": success, "ord_id": ord_id, "code": code, "msg": result.get("msg", ""), "snapshot": { "factors": details.get("scores", {}), "raw": {k: v for k, v in details.get("raw", {}).items() if k not in ("headlines",)}, "score": score, "equity": equity, "ts": time.time(), } if success else None, } # ────────────────────────────────────── # 主循环单轮 # ────────────────────────────────────── def run_cycle(self): """执行一轮分析和交易""" from config import MAX_CONCURRENT, MAX_TOTAL_PCT avail, equity = self.okx.balance() if self.start_equity == 0: self.start_equity = equity if equity > self.peak_equity: self.peak_equity = equity dd_pct = (self.peak_equity - equity) / self.peak_equity if self.peak_equity > 0 else 0 in_us, us_phase = self._is_us_session() logger.info(f"[账户] 权益:${equity:.4f} 可用:${avail:.4f} 峰值:${self.peak_equity:.4f} " f"回撤:{dd_pct:.2%} | {us_phase}") # 回撤硬暂停检查 if self.trading_paused: # 检查是否已恢复（创新高会自动解除） if equity > self.peak_equity: self.trading_paused = False else: results = {"equity": round(equity, 4), "available": round(avail, 4), "signals": [{"direction": "PAUSED", "action": "PAUSED", "reason": "drawdown hard-stop, /reset to continue", "price": 0, "score": 0, "factors": {}, "raw": {}}]} self.check_drawdown(equity) self.save_state() return results results = {"equity": round(equity, 4), "available": round(avail, 4), "signals": []} active = self.get_active_positions() results["positions"] = active # 写入 MySQL 权益历史和持仓快照 if DB_ENABLED: try: from storage.mysql_client import insert_equity, insert_position_snapshot total_margin_val = sum(p["margin"] for p in active) if active else 0 insert_equity(equity, avail, self.peak_equity, dd_pct, total_margin_val, len(active) if active else 0) if active: insert_position_snapshot(active) except Exception as e: logger.warning(f'[DB] Equity insert error: {e}') if active: for p in active: logger.info( f"[持仓] {p['symbol']} {p['side']} {p['qty']}张 @ ${p['entry']:.2f} " f"P&L:${p['upl']:.4f} margin:${p['margin']:.2f} {p['lever']}x" ) else: logger.info("[持仓] 无") # ── 持仓日志快照 ── for p in active: inst_id = p.get("inst_id", "") journal = self._active_journals.get(inst_id) if journal: try: ob = self.okx.orderbook(inst_id, 20) journal.snapshot_ob(p.get("last", 0), ob.get("asks", []), ob.get("bids", [])) liq_orders = self.okx.liquidations(inst_id) for lo in liq_orders[:5]: journal.snapshot_liq( float(lo.get("bkPx", 0)), lo.get("side", ""), float(lo.get("sz", 0))) ccy = p.get("symbol", "ETH") tv = self.okx.taker_volume(ccy, "5m") if tv: journal.snapshot_taker( float(tv[0][1]) if len(tv) > 0 else 0, float(tv[0][2]) if len(tv) > 0 else 0) lsr = self.okx.long_short_ratio(ccy, "5m") if lsr: journal.snapshot_ls_ratio(float(lsr[0][1])) except Exception: pass total_margin = sum(p["margin"] for p in active) margin_pct = total_margin / equity if equity > 0 else 0 logger.info(f"[风控] 持仓数:{len(active)}/{MAX_CONCURRENT} 保证金占比:{margin_pct:.1%}/{MAX_TOTAL_PCT:.0%}") # 浮亏超30%强制平仓（仅机器人开仓） for p in active: if not p.get("is_bot", False): continue pnl_pct = p["upl"] / p["margin"] if p["margin"] > 0 else 0 if pnl_pct < -0.3: logger.warning(f"[风控] {p['symbol']} {p['side']} 浮亏{pnl_pct:.0%}，强制平仓!") self.okx.close_position(p["inst_id"], p["side"]) self.last_loss_time = time.time() self.consecutive_losses += 1 tg_send(f"🚨 强制平仓 {p['symbol']} {p['side']}，浮亏{pnl_pct:.0%}，连亏{self.consecutive_losses}") time.sleep(1) active = self.get_active_positions() # 止损/止盈触发检测（用开仓权益对比，扣除手续费容差） # 仅当上一轮有机器人持仓、本轮全部平仓时触发 had_bot = getattr(self, '_last_had_bot_position', False) if not active and had_bot: avail_now, eq_now = self.okx.balance() open_eq = getattr(self, '_open_equity', self._last_equity) pnl = eq_now - open_eq fee_est = open_eq * 0.001 # 预估手续费 if pnl < -fee_est: # 真实亏损（超过手续费） self.last_loss_time = time.time() self.consecutive_losses += 1 # 找到平仓的币种，触发两振检测 last_closed = getattr(self, '_last_had_positions', []) for lp in last_closed: self._record_coin_loss(lp) logger.warning(f"[止损] 检测到止损触发，亏损${abs(pnl):.4f}，连亏{self.consecutive_losses}") tg_send(f"🔴 止损触发，亏损${abs(pnl):.2f}，连亏{self.consecutive_losses}次，" f"冷却{SIGNAL_COOLDOWN*(1+self.consecutive_losses)//60}分钟") elif pnl > fee_est: # 真实盈利（超过手续费） self.consecutive_losses = 0 last_closed = getattr(self, '_last_had_positions', []) for lp in last_closed: self._record_coin_win(lp) logger.info(f"[止盈] 检测到止盈触发，盈利${pnl:.4f}") tg_send(f"🟢 止盈触发，盈利${pnl:.2f}") else: logger.info(f"[平仓] 手续费内收支平衡，pnl=${pnl:.4f}") self._open_equity = 0 # 重置 self._last_had_bot_position = any(p.get("is_bot") for p in active) self._last_had_positions = [p["symbol"] for p in active if p.get("is_bot")] self._last_had_position = len(active) > 0 self._last_equity = equity # 分析每个标的 for symbol in INSTRUMENTS: try: score, details = self.analyzer.analyze(symbol) direction = details["direction"] # 动态持仓管理（跳过手动单，manage_position内部也有二重保护） for pos in active: if pos["symbol"] == symbol and pos.get("is_bot", False): closed = self.manage_position(pos, details) if closed: active = self.get_active_positions() break sig = { "symbol": symbol, "price": details["price"], "score": details["total"], "direction": direction, "factors": details["scores"], "raw": details.get("raw", {}), "funding_rate": details.get("funding_rate", 0), "max_pain": details.get("max_pain"), } # ── 幌骗检测状态注入 ── spoof_summary = self.ws_book.get_summary() if self.ws_book else None spoof_score = spoof_summary["score"] if spoof_summary else 0.0 spoof_risk = spoof_summary["risk_level"] if spoof_summary else "normal" spoof_side = spoof_summary["wall_side"] if spoof_summary else None sig["spoof"] = spoof_summary # 幌骗评分叠加到 OB 因子（权重 30%） if spoof_summary and abs(spoof_score) > 0.1: ob_raw = details["scores"].get("orderbook", 0) # 幌骗卖墙→预期上涨→增强多头 OB；幌骗买墙→增强空头 OB ob_adjusted = ob_raw * 0.7 + spoof_score * 0.3 details["scores"]["orderbook"] = round(ob_adjusted, 4) # 同步更新 total（简单重算，保持量级） from config import W_ORDERBOOK delta = (ob_adjusted - ob_raw) * W_ORDERBOOK details["total"] = round(details["total"] + delta, 4) score = details["total"] direction = "LONG" if score >= 0.25 else ("SHORT" if score <= -0.25 else "WAIT") sig["score"] = score sig["direction"] = direction if abs(spoof_score) > 0.3: wall_desc = f"{spoof_side}墙" if spoof_side else "未知墙" logger.info( f"[幌骗] OB调整: {ob_raw:+.3f} → {ob_adjusted:+.3f}" f" 总分: {sig['score']:+.3f} 方向: {direction}" f" 风险: {spoof_risk} ({wall_desc})" ) if direction != "WAIT": existing = [p for p in active if p["symbol"] == symbol] has_same = any((p["side"] == "long" if direction == "LONG" else p["side"] == "short") for p in existing) has_opposite = any((p["side"] == "short" if direction == "LONG" else p["side"] == "long") for p in existing) if has_same: sig["action"] = "HOLD" logger.info(f"[交易] {symbol} {direction} -> HOLD (决策树管理中)") elif has_opposite: opp_side = "short" if direction == "LONG" else "long" logger.info(f"[交易] {symbol} 信号翻转 -> 先平{opp_side}仓") close_r = self.okx.close_position(INSTRUMENTS[symbol]["inst"], opp_side) close_code = close_r.get("code", "?") if close_code == "0": logger.info(f"[交易] ✅ 平{opp_side}仓成功") time.sleep(1) active = self.get_active_positions() trade = self.execute_trade(symbol, direction, details) sig["trade"] = trade else: logger.warning(f"[交易] ❌ 平仓失败: {close_r.get('msg')}") sig["action"] = f"CLOSE_FAILED ({close_r.get('msg', '')})" else: # 幌骗 danger 且方向相反时阻止开仓（spoof_side 必须有效） if spoof_risk == "danger" and spoof_side: spoof_blocks = ( (direction == "LONG" and spoof_side == "buy") # 幌骗买墙→预期下跌，不做多 or (direction == "SHORT" and spoof_side == "sell") # 幌骗卖墙→预期上涨，不做空 ) if spoof_blocks: logger.warning( f"[幌骗] ⛔ 开仓被阻止: {direction} 与幌骗{spoof_side}墙方向冲突" f"（score={spoof_score:+.2f}）" ) sig["action"] = f"BLOCKED_SPOOF ({spoof_side}墙幌骗，预期反转)" results["signals"].append(sig) continue logger.info(f"[交易] {symbol} 触发 {direction}，执行开仓...") trade = self.execute_trade(symbol, direction, details) sig["trade"] = trade if "trade" in sig: t = sig["trade"] if t.get("success"): logger.info( f"[交易] ✅ {symbol} {direction} {t['size']}张 " f"@ ${t['price']:.2f} TP:${t['tp']} SL:${t['sl']} {t['leverage']}x" ) # 保存因子快照 if t.get("snapshot"): try: snap_file = "trade_snapshots.json" snaps = [] if os.path.exists(snap_file): with open(snap_file, "r", encoding="utf-8") as sf: snaps = json.load(sf) snaps.append({ "time": time.strftime("%Y-%m-%d %H:%M:%S"), "action": direction, "symbol": symbol, "size": t["size"], "price": t["price"], "tp": t["tp"], "sl": t["sl"], "lever": t["leverage"], **t["snapshot"] }) snaps = snaps[-100:] with open(snap_file, "w", encoding="utf-8") as sf: json.dump(snaps, sf, ensure_ascii=False, indent=1) except Exception as e: logger.warning(f"快照保存失败: {e}") elif t.get("action") in ("BLOCKED", "COOLDOWN"): logger.info(f"[交易] ⏸ {symbol} {t.get('reason')}") else: logger.warning(f"[交易] ❌ {symbol} 失败: {t.get('msg', t.get('code'))}") results["signals"].append(sig) time.sleep(1) except Exception as e: logger.error(f"Error analyzing {symbol}: {e}") self.save_state() return results