bizzle/btc-accumulation-monitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: publish historical metric coverage

Browse Source
This commit is contained in:
Hermes Agent
2026-07-26 23:07:24 +00:00
parent 510b2587ca
commit 661579abf9
3 changed files with 192 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backtesting/engine.py
+127 -12
View File
@@ -54,6 +54,22 @@ RATIO_SCORERS = {
},
}
BACKTEST_METRIC_PANEL = tuple(METRIC_SCORERS) + tuple(RATIO_SCORERS) + ("drawdown",)
METRIC_MAX_AGE_DAYS = {
"fear_greed": 2,
"puell_multiple": 7,
"mvrv_zscore": 7,
"reserve_risk": 7,
"rhodl_ratio": 7,
"nupl": 7,
"btc_price": 3,
"btc_price_coingecko": 3,
"btc_price_sma": 3,
"btc_price_lth": 3,
"200w_sma": 7,
"lth_realized_price": 7,
}
DRAWDOWN_RANGES = _THRESH.get("drawdown", {}).get("ranges", [[60, None, 10], [40, 60, 8], [25, 40, 6], [15, 25, 4], [5, 15, 2], [None, 5, 0]])
@@ -90,8 +106,8 @@ def _get_all_dates(index):
return sorted(all_dates)
def _last_known_value(lookup, date, max_lookback=30):
"""Get value for date, or most recent prior value within lookback window."""
def _last_known_value(lookup, date, max_lookback=0):
"""Get value for date, or a prior value within an explicit lookback."""
if date in lookup:
return lookup[date]
d = datetime.strptime(date, "%Y-%m-%d")
@@ -102,6 +118,17 @@ def _last_known_value(lookup, date, max_lookback=30):
return None
def _metric_observation(lookup, date, metric_key):
"""Return value, source date, and age under a metric-specific freshness rule."""
max_age = METRIC_MAX_AGE_DAYS.get(metric_key, 0)
target = datetime.strptime(date, "%Y-%m-%d")
for age in range(max_age + 1):
source_date = (target - timedelta(days=age)).strftime("%Y-%m-%d")
if source_date in lookup:
return lookup[source_date], source_date, age
return None, None, None
def _compute_ath_series(price_lookup, dates):
"""Compute running ATH and drawdown for each date."""
ath = 0
@@ -196,6 +223,58 @@ _BT_ML_KEY_MAP = {
}
def _common_panel_current_score(scored, ml_weights=None):
"""Recompute the current score using only metrics present historically."""
by_key = {
metric.get("key"): metric.get("score")
for metric in scored.get("metrics", [])
if metric.get("key") in BACKTEST_METRIC_PANEL and metric.get("score") is not None
}
available_keys = [key for key in BACKTEST_METRIC_PANEL if key in by_key]
coverage = {
"available_count": len(available_keys),
"panel_count": len(BACKTEST_METRIC_PANEL),
"available_keys": available_keys,
}
if not available_keys:
return None, coverage
if ml_weights:
weighted = [
(by_key[key], ml_weights.get(_BT_ML_KEY_MAP[key], 0.0))
for key in available_keys
]
weight_total = sum(weight for _, weight in weighted)
if weight_total > 0:
return round(sum(score * weight for score, weight in weighted) / weight_total * 10, 1), coverage
return round(sum(by_key[key] for key in available_keys) / len(available_keys) * 10, 1), coverage
def _backtest_data_quality_metadata(metric_counts):
"""Describe historical panel, coverage, and freshness assumptions."""
coverage = {
"minimum_metrics": min(metric_counts),
"maximum_metrics": max(metric_counts),
"average_metrics": round(sum(metric_counts) / len(metric_counts), 1),
"panel_count": len(BACKTEST_METRIC_PANEL),
} if metric_counts else {
"minimum_metrics": 0,
"maximum_metrics": 0,
"average_metrics": 0,
"panel_count": len(BACKTEST_METRIC_PANEL),
}
return {
"metric_panel": {
"id": "historical-common-v1",
"keys": list(BACKTEST_METRIC_PANEL),
"count": len(BACKTEST_METRIC_PANEL),
},
"coverage": coverage,
"staleness_days": dict(METRIC_MAX_AGE_DAYS),
}
def score_day(date, index, drawdowns, ml_weights=None):
"""Score a single day using all available metrics. Returns (composite_score, details, n_metrics).
@@ -207,29 +286,47 @@ def score_day(date, index, drawdowns, ml_weights=None):
# Simple range-based metrics
for metric_key, cfg in METRIC_SCORERS.items():
val = _last_known_value(index.get(metric_key, {}), date)
val, observed_date, age_days = _metric_observation(
index.get(metric_key, {}), date, metric_key
)
if val is not None:
s = _score_range(val, cfg["ranges"])
if s is not None:
scores.append(s)
details[metric_key] = {"value": val, "score": s, "raw": val}
details[metric_key] = {
"value": val,
"score": s,
"raw": val,
"observed_date": observed_date,
"age_days": age_days,
}
# Ratio-based metrics (price vs reference)
for metric_key, cfg in RATIO_SCORERS.items():
price_val = _last_known_value(index.get(cfg["price_key"], {}), date)
# Try alternate price keys
price_val, price_date, price_age = _metric_observation(
index.get(cfg["price_key"], {}), date, cfg["price_key"]
)
# Try alternate price keys, each with an explicit freshness rule.
if price_val is None:
for pk in ["btc_price_coingecko", "btc_price_sma", "btc_price_lth"]:
price_val = _last_known_value(index.get(pk, {}), date)
price_val, price_date, price_age = _metric_observation(index.get(pk, {}), date, pk)
if price_val is not None:
break
ref_val = _last_known_value(index.get(cfg["ref_key"], {}), date)
ref_val, ref_date, ref_age = _metric_observation(
index.get(cfg["ref_key"], {}), date, cfg["ref_key"]
)
if price_val is not None and ref_val is not None and ref_val > 0:
pct_above = ((price_val - ref_val) / ref_val) * 100
s = _score_range(pct_above, cfg["ranges"])
if s is not None:
scores.append(s)
details[metric_key] = {"value": pct_above, "score": s, "raw": pct_above}
details[metric_key] = {
"value": pct_above,
"score": s,
"raw": pct_above,
"observed_date": min(price_date, ref_date),
"age_days": max(price_age, ref_age),
}
# Drawdown
dd = drawdowns.get(date)
@@ -339,6 +436,7 @@ def run_backtest(ml_mode=False):
# Build an explicit evaluation plan. Fold-specific validation weights are OOS;
# final weights fitted on full history are never represented as OOS.
ml_plan = None
ml_artifact = None
ml_artifact_status = None
if ml_mode:
ml_artifact, ml_artifact_status = _load_ml_artifact()
@@ -447,23 +545,30 @@ def run_backtest(ml_mode=False):
all_scores_list = [d["score"] for d in daily_scores]
all_scores_list.sort()
# Get current score from cache
# Get current score from cache, recomputed on the common historical panel.
current_score = None
current_price = None
current_coverage = None
if os.path.exists(CACHE_PATH):
try:
with open(CACHE_PATH) as f:
cache = json.load(f)
scored = cache.get("_scored", {})
current_score = scored.get("composite_score")
current_ml_weights = ml_artifact.get("weights") if ml_mode and ml_artifact else None
current_score, current_coverage = _common_panel_current_score(scored, current_ml_weights)
current_price = cache.get("price", {}).get("price")
except Exception:
pass
# If no cache, use latest daily score
# If no comparable cache panel is available, use latest historical score.
if current_score is None and daily_scores:
current_score = daily_scores[-1]["score"]
current_price = daily_scores[-1].get("price")
current_coverage = {
"available_count": daily_scores[-1]["n_metrics"],
"panel_count": len(BACKTEST_METRIC_PANEL),
"available_keys": list(daily_scores[-1].get("metric_values", {})),
}
current_context = None
if current_score is not None:
@@ -523,6 +628,12 @@ def run_backtest(ml_mode=False):
current_context = {
"current_score": current_score,
"current_price": current_price,
"score_version": SCORE_VERSION,
"metric_panel_id": "historical-common-v1",
"coverage": current_coverage,
"current_weighting_source": (
"final_full_history_weights" if ml_mode and ml_artifact else "equal_weight"
),
"percentile": percentile,
"comparable_days": len(comparable),
"avg_1yr_return": avg_1yr,
@@ -583,9 +694,13 @@ def run_backtest(ml_mode=False):
"artifact": ml_artifact_status,
}
data_quality = _backtest_data_quality_metadata([day["n_metrics"] for day in daily_scores])
result = {
"date_range": {"start": daily_scores[0]["date"], "end": daily_scores[-1]["date"]},
"total_days_scored": len(daily_scores),
"metric_panel": data_quality["metric_panel"],
"coverage": data_quality["coverage"],
"staleness_days": data_quality["staleness_days"],
"bracket_stats": bracket_stats,
"signal_events": signal_events,
"current_context": current_context,