"""Playwright scraper for LookIntoBitcoin / BitcoinMagazinePro charts.""" import logging import traceback log = logging.getLogger(__name__) BASE_URL = "https://www.lookintobitcoin.com" CHARTS = { "puell_multiple": { "path": "/charts/puell-multiple/", "traces": ["Puell Multiple"], }, "mvrv_zscore": { "path": "/charts/mvrv-zscore/", "traces": ["Z-Score"], }, "reserve_risk": { "path": "/charts/reserve-risk/", "traces": ["Reserve Risk"], }, "rhodl_ratio": { "path": "/charts/rhodl-ratio/", "traces": ["RHODL Ratio"], }, "nupl": { "path": "/charts/relative-unrealized-profit--loss/", "traces": ["NUPL"], }, "200w_sma": { "path": "/charts/200-week-moving-average-heatmap/", "traces": ["200 Week Moving Average"], }, "lth_realized_price": { "path": "/charts/long-term-holder-realized-price/", "traces": ["Long-Term Holder Realized Price", "BTC Price"], }, "hash_ribbons": { "path": "/charts/hash-ribbons/", "traces": None, }, "pi_cycle_bottom": { "path": "/charts/pi-cycle-top-bottom-indicator/", "traces": None, }, "lth_supply": { "path": "/charts/long-term-holder-supply/", "traces": None, }, } def scrape_chart(chart_path, timeout=25000): """Scrape a single chart from LookIntoBitcoin. Returns list of trace dicts or None.""" from playwright.sync_api import sync_playwright store = {"data": None} with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() def handle_response(response): if "_dash-update-component" in response.url: try: store["data"] = response.json() except Exception: pass page.on("response", handle_response) try: page.goto(f"{BASE_URL}{chart_path}", timeout=timeout) page.wait_for_timeout(6000) except Exception as e: log.warning("Navigation error for %s: %s", chart_path, e) finally: browser.close() if store["data"]: try: return store["data"]["response"]["chart"]["figure"]["data"] except (KeyError, TypeError): # Try alternate response structures try: resp = store["data"] if isinstance(resp, dict): for key in resp: val = resp[key] if isinstance(val, dict) and "figure" in val: return val["figure"]["data"] if isinstance(val, dict) and "chart" in val: return val["chart"]["figure"]["data"] except Exception: pass return None def _find_trace(traces, name): """Find a trace by name (case-insensitive partial match).""" if not traces: return None name_lower = name.lower() # First pass: exact or substring match for t in traces: trace_name = t.get("name", "").lower() if name_lower in trace_name or trace_name in name_lower: return t # Second pass: check if all words in name appear in trace name words = name_lower.split() for t in traces: trace_name = t.get("name", "").lower() if all(w in trace_name for w in words): return t return None def _get_latest_value(trace): """Get the most recent non-null y value from a trace.""" if not trace: return None y = trace.get("y", []) for val in reversed(y): if val is not None: try: return float(val) except (ValueError, TypeError): continue return None def _get_recent_values(trace, n=30): """Get the last n non-null values from a trace.""" if not trace: return [] y = trace.get("y", []) values = [] for val in reversed(y): if val is not None: try: values.append(float(val)) except (ValueError, TypeError): continue if len(values) >= n: break values.reverse() return values def scrape_all(): """Scrape all charts and return parsed metric values.""" results = {} for metric_key, chart_info in CHARTS.items(): log.info("Scraping %s ...", metric_key) try: traces = scrape_chart(chart_info["path"]) if not traces: log.warning("No data for %s", metric_key) results[metric_key] = {"value": None, "error": "No data returned"} continue wanted = chart_info.get("traces") if metric_key == "puell_multiple": t = _find_trace(traces, "Puell Multiple") val = _get_latest_value(t) results[metric_key] = { "value": val, "recent": _get_recent_values(t), } elif metric_key == "mvrv_zscore": t = _find_trace(traces, "Z-Score") val = _get_latest_value(t) results[metric_key] = { "value": val, "recent": _get_recent_values(t), } elif metric_key == "200w_sma": t = _find_trace(traces, "200 Week Moving Average") or _find_trace(traces, "200 Week MA") or _find_trace(traces, "200W") val = _get_latest_value(t) # Also try to find BTC price trace price_t = _find_trace(traces, "BTC Price") or _find_trace(traces, "Price") price_val = _get_latest_value(price_t) results[metric_key] = { "value": val, "btc_price": price_val, "recent": _get_recent_values(t), } elif metric_key == "lth_realized_price": lth_t = _find_trace(traces, "Long-Term Holder Realized Price") or _find_trace(traces, "LTH Realized Price") or _find_trace(traces, "LTH") price_t = _find_trace(traces, "BTC Price") or _find_trace(traces, "Price") lth_val = _get_latest_value(lth_t) price_val = _get_latest_value(price_t) results[metric_key] = { "value": lth_val, "btc_price": price_val, "recent": _get_recent_values(lth_t), } elif metric_key == "hash_ribbons": # Look for buy/sell signal traces or MA crossover results[metric_key] = { "traces": [ {"name": t.get("name", ""), "latest": _get_latest_value(t)} for t in traces[:6] ], "value": None, } # Try to detect buy signal from trace names/colors for t in traces: name = t.get("name", "").lower() if "buy" in name or "signal" in name: results[metric_key]["buy_signal"] = True break elif metric_key == "lth_supply": # Get main supply trace t = traces[0] if traces else None for candidate in traces: name = candidate.get("name", "").lower() if "supply" in name or "lth" in name: t = candidate break recent = _get_recent_values(t, 60) # Determine trend: compare recent avg to older avg trend = None if len(recent) >= 30: old_avg = sum(recent[:15]) / 15 new_avg = sum(recent[-15:]) / 15 trend = "increasing" if new_avg > old_avg else "decreasing" results[metric_key] = { "value": _get_latest_value(t), "trend": trend, "recent": _get_recent_values(t), } else: # Generic: grab first non-layout trace with numeric data t = None if wanted: for name in wanted: t = _find_trace(traces, name) if t: break if not t: for candidate in traces: y = candidate.get("y", []) if y and any(v is not None for v in y[-10:]): t = candidate break val = _get_latest_value(t) results[metric_key] = { "value": val, "recent": _get_recent_values(t), } except Exception as e: log.error("Error scraping %s: %s\n%s", metric_key, e, traceback.format_exc()) results[metric_key] = {"value": None, "error": str(e)} return results