"""Reconstruct saved research measurements from complete execution traces.""" from __future__ import annotations import argparse import json from pathlib import Path import sys import numpy as np import pandas as pd REPO = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO)) from strategies.mlp_comparison import digest from strategies.mlp_score_structure import confirmed_score_structure from strategies.strategy_mlp_scores import _score_to_signals from tools.run_mlp_revisit import DELTAS, THRESHOLDS, TRACE_KEYS, write_json from tools.run_mlp_entry_loss_experiment import evaluator, hashes, protected_hashes def measure(times, trace, start, end): intervals = np.diff(times).astype('timedelta64[ns]').astype(float) / 1e9 durations = np.r_[np.median(intervals), intervals] mask = times >= np.datetime64(start) if end is not None: bound = np.datetime64(end) mask &= times < bound + np.timedelta64(1, 'D') if len(end) == 10 else times <= bound returns = trace['net_return'][mask] equity = np.r_[1., np.cumprod(1 + returns)] dd = (equity / np.maximum.accumulate(equity) - 1).min() years = durations[mask].sum() / (365.25 * 86400) cagr = np.expm1(np.log(equity[-1]) / years) if equity[-1] > 0 else -1. return dict(pnl_pct=(equity[-1] - 1) * 100, dd_pct=dd * 100, calmar=cagr / abs(dd) if dd < 0 else None, entries=int(trace['execute_entry'][mask].sum()), exits=int(trace['execute_exit'][mask].sum())) def verify_trace(times, close, trace, row): entry = trace['execute_entry'].astype(bool) exit_ = trace['execute_exit'].astype(bool) position = trace['position'] previous = np.r_[0, position[:-1]] assert not np.any(entry & exit_) np.testing.assert_array_equal(position, previous + entry.astype(int) - exit_.astype(int)) market_returns = np.r_[0., close[1:] / close[:-1] - 1] expected = previous * market_returns - .005 * (np.r_[False, entry[:-1]].astype(float) + exit_.astype(float)) np.testing.assert_allclose(trace['net_return'], expected, rtol=0, atol=1e-14) for period in json.loads(row['Period_Measurements']): assert 'coverage' not in period, 'Unexpected missing period' metrics = measure(times, trace, period['start'], period['end']) for key, value in metrics.items(): if value is None: assert period[key] is None else: np.testing.assert_allclose(value, period[key], atol=1e-8, rtol=1e-10) def verify(directory): verification = json.loads((directory / 'verification.json').read_text()) assert verification['complete'] manifest = json.loads((directory / 'comparison_manifest.json').read_text()) for path, expected in manifest['source_hashes'].items(): assert digest((REPO / path).read_bytes()) == expected, path protected = json.loads((directory / 'protected_hashes.json').read_text()) assert protected == protected_hashes() references = json.loads((directory / 'reference_hashes.json').read_text()) assert references == hashes([REPO / p for p in references]) source = REPO / 'data/mlp/COINBASE_BTCUSD, 360.csv' assert digest(source.read_bytes()) == manifest['data_sha256'] frame = pd.read_csv(source) frame.columns = frame.columns.str.lower().str.strip() frame['time'] = pd.to_datetime(frame.time, unit='s' if pd.api.types.is_numeric_dtype(frame.time) else None) frame = frame.iloc[:-1] close = frame.close.to_numpy() times = frame.time.to_numpy(dtype='datetime64[ns]') rows = json.loads((directory / 'results.json').read_text()) assert len({r['Evaluation_Key'] for r in rows}) == len(rows) module = evaluator() module._COMPARISON = module.capture_comparison('6H', [r['mlp_weights_file'] for r in rows]) scored_cache = {} def reconstruct_actions(trace, row): path = row['mlp_weights_file'] if path not in scored_cache: scored_cache[path] = module.load_scored_frame('6H', path) scored, stoch, prices, highs = scored_cache[path] assert module._COMPARISON.artifact_hashes[path] == row['Artifact_SHA256'] params = {key: row[key] for key in module.PARAM_COLS} params['_pine_time_start'] = module.SCORE_START kwargs = {} if row['Policy'].startswith('higher_low_'): width = int(row['Policy'].rsplit('_', 1)[1]) kwargs['entry_gate'] = confirmed_score_structure(scored.activation_score.to_numpy(), left=width, right=width).bullish_state rebuilt = _score_to_signals(scored.copy(), params, stoch, prices, highs, **kwargs) for key in TRACE_KEYS[:3]: np.testing.assert_array_equal(trace[key], rebuilt[key]) train_score = scored.loc[scored.time < pd.Timestamp(module.TRAIN_END) + pd.Timedelta(days=1), 'activation_score'].to_numpy() nearest = np.min(np.abs(train_score[:, None] - np.array([params[k] for k in THRESHOLDS])[None, :]), axis=1) assert row['Fragile_0.05'] == int((nearest < .05).sum()) confusion = [] if 'center_id' in rows[0]: centers = json.loads((directory / 'center_summary.json').read_text()) differences = json.loads((directory / 'sensitivity.json').read_text()) indexed_diff = {(d['center_id'], d['threshold'], d['delta']): d for d in differences} index = {(r['center_id'], r['threshold'], r['delta']): r for r in rows} for center in centers: cid = center['center_id'] with np.load(directory / f'{cid}.npz') as saved: np.testing.assert_array_equal(times.astype('int64'), saved['time_ns']) base = {key: saved[f'center_{key}'] for key in TRACE_KEYS} base_row = index[cid, None, 0.] assert center['flagged'] == (base_row['Fragile_0.05'] > 0) verify_trace(times, close, base, base_row) reconstruct_actions(base, base_row) outcomes = {m: [] for m in (.05, 1.)} for ti, threshold in enumerate(THRESHOLDS): for di, delta in enumerate(DELTAS): trace = {key: saved[f'{ti}_{di}_{key}'] for key in TRACE_KEYS} row = index[cid, threshold, delta] for key in module.PARAM_COLS: if key == threshold: assert row[key] == base_row[key] + delta else: assert row[key] == base_row[key] verify_trace(times, close, trace, row) reconstruct_actions(trace, row) difference = indexed_diff[cid, threshold, delta] for name in ('IS', 'OOS'): period = next(p for p in json.loads(row['Period_Measurements']) if p.get('name') == name) a = measure(times, base, period['start'], period['end']) b = measure(times, trace, period['start'], period['end']) mask = times >= np.datetime64(period['start']) if period['end']: mask &= times < np.datetime64(period['end']) + np.timedelta64(1, 'D') changed = ((base['execute_entry'] != trace['execute_entry']) | (base['execute_exit'] != trace['execute_exit'])) & mask expected = difference[name] assert expected['changed_action_bars'] == int(changed.sum()) assert expected['changed_position_bars'] == int(((base['position'] != trace['position']) & mask).sum()) assert expected['changed_return_bars'] == int(((base['net_return'] != trace['net_return']) & mask).sum()) material = False for key, limit in (('pnl_pct', 1.), ('dd_pct', 1.), ('calmar', .05)): value = b[key] - a[key] if a[key] is not None and b[key] is not None else None if value is not None: np.testing.assert_allclose(value, expected['deltas'][key], atol=1e-8, rtol=1e-10) material |= abs(value) >= limit else: assert expected['deltas'][key] is None assert material == expected['material'] if name == 'IS': outcomes[abs(delta)].append((int(changed.sum()), material)) for magnitude, outcomes_at_scale in outcomes.items(): expected = center['sensitivity'][str(magnitude)] assert expected == dict(any_actions=any(a > 0 for a, _ in outcomes_at_scale), any_material=any(m for _, m in outcomes_at_scale), max_action_bars=max(a for a, _ in outcomes_at_scale)) sample = [c for c in centers if c['sample']] for model in ['ALL'] + sorted({c['model'] for c in sample}): subset = sample if model == 'ALL' else [c for c in sample if c['model'] == model] for magnitude in (.05, 1.): for outcome in ('any_actions', 'any_material'): counts = {key: 0 for key in ('TP', 'FP', 'TN', 'FN')} for center in subset: positive = center['sensitivity'][str(magnitude)][outcome] counts[('T' if center['flagged'] == positive else 'F') + ('P' if center['flagged'] else 'N')] += 1 confusion.append(dict(model=model, magnitude=magnitude, outcome=outcome, n=len(subset), **counts)) assert len(rows) == len(centers) * 13 else: for row in rows: trace = pd.read_csv(directory / f"{row['Policy']}_bars.csv") np.testing.assert_array_equal(times, pd.to_datetime(trace.time).to_numpy()) verify_trace(times, close, {key: trace[key].to_numpy() for key in TRACE_KEYS}, row) reconstruct_actions({key: trace[key].to_numpy() for key in TRACE_KEYS}, row) assert len(rows) == 4 receipt = dict(passed=True, rows_reconstructed=len(rows), period_rows_reconstructed=len(rows)*10, accounting_reconstructed=True, actions_reconstructed=True, protected_files_unchanged=len(protected), confusion=confusion, verifier_sha256=digest(Path(__file__).read_bytes()), evidence_hashes={p.name: digest(p.read_bytes()) for p in sorted(directory.iterdir()) if p.is_file() and p.name != 'audit.json'}) write_json(directory / 'audit.json', receipt) print(json.dumps({k: v for k, v in receipt.items() if k not in ('confusion', 'evidence_hashes')}, indent=2)) if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('directory', type=Path) verify(parser.parse_args().directory.resolve())