# -*- coding: utf-8 -*- """ Claude's Trading System — 主入口 综合策略: 插针捕捉 + 动量突破 + 均值回归 数据源: OKX + Deribit 模块结构: exchange/okx.py — OKX REST/algo 封装 exchange/deribit.py — Deribit 期权数据 strategy/analyzer.py — MarketAnalyzer (15因子) strategy/trader.py — TradeManager (仓位/风控) strategy/formatter.py— format_analysis / format_cycle strategy/handler.py — handle_command (Telegram命令) notify/telegram.py — tg_send / tg_get_updates config.py — 全局配置 """ import sys import time import threading import os import subprocess # Windows 编码修复 if sys.stdout and hasattr(sys.stdout, 'reconfigure'): try: sys.stdout.reconfigure(encoding='utf-8') except: pass if sys.stderr and hasattr(sys.stderr, 'reconfigure'): try: sys.stderr.reconfigure(encoding='utf-8') except: pass from config import ( INSTRUMENTS, DEFAULT_LEVERAGE, CHECK_INTERVAL, SCORE_THRESHOLD, TG_CHAT, PROXIES, logger, TG_API_ID, TG_API_HASH, TG_NEWS_CHANNELS, DB_ENABLED, DB_HOST, DB_PORT, DB_USER, DB_PASSWORD, DB_NAME, PAPER_TRADING, PAPER_INITIAL_EQUITY, GOLD_TRADING_ENABLED, ) from exchange.okx import OKX from exchange.deribit import Deribit from exchange.okx_ws import OKXBookWS from strategy.analyzer import MarketAnalyzer from strategy.gold_analyzer import GoldAnalyzer from strategy.trader import TradeManager from strategy.formatter import format_analysis, format_gold_analysis from strategy.handler import handle_command from strategy.runner import StrategyRunner from exchange.order_executor import OrderExecutor from exchange.data_pusher import DataPusher from exchange.anomaly_detector import PriceAnomalyDetector try: from strategy.factor_engine import FactorEngine except ImportError: FactorEngine = None from notify.telegram import tg_send, tg_send_photo, tg_get_updates from notify.tg_channel_listener import start_channel_listener _DASHBOARD_PORTS = [8765, 8766, 8767] # 本地 HTTP 端口（依次尝试） def _start_http_server(root, port): """在 root 目录启动简易 HTTP 服务（供浏览器访问 chart_data.json）。返回实际端口。""" import http.server os.chdir(root) class _H(http.server.SimpleHTTPRequestHandler): def log_message(self, *a): pass try: with http.server.HTTPServer(('', port), _H) as srv: srv.serve_forever() except Exception as e: logger.warning(f"[Dashboard] HTTP服务异常: {e}") def _start_dashboard(): """后台线程：生成 detail 图表并打开浏览器，同时持续刷新 live 图表（每30秒）。""" import subprocess, webbrowser _root = os.path.dirname(os.path.abspath(__file__)) # 端口选择（尝试多个端口） port = _DASHBOARD_PORTS[0] for p in _DASHBOARD_PORTS: try: import socket s = socket.socket() s.bind(('', p)) s.close() port = p break except OSError: continue else: logger.warning(f"[Dashboard] 所有端口被占用: {_DASHBOARD_PORTS}") # HTTP 服务优先启动，让 detail 也能通过 localhost 访问 t_http = threading.Thread( target=_start_http_server, args=(_root, port), name="DashHTTP", daemon=True ) t_http.start() time.sleep(0.5) # 等 HTTP 服务绑定端口 try: # 生成 detail 图表（默认启动页） subprocess.run( [sys.executable, "tools/dashboard.py", "--mode=detail", "--no-open"], cwd=_root, timeout=120, capture_output=True ) url = f"http://localhost:{port}/factor_detail_charts.html" webbrowser.open(url) logger.info(f"[Dashboard] factor_detail_charts 图表已在浏览器打开: {url}") except Exception as e: logger.warning(f"[Dashboard] 初始生成失败: {e}") # 持续刷新 live 数据（chart.html 后台更新，不再自动打开） while True: try: time.sleep(30) subprocess.run( [sys.executable, "tools/dashboard.py", "--mode=live", "--data"], cwd=_root, timeout=25, capture_output=True ) except Exception as e: logger.debug(f"[Dashboard] 刷新异常: {e}") # 上次发送截图的时间戳（模块级，主循环中更新） _last_snapshot_time = 0.0 _detail_opened = True # 启动时已由 _start_dashboard 打开，后续刷新不再重复开浏览器 _detail_lock = threading.Lock() # 防止重建并发叠加 def _refresh_detail_async(): """后台线程：直接调用 generate_charts.py 生成 HTML。""" _root = os.path.dirname(os.path.abspath(__file__)) try: # Use subprocess - simpler and avoids import path issues subprocess.run( [sys.executable, os.path.join(_root, 'tools', 'generate_charts.py')], cwd=_root, timeout=30 ) logger.info('[Chart] factor_detail_charts.html updated') except Exception as e: logger.warning('[Chart] generation error: %s' % str(e)[:80]) def _try_send_snapshot(): """生成并发送4小时主图截图到 Telegram，失败静默。""" try: from tools.chart_snapshot import make_snapshot png = make_snapshot(hours=4) if png: tg_send_photo(png, caption="📊 ETH 4h 价格 + 总分走势") logger.info("[Snapshot] 4h截图已发送") except Exception as e: logger.debug(f"[Snapshot] 截图失败: {e}") def main(): logger.info("=" * 50) logger.info("Claude's Trading System starting") logger.info(f"Instruments: {list(INSTRUMENTS.keys())}") logger.info(f"Leverage: {DEFAULT_LEVERAGE}x | Check: {CHECK_INTERVAL}s") logger.info("=" * 50) okx = OKX() deribit = Deribit() analyzer = MarketAnalyzer(okx, deribit) manager = TradeManager(okx, analyzer) # 策略引擎（5策略：海龟/均值回归/多因子/网格/趋势） paper_trader = None if PAPER_TRADING: from exchange.paper_executor import PaperExecutor paper_trader = PaperExecutor(okx, INSTRUMENTS, PAPER_INITIAL_EQUITY, DEFAULT_LEVERAGE) executor = paper_trader # PaperExecutor 复刻了 OrderExecutor 接口 logger.info(f"[Paper] 模拟交易模式 初始权益=${PAPER_INITIAL_EQUITY}") else: executor = OrderExecutor(okx, INSTRUMENTS, DEFAULT_LEVERAGE) logger.info("[Live] 实盘交易模式") runner = StrategyRunner(executor, INSTRUMENTS, analyzer) runner.create_default_strategies() # ── 连接最高统帅部网关 ── gateway_url = os.environ.get("GATEWAY_URL", "http://127.0.0.1:8000") gateway_token = os.environ.get("GATEWAY_TOKEN", "supreme-internal-token-2026") gw = None try: from exchange.gateway_hooks import attach_gateway, gateway_send_status gw = attach_gateway(manager, gateway_url, gateway_token) # 扩展回调：战略部信号 → 通知策略引擎 def _on_strategy_signal(cmd): logger.info(f"[GW] 战略部信号: {cmd.asset} {cmd.direction} conviction={cmd.conviction}") if cmd.should_adjust_position(): ratio = cmd.get_max_position_ratio() for s in runner.strategies.values(): s.cfg.max_position_pct = ratio logger.info(f"[GW] 仓位上限调整为: {ratio*100:.0f}%") def _on_risk_alert(cmd): logger.warning(f"[GW] 风控告警: [{cmd.severity}] {cmd.rule}") if cmd.should_force_close(): # 关闭所有策略 for s in runner.strategies.values(): s.cfg.enabled = False # 平掉所有持仓 pos = okx.positions() for p in pos: if float(p.get('pos', 0)) != 0: okx.close_position(p['instId'], p['posSide']) logger.critical(f"[GW] 风控触发强制平仓: {cmd.rule}") def _on_infra_command(cmd): logger.warning(f"[GW] 基础设施指令: {cmd.action}") if cmd.action == "pause": for s in runner.strategies.values(): s.cfg.enabled = False elif cmd.action == "resume": for s in runner.strategies.values(): s.cfg.enabled = True elif cmd.action == "shutdown": runner.stop_background() # 叠加回调（保留原有 attach_gateway 设置的回调） gw.on_strategy_signal = _on_strategy_signal gw.on_risk_alert = _on_risk_alert gw.on_infra_command = _on_infra_command if gw.ping(): logger.info("[统帅部] 交易部已连接网关") else: logger.warning("[统帅部] 网关Ping失败，交易部独立运行") except Exception as e: logger.warning(f"[统帅部] 网关连接跳过: {e}") # ── 数据推送器：交易部 → 情报部非结构化市场数据 ── pusher = DataPusher( gateway_url=gateway_url, internal_token=gateway_token, poll_interval=60.0, # 每60秒推送一次 ) if pusher.ping(): logger.info("[情报部] 数据推送器已就绪，将每60秒推送微观数据") else: logger.warning("[情报部] 网关不可达，数据推送器将跳过推送") # 黄金分析器（按配置启停） gold_analyzer = None if GOLD_TRADING_ENABLED: from strategy.gold_analyzer import GoldAnalyzer gold_analyzer = GoldAnalyzer() logger.info("[Gold] XAU分析器已启用") else: logger.info("[Gold] XAU分析器已关闭") # 价格异动检测器 → 通知战略部 anomaly_detector = PriceAnomalyDetector(okx, INSTRUMENTS) # 止损区识别器 try: from strategy.stop_loss_zones import StopLossZoneDetector stop_zone_detector = StopLossZoneDetector(lookback=100) except ImportError: stop_zone_detector = None # 主动买卖量分析器 try: from exchange.taker_flow import TakerFlowAnalyzer, fetch_and_analyze, format_taker_flow taker_analyzer = TakerFlowAnalyzer(max_history=200) except ImportError: taker_analyzer = None # 因子计算后台引擎 (每60秒全品种全因子) factor_engine = None if FactorEngine: factor_engine = FactorEngine(okx, INSTRUMENTS, deribit, symbol="ETH", interval=60) factor_engine.start() logger.info("[FactorEngine] 后台因子计算已启动") # Telegram 频道监听（Telethon，后台线程） start_channel_listener(TG_API_ID, TG_API_HASH, TG_NEWS_CHANNELS, PROXIES) # 实时图表（后台线程，30秒刷新，启动时自动打开浏览器） t_dash = threading.Thread(target=_start_dashboard, name="Dashboard", daemon=True) t_dash.start() # WS 订单簿 + 幌骗检测（后台线程，每5秒检测一次） inst_id_swap = list(INSTRUMENTS.values())[0].get("inst", "ETH-USDT-SWAP") def _on_spoof_alert(alert): tg_send( f"‼️ 幌骗预警\n" f"{alert['msg']}\n" f"撤单率: {alert['withdraw_pct']}% 耗时: {alert['elapsed']}s\n" f"预期方向: {alert['direction']} 评分: {alert['score']:+.2f}" ) # 有毒订单流检测器初始化 from exchange.toxic_flow import ToxicFlowDetector toxic_detector = ToxicFlowDetector(max_history=100) def _on_toxic_alert(alert): if alert.get("type") == "toxic_flow": tg_send( f"☣️ 有毒订单流告警\n" f"等级: {alert.get('level', '?')} 评分: {alert.get('score', 0):.2f}\n" f"预测: {alert.get('direction', '?')} 置信度: {alert.get('confidence', 0):.1%}\n" f"详情: {alert.get('msg', '')[:200]}" ) ws_book = OKXBookWS(inst_id_swap, on_alert=_on_spoof_alert, proxies=PROXIES if PROXIES else None) ws_book.toxic_flow = toxic_detector # 注入有毒流检测器 ws_book.start() manager.ws_book = ws_book # 注入决策树 manager.toxic_detector = toxic_detector # 注入TradeManager # 初始化余额（带重试） for attempt in range(5): try: avail, equity = okx.balance() break except Exception as e: logger.warning(f"OKX余额获取失败 (attempt {attempt+1}/5): {e}") time.sleep(5) else: logger.critical("OKX余额获取全部重试失败，权益设为0") equity, avail = 0, 0 manager.peak_equity = equity if manager.start_equity == 0: manager.start_equity = equity # 初始化 MySQL if DB_ENABLED: try: from storage.mysql_client import init_pool, insert_event init_pool(DB_HOST, DB_PORT, DB_USER, DB_PASSWORD, DB_NAME) insert_event('INFO', 'system', 'startup', 'Trading system started', {'instruments': list(INSTRUMENTS.keys()), 'leverage': DEFAULT_LEVERAGE, 'equity': equity}) logger.info("[DB] MySQL initialized successfully") except Exception as e: logger.warning(f"[DB] MySQL init failed: {e}") mode_str = "🧪 模拟交易" if PAPER_TRADING else "🔴 实盘交易" tg_send( f"🤖 交易系统启动 ({mode_str})\n" f"标的: {', '.join(INSTRUMENTS.keys())}\n" f"杠杆: {DEFAULT_LEVERAGE}x | 阈值: ±{SCORE_THRESHOLD}\n" f"权益: ${equity:.2f} | 可用: ${avail:.2f}\n" + (f"🧪 模拟账户: ${paper_trader.equity:.2f}\n" if paper_trader else "") + f"间隔: {CHECK_INTERVAL}s\n" f"命令: /s 状态 /a 分析 /c 平仓 /st 策略 /help 帮助" ) # 启动后立即发一次截图 global _last_snapshot_time threading.Thread(target=_try_send_snapshot, daemon=True, name="Snapshot0").start() _last_snapshot_time = time.time() cycle = 0 tg_offset = 0 last_detail_time = 0 # 跳过旧消息 old = tg_get_updates(0) if old: tg_offset = old[-1]["update_id"] + 1 while True: try: # 1. 处理 Telegram 命令 updates = tg_get_updates(tg_offset) for upd in updates: tg_offset = upd["update_id"] + 1 msg = upd.get("message", {}) text = msg.get("text", "") chat_id = str(msg.get("chat", {}).get("id", "")) if chat_id == TG_CHAT and text: reply = handle_command(text, okx, manager, runner) if reply: tg_send(reply) # 2. 模拟账户更新 if paper_trader: paper_trader.update() # ── 原始数据第一时间入库 (write before process) ── if DB_ENABLED: try: from storage.mysql_client import insert_raw_ticker, insert_raw_candle for sym, cfg in INSTRUMENTS.items(): try: tk = okx.ticker(cfg["inst"]) price = float(tk.get("last", 0)) if price > 0: insert_raw_ticker( symbol=sym, price=price, bid=float(tk.get("bidPx", 0) or 0), ask=float(tk.get("askPx", 0) or 0), volume_24h=float(tk.get("vol24h", 0) or 0), ) # 5m K线 candles = okx.candles(cfg["inst"], bar="5m", limit=1) if candles: c = candles[0] insert_raw_candle(sym, "5m", int(c[0]), float(c[1]), float(c[2]), float(c[3]), float(c[4]), float(c[5])) except Exception: pass except Exception: pass # 3. 主分析周期 cycle += 1 logger.info(f"--- Cycle {cycle} ---") result = manager.run_cycle() # ── 推送非结构化数据到情报部 ── pusher.update_result(result) if pusher.should_push(): push_result = pusher.push() if push_result.get("status") == "sent": logger.debug(f"[情报部] 微观数据已推送: {push_result.get('symbols')}") elif push_result.get("status") != "skip": logger.warning(f"[情报部] 数据推送异常: {push_result}") # 3. 推送报告 has_trade = any("trade" in s for s in result.get("signals", [])) now = time.time() if has_trade or now - last_detail_time >= 60: tg_send(format_analysis(result)) last_detail_time = now # XAU 黄金并行分析 (每60s，仅当启用时) if gold_analyzer: try: from strategy.formatter import format_gold_analysis xau_result = gold_analyzer.analyze() if xau_result and xau_result.get("direction", "WAIT") != "WAIT": tg_send(format_gold_analysis(xau_result)) except Exception as e: logger.warning(f"Gold analysis error: {e}") # 每次发报告都附带截图 threading.Thread(target=_try_send_snapshot, daemon=True, name="Snapshot").start() logger.info(f"Cycle {cycle}: equity=${result['equity']}") # ── 价格异动检测 → 通知战略部 ── try: eth_price = float(result.get('latest_price', 0) or result.get('price', 0)) if eth_price > 0: anomaly_detector.feed_price("ETH", eth_price) # 获取成交量 try: candles = okx.candles(INSTRUMENTS["ETH"]["inst"], bar="5m", limit=1) if candles: anomaly_detector.feed_price("ETH", eth_price, float(candles[0][5])) except Exception: pass # 检查异动 anomalies = anomaly_detector.check_all("ETH") for event in anomalies: if event.severity in ("warning", "critical"): tg_send(f"⚠️ 价格异动 [{event.severity.upper()}]\n{event.detail}") # 异动事件入库 if DB_ENABLED: try: from storage.mysql_client import insert_raw_event, insert_flash_event insert_raw_event( event_type=event.type, symbol=event.symbol, severity=event.severity, direction=None, price=event.price, detail=event.detail[:500], metrics=event.metrics, source=f"anomaly_detector({event.metrics.get('data_source', 'okx_rest')})", ) # 同时写入战略部 macro_decision.flash_events insert_flash_event( event_type=event.type, symbol=event.symbol, severity=event.severity, price=event.price, detail=event.detail[:500], metrics=event.metrics, ) except Exception: pass # 通知战略部 if gw: try: from exchange.gateway_hooks import gateway_alert_strategy gateway_alert_strategy( alert_type=event.type, symbol=event.symbol, severity=event.severity, detail=event.detail, metrics=event.metrics, ) anomaly_detector.mark_relayed(event) logger.info(f"[异动] 已通知战略部: {event.type} {event.detail[:80]}") except Exception: pass except Exception as e: logger.debug(f"[异动] 检测异常: {e}") # ── 主动买卖量更新 (每周期) ── if taker_analyzer: taker_analyzer.feed_raw(okx.taker_volume_ratio("ETH", "CONTRACTS", "5m", 12)) # ── 止损区狩猎预警 (每5周期) ── if cycle % 5 == 0 and stop_zone_detector: try: inst = INSTRUMENTS.get("ETH", {}).get("inst", "ETH-USDT-SWAP") candles = okx.candles(inst, bar="5m", limit=100) if candles and len(candles) >= 20: closes = [float(c[4]) for c in reversed(candles)] highs = [float(c[2]) for c in reversed(candles)] lows = [float(c[3]) for c in reversed(candles)] vols = [float(c[5]) for c in reversed(candles)] sz = stop_zone_detector.analyze(closes, highs, lows, vols, closes[-1], "BTC") if sz.hunt_risk == "high": tg_send(f"🛑 止损狩猎预警\n" f"风险: {sz.hunt_risk} ({sz.hunt_risk_score:.0%})\n" f"{sz.recommendation[:120]}") except Exception: pass # ── 主动买卖量极端预警 (每3周期) ── if cycle % 3 == 0: try: flow = taker_analyzer.analyze(lookback=12) if taker_analyzer else None if flow and flow.flow_bias != "neutral" and abs(flow.delta_pct) > 0.3: tg_send(format_taker_flow(flow)) except Exception: pass # ── 每周期向统帅部汇报状态 ── if gw and cycle % 3 == 0: try: from exchange.gateway_hooks import gateway_send_status if PAPER_TRADING and paper_trader: ps = paper_trader.get_summary() gateway_send_status( equity=ps['equity'], pnl_24h=ps['realized_pnl'], positions=[{ 'inst': f"{s}.{side}", 'side': side, 'qty': p.size, 'upl': 0, } for s, sides in paper_trader.positions.items() for side, p in sides.items() if p and p.size > 0], score=total_score, strategy_name='paper_trading', ) else: positions = okx.positions() active_pos = [p for p in positions if float(p.get('pos', 0)) != 0] pnl = sum(float(p.get('upl', 0)) for p in active_pos) gateway_send_status( equity=float(result['equity']), pnl_24h=pnl, positions=[{ 'inst': p['instId'], 'side': p['posSide'], 'qty': float(p['pos']), 'upl': float(p.get('upl', 0)), } for p in active_pos], score=total_score, strategy_name='hunting_reversal', ) except Exception: pass except Exception as e: logger.error(f"Cycle error: {e}", exc_info=True) if cycle % 60 == 0: tg_send(f"❌ 系统错误: {str(e)[:200]}") # 向网关报告异常 if gw: try: from exchange.gateway_hooks import gateway_send_pnl gateway_send_pnl(pnl=0, pnl_pct=0, strategy_name='error') except Exception: pass # 分段 sleep，中间插入快讯监控 & TG 轮询（15 秒一段） # 幌骗检测由 ws_book 后台线程每 5s 自动运行，无需主循环干预 for seg in range(CHECK_INTERVAL // 15): time.sleep(15) try: analyzer.check_flash_news() except: pass try: updates = tg_get_updates(tg_offset) for u in updates: tg_offset = u["update_id"] + 1 txt = u.get("message", {}).get("text", "").strip() if txt.startswith("/") or txt in ("状态", "分析", "平仓", "帮助", "新闻"): reply = handle_command(txt, okx, manager, runner) if reply: tg_send(reply) except: pass if __name__ == "__main__": main()