"""
# ── Backtest API ───────────────────────────────────────────────────────
@app.get("/api/backtest")
def api_backtest(mode: str = "classic"):
"""Run backtest and return full results.
mode=classic (default) or mode=ml for ML-optimized scoring.
"""
try:
from backtesting.engine import run_backtest
return run_backtest(ml_mode=(mode == "ml"))
except Exception as e:
log.error("Backtest error: %s", traceback.format_exc())
return JSONResponse({"error": str(e)}, status_code=500)
@app.get("/api/backtest/history")
def api_backtest_history():
"""Return historical daily scores + prices for charting."""
try:
from backtesting.engine import run_backtest
result = run_backtest()
return {"chart_data": result.get("chart_data", []), "date_range": result.get("date_range")}
except Exception as e:
return JSONResponse({"error": str(e)}, status_code=500)
@app.post("/api/backtest/collect", status_code=202)
def api_backtest_collect():
"""Atomically reserve and start historical data collection."""
initial_progress = {"status": "starting", "current": "", "step": 0, "total": 0}
job = _jobs.reserve("history")
if job is None:
active = _jobs.active("history")
return JSONResponse(
{"error": "Collection already in progress", "job": active}, status_code=409
)
_jobs.update_progress(job["id"], initial_progress)
def _run_collector():
from scrapers.history_collector import collect_all_history
def progress_cb(metric, step, total):
_jobs.update_progress(job["id"], {
"status": "scraping", "current": metric,
"step": step + 1, "total": total,
})
collect_all_history(progress_cb=progress_cb)
_jobs.update_progress(job["id"], {"status": "complete"})
return {"collected": True}
_spawn_job(job, _run_collector)
return {"ok": True, "job_id": job["id"], "status": job["status"],
"message": "Collection started"}
@app.get("/api/backtest/status")
def api_backtest_status():
"""Check historical data and expose only the active collection job's progress."""
from scrapers.history_collector import history_status
status = history_status()
active = _jobs.active("history")
status["collecting"] = active is not None
status["job_id"] = active.get("id") if active else None
status["progress"] = active.get("progress", {}) if active else {}
return status
@app.get("/api/metric-context")
def api_metric_context(metric: str, margin: float = 0.0, mode: str = "classic"):
"""Find historical periods where a specific metric was at a similar level.
Returns forward returns for those periods, analogous to the composite-score
current_context but filtered to a single metric's historical similarity.
margin: absolute tolerance for "similar" (auto-computed from metric scale if 0).
"""
try:
from backtesting.engine import run_backtest, HISTORY_PATH, _build_daily_index, _get_all_dates, _last_known_value, METRIC_SCORERS, RATIO_SCORERS, DRAWDOWN_RANGES, _score_range
import os as _os
if not _os.path.exists(HISTORY_PATH):
return JSONResponse({"error": "No historical data. Run history collector first."}, status_code=404)
with open(HISTORY_PATH) as f:
history = json.load(f)
index = _build_daily_index(history)
all_dates = _get_all_dates(index)
# Get current metric value from cache
cache = {}
if _os.path.exists(CACHE_PATH):
with open(CACHE_PATH) as f:
cache = json.load(f)
current_raw = _get_current_metric_raw(metric, cache)
if current_raw is None:
return JSONResponse({"error": f"No current value for metric '{metric}'"}, status_code=404)
# Auto-compute margin from metric scale
if margin <= 0:
margin = _auto_metric_margin(metric, current_raw)
# Build price lookup
price_lookup = {}
for pk in ["btc_price_coingecko", "btc_price", "btc_price_sma", "btc_price_lth"]:
if pk in index:
for d, v in index[pk].items():
if d not in price_lookup:
price_lookup[d] = v
# Find historical days where this metric was similar
comparable = []
for d in all_dates:
raw_val = _get_historical_metric_raw(metric, index, d)
if raw_val is not None and abs(raw_val - current_raw) <= margin:
price = price_lookup.get(d)
fwd = _compute_day_forward_returns(price_lookup, d)
if fwd:
comparable.append({
"date": d,
"raw_value": round(raw_val, 6) if isinstance(raw_val, float) else raw_val,
"price": price,
"forward_returns": fwd,
})
# Compute average returns across comparable periods
avg_returns = {}
for period in ["30d", "90d", "180d", "365d"]:
vals = [c["forward_returns"][period] for c in comparable if period in c["forward_returns"]]
if vals:
avg_returns[period] = round(sum(vals) / len(vals), 2)
# Pick best examples (one per market cycle)
cycle_bins = [
("pre-2016", "2010-01-01", "2015-12-31"),
("2016-17 Bull", "2016-01-01", "2017-12-31"),
("2018-19 Bear", "2018-01-01", "2019-12-31"),
("2020-21 Bull", "2020-01-01", "2021-12-31"),
("2022-23 Bear", "2022-01-01", "2023-12-31"),
("2024+", "2024-01-01", "2099-12-31"),
]
examples = []
used_cycles = set()
sorted_comp = sorted(comparable, key=lambda c: abs(c["raw_value"] - current_raw))
for c in sorted_comp:
for label, start, end in cycle_bins:
if start <= c["date"] <= end and label not in used_cycles:
used_cycles.add(label)
examples.append({
"date": c["date"],
"raw_value": c["raw_value"],
"price": c["price"],
"forward_returns": c["forward_returns"],
"cycle": label,
})
break
if len(examples) >= 6:
break
examples.sort(key=lambda e: e["date"])
# Percentile: what % of all days had this metric at or below current value
all_raw_vals = []
for d in all_dates:
rv = _get_historical_metric_raw(metric, index, d)
if rv is not None:
all_raw_vals.append(rv)
all_raw_vals.sort()
below = len([v for v in all_raw_vals if v <= current_raw])
percentile = round(below / len(all_raw_vals) * 100, 1) if all_raw_vals else 50.0
return {
"metric": metric,
"current_raw": current_raw,
"margin": margin,
"comparable_days": len(comparable),
"percentile": percentile,
"avg_30d_return": avg_returns.get("30d"),
"avg_90d_return": avg_returns.get("90d"),
"avg_180d_return": avg_returns.get("180d"),
"avg_1yr_return": avg_returns.get("365d"),
"examples": examples,
}
except Exception as e:
log.error("Metric context error: %s", traceback.format_exc())
return JSONResponse({"error": str(e)}, status_code=500)
def _get_current_metric_raw(metric, cache):
"""Get the current raw value for a metric from the cache."""
# Direct cache keys
direct_keys = {
"fear_greed": ("fear_greed", "value"),
"puell_multiple": ("puell_multiple", "value"),
"mvrv_zscore": ("mvrv_zscore", "value"),
"reserve_risk": ("reserve_risk", "value"),
"rhodl_ratio": ("rhodl_ratio", "value"),
"nupl": ("nupl", "value"),
"drawdown": ("drawdown", "value"),
"hash_ribbons": ("hash_ribbons", "value"),
"sopr": ("sopr", "value"),
"sellside_risk": ("sellside_risk", "value"),
"active_address_momentum": ("active_address_momentum", "value"),
"txcount_momentum": ("txcount_momentum", "value"),
"nvt_price": ("nvt_price", "value"),
"vdd_multiple": ("vdd_multiple", "value"),
"lth_supply": ("lth_supply", "value"),
}
# Ratio-based metrics: compute from price vs reference
ratio_metrics = {
"price_vs_200w_sma": ("price", "200w_sma"),
"lth_realized_price": ("price", "lth_realized_price"),
}
if metric in direct_keys:
k, sub = direct_keys[metric]
val = cache.get(k, {})
if isinstance(val, dict):
return val.get(sub)
return val
elif metric in ratio_metrics:
price_key, ref_key = ratio_metrics[metric]
price_val = cache.get(price_key, {}).get("price") or cache.get(price_key, {}).get("value")
ref_val = cache.get(ref_key, {}).get("value")
if price_val and ref_val and ref_val > 0:
return ((price_val - ref_val) / ref_val) * 100
return None
def _get_historical_metric_raw(metric, index, date):
"""Get the raw value for a metric on a specific historical date."""
from backtesting.engine import _last_known_value
direct_keys = {
"fear_greed": "fear_greed",
"puell_multiple": "puell_multiple",
"mvrv_zscore": "mvrv_zscore",
"reserve_risk": "reserve_risk",
"rhodl_ratio": "rhodl_ratio",
"nupl": "nupl",
"drawdown": "drawdown",
"hash_ribbons": "hash_ribbons",
"sopr": "sopr",
"sellside_risk": "sellside_risk",
"active_address_momentum": "active_address_momentum",
"txcount_momentum": "txcount_momentum",
"nvt_price": "nvt_price",
"vdd_multiple": "vdd_multiple",
"lth_supply": "lth_supply",
}
if metric in direct_keys:
return _last_known_value(index.get(direct_keys[metric], {}), date)
# Ratio-based
if metric == "price_vs_200w_sma":
price_val = _last_known_value(index.get("btc_price", {}), date)
ref_val = _last_known_value(index.get("200w_sma", {}), date)
if price_val and ref_val and ref_val > 0:
return ((price_val - ref_val) / ref_val) * 100
if metric == "lth_realized_price":
price_val = _last_known_value(index.get("btc_price", {}), date)
ref_val = _last_known_value(index.get("lth_realized_price", {}), date)
if price_val and ref_val and ref_val > 0:
return ((price_val - ref_val) / ref_val) * 100
return None
def _auto_metric_margin(metric, current_val):
"""Compute a reasonable similarity margin based on metric type and scale."""
margins = {
"fear_greed": 5.0,
"puell_multiple": 0.15,
"mvrv_zscore": 0.5,
"reserve_risk": 0.002,
"rhodl_ratio": 300,
"nupl": 0.1,
"drawdown": 8.0,
"sopr": 0.02,
"sellside_risk": 0.001,
"active_address_momentum": 0.05,
"txcount_momentum": 0.05,
"nvt_price": 5000,
"vdd_multiple": 0.15,
"price_vs_200w_sma": 10.0,
"lth_realized_price": 10.0,
}
if metric in margins:
return margins[metric]
# Fallback: 15% of current value
return abs(current_val) * 0.15 if current_val != 0 else 1.0
def _compute_day_forward_returns(price_lookup, date):
"""Compute forward returns for a single date."""
from datetime import datetime as _dt, timedelta as _td
p0 = price_lookup.get(date)
if p0 is None or p0 <= 0:
return {}
r = {}
dt = _dt.strptime(date, "%Y-%m-%d")
for days in [30, 90, 180, 365]:
future = (dt + _td(days=days)).strftime("%Y-%m-%d")
pf = price_lookup.get(future)
if pf is not None:
r[f"{days}d"] = round(((pf - p0) / p0) * 100, 2)
return r
# ── Backtest HTML Page ─────────────────────────────────────────────────
BACKTEST_HTML = """
""" + SHARED_HEAD + """
Historical Backtest — Bitcoin Accumulation Zone Monitor
₿ Accumulation Zone Monitor
""" + NAV_HTML + """
Historical Data Required
Scrape full historical time series from LookIntoBitcoin charts, CoinGecko, and Fear & Greed Index. This takes several minutes (10+ charts to scrape).