diff --git a/backtesting/engine.py b/backtesting/engine.py index 1ce03fb..6b1dd86 100644 --- a/backtesting/engine.py +++ b/backtesting/engine.py @@ -7,6 +7,8 @@ import sys from collections import defaultdict from datetime import datetime, timedelta +from scoring.policy import SCORE_BRACKETS, SCORE_VERSION, score_in_bracket + log = logging.getLogger(__name__) BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) @@ -16,14 +18,7 @@ HISTORY_PATH = os.path.join(BASE_DIR, "data", "history.json") CACHE_PATH = os.path.join(BASE_DIR, "data", "cache.json") # Score brackets matching the dashboard assessment levels -BRACKETS = [ - (0, 20, "Extreme Caution"), - (21, 40, "Caution"), - (41, 55, "Neutral"), - (56, 70, "Moderate Opportunity"), - (71, 85, "Strong Accumulation"), - (86, 100, "Extreme Accumulation"), -] +BRACKETS = SCORE_BRACKETS # Scoring thresholds — load from config/thresholds.json (single source of truth) import os as _os @@ -322,7 +317,7 @@ def run_backtest(ml_mode=False): # --- Bracket statistics --- bracket_stats = [] for low, high, label in BRACKETS: - days_in = [d for d in daily_scores if low <= d["score"] <= high] + days_in = [d for d in daily_scores if score_in_bracket(d["score"], (low, high, label))] if not days_in: bracket_stats.append({ "range": f"{low}-{high}", "label": label, "days": 0, @@ -506,6 +501,7 @@ def run_backtest(ml_mode=False): "current_context": current_context, "chart_data": chart_data, "ml_mode": ml_mode, + "score_version": SCORE_VERSION, "computed_at": datetime.utcnow().isoformat() + "Z", } diff --git a/ml/optimizer.py b/ml/optimizer.py index 7fea652..67c0ca5 100644 --- a/ml/optimizer.py +++ b/ml/optimizer.py @@ -26,6 +26,8 @@ from sklearn.metrics import ( from sklearn.model_selection import TimeSeriesSplit from sklearn.preprocessing import StandardScaler +from scoring.policy import SCORE_BRACKETS, score_in_bracket + logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", @@ -117,14 +119,7 @@ INTERACTION_FEATURES = ["mvrv_x_nupl", "puell_x_reserve"] CYCLE_FEATURES = ["days_since_ath"] FEATURE_COLS = SCORE_FEATURES + RAW_FEATURES + DELTA_FEATURES + INTERACTION_FEATURES + CYCLE_FEATURES -BRACKETS = [ - (0, 20, "Extreme Caution"), - (21, 40, "Caution"), - (41, 55, "Neutral"), - (56, 70, "Moderate Opportunity"), - (71, 85, "Strong Accumulation"), - (86, 100, "Extreme Accumulation"), -] +BRACKETS = SCORE_BRACKETS def _row_date(row): @@ -526,7 +521,7 @@ def _composite_score(row, mode, ml_weights=None): def _summarize_brackets(scored_rows, score_key): results = [] for low, high, label in BRACKETS: - days_in = [r for r in scored_rows if low <= r[score_key] <= high] + days_in = [r for r in scored_rows if score_in_bracket(r[score_key], (low, high, label))] if not days_in: results.append({ "range": f"{low}-{high}", "label": label, diff --git a/scoring/engine.py b/scoring/engine.py index 8ee41df..c8ea9d5 100644 --- a/scoring/engine.py +++ b/scoring/engine.py @@ -4,6 +4,8 @@ import json import os import logging +from scoring.policy import SCORE_VERSION, assessment_for_score + log = logging.getLogger(__name__) THRESHOLDS_PATH = os.path.join( @@ -544,19 +546,7 @@ def score_all(metrics): else: composite = 0 - # Assessment text — calibrated for cycle-aware scoring - if composite >= 80: - assessment = "EXTREME ACCUMULATION ZONE" - elif composite >= 65: - assessment = "STRONG ACCUMULATION ZONE" - elif composite >= 50: - assessment = "MODERATE OPPORTUNITY" - elif composite >= 35: - assessment = "NEUTRAL" - elif composite >= 20: - assessment = "CAUTION — OVERHEATED" - else: - assessment = "EXTREME CAUTION" + assessment = assessment_for_score(composite) return { "metrics": results, @@ -564,6 +554,7 @@ def score_all(metrics): "assessment": assessment, "scored_count": len(valid_scores), "total_count": len(results), + "score_version": SCORE_VERSION, } @@ -646,19 +637,7 @@ def score_all_ml(metrics): m["ml_weight"] = round(effective_weight, 4) m["ml_contribution"] = round(m["score"] * effective_weight * 10, 2) - # Assessment text (same thresholds as classic) - if composite >= 80: - assessment = "EXTREME ACCUMULATION ZONE" - elif composite >= 65: - assessment = "STRONG ACCUMULATION ZONE" - elif composite >= 50: - assessment = "MODERATE OPPORTUNITY" - elif composite >= 35: - assessment = "NEUTRAL" - elif composite >= 20: - assessment = "CAUTION — OVERHEATED" - else: - assessment = "EXTREME CAUTION" + assessment = assessment_for_score(composite) return { "metrics": results, @@ -669,4 +648,5 @@ def score_all_ml(metrics): "ml_mode": True, "classic_score": classic["composite_score"], "ml_weight_total": round(weight_total, 4), + "score_version": SCORE_VERSION, } diff --git a/scoring/policy.py b/scoring/policy.py new file mode 100644 index 0000000..ba19ba0 --- /dev/null +++ b/scoring/policy.py @@ -0,0 +1,35 @@ +"""Canonical score version, brackets, and assessment semantics.""" + +SCORE_VERSION = "accumulation-score-v2" + +# Half-open intervals [low, high), except the final bracket includes 100. +# Keep labels canonical because they are persisted in live and backtest output. +SCORE_BRACKETS = [ + (0, 20, "EXTREME CAUTION"), + (20, 35, "CAUTION — OVERHEATED"), + (35, 50, "NEUTRAL"), + (50, 65, "MODERATE OPPORTUNITY"), + (65, 80, "STRONG ACCUMULATION ZONE"), + (80, 100, "EXTREME ACCUMULATION ZONE"), +] + + +def score_in_bracket(score, bracket): + """Return whether a 0-100 score belongs to a canonical bracket.""" + low, high, _ = bracket + if not 0 <= score <= 100: + return False + return low <= score < high or (high == 100 and score == 100) + + +def bracket_for_score(score): + """Return the one canonical bracket for a 0-100 score.""" + for bracket in SCORE_BRACKETS: + if score_in_bracket(score, bracket): + return bracket + raise ValueError(f"score must be between 0 and 100, got {score!r}") + + +def assessment_for_score(score): + """Return the canonical assessment label for a score.""" + return bracket_for_score(score)[2] diff --git a/tests/test_ml_optimizer_validation.py b/tests/test_ml_optimizer_validation.py index 365f01f..726a848 100644 --- a/tests/test_ml_optimizer_validation.py +++ b/tests/test_ml_optimizer_validation.py @@ -66,9 +66,9 @@ def test_run_out_of_sample_comparison_scores_only_validation_rows_with_fold_weig assert sum(bucket["days"] for bucket in comparison["ml_weighted"]) == 2 assert sum(bucket["days"] for bucket in comparison["equal_weight"]) == 2 - extreme_ml = next(bucket for bucket in comparison["ml_weighted"] if bucket["label"] == "Extreme Accumulation") + extreme_ml = next(bucket for bucket in comparison["ml_weighted"] if bucket["label"] == "EXTREME ACCUMULATION ZONE") assert extreme_ml["days"] == 2 assert extreme_ml["avg_365d"] == 110.0 - caution_equal = next(bucket for bucket in comparison["equal_weight"] if bucket["label"] == "Caution") + caution_equal = next(bucket for bucket in comparison["equal_weight"] if bucket["label"] == "CAUTION — OVERHEATED") assert caution_equal["days"] == 2 diff --git a/tests/test_score_policy.py b/tests/test_score_policy.py new file mode 100644 index 0000000..a641c1f --- /dev/null +++ b/tests/test_score_policy.py @@ -0,0 +1,38 @@ +import pytest + +from backtesting import engine as backtest_engine +from ml import optimizer +from scoring import engine as scoring_engine +from scoring import policy + + +def test_score_brackets_are_contiguous_and_shared_by_all_scoring_paths(): + assert backtest_engine.BRACKETS is policy.SCORE_BRACKETS + assert optimizer.BRACKETS is policy.SCORE_BRACKETS + + for left, right in zip(policy.SCORE_BRACKETS, policy.SCORE_BRACKETS[1:]): + assert left[1] == right[0] + + for tenth in range(0, 1001): + score = tenth / 10 + matches = [bracket for bracket in policy.SCORE_BRACKETS if policy.score_in_bracket(score, bracket)] + assert len(matches) == 1, f"score {score} matched {matches}" + assert policy.assessment_for_score(score) == matches[0][2] + + +@pytest.mark.parametrize( + ("score", "assessment"), + [ + (0, "EXTREME CAUTION"), + (19.999, "EXTREME CAUTION"), + (20, "CAUTION — OVERHEATED"), + (35, "NEUTRAL"), + (50, "MODERATE OPPORTUNITY"), + (65, "STRONG ACCUMULATION ZONE"), + (80, "EXTREME ACCUMULATION ZONE"), + (100, "EXTREME ACCUMULATION ZONE"), + ], +) +def test_assessment_boundaries_match_live_scoring(score, assessment): + assert policy.assessment_for_score(score) == assessment + assert scoring_engine.assessment_for_score(score) == assessment