ddrwode/lm_code
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

日志展示优化

Browse Source
This commit is contained in:
ddrwode
2026-02-06 16:22:10 +08:00
parent eae69f7c8a
commit 3b4b5d3a58
2 changed files with 204 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
bitmart/四分之一，五分钟，反手条件充足修改版.py
View File

@@ -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__':