Spaces:

sadovsky
/

FIN_ASSISTANT

Sleeping

File size: 9,180 Bytes

8f39086

"""
Crypto Dashboard — Plotly Edition (clean layout)
• убраны colorbar заголовки (percent_change_*)
• уменьшены отступы KPI
• без глобального Markdown-заголовка
"""
import requests

import numpy as np
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go

from config import CACHE_RETRY_SECONDS, CACHE_TTL_SECONDS
from infrastructure.cache import CacheUnavailableError, TTLCache
from infrastructure.llm_client import llm_service


_coinlore_cache = TTLCache(CACHE_TTL_SECONDS, CACHE_RETRY_SECONDS)


def _load_coinlore() -> pd.DataFrame:
    url = "https://api.coinlore.net/api/tickers/"
    try:
        response = requests.get(url, timeout=20)
        response.raise_for_status()
        payload = response.json()
        data = payload.get("data")
        if not isinstance(data, list):
            raise ValueError("Unexpected Coinlore payload structure")
    except requests.RequestException as exc:  # noqa: PERF203 - propagate meaningful message
        raise CacheUnavailableError(
            "Coinlore API request failed.",
            CACHE_RETRY_SECONDS,
        ) from exc
    except ValueError as exc:
        raise CacheUnavailableError(
            "Coinlore API returned unexpected response.",
            CACHE_RETRY_SECONDS,
        ) from exc

    df = pd.DataFrame(data)
    for col in [
        "price_usd",
        "market_cap_usd",
        "volume24",
        "percent_change_1h",
        "percent_change_24h",
        "percent_change_7d",
    ]:
        df[col] = pd.to_numeric(df[col], errors="coerce")
    return df


def fetch_coinlore_data(limit: int = 100) -> pd.DataFrame:
    """Return cached Coinlore data limited to the requested number of rows."""

    base = _coinlore_cache.get("coinlore", _load_coinlore)
    return base.head(limit).copy()


def _kpi_line(df) -> str:
    """Формирует компактную KPI-строку без лишних пробелов"""
    tracked = ["BTC", "ETH", "SOL", "DOGE"]
    parts = []
    for sym in tracked:
        row = df[df["symbol"] == sym]
        if row.empty:
            continue
        price = float(row["price_usd"])
        ch = float(row["percent_change_24h"])
        arrow = "↑" if ch > 0 else "↓"
        color = "#4ade80" if ch > 0 else "#f87171"
        parts.append(
            f"<b>{sym}</b> ${price:,.0f} "
            f"<span style='color:{color}'>{arrow} {abs(ch):.2f}%</span>"
        )
    return " , ".join(parts)


def build_crypto_dashboard(top_n=50):
    try:
        df = fetch_coinlore_data(top_n)
    except CacheUnavailableError as e:
        wait = int(e.retry_in) + 1
        message = f"⚠️ Coinlore API cooling down. Retry in ~{wait} seconds."
        return (
            _error_figure("Market Composition", message),
            _error_figure("Top Movers", message),
            _error_figure("Market Cap vs Volume", message),
            message,
            message,
        )
    except Exception:  # noqa: BLE001 - surface unexpected failures
        message = "❌ Failed to load market data. Please try again later."
        return (
            _error_figure("Market Composition", message),
            _error_figure("Top Movers", message),
            _error_figure("Market Cap vs Volume", message),
            message,
            message,
        )

    # === Treemap ===
    fig_treemap = px.treemap(
        df,
        path=["symbol"],
        values="market_cap_usd",
        color="percent_change_24h",
        color_continuous_scale="RdYlGn",
        height=420,
    )
    fig_treemap.update_layout(
        title=None,
        template="plotly_dark",
        coloraxis_colorbar=dict(title=None),  # 🔹 убираем надпись percent_change_24h
        margin=dict(l=5, r=5, t=5, b=5),
        paper_bgcolor="rgba(0,0,0,0)",
        plot_bgcolor="rgba(0,0,0,0)",
    )

    # === Bar chart (Top gainers) ===
    top = df.sort_values("percent_change_24h", ascending=False).head(12)
    fig_bar = px.bar(
        top,
        x="percent_change_24h",
        y="symbol",
        orientation="h",
        color="percent_change_24h",
        color_continuous_scale="Blues",
        height=320,
    )
    fig_bar.update_layout(
        title=None,
        template="plotly_dark",
        coloraxis_colorbar=dict(title=None),  # 🔹 убираем надпись percent_change_24h
        margin=dict(l=40, r=10, t=5, b=18),
        paper_bgcolor="rgba(0,0,0,0)",
        plot_bgcolor="rgba(0,0,0,0)",
    )

    # === Scatter (Market Cap vs Volume) ===
    bubble_df = df.head(60).copy()
    if not bubble_df.empty:
        cap = bubble_df["market_cap_usd"].fillna(0).clip(lower=1.0)
        rank = cap.rank(pct=True)

        sqrt_cap = np.sqrt(cap)
        sqrt_min, sqrt_max = float(sqrt_cap.min()), float(sqrt_cap.max())
        if sqrt_max - sqrt_min > 0:
            sqrt_norm = (sqrt_cap - sqrt_min) / (sqrt_max - sqrt_min)
        else:
            sqrt_norm = pd.Series(0.0, index=bubble_df.index)

        log_cap = np.log1p(cap)
        log_min, log_max = float(log_cap.min()), float(log_cap.max())
        if log_max - log_min > 0:
            log_norm = (log_cap - log_min) / (log_max - log_min)
        else:
            log_norm = pd.Series(0.0, index=bubble_df.index)

        hybrid = 0.55 * rank + 0.30 * sqrt_norm + 0.15 * log_norm
        hybrid = np.power(hybrid, 0.85)
        bubble_df["bubble_size"] = 10 + (56 - 10) * hybrid
    else:
        bubble_df["bubble_size"] = 10

    fig_bubble = px.scatter(
        bubble_df,
        x="market_cap_usd",
        y="volume24",
        size="bubble_size",
        color="percent_change_7d",
        hover_name="symbol",
        log_x=True,
        log_y=True,
        color_continuous_scale="RdYlGn",
        height=320,
    )
    fig_bubble.update_layout(
        title=None,
        template="plotly_dark",
        coloraxis_colorbar=dict(title=None),  # 🔹 убираем надпись percent_change_7d
        margin=dict(l=36, r=10, t=5, b=18),
        paper_bgcolor="rgba(0,0,0,0)",
        plot_bgcolor="rgba(0,0,0,0)",
    )

    # === LLM summary ===
    summary = _ai_summary(df)
    kpi_text = _kpi_line(df)
    return fig_treemap, fig_bar, fig_bubble, summary, kpi_text


def _ai_summary(df):
    timestamp = pd.Timestamp.utcnow().strftime("%Y-%m-%d %H:%M UTC")
    leaders = df.sort_values("percent_change_24h", ascending=False).head(3)["symbol"].tolist()
    laggards = df.sort_values("percent_change_24h").head(3)["symbol"].tolist()

    total_cap = float(df["market_cap_usd"].sum()) if not df.empty else 0.0
    total_volume = float(df["volume24"].sum()) if not df.empty else 0.0
    btc_cap = float(df.loc[df["symbol"] == "BTC", "market_cap_usd"].sum()) if total_cap else 0.0
    btc_dominance = (btc_cap / total_cap * 100) if total_cap else 0.0

    snapshot_rows = (
        df.sort_values("market_cap_usd", ascending=False)
        .head(12)
        [["symbol", "price_usd", "percent_change_24h", "percent_change_7d", "volume24"]]
    )
    lines = []
    for row in snapshot_rows.itertuples(index=False):
        lines.append(
            (
                f"{row.symbol}: price ${row.price_usd:,.2f}, "
                f"24h {row.percent_change_24h:+.2f}%, "
                f"7d {row.percent_change_7d:+.2f}%, "
                f"24h volume ${row.volume24:,.0f}"
            )
        )
    snapshot_text = "\n".join(lines)

    system_prompt = (
        "You are a crypto market strategist receiving a fresh Coinlore snapshot. "
        "Use only the provided metrics to deliver an actionable analysis. "
        "Do not mention training cutoffs or missing live access—assume the snapshot reflects the current market."
    )
    user_prompt = f"""
Coinlore snapshot captured at {timestamp}.
Aggregate totals:
- Total market cap (tracked set): ${total_cap:,.0f}
- 24h traded volume: ${total_volume:,.0f}
- BTC dominance: {btc_dominance:.2f}%

Key movers by 24h change:
{snapshot_text or 'No data available.'}

Top gainers (24h): {', '.join(leaders) if leaders else 'n/a'}
Top laggards (24h): {', '.join(laggards) if laggards else 'n/a'}

Provide:
1. Market sentiment and breadth.
2. Liquidity and volatility observations.
3. Short-term outlook and immediate risks, grounded in this snapshot.
"""

    text = ""
    for delta in llm_service.stream_chat(
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        model="meta-llama/Meta-Llama-3.1-8B-Instruct",
    ):
        text += delta
    return text


def _error_figure(title: str, message: str) -> go.Figure:
    fig = go.Figure()
    fig.add_annotation(
        text=message,
        showarrow=False,
        font=dict(color="#ff6b6b", size=16),
        xref="paper",
        yref="paper",
        x=0.5,
        y=0.5,
    )
    fig.update_layout(
        template="plotly_dark",
        title=title,
        xaxis=dict(visible=False),
        yaxis=dict(visible=False),
        height=360,
        paper_bgcolor="rgba(0,0,0,0)",
        plot_bgcolor="rgba(0,0,0,0)",
    )
    return fig