import json from dataclasses import replace import numpy as np import pytest from strategies.mlp_comparison import ComparisonSnapshot, require_replay import tools.run_mlp_deep_sweep as sweep from strategies.mlp_execution_stability import VERSION, THRESHOLDS, DELTAS, stability_periods def valid(**overrides): row = dict(Comparison_Fingerprint='same', Evaluation_Status='complete', source='candidate', **{'P&L/DD Ratio': 10., 'Calmar Ratio': 2., 'Total P&L %': 100., 'Fragile_0.05': 0, 'Sub1_Calmar': 1., 'Sub2_Calmar': 1., 'WFO_Neg_Folds': 0, 'OOS_Total Trades': 2, 'OOS_Sortino Ratio': 3., 'OOS_Bear_Pct': 0.}) row.update(overrides) period = dict(center={'pnl_pct': 100., 'dd_pct': -10.}, perturbed={'pnl_pct': 100., 'dd_pct': -10.}) row.update(Evaluation_Key='configuration', Stability_Evaluation_Key='configuration', Stability_Context=row['Comparison_Fingerprint'], Stability_Status='passed', Stability_Report=json.dumps(dict(version=VERSION, passed=True, trials=[dict(threshold=k, delta=d, periods=[dict(period, name=n, start=s, end=e) for n, s, e in stability_periods()]) for k in THRESHOLDS for d in DELTAS]))) return row def test_incompatible_high_history_is_provenance_only(): candidate = valid() incumbent = valid(**{'Calmar Ratio': 1., 'P&L/DD Ratio': 9.}) historical = valid(**{'Calmar Ratio': 1000., 'P&L/DD Ratio': 1000., 'OOS_Sortino Ratio': 999.}) assert sweep.promotable(candidate, incumbent, historical) assert not sweep.promotable(valid(**{'OOS_Sortino Ratio': 1.}), incumbent, historical) assert not sweep.promotable(valid(WFO_Neg_Folds=1), incumbent, historical) assert not sweep.promotable(valid(**{'Sub1_Calmar': 0.}), incumbent, historical) assert not sweep.promotable(valid(Comparison_Fingerprint='other'), incumbent) def test_execution_guard_replaces_raw_proximity_without_weakening_other_gates(): candidate = valid(**{'Fragile_0.05': 99}) incumbent = valid() assert sweep.promotable(candidate, incumbent) assert not sweep.promotable(candidate, dict(incumbent, Stability_Report='{}')) for change in ({'Stability_Status': 'failed'}, {'Stability_Report': '{}'}, {'Stability_Evaluation_Key': 'other'}, {'P&L/DD Ratio': 0}, {'Sub1_Calmar': 0}, {'Sub2_Calmar': -1}, {'WFO_Neg_Folds': 6}, {'Calmar Ratio': 1}, {'Evaluation_Status': 'incomplete'}, {'OOS_Sortino Ratio': 1}): assert not sweep.promotable(dict(candidate, **change), incumbent), change @pytest.mark.parametrize('status', ['failed', 'incomplete']) def test_failed_or_incomplete_incumbent_fails_closed(status): incumbent = valid(Evaluation_Status=status) assert not sweep.promotable(valid(), incumbent) with pytest.raises(ValueError): require_replay(incumbent) def test_snapshot_and_compatible_cache(monkeypatch): snapshot = sweep.capture_comparison('6H', [sweep.load_winner('6H')['mlp_weights_file']]) params = sweep.normalise_param(sweep.load_winner('6H')) monkeypatch.setattr(sweep, '_COMPARISON', snapshot) cache = {} first = sweep.evaluate('6H', params, cache) assert len(cache) == 1 assert sweep.evaluate('6H', params, cache) == first assert len(cache) == 1 changed = replace(snapshot, manifest_json=snapshot.manifest_json + ' ') monkeypatch.setattr(sweep, '_COMPARISON', changed) assert sweep.evaluate('6H', params, cache)['Comparison_Fingerprint'] != first['Comparison_Fingerprint'] assert len(cache) == 2 periods = json.loads(first['Period_Measurements']) assert len(periods) == len(sweep.WFO_FOLDS) + 4 assert all('entries' in period or period.get('coverage') == 'empty' for period in periods) def test_nan_model_is_failed_replay(monkeypatch): path = sweep.load_winner('6H')['mlp_weights_file'] snapshot = sweep.capture_comparison('6H', [path]) payload = json.loads(snapshot.artifacts[0][1]) payload['layers'][0]['W'][0][0] = np.nan snapshot = replace(snapshot, artifacts=((path, json.dumps(payload).encode()),)) monkeypatch.setattr(sweep, '_COMPARISON', snapshot) with pytest.raises(ValueError): require_replay(sweep.evaluate('6H', sweep.normalise_param(sweep.load_winner('6H')), {})) def test_equal_additional_threshold_budget(): candidates = sweep.generate_candidates('6H', 20, 77, extra_weights=[ 'strategies/params/mlp/mlp_weights_COINBASE_BTCUSD_6H_grouped_v1_seed9090.json']) per_model = {} for row in candidates: if row['source'] == 'equal_tuning': per_model.setdefault(row['mlp_weights_file'], []).append({k:v for k,v in row.items() if k != 'mlp_weights_file'}) assert len(per_model) >= 2 values = list(per_model.values()) assert all(rows == values[0] for rows in values) def test_real_data_and_source_hashes_change_context(tmp_path): repo = sweep.REPO original = sweep.capture_comparison('6H', [sweep.load_winner('6H')['mlp_weights_file']]) source = tmp_path / 'bars.csv' source.write_bytes(original.data + b'\n') contract = json.loads(original.manifest_json)['contract'] artifact = original.artifacts[0][0] changed_data = ComparisonSnapshot.capture(source, [artifact], repo=repo, asset=sweep.ASSET, tf='6H', contract=contract) assert changed_data.fingerprint != original.fingerprint manifest = json.loads(original.manifest_json) for path in manifest['source_hashes']: target = tmp_path / path target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes((repo / path).read_bytes()) (tmp_path / 'strategies/mlp_accounting.py').write_text('# changed accounting') source.write_bytes(original.data) changed_accounting = ComparisonSnapshot.capture(source, [artifact], repo=tmp_path, asset=sweep.ASSET, tf='6H', contract=contract) assert changed_accounting.fingerprint != original.fingerprint def test_real_truncated_data_is_incomplete(monkeypatch): import io import pandas as pd winner = sweep.load_winner('6H') snapshot = sweep.capture_comparison('6H', [winner['mlp_weights_file']]) frame = pd.read_csv(io.BytesIO(snapshot.data)).iloc[:100] snapshot = replace(snapshot, data=frame.to_csv(index=False).encode()) monkeypatch.setattr(sweep, '_COMPARISON', snapshot) row = sweep.evaluate('6H', sweep.normalise_param(winner), {}) assert row['Evaluation_Status'] == 'incomplete' with pytest.raises(ValueError): require_replay(row) def test_force_does_not_bypass_failed_incumbent(monkeypatch): original = sweep.evaluate def failed(tf, params, cache): row = original(tf, params, cache) if params.get('source') == 'current': row['Evaluation_Status'] = 'failed' return row monkeypatch.setattr(sweep, 'evaluate', failed) monkeypatch.setattr(sweep, 'write_winner', lambda *args: pytest.fail('must not write')) with pytest.raises(ValueError, match='Incumbent replay'): sweep.run_tf('6H', 0, 1, 10, False, force_promote=True) def test_run_tf_rejects_nonreproducing_incumbent(monkeypatch): original = sweep.evaluate count = 0 def unstable(tf, params, cache): nonlocal count row = original(tf, params, cache) if params.get('source') == 'current': count += 1 row['Calmar Ratio'] += count return row monkeypatch.setattr(sweep, 'evaluate', unstable) monkeypatch.setattr(sweep, 'write_winner', lambda *args: pytest.fail('must not write')) with pytest.raises(ValueError, match='not reproducible'): sweep.run_tf('6H', 0, 1, 10, False) def test_nonfinite_incumbent_parameter_is_rejected(): winner = sweep.load_winner('6H') winner['i_long_entry_activation_threshold'] = 'nan' with pytest.raises(ValueError, match='Nonfinite parameter'): sweep.normalise_param(winner) def test_existing_empty_winner_is_not_a_new_asset(tmp_path, monkeypatch): path = tmp_path / 'winner.csv' path.write_text('mlp_weights_file,Calmar Ratio\n') monkeypatch.setattr(sweep, 'winner_path', lambda tf: path) with pytest.raises(ValueError, match='no replayable row'): sweep.load_winner('6H')