btc-accumulation-monitor/scrapers/checkonchain.py

"""Scraper for static CheckOnChain Plotly chart HTML pages."""

from __future__ import annotations

import array
import base64
import json
import logging
import re
from html import unescape

import requests

log = logging.getLogger(__name__)

CHARTS = {
    "sopr": {
        "url": "https://charts.checkonchain.com/btconchain/realised/sopr/sopr_light.html",
        "traces": ["SOPR"],
    },
    "sellside_risk": {
        "url": "https://charts.checkonchain.com/btconchain/realised/sellsideriskratio_all/sellsideriskratio_all_light.html",
        "traces": ["Sell-side Risk Ratio", "Sellside Risk Ratio", "SSR"],
    },
    "active_address_momentum": {
        "url": "https://charts.checkonchain.com/btconchain/adoption/actaddress_momentum/actaddress_momentum_light.html",
        "traces": ["30DMA", "30 Day", "Active Address"],
    },
    "txcount_momentum": {
        "url": "https://charts.checkonchain.com/btconchain/adoption/txcount_momentum/txcount_momentum_light.html",
        "traces": ["30DMA", "30 Day", "Transaction"],
    },
    "nvt_price": {
        "url": "https://charts.checkonchain.com/btconchain/pricing/pricing_nvtprice/pricing_nvtprice_light.html",
        "traces": ["NVT Price", "NVT"],
    },
    "vdd_multiple": {
        "url": "https://charts.checkonchain.com/btconchain/lifespan/vddmultiple/vddmultiple_light.html",
        "traces": ["VDD Multiple", "Value Days Destroyed"],
    },
}


def _extract_plotly_traces(html_text: str):
    """Extract first Plotly.newPlot trace array from a static Plotly HTML page."""
    marker = "Plotly.newPlot("
    start = html_text.find(marker)
    if start < 0:
        return []
    first_array = html_text.find("[", start)
    if first_array < 0:
        return []

    depth = 0
    in_string = False
    escape = False
    quote = ""
    for idx in range(first_array, len(html_text)):
        ch = html_text[idx]
        if in_string:
            if escape:
                escape = False
            elif ch == "\\":
                escape = True
            elif ch == quote:
                in_string = False
            continue
        if ch in {'"', "'"}:
            in_string = True
            quote = ch
        elif ch == "[":
            depth += 1
        elif ch == "]":
            depth -= 1
            if depth == 0:
                raw = html_text[first_array:idx + 1]
                return json.loads(raw)
    return []


def scrape_chart(url: str, timeout=30):
    resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=timeout)
    resp.raise_for_status()
    return _extract_plotly_traces(unescape(resp.text))


def _find_trace(traces, names):
    names = [n.lower() for n in names if n]
    # Prefer non-price traces with the requested terms.
    for trace in traces:
        trace_name = str(trace.get("name", "")).lower()
        if "price" in trace_name and not any("price" in n for n in names):
            continue
        if any(n in trace_name for n in names):
            return trace
    # Fallback: first numeric non-price trace.
    for trace in traces:
        trace_name = str(trace.get("name", "")).lower()
        if "price" in trace_name:
            continue
        y = trace.get("y") or []
        if any(v is not None for v in y[-30:]):
            return trace
    return None


def _decode_plotly_array(values):
    """Decode Plotly typed-array JSON ({dtype, bdata}) or return plain values."""
    if not isinstance(values, dict) or "bdata" not in values:
        return values or []

    dtype = values.get("dtype")
    typecodes = {
        "f8": "d", "float64": "d",
        "f4": "f", "float32": "f",
        "i8": "q", "int64": "q",
        "i4": "i", "int32": "i",
        "u8": "Q", "uint64": "Q",
        "u4": "I", "uint32": "I",
    }
    typecode = typecodes.get(dtype)
    if not typecode:
        return []
    decoded = base64.b64decode(values["bdata"])
    arr = array.array(typecode)
    arr.frombytes(decoded)
    if values.get("byteorder") == "big":
        arr.byteswap()
    return arr.tolist()


def _numeric_values(trace):
    values = []
    for value in _decode_plotly_array((trace or {}).get("y", [])):
        if value is None:
            continue
        try:
            values.append(float(value))
        except (TypeError, ValueError):
            pass
    return values


def _latest(values):
    return values[-1] if values else None


def _momentum(values, window=30):
    if len(values) <= window or values[-window] == 0:
        return None
    return (values[-1] - values[-window]) / values[-window]


def scrape_all():
    results = {}
    for key, cfg in CHARTS.items():
        log.info("Scraping CheckOnChain %s ...", key)
        try:
            traces = scrape_chart(cfg["url"])
            trace = _find_trace(traces, cfg.get("traces", []))
            values = _numeric_values(trace)
            value = _latest(values)
            if key in {"active_address_momentum", "txcount_momentum", "vdd_multiple"}:
                # The card value is momentum, while the sparkline shows the raw metric.
                value = _momentum(values)
            results[key] = {"value": value, "recent": values[-30:]}
        except Exception as exc:
            log.error("CheckOnChain scrape failed for %s: %s", key, exc)
            results[key] = {"value": None, "error": str(exc)}
    return results