fix: remove leakage from legacy ML evaluation

config/best_config.json

@@ -27,6 +27,11 @@
       0.3,
       0.5
     ],
+    "return_scales_pct": [
+      10.0,
+      30.0,
+      60.0
+    ],
     "score_range": [
       0,
       100
@@ -61,7 +66,7 @@
     "rolling_test_size": 300,
     "walk_forward_windows": 5,
     "train_pct": 0.7,
-    "validation_pct": 0.15,
+    "validation_pct": 0.3,
     "test_pct": 0.15
   },
   "timeframe": "4h"

config/current_config.json

@@ -27,6 +27,11 @@
       0.3,
       0.5
     ],
+    "return_scales_pct": [
+      10.0,
+      30.0,
+      60.0
+    ],
     "score_range": [
       0,
       100
@@ -61,7 +66,7 @@
     "rolling_test_size": 300,
     "walk_forward_windows": 5,
     "train_pct": 0.7,
-    "validation_pct": 0.15,
+    "validation_pct": 0.3,
     "test_pct": 0.15
   },
   "timeframe": "4h"

config/initial_config.json

@@ -27,6 +27,11 @@
       0.3,
       0.5
     ],
+    "return_scales_pct": [
+      10.0,
+      30.0,
+      60.0
+    ],
     "score_range": [
       0,
       100
@@ -61,7 +66,7 @@
     "rolling_test_size": 300,
     "walk_forward_windows": 5,
     "train_pct": 0.7,
-    "validation_pct": 0.15,
+    "validation_pct": 0.3,
     "test_pct": 0.15
   },
   "timeframe": "4h"

ml_engine/train_and_backtest.py

@@ -190,30 +190,22 @@ def create_accumulation_target(df, config):
             fwd[i] = (close[i + period] - close[i]) / close[i] * 100
         forward_returns.append(fwd)
 
-    # Rank each forward return (percentile rank, 0-1)
+    # Convert each return to a deterministic 0-100 quality score. Global
-    # Higher rank = better buy point (higher future return)
+    # percentile ranks leak the distribution of future validation/test rows into
-    ranked = []
+    # earlier training labels; a fixed tanh transform is invariant to rows added
-    for fwd in forward_returns:
+    # outside the observation's own forward horizons.
-        valid_mask = ~np.isnan(fwd)
+    scales = tgt.get("return_scales_pct", [10.0, 30.0, 60.0])
-        ranks = np.full(n, np.nan)
+    if len(scales) != len(forward_periods) or any(scale <= 0 for scale in scales):
-        valid_vals = fwd[valid_mask]
+        raise ValueError("target.return_scales_pct must contain one positive scale per forward period")
-        if len(valid_vals) > 0:
-            from scipy.stats import rankdata
-            r = rankdata(valid_vals, method="average") / len(valid_vals)
-            ranks[valid_mask] = r
-        ranked.append(ranks)
 
-    # Weighted combination of ranks -> accumulation score (0-100)
     score = np.zeros(n)
     valid = np.ones(n, dtype=bool)
-    for r, w in zip(ranked, weights):
+    for fwd, weight, scale in zip(forward_returns, weights, scales):
-        nan_mask = np.isnan(r)
+        nan_mask = np.isnan(fwd)
         valid &= ~nan_mask
-        r_filled = np.where(nan_mask, 0, r)
+        quality = 50.0 + 50.0 * np.tanh(np.where(nan_mask, 0.0, fwd) / scale)
-        score += w * r_filled
+        score += weight * quality
 
-    # Scale to 0-100
-    score = score * 100
     score[~valid] = np.nan
 
     return pd.Series(score, index=df.index, name="target")
@@ -488,6 +480,27 @@ def apply_scaling_pca(X_train, X_val, X_test, config):
     return X_train, X_val, X_test, scaler, pca
 
 
+# ---------------------------------------------------------------------------
+# Walk-forward split helpers
+# ---------------------------------------------------------------------------
+
+def _max_forward_horizon(config):
+    """Return the longest forward-label horizon in candles."""
+    target = config.get("target", {})
+    key = "forward_periods_1h" if config.get("timeframe", "4h") == "1h" else "forward_periods_4h"
+    periods = target.get(key, [168, 720, 2160] if key.endswith("1h") else [42, 180, 540])
+    return max(int(period) for period in periods)
+
+
+def _purge_label_overlap(frame, horizon):
+    """Remove rows whose forward-return labels cross the next split boundary."""
+    if horizon <= 0:
+        return frame
+    if len(frame) <= horizon:
+        return frame.iloc[0:0]
+    return frame.iloc[:-horizon]
+
+
 # ---------------------------------------------------------------------------
 # Rolling Window Validation
 # ---------------------------------------------------------------------------
@@ -499,6 +512,7 @@ def rolling_window_train_test(df, feature_cols, config):
     test_size = training_cfg.get("rolling_test_size", 300)
     val_pct = training_cfg.get("validation_pct", 0.15)
     model_type = config.get("model_type", "xgboost")
+    purge_horizon = _max_forward_horizon(config)
 
     n = len(df)
     all_predictions = []  # list of (predicted_score, actual_score, close_price)
@@ -527,10 +541,16 @@ def rolling_window_train_test(df, feature_cols, config):
             start += test_size
             continue
 
-        # Split train into train/val
+        # Purge labels whose longest forward-return horizon overlaps the next
+        # split. Without this embargo, training and validation targets consume
+        # prices from the following validation/test partition.
         val_split = int(len(train_full) * (1.0 - val_pct))
-        train_df = train_full.iloc[:val_split]
+        train_df = _purge_label_overlap(train_full.iloc[:val_split], purge_horizon)
-        val_df = train_full.iloc[val_split:]
+        val_df = _purge_label_overlap(train_full.iloc[val_split:], purge_horizon)
+
+        if len(train_df) < 10 or len(val_df) < 1:
+            start += test_size
+            continue
 
         X_train = train_df[feature_cols].values
         y_train = train_df["target"].values
@@ -632,6 +652,7 @@ def walk_forward_train_test(df, feature_cols, config):
     n_windows = training_cfg.get("walk_forward_windows", 5)
     train_pct = training_cfg.get("train_pct", 0.7)
     val_pct = training_cfg.get("validation_pct", 0.15)
+    purge_horizon = _max_forward_horizon(config)
 
     n = len(df)
     window_size = n // n_windows
@@ -655,8 +676,8 @@ def walk_forward_train_test(df, feature_cols, config):
         train_end = int(wn * train_pct)
         val_end = int(wn * (train_pct + val_pct))
 
-        train_df = window_data.iloc[:train_end]
+        train_df = _purge_label_overlap(window_data.iloc[:train_end], purge_horizon)
-        val_df = window_data.iloc[train_end:val_end]
+        val_df = _purge_label_overlap(window_data.iloc[train_end:val_end], purge_horizon)
         test_df = window_data.iloc[val_end:]
 
         if len(test_df) < 10:
@@ -842,6 +863,46 @@ def _extract_feature_importances(model, n_features):
 # Results Compilation
 # ---------------------------------------------------------------------------
 
+def simulate_periodic_accumulation(predicted_scores, close_prices, buy_threshold, contribution=1.0):
+    """Compare DCA and signal strategies with equal periodic contributions.
+
+    Both strategies receive the same cash on every observation. DCA invests the
+    contribution immediately; the signal strategy retains cash until a buy
+    signal, then deploys its available balance. Terminal wealth includes cash.
+    """
+    scores = np.asarray(predicted_scores, dtype=float)
+    prices = np.asarray(close_prices, dtype=float)
+    if len(scores) != len(prices):
+        raise ValueError("predicted_scores and close_prices must have equal length")
+    if len(prices) == 0 or contribution <= 0 or np.any(prices <= 0):
+        raise ValueError("prices must be positive and contribution must be greater than zero")
+
+    dca_btc = float(np.sum(contribution / prices))
+    model_btc = 0.0
+    model_cash = 0.0
+    for score, price in zip(scores, prices):
+        model_cash += contribution
+        if score >= buy_threshold:
+            model_btc += model_cash / price
+            model_cash = 0.0
+
+    contributed = float(len(prices) * contribution)
+    terminal_price = float(prices[-1])
+    dca_terminal = dca_btc * terminal_price
+    model_terminal = model_btc * terminal_price + model_cash
+    improvement = (model_terminal - dca_terminal) / dca_terminal * 100 if dca_terminal else 0.0
+    return {
+        "dca_contributed": contributed,
+        "model_contributed": contributed,
+        "dca_btc": dca_btc,
+        "model_btc": model_btc,
+        "model_cash": model_cash,
+        "dca_terminal_value": dca_terminal,
+        "model_terminal_value": model_terminal,
+        "terminal_wealth_improvement_pct": improvement,
+    }
+
+
 def compile_results(predictions, per_window_cost_improvement,
                     fi_sum, fi_count, feature_cols, config):
     """Compile accumulation signal results into output JSON."""
@@ -896,6 +957,13 @@ def compile_results(predictions, per_window_cost_improvement,
         model_avg = dca_avg
         cost_basis_improvement = 0.0
 
+    portfolio = simulate_periodic_accumulation(
+        pred_scores,
+        close_prices,
+        buy_threshold=good_threshold,
+        contribution=1.0,
+    )
+
     # --- Signal Frequency ---
     signal_frequency = strong_buy_count / total_candles * 100 if total_candles > 0 else 0
 
@@ -946,7 +1014,14 @@ def compile_results(predictions, per_window_cost_improvement,
         quality_good = False
 
     return {
+        # Retained for backward compatibility; model selection uses the equal-
+        # capital terminal wealth metric below.
         "cost_basis_improvement_pct": round(cost_basis_improvement, 2),
+        "terminal_wealth_improvement_pct": round(portfolio["terminal_wealth_improvement_pct"], 2),
+        "model_terminal_value": round(portfolio["model_terminal_value"], 6),
+        "dca_terminal_value": round(portfolio["dca_terminal_value"], 6),
+        "model_cash": round(portfolio["model_cash"], 6),
+        "backtest_objective": "equal_periodic_contribution_terminal_wealth",
         "avg_cost_basis_model": round(model_avg, 2),
         "avg_cost_basis_dca": round(dca_avg, 2),
         "strong_buy_signal_count": strong_buy_count,
@@ -968,6 +1043,11 @@ def compile_results(predictions, per_window_cost_improvement,
 def _empty_results(per_window):
     return {
         "cost_basis_improvement_pct": 0.0,
+        "terminal_wealth_improvement_pct": 0.0,
+        "model_terminal_value": 0.0,
+        "dca_terminal_value": 0.0,
+        "model_cash": 0.0,
+        "backtest_objective": "equal_periodic_contribution_terminal_wealth",
         "avg_cost_basis_model": 0.0,
         "avg_cost_basis_dca": 0.0,
         "strong_buy_signal_count": 0,

orchestrator.py

@@ -28,7 +28,7 @@ MAC_MINI_HOST = "bizzle@bizzles-mac-mini-1"
 MAX_ITERATIONS = 50
 CONVERGENCE_WINDOW = 5
 CONVERGENCE_THRESHOLD = 0.01  # 1% improvement
-TARGET_COST_IMPROVEMENT = 20.0  # 20% cost basis improvement = exceptional
+TARGET_COST_IMPROVEMENT = 20.0  # Backward-compatible name: terminal wealth objective
 MIN_SIGNAL_COUNT = 30  # Minimum strong buy signals for valid results
 ML_TIMEOUT = 600  # 10 minutes
 
@@ -49,6 +49,11 @@ def log(msg, color=""):
     print(f"{C.DIM}[{ts}]{C.RESET} {color}{msg}{C.RESET}")
 
 
+def objective_score(results):
+    """Return the equal-capital portfolio objective used for model selection."""
+    return float(results.get("terminal_wealth_improvement_pct", 0.0))
+
+
 def run_cmd(cmd, timeout=120, check=True):
     """Run a shell command and return stdout."""
     result = subprocess.run(
@@ -160,11 +165,12 @@ def print_header():
 
 
 def print_results(results, iteration):
-    cost_imp = results.get("cost_basis_improvement_pct", 0)
+    objective = objective_score(results)
-    color = C.GREEN if cost_imp > 15 else C.YELLOW if cost_imp > 10 else C.RED
+    color = C.GREEN if objective > 15 else C.YELLOW if objective > 10 else C.RED
     print(f"""
 {C.BOLD}--- Iteration {iteration} Results ---{C.RESET}
-  Cost Improvement: {color}{C.BOLD}{cost_imp:.1f}%{C.RESET}
+  Terminal Wealth vs DCA: {color}{C.BOLD}{objective:.1f}%{C.RESET}
+  Legacy Cost Basis Delta: {results.get('cost_basis_improvement_pct', 0):.1f}%
   Avg Cost (Model): ${results.get('avg_cost_basis_model', 0):,.2f}
   Avg Cost (DCA):   ${results.get('avg_cost_basis_dca', 0):,.2f}
   Strong Signals:   {results.get('strong_buy_signal_count', 0)}
@@ -244,7 +250,7 @@ def main():
 
         print_results(results, iteration)
 
-        current_score = results.get("cost_basis_improvement_pct", 0)
+        current_score = objective_score(results)
         signal_count = results.get("strong_buy_signal_count", 0)
         is_best = current_score > best_score and signal_count >= MIN_SIGNAL_COUNT
 
@@ -252,12 +258,14 @@ def main():
             best_score = current_score
             with open(best_config_path, "w") as f:
                 json.dump(config, f, indent=2)
-            log(f"NEW BEST! Cost Improvement: {best_score:.1f}%", f"{C.BOLD}{C.GREEN}")
+            log(f"NEW BEST! Terminal Wealth Improvement: {best_score:.1f}%", f"{C.BOLD}{C.GREEN}")
 
         iter_data = {
             "iteration": iteration,
             "timestamp": datetime.now(timezone.utc).isoformat(),
             "cost_improvement": current_score,
+            "objective_improvement": current_score,
+            "objective": "equal_periodic_contribution_terminal_wealth",
             "avg_30d_return": results.get("avg_quality_score_strong_buy", 0),
             "avg_90d_return": results.get("pct_quality_strong_buy", 0),
             "signal_count": signal_count,
@@ -312,7 +320,7 @@ def main():
 ========================================================{C.RESET}
 
   Total Iterations:     {len(history)}
-  Best Cost Improvement: {C.BOLD}{best_score:.1f}%{C.RESET}
+  Best Terminal Wealth Improvement: {C.BOLD}{best_score:.1f}%{C.RESET}
   Best Config:          {best_config_path}
   Iteration Log:        {ITERATIONS_LOG}
 """)

tests/test_legacy_ml_validation.py

@@ -0,0 +1,103 @@
+import numpy as np
+import pandas as pd
+
+import orchestrator
+from ml_engine import train_and_backtest as legacy
+from ml_engine.train_and_backtest import create_accumulation_target
+
+
+def _frame(prices):
+    return pd.DataFrame({"close": prices})
+
+
+def test_accumulation_target_for_existing_row_is_invariant_to_unrelated_future_rows():
+    config = {
+        "timeframe": "4h",
+        "target": {
+            "forward_periods_4h": [1, 2, 3],
+            "weights": [0.2, 0.3, 0.5],
+            "return_scales_pct": [5, 10, 20],
+        },
+    }
+    base = _frame([100, 102, 104, 106, 108, 110, 112, 114])
+    extended = _frame([100, 102, 104, 106, 108, 110, 112, 114, 1000, 1, 2000])
+
+    base_target = create_accumulation_target(base, config)
+    extended_target = create_accumulation_target(extended, config)
+
+    assert np.isclose(base_target.iloc[0], extended_target.iloc[0])
+    assert 0 <= base_target.iloc[0] <= 100
+
+
+def test_rolling_validation_purges_forward_label_horizon_at_train_boundaries(monkeypatch):
+    rows = 200
+    frame = pd.DataFrame({
+        "feature": np.linspace(0, 1, rows),
+        "target": np.arange(rows, dtype=float) % 100,
+        "close": np.linspace(10_000, 20_000, rows),
+    })
+    observed = []
+
+    def fake_train(X_train, y_train, X_val, y_val, X_test, *args):
+        observed.append((len(X_train), len(X_val), len(X_test)))
+        return np.full(len(X_test), 50.0), np.array([1.0])
+
+    monkeypatch.setattr(legacy, "_train_and_predict_window", fake_train)
+    config = {
+        "model_type": "xgboost",
+        "target": {"forward_periods_4h": [1, 2, 3]},
+        "training": {
+            "rolling_train_size": 120,
+            "rolling_test_size": 40,
+            "validation_pct": 0.25,
+        },
+        "features": {"use_scaler": False, "use_pca": False},
+        "strategy": {},
+    }
+
+    legacy.rolling_window_train_test(frame, ["feature"], config)
+
+    assert observed[0] == (87, 27, 40)
+
+
+def test_periodic_accumulation_compares_equal_contributions_and_retains_cash():
+    result = legacy.simulate_periodic_accumulation(
+        predicted_scores=np.array([90, 10, 90, 10], dtype=float),
+        close_prices=np.array([100, 300, 100, 200], dtype=float),
+        buy_threshold=70,
+        contribution=100,
+    )
+
+    assert np.isclose(result["dca_contributed"], 400)
+    assert np.isclose(result["model_contributed"], 400)
+    assert np.isclose(result["model_cash"], 100)
+    assert np.isclose(result["model_btc"], 3)
+    assert result["model_terminal_value"] > result["dca_terminal_value"]
+
+
+def test_compiled_results_publish_equal_capital_terminal_wealth_metric():
+    predictions = [
+        {"predicted": score, "actual": 50.0, "close": price}
+        for score, price in zip([90, 10, 90, 10], [100, 300, 100, 200])
+    ]
+
+    result = legacy.compile_results(
+        predictions,
+        per_window_cost_improvement=[],
+        fi_sum=np.array([1.0]),
+        fi_count=1,
+        feature_cols=["feature"],
+        config={"model_type": "xgboost", "strategy": {"good_buy_threshold": 70}},
+    )
+
+    assert result["terminal_wealth_improvement_pct"] > 0
+    assert result["backtest_objective"] == "equal_periodic_contribution_terminal_wealth"
+
+
+def test_orchestrator_selects_models_by_terminal_wealth_not_cost_basis():
+    results = {
+        "terminal_wealth_improvement_pct": 4.5,
+        "cost_basis_improvement_pct": 99.0,
+    }
+
+    assert orchestrator.objective_score(results) == 4.5