104 lines
3.4 KiB
Python
104 lines
3.4 KiB
Python
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
|