p1otek/btc-trading
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: use ANSI cursor reset for flicker-free dashboard (v1.3.8)

Browse Source
This commit is contained in:
Gemini CLI
2026-03-05 22:39:48 +01:00
parent ca0e74d402
commit 08a4639dad
1 changed files with 17 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
src/strategies/ping_pong_bot.py
View File

@ -11,7 +11,7 @@ import numpy as np
from datetime import datetime, timezone
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from rich.console import Console
from rich.console import Console, Group
from rich.table import Table
from rich.panel import Panel
from rich.layout import Layout
@ -68,7 +68,7 @@ class DatabaseManager:
class PingPongBot:
def __init__(self, config_path="config/ping_pong_config.yaml"):
self.version = "1.3.7"
self.version = "1.3.8"
with open(config_path, 'r') as f:
self.config = yaml.safe_load(f)
@ -107,7 +107,6 @@ class PingPongBot:
self.status_msg = "Initializing..."
self.last_signal = None
self.start_time = datetime.now()
self.last_heartbeat = datetime.now()
self.console = Console()
# Parameters
@ -157,9 +156,7 @@ class PingPongBot:
return df
async def update_exchange_data(self):
"""Fetch Price, Balance, Position every 15s with timeout"""
try:
# Wrap synchronous pybit calls in asyncio.to_thread for better async behavior
ticker = await asyncio.to_thread(self.session.get_tickers, category="linear", symbol=self.symbol)
if ticker['retCode'] == 0:
self.market_price = float(ticker['result']['list'][0]['lastPrice'])
@ -234,17 +231,15 @@ class PingPongBot:
except Exception as e:
logger.error(f"Trade Error: {e}")
def print_dashboard(self):
"""Prints a clean summary to the logs"""
now = datetime.now().strftime("%H:%M:%S")
self.last_heartbeat = datetime.now()
def get_dashboard_content(self):
"""Prepares the dashboard content as a Group of tables"""
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 1. Header
header = Table(title=f"PING-PONG BOT v{self.version} Dashboard", box=box.ROUNDED, expand=True)
header = Table(title=f"PING-PONG BOT v{self.version} [{self.direction.upper()}]", box=box.ROUNDED, expand=True)
header.add_column("Property"); header.add_column("Value")
header.add_row("Symbol", self.symbol)
header.add_row("Market Price", f"${self.market_price:,.2f}")
header.add_row("Direction", self.direction.upper())
header.add_row("Last Candle", f"{self.last_candle_time} (@${self.last_candle_price:,.2f})")
# 2. Indicators
@ -264,31 +259,28 @@ class PingPongBot:
pos.add_row("Entry Price", f"${float(self.position['avgPrice']):,.2f}")
pos.add_row("Unrealized PnL", f"[bold {'green' if pnl>=0 else 'red'}]${pnl:,.2f}")
else:
pos_row = "NONE"
pos.add_row("Position", "NONE")
pos.add_row("Status", f"[bold blue]{self.status_msg}[/]")
pos.add_row("Last Signal", str(self.last_signal or "None"))
pos.add_row("Heartbeat", f"[italic]{now}[/]")
self.console.print("\n")
self.console.print(header)
self.console.print(inds)
self.console.print(pos)
self.console.print("-" * 50)
return Group(header, inds, pos)
async def run(self):
logger.info("Bot starting...")
await self.db.connect()
last_exchange_update = 0
# Clear screen once at start
self.console.clear()
while True:
try:
now_ts = time.time()
# 1. Exchange Sync (15s)
if now_ts - last_exchange_update >= 15:
# Wrapped in timeout to prevent hanging
await asyncio.wait_for(self.update_exchange_data(), timeout=10)
last_exchange_update = now_ts
@ -306,16 +298,18 @@ class PingPongBot:
await self.execute_trade(signal)
self.last_candle_time = latest['time']
self.last_candle_price = latest['close']
self.status_msg = f"New Candle: {latest['time']}"
self.status_msg = f"New Candle: {latest['time'].strftime('%H:%M:%S')}"
else:
self.status_msg = f"No candles found for {self.db_symbol}/{self.db_interval}"
# 3. Print Dashboard
self.print_dashboard()
# 3. Print Dashboard - Overwrite using ANSI
# Moving cursor to top-left (0,0) and printing the group
# This is much smoother than clear()
self.console.print("\033[H", end="")
self.console.print(self.get_dashboard_content())
except asyncio.TimeoutError:
logger.error("Exchange update timed out")
self.status_msg = "Exchange Update Timeout"
self.status_msg = "Exchange Sync Timeout"
except Exception as e:
logger.exception(f"Loop Error: {e}")
self.status_msg = f"Error: {str(e)[:50]}"