terminal_dashboard.py

#!/usr/bin/env python3
"""
terminal_dashboard.py — Live terminal dashboard for SignalScanner.

Shows:
    - Market Pulse banner (BTC dominance, trend, decorrelation count)
    - Per-pair table with full indicator state + BTC correlation
Refreshes every 10s. Ctrl-C to exit.
"""
from __future__ import annotations

import json
import sqlite3
import time
from datetime import datetime
from pathlib import Path

from rich.console import Console, Group
from rich.live import Live
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from rich import box


BASE_DIR = Path(__file__).parent.resolve()
DB_PATH = BASE_DIR / "state.sqlite"
REFRESH_SECONDS = 10


def fetch_signals():
    if not DB_PATH.exists():
        return [], {}, {}
    with sqlite3.connect(DB_PATH) as conn:
        conn.row_factory = sqlite3.Row
        signals = conn.execute("""
            SELECT s.*
            FROM signals s
            INNER JOIN (
                SELECT pair, MAX(timestamp_utc) AS max_ts
                FROM signals GROUP BY pair
            ) latest ON s.pair = latest.pair AND s.timestamp_utc = latest.max_ts
            ORDER BY s.rank_24h ASC
        """).fetchall()
        last_run = conn.execute(
            "SELECT * FROM scanner_runs ORDER BY id DESC LIMIT 1"
        ).fetchone()
        pulse = conn.execute(
            "SELECT * FROM market_pulse ORDER BY timestamp_utc DESC LIMIT 1"
        ).fetchone()
        pulse_dict = dict(pulse) if pulse else {}
        if pulse_dict:
            try:
                pulse_dict["decorrelation_events"] = json.loads(
                    pulse_dict.get("decorrelation_events_json") or "[]"
                )
            except Exception:
                pulse_dict["decorrelation_events"] = []
        return [dict(s) for s in signals], dict(last_run) if last_run else {}, pulse_dict


def color_rsi(rsi):
    if rsi is None:
        return Text("-", style="dim")
    if rsi < 25:
        return Text(f"{rsi:.0f}", style="bold green")
    if rsi > 75:
        return Text(f"{rsi:.0f}", style="bold red")
    return Text(f"{rsi:.0f}", style="white")


def color_change(pct):
    if pct is None:
        return Text("-")
    color = "green" if pct >= 0 else "red"
    return Text(f"{pct:+.2f}%", style=color)


def color_alignment(val):
    colors = {"bullish": "green", "bearish": "red", "mixed": "yellow"}
    return Text(val or "-", style=colors.get(val, "white"))


def color_correlation(corr):
    if corr is None:
        return Text("-", style="dim")
    if abs(corr) > 0.8:
        return Text(f"{corr:+.2f}", style="bold white")  # strong
    if abs(corr) > 0.5:
        return Text(f"{corr:+.2f}", style="white")
    if abs(corr) < 0.3:
        return Text(f"{corr:+.2f}", style="yellow")  # decorrelated
    return Text(f"{corr:+.2f}", style="dim white")


def build_table(signals: list[dict]) -> Table:
    table = Table(
        title="SignalScanner — live indicator snapshot",
        box=box.SIMPLE_HEAD,
        expand=True,
        pad_edge=False,
        header_style="bold cyan",
    )
    table.add_column("Pair", style="bold white", no_wrap=True)
    table.add_column("Price", justify="right")
    table.add_column("24h%", justify="right")
    table.add_column("RSI1h", justify="right")
    table.add_column("RSI4h", justify="right")
    table.add_column("RSI1d", justify="right")
    table.add_column("vs EMA200", justify="right")
    table.add_column("Trend", justify="center")
    table.add_column("BB", justify="right")
    table.add_column("Vol×", justify="right")
    table.add_column("BTC corr", justify="right")
    table.add_column("#Sig", justify="right")
    table.add_column("Tags", style="yellow", no_wrap=False, overflow="fold")

    for s in signals:
        price = format(s.get("close_price") or 0, ",.4f")
        dist_ema = s.get("dist_from_ema200_pct") or 0
        dist_style = "green" if dist_ema > 0 else "red"
        active = s.get("active_signal_count", 0)
        sig_style = "bold red" if active >= 3 else "bold yellow" if active >= 1 else "dim"

        table.add_row(
            s["pair"],
            f"${price}",
            color_change(s.get("pct_change_24h")),
            color_rsi(s.get("rsi_1h")),
            color_rsi(s.get("rsi_4h")),
            color_rsi(s.get("rsi_1d")),
            Text(f"{dist_ema:+.1f}%", style=dist_style),
            color_alignment(s.get("ema_alignment")),
            f"{(s.get('bb_position') or 0) * 100:.0f}%",
            f"{(s.get('volume_ratio_20') or 0):.1f}",
            color_correlation(s.get("btc_correlation")),
            Text(str(active), style=sig_style),
            (s.get("signal_tags") or "").replace(",", " · "),
        )
    return table


def build_header(last_run: dict, pulse: dict, num_signals: int) -> Panel:
    last_scan = last_run.get("end_utc", "never")
    if last_scan and last_scan != "never":
        dt = datetime.fromisoformat(last_scan.replace("Z", "+00:00"))
        last_scan = dt.strftime("%H:%M:%S UTC")

    dom = pulse.get("btc_dominance_pct")
    delta = pulse.get("btc_dominance_delta_24h")
    trend = pulse.get("btc_dominance_trend") or "—"
    decorrel_count = len(pulse.get("decorrelation_events", []))

    dom_str = f"{dom:.2f}%" if dom is not None else "—"
    delta_str = f"({delta:+.2f}pp 24h)" if delta is not None else ""
    trend_color = {"rising": "red", "falling": "green", "flat": "white"}.get(trend, "white")
    decorrel_color = "red" if decorrel_count > 0 else "white"

    text = (
        f"Last scan: [cyan]{last_scan}[/] | "
        f"Pairs: [cyan]{num_signals}[/] | "
        f"BTC dom: [bold]{dom_str}[/] {delta_str} [{trend_color}]{trend}[/] | "
        f"Decorrel: [{decorrel_color}]{decorrel_count}[/] | "
        f"Refresh: {REFRESH_SECONDS}s | Ctrl-C to exit"
    )
    return Panel(text, title="[bold]SignalScanner v1 + Market Pulse[/]", border_style="cyan")


def render():
    signals, last_run, pulse = fetch_signals()
    if not signals:
        return Panel(
            "[yellow]No data yet.[/] Run scanner.py first — state.sqlite is empty.",
            title="[bold]SignalScanner v1[/]",
            border_style="yellow",
        )
    header = build_header(last_run, pulse, len(signals))
    table = build_table(signals)
    return Group(header, table)


def main():
    console = Console()
    try:
        with Live(render(), console=console, refresh_per_second=1, screen=False) as live:
            while True:
                time.sleep(REFRESH_SECONDS)
                live.update(render())
    except KeyboardInterrupt:
        console.print("\n[cyan]Dashboard closed.[/]")


if __name__ == "__main__":
    main()