""" TDD tests for strategy_mlp_scores.generate_signals. Core guarantee under test (linear degeneracy): an MLP constructed to be an (arbitrarily good) linear approximation of the existing perceptron score must reproduce the existing strategy's behaviour — near-perfect score correlation and identical entry/exit bars when thresholds are scaled by the same factor. This pins the new module's crossunder/trailing/position logic to the old one. """ import os import numpy as np import pandas as pd import pytest import strategies.strategy_activation_scores as perceptron from strategies.strategy_mlp_scores import ( FEATURE_COLS, FEATURE_WEIGHT_PARAMS, _score_to_signals, generate_signals, save_mlp_artifact, ) DATA_CSV = os.path.join(os.path.dirname(__file__), "..", "data", "mlp", "COINBASE_BTCUSD, 360.csv") # Small linearisation constants: tanh(eps*s) ~ eps*s with relative error eps^2/3. EPS = 1e-3 DELTA = 1e-3 # output-layer scale; mlp_score ~ DELTA * perceptron_score @pytest.fixture(scope="module") def df_real(): df = pd.read_csv(DATA_CSV) df.columns = df.columns.str.lower().str.strip() df["time"] = pd.to_datetime(df["time"], utc=True).dt.tz_localize(None) # The 49-column compatibility artifact predates the MLP repurposing of # qqq_spy/fear-greed slots to SOPR/CVD. Keep this fixture on the original # perceptron sources so the linear-degeneracy contract remains meaningful. df["sopr_norm"] = df.get("qqq_spy_roc_sign", 0.0) df["cvd_norm"] = df.get("fear_greed_norm", 0.0) # Recent 3000 bars: enough threshold crossings, fast to run. return df.tail(3000).reset_index(drop=True) def _perceptron_weights(seed=11): """Random nonzero weight per feature for the 49 features shared with the perceptron. Uses only the params that the perceptron strategy actually reads (config.WEIGHT_COLS). bb_pct_b_norm (col 49) is MLP-only and excluded so both strategies normalise by the same denominator — keeping the linear-degeneracy equivalence intact. """ import config n = len(config.WEIGHT_COLS) # 49 rng = np.random.default_rng(seed) w = rng.uniform(-100, 100, size=n) w[np.abs(w) < 5.0] = 5.0 # keep every weight clearly nonzero w_params = dict(zip(config.WEIGHT_COLS, w)) artifact_w = np.array([ w_params[param] for param in FEATURE_WEIGHT_PARAMS if param in w_params ]) return artifact_w, w_params def _degenerate_artifact(tmp_path, w): """[N, 2, 1] MLP that linearises to DELTA * perceptron score. Uses the same N features as the perceptron (config.WEIGHT_COLS) so normalisation denominators match. bb_pct_b_norm (col 49) is excluded — the artifact saves its own feature_cols list so generate_signals uses the right subset. Hidden unit 0 = tanh(EPS * x.w_hat) ~ EPS * s (w_hat = w / sum|w|, s in [-1,1]) Output = tanh((DELTA/EPS) * h0) ~ DELTA * s => mlp_score ~ DELTA * 1000 * s = DELTA * perceptron_score """ import config feat_cols = [FEATURE_COLS[i] for i, p in enumerate(FEATURE_WEIGHT_PARAMS) if p in set(config.WEIGHT_COLS)] n = len(feat_cols) assert n == len(w), f"weight vector length {len(w)} != feature subset length {n}" w_hat = w / np.sum(np.abs(w)) W1 = np.zeros((2, n)) W1[0, :] = EPS * w_hat b1 = np.zeros(2) W2 = np.array([[DELTA / EPS, 0.0]]) b2 = np.zeros(1) path = str(tmp_path / "mlp_weights_DEGEN_6H.json") save_mlp_artifact(path, [(W1, b1), (W2, b2)], asset="TEST", timeframe="6H", feature_cols=feat_cols) return path THRESHOLDS = { "i_long_entry_activation_threshold": 100.0, "i_long_exit_activation_threshold": 140.0, "i_long_exit_activation_confirmation_threshold": 30.0, "i_use_long_exit_confirmation": 1.0, "i_use_long_entry_confirmation": False, } def _scaled_thresholds(): out = dict(THRESHOLDS) for k in ("i_long_entry_activation_threshold", "i_long_exit_activation_threshold", "i_long_exit_activation_confirmation_threshold"): out[k] = THRESHOLDS[k] * DELTA return out class TestLinearDegeneracy: def test_score_correlation_and_identical_trades(self, tmp_path, df_real): w, w_params = _perceptron_weights() artifact = _degenerate_artifact(tmp_path, w) old = perceptron.generate_signals(df_real.copy(), **{**w_params, **THRESHOLDS}) new = generate_signals(df_real.copy(), mlp_weights_file=artifact, **_scaled_thresholds()) corr = np.corrcoef(old["activation_score"], new["activation_score"])[0, 1] assert corr > 0.999999 np.testing.assert_array_equal( old["execute_entry"].values, new["execute_entry"].values) np.testing.assert_array_equal( old["execute_exit"].values, new["execute_exit"].values) np.testing.assert_array_equal(old["position"].values, new["position"].values) def test_trailing_stop_path_equivalence(self, tmp_path, df_real): w, w_params = _perceptron_weights(seed=12) artifact = _degenerate_artifact(tmp_path, w) trail = {"i_trailing_stop_threshold": 10.0} old = perceptron.generate_signals(df_real.copy(), **{**w_params, **THRESHOLDS, **trail}) new = generate_signals(df_real.copy(), mlp_weights_file=artifact, **{**_scaled_thresholds(), **trail}) np.testing.assert_array_equal(old["position"].values, new["position"].values) np.testing.assert_array_equal( old["execute_entry"].values, new["execute_entry"].values) np.testing.assert_array_equal( old["execute_exit"].values, new["execute_exit"].values) class TestParamValidation: def test_missing_weights_file_raises(self, df_real): with pytest.raises(ValueError, match="mlp_weights_file"): generate_signals(df_real.copy(), **THRESHOLDS) def test_mismatched_feature_cols_raises(self, tmp_path, df_real): path = str(tmp_path / "mlp_weights_BAD_6H.json") layers = [(np.zeros((2, 3)), np.zeros(2)), (np.zeros((1, 2)), np.zeros(1))] save_mlp_artifact(path, layers, feature_cols=["a", "b", "c"]) with pytest.raises(ValueError, match="feature_cols"): generate_signals(df_real.copy(), mlp_weights_file=path, **THRESHOLDS) class TestPineTimeGate: def test_pre_start_crossunder_does_not_open_position(self): df = pd.DataFrame({ "time": pd.to_datetime([ "2017-11-30 16:00", "2017-12-01 00:00", "2017-12-01 08:00", "2017-12-01 16:00", ]), "activation_score": [200.0, 50.0, 200.0, 50.0], "close": [100.0, 101.0, 102.0, 103.0], "high": [100.0, 101.0, 102.0, 103.0], }) params = { "i_long_entry_activation_threshold": 100.0, "i_long_exit_activation_threshold": 900.0, "i_long_exit_activation_confirmation_threshold": -900.0, "i_use_long_exit_confirmation": 1.0, "i_use_long_entry_confirmation": False, "_pine_time_start": "2017-12-01 08:00", } signals = _score_to_signals( df.copy(), params, np.zeros(len(df)), df["close"].to_numpy(np.float64), df["high"].to_numpy(np.float64), ) assert signals["execute_entry"].tolist() == [False, False, False, True] assert signals["position"].tolist() == [0, 0, 0, 1] class TestFeatureWeightParams: def test_one_param_per_feature(self): # FEATURE_COLS is 55 for the bb50_bull arch (5 new on-chain signals added). # FEATURE_WEIGHT_PARAMS covers the 50 base features used by the perceptron # strategy; the 5 new MLP-only columns (sopr/cvd/rvol/active1y/hr) have no # separate weight param and sit at cols 32,46,50,51,52,53,54. assert len(FEATURE_WEIGHT_PARAMS) == 50 assert len(FEATURE_COLS) == 55 assert all(p.startswith("i_w_") for p in FEATURE_WEIGHT_PARAMS) assert len(set(FEATURE_WEIGHT_PARAMS)) == 50 def test_config_weight_cols_is_subset(self): # Every perceptron weight param must have an MLP counterpart. # FEATURE_WEIGHT_PARAMS may have extra MLP-only params (e.g. i_w_bb_pct_b). import config assert set(config.WEIGHT_COLS).issubset(set(FEATURE_WEIGHT_PARAMS))