348 lines
13 KiB
Python
348 lines
13 KiB
Python
"""
|
|
Dashboard rendering module using rich.
|
|
Provides DashboardRenderer for building rich terminal tables and layouts.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
try:
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
from rich.live import Live
|
|
from rich.layout import Layout
|
|
from rich.text import Text
|
|
from rich.padding import Padding
|
|
RICH_AVAILABLE = True
|
|
except ImportError:
|
|
RICH_AVAILABLE = False
|
|
|
|
|
|
class DashboardRenderer:
|
|
"""Encapsulates all rich-based dashboard rendering logic."""
|
|
|
|
def __init__(self, console=None, table_visibility=None):
|
|
if not RICH_AVAILABLE:
|
|
raise ImportError("rich is not available. Install with: pip install rich")
|
|
self.console = console or Console()
|
|
self.previous_prices = {}
|
|
self.table_visibility = table_visibility or {
|
|
"market": True,
|
|
"strategies": False,
|
|
"indicators": True,
|
|
"balances": True,
|
|
}
|
|
|
|
def toggle_table(self, table_name, enabled=None):
|
|
"""Toggle a table's visibility on the dashboard.
|
|
|
|
Args:
|
|
table_name: The key of the table to toggle (e.g. "market", "strategies").
|
|
enabled: If None, flips the current state. Otherwise sets to the given value.
|
|
|
|
Returns:
|
|
The new visibility state for the table.
|
|
"""
|
|
if table_name not in self.table_visibility:
|
|
raise ValueError(f"Unknown table: {table_name}")
|
|
if enabled is None:
|
|
self.table_visibility[table_name] = not self.table_visibility[table_name]
|
|
else:
|
|
self.table_visibility[table_name] = enabled
|
|
return self.table_visibility[table_name]
|
|
|
|
def _format_price(self, price_val, width=10):
|
|
"""Format a price value with appropriate precision."""
|
|
try:
|
|
price_float = float(price_val)
|
|
if price_float < 1:
|
|
return f"{price_float:>{width}.6f}"
|
|
elif price_float < 100:
|
|
return f"{price_float:>{width}.4f}"
|
|
else:
|
|
return f"{price_float:>{width}.2f}"
|
|
except (ValueError, TypeError):
|
|
return f"{'Loading...':>{width}}"
|
|
|
|
def build_market_table(self, watched_coins, prices, display_names):
|
|
"""Build the market dashboard table."""
|
|
table = Table(title="Market Dashboard", show_header=True, header_style="bold cyan", title_style="bold white")
|
|
table.add_column("#", justify="right", style="dim", width=3)
|
|
table.add_column("Coin", justify="center", width=8)
|
|
table.add_column("Best Bid", justify="right")
|
|
table.add_column("Live Price", justify="right")
|
|
table.add_column("Best Ask", justify="right")
|
|
table.add_column("Gap", justify="right")
|
|
table.add_column("Dir", justify="center", width=3)
|
|
|
|
for i, coin in enumerate(watched_coins, 1):
|
|
display_name = display_names.get(coin, coin)
|
|
mid = prices.get(coin)
|
|
bid = prices.get(f"{coin}_bid")
|
|
ask = prices.get(f"{coin}_ask")
|
|
|
|
formatted_mid = self._format_price(mid)
|
|
formatted_bid = self._format_price(bid)
|
|
formatted_ask = self._format_price(ask)
|
|
|
|
gap_str = "Loading..."
|
|
gap_style = "dim"
|
|
try:
|
|
gap_val = float(ask) - float(bid)
|
|
if gap_val < 1:
|
|
gap_str = f"{gap_val:.6f}"
|
|
else:
|
|
gap_str = f"{gap_val:.4f}"
|
|
gap_style = "green" if gap_val > 0 else "red"
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
direction = " "
|
|
direction_style = "dim"
|
|
prev_mid = self.previous_prices.get(coin)
|
|
if prev_mid is not None and mid is not None:
|
|
try:
|
|
if float(mid) > float(prev_mid):
|
|
direction = "↑"
|
|
direction_style = "green"
|
|
elif float(mid) < float(prev_mid):
|
|
direction = "↓"
|
|
direction_style = "red"
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
table.add_row(
|
|
str(i), display_name, formatted_bid, formatted_mid, formatted_ask,
|
|
Text(gap_str, style=gap_style),
|
|
Text(direction, style=direction_style)
|
|
)
|
|
|
|
if coin == "SUI":
|
|
table.add_section()
|
|
|
|
if mid is not None:
|
|
self.previous_prices[coin] = mid
|
|
|
|
return table
|
|
|
|
def build_strategy_table(self, strategy_statuses, strategy_configs):
|
|
"""Build the strategies table."""
|
|
table = Table(title="Strategies", show_header=True, header_style="bold cyan", title_style="bold white")
|
|
table.add_column("#", justify="center", width=3)
|
|
table.add_column("Strategy Name", width=25)
|
|
table.add_column("Coin", justify="center", width=8)
|
|
table.add_column("Signal", justify="center", width=10)
|
|
table.add_column("Signal Price", justify="right", width=14)
|
|
table.add_column("Last Change", justify="right", width=19)
|
|
table.add_column("TF", justify="center", width=7)
|
|
table.add_column("Size", justify="center", width=10)
|
|
|
|
for i, (name, status) in enumerate(strategy_statuses.items(), 1):
|
|
signal = status.get('current_signal', 'N/A')
|
|
price = status.get('signal_price')
|
|
price_display = f"{price:.4f}" if isinstance(price, (int, float)) else "-"
|
|
last_change = status.get('last_signal_change_utc')
|
|
last_change_display = 'Never'
|
|
if last_change:
|
|
dt_utc = datetime.fromisoformat(last_change.replace('Z', '+00:00')).replace(tzinfo=timezone.utc)
|
|
dt_local = dt_utc.astimezone(None)
|
|
last_change_display = dt_local.strftime('%Y-%m-%d %H:%M')
|
|
|
|
config_params = strategy_configs.get(name, {}).get('parameters', {})
|
|
coin = status.get('coin', config_params.get('coin', 'N/A'))
|
|
|
|
size = status.get('size')
|
|
if not size:
|
|
if 'coins_to_copy' in config_params:
|
|
size = 'Multi'
|
|
else:
|
|
size = config_params.get('size', 'N/A')
|
|
|
|
timeframe = config_params.get('timeframe', 'N/A')
|
|
|
|
signal_style = ""
|
|
if signal == "BUY":
|
|
signal_style = "green"
|
|
elif signal == "SELL":
|
|
signal_style = "red"
|
|
elif signal == "NEUTRAL":
|
|
signal_style = "yellow"
|
|
|
|
table.add_row(
|
|
str(i), name, coin,
|
|
Text(signal, style=signal_style) if signal_style else signal,
|
|
price_display, last_change_display, timeframe, str(size)
|
|
)
|
|
|
|
return table
|
|
|
|
def _format_change_value(self, value):
|
|
"""Format a percentage change value with color styling."""
|
|
if value is None:
|
|
return Text("N/A", style="dim")
|
|
if value > 0:
|
|
return Text(f"+{value:.2f}%", style="green")
|
|
elif value < 0:
|
|
return Text(f"{value:.2f}%", style="red")
|
|
else:
|
|
return Text(f"{value:.2f}%", style="yellow")
|
|
|
|
def _format_value(self, value, width=12):
|
|
"""Format a numeric value for display."""
|
|
if value is None:
|
|
return Text("N/A", style="dim")
|
|
try:
|
|
val = float(value)
|
|
if abs(val) < 1:
|
|
return Text(f"{val:>{width}.6f}")
|
|
elif abs(val) < 100:
|
|
return Text(f"{val:>{width}.4f}")
|
|
else:
|
|
return Text(f"{val:>{width}.2f}")
|
|
except (ValueError, TypeError):
|
|
return Text("N/A", style="dim")
|
|
|
|
def build_indicators_table(self, indicators_status):
|
|
"""Build the indicators dashboard table."""
|
|
table = Table(title="Indicators", show_header=True, header_style="bold cyan", title_style="bold white")
|
|
table.add_column("#", justify="right", style="dim", width=3)
|
|
table.add_column("Indicator", width=20)
|
|
table.add_column("Value", justify="right")
|
|
table.add_column("1h Change", justify="right", width=12)
|
|
table.add_column("1D Change", justify="right", width=12)
|
|
table.add_column("Deviation", justify="right", width=12)
|
|
|
|
if not indicators_status:
|
|
table.add_row("1", "Loading...", "N/A", "N/A", "N/A", "N/A")
|
|
return table
|
|
|
|
indicators = indicators_status.get("indicators", {})
|
|
for i, (name, data) in enumerate(indicators.items(), 1):
|
|
display_name = data.get("display_name", name)
|
|
value = data.get("value")
|
|
changes = data.get("changes", {})
|
|
deviation = data.get("deviation")
|
|
|
|
formatted_value = self._format_value(value)
|
|
change_1h = self._format_change_value(changes.get("1h"))
|
|
change_1d = self._format_change_value(changes.get("1d"))
|
|
|
|
if deviation is not None:
|
|
if deviation > 0:
|
|
deviation_str = Text(f"+{deviation:.2f}%", style="green")
|
|
elif deviation < 0:
|
|
deviation_str = Text(f"{deviation:.2f}%", style="red")
|
|
else:
|
|
deviation_str = Text(f"{deviation:.2f}%", style="yellow")
|
|
else:
|
|
deviation_str = Text("N/A", style="dim")
|
|
|
|
table.add_row(
|
|
str(i), display_name, formatted_value,
|
|
change_1h, change_1d, deviation_str
|
|
)
|
|
|
|
return table
|
|
|
|
def build_balances_table(self, account_data, prices=None):
|
|
"""Build a combined balances and open positions table.
|
|
|
|
Args:
|
|
account_data: dict with keys:
|
|
- spot_balances: list of {coin, total}
|
|
- positions: list of position dicts with position data
|
|
- account_value: float
|
|
- margin_used: float
|
|
- utilization: float
|
|
prices: dict mapping coin names to current mark prices
|
|
"""
|
|
if prices is None:
|
|
prices = {}
|
|
table = Table(show_header=True, header_style="bold cyan", title="Account Summary")
|
|
table.add_column("Type", justify="center", width=8)
|
|
table.add_column("Coin", justify="center", width=8)
|
|
table.add_column("Size", justify="right", width=12)
|
|
table.add_column("Value", justify="right", width=12)
|
|
|
|
spot_balances = account_data.get('spot_balances', [])
|
|
for bal in spot_balances:
|
|
total = float(bal.get('total', 0))
|
|
if total > 0:
|
|
coin = bal.get('coin', 'Unknown')
|
|
mark_price = float(prices.get(coin, 0))
|
|
usd_value = total * mark_price
|
|
table.add_row(
|
|
Text("Spot", style="blue"),
|
|
coin,
|
|
f"{total:,.4f}",
|
|
f"${usd_value:,.2f}"
|
|
)
|
|
|
|
positions = account_data.get('positions', [])
|
|
for pos in positions:
|
|
position = pos.get('position', {})
|
|
coin = position.get('coin', 'Unknown')
|
|
size = float(position.get('szi', 0))
|
|
if size != 0:
|
|
position_value = float(position.get('positionValue', 0))
|
|
side = "LONG" if size > 0 else "SHORT"
|
|
side_style = "green" if size > 0 else "red"
|
|
|
|
table.add_row(
|
|
Text(f"P({side})", style=side_style),
|
|
coin,
|
|
f"{size:,.4f}",
|
|
f"${position_value:,.2f}"
|
|
)
|
|
|
|
if not spot_balances and not positions:
|
|
table.add_row("None", "-", "-", "-")
|
|
|
|
account_value = account_data.get('account_value', 0)
|
|
margin_used = account_data.get('margin_used', 0)
|
|
utilization = account_data.get('utilization', 0)
|
|
|
|
# table.add_section()
|
|
table.add_row(
|
|
Text("Acct", style="bold"),
|
|
"-", "-",
|
|
f"${account_value:,.2f}"
|
|
)
|
|
table.add_row(
|
|
Text("Util", style="bold"),
|
|
"-", "-",
|
|
f"{utilization:.2f}%"
|
|
)
|
|
|
|
return table
|
|
|
|
def build_layout(self, watched_coins, prices, display_names, strategy_statuses, strategy_configs, indicators_status=None, account_data=None):
|
|
"""Build the complete dashboard layout in a 2x2 grid."""
|
|
from rich.layout import Layout as RichLayout
|
|
|
|
tables = []
|
|
|
|
if self.table_visibility.get("market", True):
|
|
tables.append(self.build_market_table(watched_coins, prices, display_names))
|
|
if self.table_visibility.get("indicators", True):
|
|
tables.append(Padding(self.build_indicators_table(indicators_status), (0, 0, 0, 2)))
|
|
if account_data is not None and self.table_visibility.get("balances", True):
|
|
tables.append(Padding(self.build_balances_table(account_data, prices), (0, 0, 0, 2)))
|
|
if self.table_visibility.get("strategies", True):
|
|
tables.append(self.build_strategy_table(strategy_statuses, strategy_configs))
|
|
|
|
if not tables:
|
|
return RichLayout()
|
|
|
|
if len(tables) <= 2:
|
|
layout = RichLayout()
|
|
layout.split_row(*tables)
|
|
return layout
|
|
|
|
top = RichLayout(ratio=1)
|
|
bottom = RichLayout(ratio=2)
|
|
top.split_row(*tables[:2])
|
|
bottom.split_row(*tables[2:])
|
|
layout = RichLayout()
|
|
layout.split_column(top, bottom)
|
|
return layout
|