feat: replace ML optimizer with on-chain accumulation zone monitor
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>
This commit is contained in:
BizzleBot
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
aba30f7718
commit
62e32fc655
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,34 @@
|
||||
"""Fear & Greed Index from alternative.me API."""
|
||||
|
||||
import logging
|
||||
import requests
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
FNG_URL = "https://api.alternative.me/fng/?limit=30"
|
||||
|
||||
|
||||
def fetch():
|
||||
"""Fetch Fear & Greed data. Returns dict with value, classification, and recent history."""
|
||||
try:
|
||||
resp = requests.get(FNG_URL, timeout=15)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
entries = data.get("data", [])
|
||||
if not entries:
|
||||
return {"value": None, "error": "No data"}
|
||||
|
||||
current = entries[0]
|
||||
value = int(current["value"])
|
||||
classification = current.get("value_classification", "")
|
||||
|
||||
recent = [int(e["value"]) for e in entries[:30]]
|
||||
|
||||
return {
|
||||
"value": value,
|
||||
"classification": classification,
|
||||
"recent": recent,
|
||||
}
|
||||
except Exception as e:
|
||||
log.error("Fear & Greed fetch error: %s", e)
|
||||
return {"value": None, "error": str(e)}
|
||||
@@ -0,0 +1,265 @@
|
||||
"""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
|
||||
@@ -0,0 +1,80 @@
|
||||
"""BTC price data from CoinGecko API."""
|
||||
|
||||
import logging
|
||||
import requests
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
PRICE_URL = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd&include_24hr_change=true"
|
||||
HISTORY_URL = "https://api.coingecko.com/api/v3/coins/bitcoin/market_chart?vs_currency=usd&days=365"
|
||||
ATH_URL = "https://api.coingecko.com/api/v3/coins/bitcoin?localization=false&tickers=false&market_data=true&community_data=false&developer_data=false"
|
||||
|
||||
|
||||
def fetch_current():
|
||||
"""Fetch current BTC price and 24h change."""
|
||||
try:
|
||||
resp = requests.get(PRICE_URL, timeout=15)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
btc = data.get("bitcoin", {})
|
||||
return {
|
||||
"price": btc.get("usd"),
|
||||
"change_24h": btc.get("usd_24h_change"),
|
||||
}
|
||||
except Exception as e:
|
||||
log.error("Price fetch error: %s", e)
|
||||
return {"price": None, "error": str(e)}
|
||||
|
||||
|
||||
def fetch_historical():
|
||||
"""Fetch 365 days of BTC price history. Returns list of [timestamp, price]."""
|
||||
try:
|
||||
resp = requests.get(HISTORY_URL, timeout=30)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
prices = data.get("prices", [])
|
||||
return prices
|
||||
except Exception as e:
|
||||
log.error("Historical price fetch error: %s", e)
|
||||
return []
|
||||
|
||||
|
||||
def fetch_ath():
|
||||
"""Fetch BTC all-time high from CoinGecko."""
|
||||
try:
|
||||
resp = requests.get(ATH_URL, timeout=15)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
market = data.get("market_data", {})
|
||||
ath = market.get("ath", {}).get("usd")
|
||||
ath_change = market.get("ath_change_percentage", {}).get("usd")
|
||||
return {
|
||||
"ath": ath,
|
||||
"ath_change_pct": ath_change,
|
||||
}
|
||||
except Exception as e:
|
||||
log.error("ATH fetch error: %s", e)
|
||||
return {"ath": None, "error": str(e)}
|
||||
|
||||
|
||||
def calculate_200d_sma(prices):
|
||||
"""Calculate 200-day SMA from historical price data."""
|
||||
if not prices or len(prices) < 200:
|
||||
return None
|
||||
# prices is [[timestamp, price], ...]
|
||||
recent_200 = [p[1] for p in prices[-200:]]
|
||||
return sum(recent_200) / len(recent_200)
|
||||
|
||||
|
||||
def calculate_mayer_multiple(current_price, sma_200d):
|
||||
"""Mayer Multiple = current price / 200-day SMA."""
|
||||
if not current_price or not sma_200d or sma_200d == 0:
|
||||
return None
|
||||
return current_price / sma_200d
|
||||
|
||||
|
||||
def calculate_drawdown(current_price, ath):
|
||||
"""Drawdown from ATH as percentage."""
|
||||
if not current_price or not ath or ath == 0:
|
||||
return None
|
||||
return (ath - current_price) / ath * 100
|
||||
Reference in New Issue
Block a user