diff --git a/config/best_config.json b/config/best_config.json index b6bbe12..e173c43 100644 --- a/config/best_config.json +++ b/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" diff --git a/config/current_config.json b/config/current_config.json index 6a476c3..8926ed4 100644 --- a/config/current_config.json +++ b/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" diff --git a/config/initial_config.json b/config/initial_config.json index b6bbe12..e173c43 100644 --- a/config/initial_config.json +++ b/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" diff --git a/ml_engine/train_and_backtest.py b/ml_engine/train_and_backtest.py index a991d68..415f7e5 100755 --- a/ml_engine/train_and_backtest.py +++ b/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) - # Higher rank = better buy point (higher future return) - ranked = [] - for fwd in forward_returns: - valid_mask = ~np.isnan(fwd) - ranks = np.full(n, np.nan) - valid_vals = fwd[valid_mask] - 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) + # Convert each return to a deterministic 0-100 quality score. Global + # percentile ranks leak the distribution of future validation/test rows into + # earlier training labels; a fixed tanh transform is invariant to rows added + # outside the observation's own forward horizons. + scales = tgt.get("return_scales_pct", [10.0, 30.0, 60.0]) + if len(scales) != len(forward_periods) or any(scale <= 0 for scale in scales): + raise ValueError("target.return_scales_pct must contain one positive scale per forward period") - # 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): - nan_mask = np.isnan(r) + for fwd, weight, scale in zip(forward_returns, weights, scales): + nan_mask = np.isnan(fwd) valid &= ~nan_mask - r_filled = np.where(nan_mask, 0, r) - score += w * r_filled + quality = 50.0 + 50.0 * np.tanh(np.where(nan_mask, 0.0, fwd) / scale) + 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] - val_df = train_full.iloc[val_split:] + train_df = _purge_label_overlap(train_full.iloc[:val_split], purge_horizon) + 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] - val_df = window_data.iloc[train_end:val_end] + train_df = _purge_label_overlap(window_data.iloc[:train_end], purge_horizon) + 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, diff --git a/orchestrator.py b/orchestrator.py index 4234a64..7c6219d 100755 --- a/orchestrator.py +++ b/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) - color = C.GREEN if cost_imp > 15 else C.YELLOW if cost_imp > 10 else C.RED + objective = objective_score(results) + 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} """) diff --git a/tests/test_legacy_ml_validation.py b/tests/test_legacy_ml_validation.py new file mode 100644 index 0000000..fbb10d2 --- /dev/null +++ b/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