import numpy as np import pandas as pd import pytest from strategies.mlp_learned_exit import (ValueHead, continuation_target, fit_head, market_features, replay, state_features, training_records) from strategies.strategy_mlp_scores import _score_to_signals def frame(n=30): return pd.DataFrame({'time': pd.date_range('2020-01-01', periods=n, freq='6h'), 'close': np.ones(n)*100, 'high': np.ones(n)*100, 'activation_score': np.resize([200., 0., 0., 0.], n), 'mvrv_regime': np.zeros(n), 'rvol_norm': np.zeros(n), 'fed_net_liq_sign': np.zeros(n)}) def ops(data, **params): return _score_to_signals(data.copy(), params, np.ones(len(data)), data.close.to_numpy(), data.high.to_numpy(), return_opportunities=True) def head(value): return ValueHead('state', np.zeros(7), np.ones(7), np.zeros(7), value) def test_constant_positive_head_holds_until_cap_and_negative_exits(): data = frame() held, decisions = replay(data, ops(data), start='2020-01-01', end='2020-02-01', head=head(.01), mode='veto') baseline, _ = replay(data, ops(data), start='2020-01-01', end='2020-02-01', head=head(-.01), mode='veto') assert np.flatnonzero(held.execute_exit)[0] == 14 assert np.flatnonzero(baseline.execute_exit)[0] == 2 assert next(d for d in decisions if d['action']=='exit')['reason']=='extension_cap' def test_stop_cannot_be_vetoed(): data = frame() data.loc[2:, 'close'] = 80 result, decisions = replay(data, ops(data, i_trailing_stop_threshold=10), start='2020-01-01', end='2020-02-01', head=head(.01), mode='veto') assert result.execute_exit.iloc[2] assert decisions[0]['reason']=='trailing_stop' @pytest.mark.parametrize('window', [1, 4]) @pytest.mark.parametrize('trail', [0, 10]) def test_replay_abstention_matches_canonical(window, trail): data = frame(200) rng = np.random.default_rng(42) data.activation_score = rng.normal(0, 200, len(data)) data.close = rng.uniform(80, 100, len(data)) params = {'i_exit_score_window': window, 'i_trailing_stop_threshold': trail, '_pine_time_start': '2020-01-05', '_pine_time_end': '2020-02-01 23:59:59'} canonical = _score_to_signals(data.copy(), params, np.ones(len(data)), data.close.to_numpy(), data.high.to_numpy()) actual, _ = replay(data, ops(data, **params), start='2020-01-05', end='2020-02-01', mode='manage') for name in ('position', 'execute_entry', 'execute_exit'): np.testing.assert_array_equal(actual[name], canonical[name]) def test_target_wealth_costs_and_stop_endpoint(): value, end = continuation_target(np.array([100., 110, 121, 121, 121]), np.array([100., 110, 121, 121, 121]), 0, -2, 100., 0.) assert end == 4 assert value == pytest.approx(1.21*.995-.995) value, end = continuation_target(np.array([100., 80, 100, 100, 100]), np.ones(5)*100, 0, -2, 100., .1) assert end == 1 assert value == pytest.approx(-.2) @pytest.mark.parametrize('model_type', ['ridge', 'tree']) def test_purge_and_training_only_scaler_ignore_future_records(model_type): times = pd.date_range('2020-01-01', periods=50) records = [{'entry_bar': i, 'label_end': i+1, 'market': np.ones(7)*i, 'state': np.ones(7)*i, 'target': .01*i} for i in range(40)] fitted, audit = fit_head(records, times, times[30], 'combined', model_type) assert audit['trades']==29 assert pd.Timestamp(audit['max_label_end']) < times[30] for r in records[29:]: r['target'] = -1000. r['market'][:] = 1000. second, _ = fit_head(records, times, times[30], 'combined', model_type) np.testing.assert_array_equal(fitted.coefficients, second.coefficients) np.testing.assert_array_equal(fitted.mean, second.mean) assert fitted.tree == second.tree def test_minimum_independent_trades_not_sample_count(): records = [{'entry_bar': 0, 'label_end': 1, 'market': np.zeros(7), 'state': np.zeros(7), 'target': 0.}] * 1000 fitted, audit = fit_head(records, pd.date_range('2020-01-01', periods=3), '2020-01-03', 'state') assert fitted is None and audit['trades']==1 def test_prefix_causality_and_actual_candidate_state(): data = frame() full, actions = replay(data, ops(data), start='2020-01-01', end='2020-02-01', head=head(.01), mode='veto') prefix, _ = replay(data.iloc[:10].copy(), ops(data.iloc[:10]), start='2020-01-01', end='2020-02-01', head=head(.01), mode='veto') np.testing.assert_array_equal(full.position[:10], prefix.position) assert next(d for d in actions if d['bar']==10)['entry_bar']==1 np.testing.assert_array_equal(market_features(data)[:10], market_features(data.iloc[:10])) def test_training_rows_include_post_exit_state_and_purged_horizon(): data = frame() records = training_records(data, ops(data)) continued = next(r for r in records if r['entry_bar']==1 and r['bar']==10) assert continued['state'][0] == .09 assert continued['state'][3] == 8/12 assert all(r['bar'] < r['label_end'] <= r['bar']+4 for r in records) def test_position_features_and_missing_market_inputs(): state = state_features(10, 0, 100, 120, 110, -1) assert state[1] == pytest.approx(.1) assert state[2] == pytest.approx(110/120-1) with pytest.raises(ValueError, match='Missing'): market_features(frame().drop(columns='rvol_norm')) def test_tree_portable_predictor_learns_nonlinear_state(): records = [{'entry_bar': i, 'label_end': i+1, 'market': np.zeros(7), 'state': np.array([float(i), 0, 0, 0, 0, 0, 0]), 'target': .01 if 10 <= i < 30 else -.01} for i in range(40)] fitted, _ = fit_head(records, pd.date_range('2020-01-01', periods=50), '2021-01-01', 'state', 'tree') assert fitted.predict(np.zeros(7), records[0]['state']) < 0 assert fitted.predict(np.zeros(7), records[20]['state']) > 0 assert fitted.predict(np.zeros(7), records[39]['state']) < 0 assert len(fitted.tree['left']) <= 7 def test_each_trade_keeps_equal_weight_when_identical_rows_repeat(): times = pd.date_range('2020-01-01', periods=50) records = [{'entry_bar': i, 'label_end': i+1, 'market': np.ones(7)*i, 'state': np.ones(7)*i, 'target': .01*i} for i in range(40)] first, _ = fit_head(records, times, '2021-01-01', 'state') second, _ = fit_head(records + [records[0]]*100, times, '2021-01-01', 'state') np.testing.assert_allclose(first.coefficients, second.coefficients, atol=1e-14) assert first.intercept == pytest.approx(second.intercept) def test_nonfinite_training_values_fail_closed(): times = pd.date_range('2020-01-01', periods=50) records = [{'entry_bar': i, 'label_end': i+1, 'market': np.zeros(7), 'state': np.zeros(7), 'target': float('nan')} for i in range(30)] with pytest.raises(ValueError, match='Nonfinite'): fit_head(records, times, '2021-01-01', 'state')