From 3b4b5d3a585edce8dfd0ad9e8805d2fbc277072c Mon Sep 17 00:00:00 2001 From: ddrwode <34234@3来 34> Date: Fri, 6 Feb 2026 16:22:10 +0800 Subject: [PATCH] =?UTF-8?q?=E6=97=A5=E5=BF=97=E5=B1=95=E7=A4=BA=E4=BC=98?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../四分之一，五分钟，反手条件充足修改版.py | 150 +++++++++++++++--- test.py | 77 ++++++++- 2 files changed, 204 insertions(+), 23 deletions(-) diff --git a/bitmart/四分之一，五分钟，反手条件充足修改版.py b/bitmart/四分之一，五分钟，反手条件充足修改版.py index 879eef4..a7e570c 100644 --- a/bitmart/四分之一，五分钟，反手条件充足修改版.py +++ b/bitmart/四分之一，五分钟，反手条件充足修改版.py @@ -1,27 +1,34 @@ import sys import time +from datetime import datetime from tqdm import tqdm from loguru import logger from bit_tools import openBrowser from DrissionPage import ChromiumPage from DrissionPage import ChromiumOptions - from bitmart.api_contract import APIContract -# Claude 风格控制台：左侧竖线 + 仅消息（无时间戳），颜色由下方 LOG_* 控制 -logger.remove() -logger.add(sys.stderr, format="\033[2m│\033[0m {message}", colorize=False) +from rich.console import Console +from rich.panel import Panel +from rich.table import Table +from rich.layout import Layout +from rich.align import Align +from rich.text import Text +from rich.live import Live -# 颜色标签 + 分块边框 -_R = "\033[0m" # reset -_B = "\033[1m" # bold -_D = "\033[2m" # dim -_C = "\033[36m" # cyan - 价格/数据 -_Y = "\033[33m" # yellow - 信号 -_G = "\033[32m" # green - 仓位/操作 -_M = "\033[90m" # gray - 系统 -_W = "\033[97m" # white - 高亮 +# 是否使用 Rich 仪表盘（否则用 loguru 普通输出） +USE_RICH_DASHBOARD = True +console = Console() + +# 颜色标签（loguru 用，无 Rich 时） +_R = "\033[0m" +_B = "\033[1m" +_C = "\033[36m" +_Y = "\033[33m" +_G = "\033[32m" +_M = "\033[90m" +_W = "\033[97m" def _tag(t: str, color: str) -> str: return color + "[" + t + "]" + _R + " " @@ -31,8 +38,12 @@ LOG_SIGNAL = _tag("信号", _Y) LOG_POSITION = _tag("仓位", _G) LOG_SYSTEM = _tag("系统", _M) +if not USE_RICH_DASHBOARD: + logger.remove() + logger.add(sys.stderr, format="\033[2m│\033[0m {message}", colorize=False) + def log_kline_header(kline_id): - """新 K 线分块头（Claude 风格面板）""" + """新 K 线分块头（仅非 Rich 模式使用）""" s = f" K 线 {kline_id} " width = max(52, len(s) + 4) line = "─" * (width - 2) @@ -43,6 +54,54 @@ def log_kline_header(kline_id): logger.info(_M + "╰" + line + "╯" + _R) +# ---------- Rich 仪表盘 ---------- +def make_header() -> Panel: + title = Text("四分之一策略 · ETHUSDT 5m", style="bold cyan") + subtitle = Text(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), style="dim") + return Panel(Align.center(title + Text("\n") + subtitle), border_style="cyan") + +def make_metrics_panel(state: dict) -> Panel: + t = Table(show_header=True, header_style="bold magenta", expand=True) + t.add_column("指标", style="cyan") + t.add_column("数值", justify="right", style="green") + t.add_row("当前价", f"{state.get('price', 0):.2f}") + pos_map = {0: "无", 1: "多", -1: "空"} + t.add_row("持仓", pos_map.get(state.get("position", 0), "?")) + t.add_row("K线 ID", str(state.get("kline_id", "-"))) + t.add_row("做多触发", f"{state.get('long_trigger', 0):.2f}") + t.add_row("做空触发", f"{state.get('short_trigger', 0):.2f}") + t.add_row("突破做多", f"{state.get('long_breakout', 0):.2f}") + t.add_row("突破做空", f"{state.get('short_breakout', 0):.2f}") + ema10 = state.get("ema10") + atr14 = state.get("atr14") + t.add_row("EMA10", f"{ema10:.2f}" if ema10 is not None else "-") + t.add_row("ATR14", f"{atr14:.2f}" if atr14 is not None else "-") + pnl = state.get("unrealized_pnl") + t.add_row("未实现盈亏", f"{pnl:.2f}$" if pnl is not None else "-") + return Panel(t, title="[bold]价格 / 指标[/bold]", border_style="magenta") + +def make_logs_panel(logs: list) -> Panel: + body = "\n".join(logs[-12:]) if logs else "等待日志..." + text = Text.from_ansi(body) if body else Text("等待日志...", style="dim") + return Panel(text, title="[bold]状态 / 日志[/bold]", border_style="green") + +def make_footer(msg: str = "Ctrl+C 退出") -> Panel: + return Panel(Align.center(Text(msg, style="bold yellow")), border_style="yellow") + +def build_dashboard_layout(state: dict, logs: list) -> Layout: + layout = Layout() + layout.split_column( + Layout(make_header(), name="header", size=5), + Layout(name="body", ratio=1), + Layout(make_footer(), name="footer", size=3), + ) + layout["body"].split_row( + Layout(make_metrics_panel(state), name="left", ratio=1), + Layout(make_logs_panel(logs), name="right", ratio=2), + ) + return layout + + class BitmartFuturesTransaction: def __init__(self, bit_id): @@ -101,10 +160,14 @@ class BitmartFuturesTransaction: self.default_order_size = 25 # 开仓/反手张数，统一在此修改 # 策略相关变量 - self.prev_kline = None # 上一根K线 - self.current_kline = None # 当前K线 - self.prev_entity = None # 上一根K线实体大小 - self.current_open = None # 当前K线开盘价 + self.prev_kline = None + self.current_kline = None + self.prev_entity = None + self.current_open = None + # Rich 仪表盘：状态与日志（供 build_dashboard_layout 使用） + self._display_state = {} + self._display_logs = [] + self._display_triggers = {} def get_klines(self): """获取最近2根K线（当前K线和上一根K线）""" @@ -453,6 +516,8 @@ class BitmartFuturesTransaction: if skip_long_by_lower_third: logger.info(LOG_PRICE + "上一根阳+当前阴(做空形态)，不按下1/4做多") + self._display_triggers = {"long_trigger": long_trigger, "short_trigger": short_trigger, "long_breakout": long_breakout, "short_breakout": short_breakout} + # 无持仓时检查开仓信号 if self.start == 0: if current_price >= long_breakout and not skip_long_by_lower_third: @@ -667,11 +732,29 @@ class BitmartFuturesTransaction: return page_start = True + if USE_RICH_DASHBOARD: + self._display_logs = [] + logger.remove() + def _sink(msg): + self._display_logs.append(msg.record["message"]) + if len(self._display_logs) > 50: + self._display_logs.pop(0) + logger.add(_sink, format="{message}") - while True: + live = None + if USE_RICH_DASHBOARD: + live = Live(console=console, refresh_per_second=8, screen=True) + live.start() + try: + while True: + if live is not None: + try: + live.update(build_dashboard_layout(self._display_state, self._display_logs)) + except Exception: + pass - if page_start: - # 打开浏览器 + if page_start: + # 打开浏览器 for i in range(5): if self.openBrowser(): logger.info(LOG_SYSTEM + "浏览器打开成功") @@ -718,6 +801,21 @@ class BitmartFuturesTransaction: logger.debug(f"当前持仓状态: {self.start} (0=无, 1=多, -1=空)") + # 更新仪表盘左侧数据（供 Rich 展示） + try: + self._display_state["price"] = current_price + self._display_state["position"] = self.start + self._display_state["kline_id"] = current_kline_time + self._display_state["unrealized_pnl"] = self.get_unrealized_pnl_usd() + kline_series = self.get_klines_series(35) + ema10, ema20, atr14 = self.get_ema_atr_for_exit(kline_series) + self._display_state["ema10"] = ema10 + self._display_state["atr14"] = atr14 + if getattr(self, "_display_triggers", None): + self._display_state.update(self._display_triggers) + except Exception: + pass + # 3.5 止损/止盈/移动止损 + EMA/ATR 平仓 + 当前K线从极值回落平仓 if self.start != 0: # 当前K线从最高/最低点回落平仓：换线重置跟踪，有持仓时更新本K线内最高/最低价并检查 @@ -877,6 +975,16 @@ class BitmartFuturesTransaction: except Exception as e: logger.error(f"主循环异常: {e}") time.sleep(5) + finally: + if live is not None: + try: + live.stop() + except Exception: + pass + + if USE_RICH_DASHBOARD: + logger.remove() + logger.add(sys.stderr, format="{message}") if __name__ == '__main__': diff --git a/test.py b/test.py index 276a930..c54cd3a 100644 --- a/test.py +++ b/test.py @@ -1,3 +1,76 @@ +from time import sleep +from datetime import datetime +import random -from pathlib import Path -print(Path(__file__).parent) \ No newline at end of file +from rich.console import Console +from rich.panel import Panel +from rich.table import Table +from rich.layout import Layout +from rich.align import Align +from rich.text import Text +from rich.live import Live + +console = Console() + +def make_header() -> Panel: + title = Text("SYSTEM DASHBOARD", style="bold cyan") + subtitle = Text(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), style="dim") + return Panel(Align.center(title + "\n" + subtitle), border_style="cyan") + +def make_metrics_table(cpu: int, mem: int, qps: int) -> Panel: + t = Table(show_header=True, header_style="bold magenta", expand=True) + t.add_column("Metric") + t.add_column("Value", justify="right") + t.add_row("CPU", f"{cpu}%") + t.add_row("Memory", f"{mem}%") + t.add_row("QPS", str(qps)) + return Panel(t, title="Metrics", border_style="magenta") + +def make_status_panel(status: str, logs: list[str]) -> Panel: + body = "\n".join(logs[-8:]) if logs else "No logs yet." + text = Text(f"[bold]Status:[/bold] {status}\n\n", style="white") + text.append(body, style="dim") + return Panel(text, title="Status / Logs", border_style="green") + +def make_footer(msg: str) -> Panel: + return Panel(Text(msg, style="bold yellow"), border_style="yellow") + +def build_layout(cpu: int, mem: int, qps: int, status: str, logs: list[str]) -> Layout: + layout = Layout() + layout.split_column( + Layout(make_header(), name="header", size=5), + Layout(name="body", ratio=1), + Layout(make_footer("Press Ctrl+C to exit"), name="footer", size=3), + ) + layout["body"].split_row( + Layout(make_metrics_table(cpu, mem, qps), name="left", ratio=1), + Layout(make_status_panel(status, logs), name="right", ratio=2), + ) + return layout + +def main(): + logs = [] + status = "OK" + with Live(console=console, refresh_per_second=8, screen=True): + while True: + cpu = random.randint(1, 100) + mem = random.randint(1, 100) + qps = random.randint(50, 5000) + + if cpu > 85 or mem > 90: + status = "WARN" + logs.append(f"{datetime.now().strftime('%H:%M:%S')} - High load detected") + else: + status = "OK" + if random.random() < 0.2: + logs.append(f"{datetime.now().strftime('%H:%M:%S')} - Heartbeat") + + layout = build_layout(cpu, mem, qps, status, logs) + Live.get_renderable = lambda self: layout # 小技巧：避免重复创建 Live + sleep(0.15) + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + console.print("\nBye!", style="bold red")