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

feat: add real-time status dashboard using rich tables

Browse Source
This commit is contained in:
Gemini CLI
2026-03-05 21:10:22 +01:00
parent ccc36da1e2
commit 9c3fa908ce
1 changed files with 76 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
src/strategies/ping_pong_bot.py
View File

@ -10,8 +10,13 @@ import pandas as pd
import numpy as np
from datetime import datetime, timezone
from dotenv import load_dotenv
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.layout import Layout
from rich import box
# Try to import pybit, if not available, we'll suggest installing it
# Try to import pybit
try:
from pybit.unified_trading import HTTP
except ImportError:
@ -55,12 +60,17 @@ class PingPongBot:
# State
self.last_candle_time = None
self.current_indicators = {}
self.current_indicators = {
"rsi": {"value": 0.0, "timestamp": "N/A"},
"hurst_lower": {"value": 0.0, "timestamp": "N/A"},
"hurst_upper": {"value": 0.0, "timestamp": "N/A"}
}
self.position = None
self.wallet_balance = 0
self.status_msg = "Initializing..."
self.last_signal = None
self.start_time = datetime.now()
self.console = Console()
# Grid parameters from config
self.tp_pct = self.config['take_profit_pct'] / 100.0
@ -97,27 +107,80 @@ class PingPongBot:
mcl = mcl_t / 2
mcl_2 = int(round(mcl / 2))
# True Range
df['h_l'] = df['high'] - df['low']
df['h_pc'] = abs(df['high'] - df['close'].shift(1))
df['l_pc'] = abs(df['low'] - df['close'].shift(1))
df['tr'] = df[['h_l', 'h_pc', 'l_pc']].max(axis=1)
# RMA of Close and ATR
df['ma_mcl'] = self.rma(df['close'], mcl)
df['atr_mcl'] = self.rma(df['tr'], mcl)
# Historical Offset
df['center'] = df['ma_mcl'].shift(mcl_2)
# Fill first values where shift produces NaN with the MA itself (as done in JS: historical_ma || src)
df['center'] = df['center'].fillna(df['ma_mcl'])
mcm_off = mcm * df['atr_mcl']
df['hurst_upper'] = df['center'] + mcm_off
df['hurst_lower'] = df['center'] - mcm_off
# Update current indicator state
last_row = df.iloc[-1]
now_str = datetime.now().strftime("%H:%M:%S")
self.current_indicators["rsi"] = {"value": last_row['rsi'], "timestamp": now_str}
self.current_indicators["hurst_lower"] = {"value": last_row['hurst_lower'], "timestamp": now_str}
self.current_indicators["hurst_upper"] = {"value": last_row['hurst_upper'], "timestamp": now_str}
return df
def render_dashboard(self):
"""Render a clean summary of the bot state"""
# We don't clear to avoid flickering in some terminals, just print a separator or use rich.Live if needed
# But for simple docker logs, clear is usually fine or just printing the table.
# 1. Config Table
cfg_table = Table(title="[bold cyan]PING-PONG BOT CONFIG[/]", box=box.ROUNDED, expand=True)
cfg_table.add_column("Property", style="dim")
cfg_table.add_column("Value", style="bold white")
cfg_table.add_row("Symbol", f"{self.symbol} ({self.interval}m)")
cfg_table.add_row("Direction", f"{self.direction.upper()}")
cfg_table.add_row("Capital/Margin", f"${self.pos_size_margin} (Lev: {self.leverage}x)")
cfg_table.add_row("Partial Exit", f"{self.partial_exit_pct*100}%")
cfg_table.add_row("TP Target", f"{self.tp_pct*100}%")
# 2. Indicators Table
ind_table = Table(title="[bold yellow]TECHNICAL INDICATORS[/]", box=box.ROUNDED, expand=True)
ind_table.add_column("Indicator", style="dim")
ind_table.add_column("Current Value", justify="right")
ind_table.add_column("Last Update", justify="center")
ind_table.add_row("RSI", f"{self.current_indicators['rsi']['value']:.2f}", self.current_indicators['rsi']['timestamp'])
ind_table.add_row("Hurst Lower", f"{self.current_indicators['hurst_lower']['value']:.2f}", self.current_indicators['hurst_lower']['timestamp'])
ind_table.add_row("Hurst Upper", f"{self.current_indicators['hurst_upper']['value']:.2f}", self.current_indicators['hurst_upper']['timestamp'])
# 3. Position Table
pos_table = Table(title="[bold green]ACTIVE POSITION & ACCOUNT[/]", box=box.ROUNDED, expand=True)
pos_table.add_column("Property", style="dim")
pos_table.add_column("Value", style="bold white")
pos_table.add_row("Wallet Balance", f"${self.wallet_balance:.2f}")
if self.position:
p = self.position
pnl = float(p.get('unrealisedPnl', 0))
pnl_color = "green" if pnl >= 0 else "red"
pos_table.add_row("Position Size", f"{p['size']}")
pos_table.add_row("Entry Price", f"{p['avgPrice']}")
pos_table.add_row("Unrealized PnL", f"[{pnl_color}]${pnl:.2f}[/]")
pos_table.add_row("Liquidation", f"{p.get('liqPrice', 'N/A')}")
else:
pos_table.add_row("Position", "NONE (Scanning...)")
pos_table.add_row("Last Signal", str(self.last_signal or "None"))
pos_table.add_row("Status", f"[bold blue]{self.status_msg}[/]")
self.console.print("\n" + "="*50)
self.console.print(cfg_table)
self.console.print(ind_table)
self.console.print(pos_table)
self.console.print(f"[dim]Uptime: {str(datetime.now() - self.start_time).split('.')[0]} | Last Run: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}[/]\n")
async def fetch_data(self):
"""Fetch latest Klines from Bybit V5"""
try:
@ -149,7 +212,7 @@ class PingPongBot:
async def update_account_info(self):
"""Update position and balance information"""
try:
# Get Position
# Get Position (with settleCoin for Hedge Mode compatibility)
pos_response = self.session.get_positions(
category="linear",
symbol=self.symbol,
@ -333,22 +396,16 @@ class PingPongBot:
df = await self.fetch_data()
if df is not None:
# 3. Check for New Candle (for signal processing)
last_price = float(df.iloc[-1]['close'])
# 4. Strategy Logic
# 3. Strategy Logic
signal = self.check_signals(df)
if signal:
logger.info(f"Signal detected: {signal} @ {last_price}")
logger.info(f"Signal detected: {signal} @ {df.iloc[-1]['close']}")
await self.execute_trade_logic(df, signal)
# 5. Simple status log
if self.position:
logger.info(f"Price: {last_price:.2f} | Pos: {self.position['size']} @ {self.position['avgPrice']} | Wallet: {self.wallet_balance:.2f}")
else:
logger.info(f"Price: {last_price:.2f} | No Position | Wallet: {self.wallet_balance:.2f}")
await asyncio.sleep(self.config.get('loop_interval_seconds', 5))
# 4. Render Summary Dashboard
self.render_dashboard()
await asyncio.sleep(self.config.get('loop_interval_seconds', 10))
if __name__ == "__main__":
try:
@ -357,5 +414,4 @@ if __name__ == "__main__":
except KeyboardInterrupt:
print("\nBot Stopped by User")
except Exception as e:
print(f"\nCritical Error: {e}")
logger.exception("Critical Error in main loop")