62e32fc655
Complete rewrite — replaces the ML-based signal optimizer with a transparent on-chain metric monitoring dashboard. Scrapes 10 metrics from LookIntoBitcoin (Playwright) and free APIs, scores each 0-10, composite 0-100. Metrics: Fear & Greed, Puell Multiple, MVRV Z-Score, Drawdown from ATH, Price vs 200W SMA, Reserve Risk, RHODL Ratio, NUPL, LTH Realized Price, Hash Ribbons. Auto-refreshes every 15 minutes. Settings page preserved. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
266 lines
9.0 KiB
Python
266 lines
9.0 KiB
Python
"""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
|