"""Synthetic preregistration checks; these never load incumbent market data.""" from types import SimpleNamespace import numpy as np import pandas as pd import pytest from strategies import mlp_entry_loss as policy from strategies.mlp_accounting import net_returns from strategies.strategy_mlp_scores import _score_to_signals from tools import run_mlp_entry_loss_experiment as runner def replay(monkeypatch, close, entry, exit_flags=None, *, high=None, loss=.10): n = len(close) frame = pd.DataFrame(dict(time=pd.date_range('2020-01-01', periods=n, freq='6h'), close=close)) op = dict(entry=np.array(entry, bool), exit=np.array(exit_flags or [False]*n, bool), fallback=np.zeros(n, bool), score_exit=np.zeros(n), exit_threshold=210, time_ok=np.ones(n, bool), dead_zone=False, trail=.20) monkeypatch.setattr(policy, '_score_to_signals', lambda *a, **kw: op) return policy.entry_loss_signals(frame, {}, np.zeros(n), np.array(close), np.array(high or close), loss_fraction=loss) def test_equality_first_held_bar_and_no_same_bar_reentry(monkeypatch): trace = replay(monkeypatch, [100, 90, 95], [True, True, True]) assert trace.execute_entry.tolist() == [True, False, True] assert trace.execute_exit.tolist() == [False, True, False] assert trace.entry_loss_overshoot_pct.iloc[1] == 0 assert net_returns(trace).tolist() == pytest.approx([0, -.11, 0]) def test_simultaneous_overshoot_fill_bar_high_exclusion(monkeypatch): trace = replay(monkeypatch, [100, 91, 70], [True, False, False], [False, False, True], high=[100, 200, 100]) assert not trace.execute_exit.iloc[1] assert trace.trade_high.iloc[1] == 100 assert trace.loc[2, ['score_exit_flag', 'trailing_exit_flag', 'entry_loss_flag']].all() assert trace.entry_loss_overshoot_pct.iloc[2] == pytest.approx(20) def test_terminal_mark_and_no_entry_bar_exit(monkeypatch): trace = replay(monkeypatch, [100, 110], [True, False], [True, False]) assert not trace.execute_exit.any() assert trace.position.tolist() == [1, 1] assert net_returns(trace).tolist() == pytest.approx([0, .095]) def synthetic_frame(n=80): rng = np.random.default_rng(733) close = 100*np.exp(np.cumsum(rng.normal(0, .06, n))) return pd.DataFrame(dict(time=pd.date_range('2020-01-01', periods=n, freq='6h'), close=close, high=close*1.04, activation_score=rng.uniform(-400, 400, n))) def test_control_equivalence_and_prefix_causality(): frame = synthetic_frame() params = dict(runner.FROZEN) args = (params, np.zeros(len(frame)), frame.close.to_numpy(), frame.high.to_numpy()) canonical = _score_to_signals(frame.copy(), *args) control = policy.entry_loss_signals(frame.copy(), *args) for key in ('position', 'execute_entry', 'execute_exit'): np.testing.assert_array_equal(canonical[key], control[key]) np.testing.assert_array_equal(net_returns(canonical), net_returns(control)) full = policy.entry_loss_signals(frame.copy(), *args, loss_fraction=.10) for n in range(1, len(frame)): prefix = policy.entry_loss_signals(frame.iloc[:n].copy(), params, np.zeros(n), frame.close.to_numpy()[:n], frame.high.to_numpy()[:n], loss_fraction=.10) pd.testing.assert_frame_equal(prefix, full.iloc[:n]) def test_diagnostics_censors_gaps_and_end(monkeypatch): trace = replay(monkeypatch, [100, 90]+[105]*30, [True]+[False]*31) trace['net_return'] = net_returns(trace) trace['fee_return'] = runner.COMMISSION_RATE*(trace.execute_entry.shift(1, fill_value=False).astype(float)+trace.execute_exit.astype(float)) diag = runner.diagnostics(trace, trace) assert all(x['censored'] is None for x in diag['exit_events'][0]['recoveries']) assert diag['exit_events'][0]['recoveries'][0]['recovered_entry'] trace.loc[3:, 'time'] += pd.Timedelta(hours=6) assert all(x['censored']=='gap' for x in runner.diagnostics(trace, trace)['exit_events'][0]['recoveries']) assert runner.diagnostics(trace.iloc[:3], trace.iloc[:3])['exit_events'][0]['recoveries'][0]['censored']=='end_of_data' def test_evaluator_integration_control_metrics_identity_and_gates(monkeypatch): original = runner.evaluator frame = synthetic_frame(14000) frame['time'] = pd.date_range('2017-01-01', periods=len(frame), freq='6h') snapshot = SimpleNamespace(fingerprint='synthetic-contract', artifact_hashes={'synthetic': 'sha'}) def isolated(snapshot=None, signal=None): module = original(snapshot, signal) module.load_scored_frame = lambda *a: (frame, np.zeros(len(frame)), frame.close.to_numpy(), frame.high.to_numpy()) return module monkeypatch.setattr(runner, 'evaluator', isolated) params = dict(runner.FROZEN, mlp_weights_file='synthetic', source='current') canonical = runner.evaluate_policy(snapshot, params, None, canonical=True) disabled = runner.evaluate_policy(snapshot, params, None) runner.assert_control(canonical, disabled) baseline = runner.identify(disabled[0], snapshot, 'baseline') candidate, _ = runner.evaluate_policy(snapshot, dict(params, source='candidate'), .10) candidate = runner.identify(candidate, snapshot, 'entry_loss_10pct') assert candidate['Evaluation_Key'] != baseline['Evaluation_Key'] assert candidate['Comparison_Fingerprint'] == baseline['Comparison_Fingerprint'] assert len(__import__('json').loads(candidate['Period_Measurements'])) == len(original().WFO_FOLDS)+4 assert runner.gate_decisions(original(), candidate, baseline)['qualifies'] == original().promotable(candidate, baseline) assert original()._score_to_signals is _score_to_signals def test_reject_other_percentages(): with pytest.raises(ValueError): policy.entry_loss_signals(None, None, None, None, None, loss_fraction=.075) def test_time_window_disallows_close_after_end(): frame = pd.DataFrame(dict(time=pd.date_range('2020-01-01', periods=4, freq='6h'), close=[100., 100., 70., 60.], high=[100., 100., 100., 100.], activation_score=[100., 0., -300., -400.])) params = dict(runner.FROZEN, _pine_time_start='2020-01-01', _pine_time_end='2020-01-01 06:00') args = (params, np.zeros(4), frame.close.to_numpy(), frame.high.to_numpy()) candidate = policy.entry_loss_signals(frame.copy(), *args, loss_fraction=.10) control = policy.entry_loss_signals(frame.copy(), *args) canonical = _score_to_signals(frame.copy(), *args) assert candidate.entry_loss_flag.tolist() == [False, False, True, True] assert candidate.position.tolist() == [0, 1, 1, 1] assert not candidate.execute_exit.any() for key in ('position', 'execute_entry', 'execute_exit'): np.testing.assert_array_equal(control[key], canonical[key])