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

feat: use rich.Live(screen=False) for flicker-free dashboard (v1.3.9)

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

91
src/strategies/ping_pong_bot.py
View File

@ -15,6 +15,7 @@ from rich.console import Console, Group
from rich.table import Table from rich.table import Table
from rich.panel import Panel from rich.panel import Panel
from rich.layout import Layout from rich.layout import Layout
from rich.live import Live
from rich import box from rich import box
import asyncpg import asyncpg
@ -40,7 +41,6 @@ logging.basicConfig(
logger = logging.getLogger("PingPongBot") logger = logging.getLogger("PingPongBot")
class DatabaseManager: class DatabaseManager:
"""Minimal Database Manager for the bot"""
def __init__(self): def __init__(self):
self.host = os.getenv('DB_HOST', '20.20.20.20') self.host = os.getenv('DB_HOST', '20.20.20.20')
self.port = int(os.getenv('DB_PORT', 5433)) self.port = int(os.getenv('DB_PORT', 5433))
@ -68,7 +68,7 @@ class DatabaseManager:
class PingPongBot: class PingPongBot:
def __init__(self, config_path="config/ping_pong_config.yaml"): def __init__(self, config_path="config/ping_pong_config.yaml"):
self.version = "1.3.8" self.version = "1.3.9"
with open(config_path, 'r') as f: with open(config_path, 'r') as f:
self.config = yaml.safe_load(f) self.config = yaml.safe_load(f)
@ -78,19 +78,19 @@ class PingPongBot:
if not self.api_key or not self.api_secret: if not self.api_key or not self.api_secret:
raise ValueError("API_KEY and API_SECRET must be set in .env file") raise ValueError("API_KEY and API_SECRET must be set in .env file")
# Corrected timeout parameter for pybit V5 HTTP
self.session = HTTP( self.session = HTTP(
testnet=False, testnet=False,
api_key=self.api_key, api_key=self.api_key,
api_secret=self.api_secret api_secret=self.api_secret,
timeout=10
) )
self.db = DatabaseManager() self.db = DatabaseManager()
self.symbol = self.config['symbol'].upper() self.symbol = self.config['symbol'].upper()
self.db_symbol = self.symbol.replace("USDT", "") self.db_symbol = self.symbol.replace("USDT", "")
self.interval = str(self.config['interval']) self.interval = str(self.config['interval'])
self.db_interval = self.interval + "m" if self.interval.isdigit() else self.interval self.db_interval = self.interval + "m" if self.interval.isdigit() else self.interval
self.direction = self.config['direction'].lower() self.direction = self.config['direction'].lower()
# State # State
@ -168,11 +168,11 @@ class PingPongBot:
wallet = await asyncio.to_thread(self.session.get_wallet_balance, category="linear", accountType="UNIFIED", coin="USDT") wallet = await asyncio.to_thread(self.session.get_wallet_balance, category="linear", accountType="UNIFIED", coin="USDT")
if wallet['retCode'] == 0: if wallet['retCode'] == 0:
result_list = wallet['result']['list'] res_list = wallet['result']['list']
if result_list: if res_list:
self.wallet_balance = float(result_list[0].get('totalWalletBalance', 0)) self.wallet_balance = float(res_list[0].get('totalWalletBalance', 0))
if self.wallet_balance == 0: if self.wallet_balance == 0:
coin_info = result_list[0].get('coin', []) coin_info = res_list[0].get('coin', [])
if coin_info: if coin_info:
self.wallet_balance = float(coin_info[0].get('walletBalance', 0)) self.wallet_balance = float(coin_info[0].get('walletBalance', 0))
except Exception as e: except Exception as e:
@ -231,50 +231,42 @@ class PingPongBot:
except Exception as e: except Exception as e:
logger.error(f"Trade Error: {e}") logger.error(f"Trade Error: {e}")
def get_dashboard_content(self): def get_dashboard(self):
"""Prepares the dashboard content as a Group of tables""" """Generates a single consolidated Panel for the dashboard"""
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") now = datetime.now().strftime("%H:%M:%S")
# 1. Header # 1. Info Table
header = Table(title=f"PING-PONG BOT v{self.version} [{self.direction.upper()}]", box=box.ROUNDED, expand=True) info = Table(box=None, expand=True, show_header=False)
header.add_column("Property"); header.add_column("Value") info.add_column("K", style="cyan"); info.add_column("V", style="white")
header.add_row("Symbol", self.symbol) info.add_row("Symbol", f"{self.symbol} [{self.direction.upper()}]")
header.add_row("Market Price", f"${self.market_price:,.2f}") info.add_row("Price", f"${self.market_price:,.2f}")
header.add_row("Last Candle", f"{self.last_candle_time} (@${self.last_candle_price:,.2f})") info.add_row("Candle", f"{self.last_candle_time} (@${self.last_candle_price:,.2f})")
# 2. Indicators # 2. Indicators Table
inds = Table(title="INDICATORS", box=box.ROUNDED, expand=True) inds = Table(box=box.SIMPLE, expand=True)
inds.add_column("Indicator"); inds.add_column("Value"); inds.add_column("Updated") inds.add_column("Indicator", style="dim"); inds.add_column("Value", justify="right"); inds.add_column("Updated", justify="center")
inds.add_row("Hurst Upper", f"{self.current_indicators['hurst_upper']['value']:.2f}", self.current_indicators['hurst_upper']['timestamp']) inds.add_row("HURST UPPER", f"{self.current_indicators['hurst_upper']['value']:.2f}", self.current_indicators['hurst_upper']['timestamp'])
inds.add_row("Hurst Lower", f"{self.current_indicators['hurst_lower']['value']:.2f}", self.current_indicators['hurst_lower']['timestamp']) inds.add_row("HURST LOWER", f"{self.current_indicators['hurst_lower']['value']:.2f}", self.current_indicators['hurst_lower']['timestamp'])
inds.add_row("RSI", f"{self.current_indicators['rsi']['value']:.2f}", self.current_indicators['rsi']['timestamp']) inds.add_row("RSI", f"{self.current_indicators['rsi']['value']:.2f}", self.current_indicators['rsi']['timestamp'])
# 3. Position # 3. Position Table
pos = Table(title="PORTFOLIO & STATUS", box=box.ROUNDED, expand=True) pos = Table(box=box.SIMPLE, expand=True)
pos.add_column("Property"); pos.add_column("Value") pos.add_column("Wallet", justify="center"); pos.add_column("Size", justify="center"); pos.add_column("PnL", justify="center")
pos.add_row("Wallet Balance", f"${self.wallet_balance:,.2f}")
if self.position: if self.position:
pnl = float(self.position['unrealisedPnl']) pnl = float(self.position['unrealisedPnl'])
pos.add_row("Position Size", str(self.position['size'])) pos.add_row(f"${self.wallet_balance:,.2f}", str(self.position['size']), f"[bold {'green' if pnl>=0 else 'red'}]${pnl:,.2f}")
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: else:
pos.add_row("Position", "NONE") pos.add_row(f"${self.wallet_balance:,.2f}", "0", "-")
pos.add_row("Status", f"[bold blue]{self.status_msg}[/]") content = Group(info, inds, pos, f"[dim]Status: {self.status_msg} | Signal: {self.last_signal or 'None'} | Heartbeat: {now}[/]")
pos.add_row("Last Signal", str(self.last_signal or "None")) return Panel(content, title=f"[bold cyan]PING-PONG BOT v{self.version}[/]", border_style="blue")
pos.add_row("Heartbeat", f"[italic]{now}[/]")
return Group(header, inds, pos)
async def run(self): async def run(self):
logger.info("Bot starting...") logger.info(f"Bot v{self.version} starting...")
await self.db.connect() await self.db.connect()
last_exchange_update = 0 last_exchange_update = 0
# Clear screen once at start with Live(self.get_dashboard(), console=self.console, refresh_per_second=1, screen=False) as live:
self.console.clear()
while True: while True:
try: try:
now_ts = time.time() now_ts = time.time()
@ -294,25 +286,22 @@ class PingPongBot:
df = self.calculate_indicators(df) df = self.calculate_indicators(df)
signal = self.check_signals(df) signal = self.check_signals(df)
if signal: if signal:
logger.info(f"Signal detected: {signal}") logger.info(f"SIGNAL: {signal}")
await self.execute_trade(signal) await self.execute_trade(signal)
self.last_candle_time = latest['time'] self.last_candle_time = latest['time']
self.last_candle_price = latest['close'] self.last_candle_price = latest['close']
self.status_msg = f"New Candle: {latest['time'].strftime('%H:%M:%S')}" self.status_msg = f"Last Candle: {latest['time'].strftime('%H:%M:%S')}"
else: else:
self.status_msg = f"No candles found for {self.db_symbol}/{self.db_interval}" self.status_msg = "Waiting for candles..."
# 3. Print Dashboard - Overwrite using ANSI # 3. Update Dashboard
# Moving cursor to top-left (0,0) and printing the group live.update(self.get_dashboard())
# This is much smoother than clear()
self.console.print("\033[H", end="")
self.console.print(self.get_dashboard_content())
except asyncio.TimeoutError: except asyncio.TimeoutError:
self.status_msg = "Exchange Sync Timeout" self.status_msg = "Exchange Timeout"
except Exception as e: except Exception as e:
logger.exception(f"Loop Error: {e}") logger.error(f"Loop error: {e}")
self.status_msg = f"Error: {str(e)[:50]}" self.status_msg = f"Error: {str(e)[:40]}"
await asyncio.sleep(5) await asyncio.sleep(5)