import json import pathlib import subprocess import tempfile import unittest ROOT = pathlib.Path(__file__).parents[1] class ResultsV2RuntimeContractTests(unittest.TestCase): def test_observer_records_actual_runtime_equity_without_mutating_order_state(self): runtime_state = { "id": 42, "timestamp": 120, "equity": 10.0, "position": 2.0, "average_price": 100.0, "mark_price": 125.0, } source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$state = json_decode({json.dumps(json.dumps(runtime_state))}, true); " "$before = json_encode($state); " "$point = ResultsV2Observer::equityPoint($state); " "echo json_encode([$point, $before === json_encode($state)]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) point, unchanged = json.loads(result.stdout) self.assertEqual(42, point["test_id"]) self.assertEqual(120, point["timestamp"]) self.assertEqual(10.0, point["wallet_balance"]) self.assertEqual(0.004, point["unrealized_pnl"]) self.assertEqual(10.004, point["equity"]) self.assertTrue(unchanged) def test_update_and_stop_observe_real_timeline_before_the_existing_terminal_transition(self): update = (ROOT / "api" / "private" / "update.php").read_text(encoding="utf-8") stop = (ROOT / "api" / "private" / "stop.php").read_text(encoding="utf-8") self.assertIn("ResultsV2Observer::fromEnvironment((int)$test_id", update) self.assertIn("ResultsV2Observer::fromEnvironment((int)$test_id", stop) self.assertGreater(update.index("$results_observer->observe($test_data);"), update.index("$order->fillThis")) natural_start = update.index("if($is_done) {", update.index("$guard_data_unavailable")) natural_terminal = update.index("$results_observer->persistTerminal($terminal_data", natural_start) self.assertGreater(natural_terminal, update.index("$order->ClosePosition", natural_start)) self.assertLess(natural_terminal, update.index("$test->stop();", natural_start)) self.assertIn("$terminal_data = $test->get();", update) self.assertIn("$results_observer->persistTerminal($terminal_data, 'end_reached', true);", update) self.assertIn("$results_observer->persistTerminal($persisted_test, $recovered_reason", update) self.assertIn("$results_observer->persistTerminal($test_data, $recovered_reason", update) self.assertGreaterEqual(update.count("$results_observer->persistTerminal"), 3) self.assertGreater(stop.index("$results_observer->persistTerminal"), stop.index("$order->ClosePosition")) self.assertLess(stop.index("$results_observer->persistTerminal"), stop.index("$test->stop(true);")) self.assertIn("$terminal_data = $test->get();", stop) self.assertIn("$results_observer->persistTerminal($terminal_data, 'stopped', false);", stop) def test_existing_indicator_endpoints_pass_their_calculated_values_to_the_observer(self): materialized = {"phases": [{"phase": 1, "setup": {"grid_type_use_ind": 1, "indicator_parameters": {"sma": {"period": 20, "z": 1}}}}]} snapshot_hash_source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"echo ResultsV2::snapshotHash(json_decode({json.dumps(json.dumps(materialized))}, true));" ) snapshot_hash = subprocess.run(["php", "-r", snapshot_hash_source], capture_output=True, text=True, check=True).stdout source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$materialized = json_decode({json.dumps(json.dumps(materialized))}, true); " f"$points = ResultsV2Observer::indicatorSnapshot(42, 'sma', ['period' => 20, 'z' => 1], 7, {json.dumps(snapshot_hash)}, " "$materialized, 1, 'signal_evaluation', 'price', 'instrument_price', ['rule' => 'sma'], 120, 101.5, ['close' => 102]); " "echo json_encode($points);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) points = json.loads(result.stdout) self.assertEqual(1, len(points)) self.assertEqual("sma", points[0]["series_type"]) self.assertEqual(102, points[0]["price_context"]["close"]) self.assertEqual(7, points[0]["strategy_revision_id"]) reordered_source = source.replace("['period' => 20]", "['z' => 1, 'period' => 20]") canonical_source = source.replace("['period' => 20]", "['period' => 20, 'z' => 1]") reordered = json.loads(subprocess.run(["php", "-r", reordered_source], capture_output=True, text=True, check=True).stdout) canonical = json.loads(subprocess.run(["php", "-r", canonical_source], capture_output=True, text=True, check=True).stdout) self.assertEqual(reordered[0]["series_id"], canonical[0]["series_id"]) for endpoint in ("get_sma.php", "get_ema.php", "get_supertrend.php"): source = (ROOT / "api" / "ind" / endpoint).read_text(encoding="utf-8") self.assertIn("ResultsV2Observer::fromEnvironment((int)$test_id", source) self.assertIn("$results_observer->observeIndicator", source) def test_indicator_parameters_are_required_from_canonical_frozen_indicator_snapshot(self): materialized = {"phases": [{"phase": 1, "setup": {"grid_type_use_ind": 1, "indicator_parameters": {"sma": {"period": 20}}}}]} source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$materialized = json_decode({json.dumps(json.dumps(materialized))}, true); " "$hash = ResultsV2::snapshotHash($materialized); " "$ok = ResultsV2::indicatorSnapshot(42, 'sma:hash', 'sma', ['period' => 20], 7, $hash, $materialized, 1, 'signal_evaluation', 'price', 'instrument_price', ['rule' => 'sma'], [[ 'timestamp' => 1, 'value' => 2 ]]); " "$tampered = false; try { ResultsV2::indicatorSnapshot(42, 'sma:hash', 'sma', ['period' => 21], 7, $hash, $materialized, 1, 'signal_evaluation', 'price', 'instrument_price', ['rule' => 'sma'], [[ 'timestamp' => 1, 'value' => 2 ]]); } catch (InvalidArgumentException $e) { $tampered = true; } " "$missing = false; try { ResultsV2::indicatorSnapshot(42, 'sma:hash', 'sma', ['period' => 20], 7, ResultsV2::snapshotHash(['other' => 1]), ['other' => 1], 1, 'signal_evaluation', 'price', 'instrument_price', ['rule' => 'sma'], [[ 'timestamp' => 1, 'value' => 2 ]]); } catch (InvalidArgumentException $e) { $missing = true; } " "echo json_encode([count($ok), $tampered, $missing]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([1, True, True], json.loads(result.stdout)) def test_interface_runtime_snapshot_phase_setup_is_the_frozen_indicator_contract(self): snapshot = {"instrument_id": 92, "compounding": {}, "phases": [{"phase": 1, "setup": {"grid_type_use_ind": 1, "grid_type_ind_type": 4, "grid_type_ind_id": 64, "indicator_parameters": {"sma": {"period": 20, "candle_size": 5}}}}]} source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); $hash=ResultsV2::snapshotHash($snapshot); " "$test=['strategy_revision_id'=>7,'strategy_snapshot'=>$snapshot,'strategy_snapshot_hash'=>$hash]; " "$points=ResultsV2::indicatorSnapshot(42,'sma:hash','sma',['period'=>20,'candle_size'=>5],7,$hash,$snapshot,1,'signal_evaluation','price','instrument_price',['rule'=>'sma'],[['timestamp'=>1,'value'=>2]]); " "$bad=false; try { ResultsV2::indicatorSnapshot(42,'sma:hash','sma',['period'=>21,'candle_size'=>5],7,$hash,$snapshot,1,'signal_evaluation','price','instrument_price',['rule'=>'sma'],[['timestamp'=>1,'value'=>2]]); } catch (InvalidArgumentException $e) {$bad=true;} " "echo json_encode([count($points),ResultsV2Observer::indicatorEvidenceRequired($test),$bad]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([1, True, True], json.loads(result.stdout)) def test_legacy_terminal_is_partial_even_with_buffered_equity_and_transactions(self): state = {"id": 42, "start_equity": 10.0, "equity": 11.0, "fee": 0.1, "position": 0.0, "average_price": 0.0, "timestamp": 180} points = [{"timestamp": 60, "wallet_balance": 10.0, "unrealized_pnl": 0.0}, {"timestamp": 180, "wallet_balance": 11.0, "unrealized_pnl": 0.0}] transactions = [{"timestamp": 60, "position": 1.0, "pnl": 0.0}, {"timestamp": 180, "position": 0.0, "pnl": 1.0}] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$state = json_decode({json.dumps(json.dumps(state))}, true); $points = json_decode({json.dumps(json.dumps(points))}, true); $tx = json_decode({json.dumps(json.dumps(transactions))}, true); " "$legacy = ResultsV2Observer::terminalSummary($state, $points, $tx, true, 'end_reached', true, 2, 4); " "$bound = $state + ['strategy_revision_id' => 7, 'strategy_snapshot' => ['indicator' => ['period' => 20]]]; $bound['strategy_snapshot_hash'] = ResultsV2::snapshotHash($bound['strategy_snapshot']); " "$complete = ResultsV2Observer::terminalSummary($bound, $points, $tx, true, 'end_reached', true, 2, 4); echo json_encode([$legacy, $complete]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) legacy, complete = json.loads(result.stdout) self.assertEqual("legacy_partial", legacy["data_completeness"]) self.assertIsNone(legacy["max_drawdown"]) self.assertEqual("complete", complete["data_completeness"]) def test_interface_runtime_snapshot_fields_are_immutable_and_require_indicator_evidence(self): state = {"id": 42, "start_equity": 10.0, "fee": 0.1, "position": 0.0, "average_price": 0.0, "timestamp": 180, "phase": 1} snapshot = {"instrument_id": 92, "compounding": {}, "phases": [{"phase": 1, "setup": {"grid_type_use_ind": 1, "grid_type_ind_type": 4, "grid_type_ind_id": 64, "indicator_parameters": {"sma": {"period": 20, "candle_size": 5}}}}]} points = [{"timestamp": 60, "wallet_balance": 10.0, "unrealized_pnl": 0.0}, {"timestamp": 180, "wallet_balance": 11.0, "unrealized_pnl": 0.0}] transactions = [{"timestamp": 60, "position": 1.0, "pnl": 0.0}, {"timestamp": 180, "position": 0.0, "pnl": 1.0}] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$state=json_decode({json.dumps(json.dumps(state))}, true); $snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); " f"$points=json_decode({json.dumps(json.dumps(points))}, true); $tx=json_decode({json.dumps(json.dumps(transactions))}, true); " "$canonical=function($value) use (&$canonical){if(!is_array($value))return $value;if(array_is_list($value))return array_map($canonical,$value);ksort($value);foreach($value as $key=>$item)$value[$key]=$canonical($item);return $value;}; " "$state += ['strategy_revision_id'=>7,'runtime_snapshot_json'=>json_encode($snapshot),'runtime_content_hash'=>hash('sha256', json_encode($canonical($snapshot), JSON_UNESCAPED_SLASHES))]; " "$summary=ResultsV2Observer::terminalSummary($state,$points,$tx,true,'end_reached',true,2,4); " "class ResultsV2TelemetryBuffer { function __construct($id){} function addIncompleteReason($reason){} function markEvidence($kind){return true;} } " "class FakeResultsStore { public $points=[]; function persistIndicatorPoints($points){$this->points=$points;return true;} } " "$observer=(new ReflectionClass(ResultsV2Observer::class))->newInstanceWithoutConstructor(); $store=new FakeResultsStore(); $property=new ReflectionProperty(ResultsV2Observer::class,'store'); $property->setAccessible(true); $property->setValue($observer,$store); " "$observed=$observer->observeIndicator($state,'sma',['period'=>20,'candle_size'=>5],101.5,['close'=>102],'signal_evaluation','price','instrument_price',['indicator'=>'sma']); " "echo json_encode([$summary['data_completeness'],ResultsV2Observer::indicatorEvidenceRequired($state),$observed,count($store->points),$store->points[0]['strategy_snapshot_hash'],$state['runtime_content_hash']]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) values = json.loads(result.stdout) self.assertEqual(["complete", True, True, 1], values[:4]) self.assertEqual(values[4], values[5]) def test_supertrend_snapshot_aliases_match_runtime_indicator_parameters(self): snapshot = { "phases": [{ "setup": { "indicator_control_enabled": 1, "indicator_parameters": { "supertrend": { "supertrend_atr_period": 20, "supertrend_candle_size": 15, "supertrend_multiplier": 16, "supertrend_deviation": 0, "supertrend_use_direction": 0, "supertrend_use_value": 0, } } } }] } source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); " "echo json_encode(ResultsV2::frozenIndicatorParameters($snapshot,'supertrend'), JSON_PRESERVE_ZERO_FRACTION);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual( {"atr_length": 20, "candle_size": 15, "multiplier": 16}, json.loads(result.stdout), ) self.assertIsInstance(json.loads(result.stdout)["multiplier"], float) def test_sma_snapshot_aliases_match_runtime_indicator_parameters(self): snapshot = { "phases": [{ "phase": 1, "setup": { "grid_type_use_ind": 1, "indicator_parameters": { "sma": { "sma_period": 30, "sma_candle_size": 1440, "sma_above": 1, "sma_below": 1, } }, }, }] } source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); " "$hash=ResultsV2::snapshotHash($snapshot); " "$points=ResultsV2::indicatorSnapshot(392707,'sma:hash','sma',['period'=>30,'candle_size'=>1440],7,$hash,$snapshot,1,'signal_evaluation','price','instrument_price',['indicator'=>'sma'],[['timestamp'=>1,'value'=>2]]); " "echo json_encode(count($points));" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual(1, json.loads(result.stdout)) def test_enabled_supertrend_payload_accepts_semantically_equal_numeric_types_and_completes(self): snapshot = { "phases": [{ "phase": 1, "setup": { "indicator_control_enabled": 1, "indicator_parameters": { "supertrend": { "supertrend_atr_period": 20, "supertrend_candle_size": 15, "supertrend_multiplier": 16, "supertrend_deviation": 0, "supertrend_use_direction": 0, "supertrend_use_value": 0, } }, }, }] } source = f""" class ResultsV2TelemetryBuffer {{ public static $reasons = []; public static $evidence = []; function __construct($id) {{}} function addIncompleteReason($reason) {{ if (!in_array($reason, self::$reasons, true)) self::$reasons[] = $reason; return true; }} function markEvidence($kind) {{ self::$evidence[$kind] = true; return true; }} function hasEvidence($kind) {{ return !empty(self::$evidence[$kind]); }} function transactions() {{ return [['id' => 'tx-1', 'timestamp' => 120, 'position' => 0, 'pnl' => 0]]; }} function ordersCount() {{ return 1; }} function terminalRuntimeSeconds() {{ return 1; }} function incompleteReasons() {{ return self::$reasons; }} }} class FakeResultsStore {{ public static $summary; public static $indicatorPoints = []; function persistIndicatorPoints($points) {{ self::$indicatorPoints = $points; return true; }} function persistEquityPoints($id, $points) {{ return true; }} function getTerminalEquityPoints($id, $initial = 0) {{ return ['points' => [], 'truncated' => false, 'point_count' => 1, 'final_equity' => 100.0, 'terminal_timestamp' => 120, 'max_drawdown' => 0.0, 'peak_timestamp' => null, 'trough_timestamp' => null]; }} function getTerminalTransactions($id) {{ return ['transactions' => [['timestamp' => 120, 'order_name' => '', 'position' => 0, 'pnl' => 0]], 'truncated' => false]; }} function persistSummary($summary) {{ self::$summary = $summary; return true; }} }} require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; $snapshot = json_decode({json.dumps(json.dumps(snapshot))}, true); $hash = ResultsV2::strategySnapshotHash($snapshot); $state = ['id' => 42, 'start_equity' => 100, 'fee' => 0, 'position' => 0, 'average_price' => 0, 'timestamp' => 120, 'phase' => 1, 'strategy_revision_id' => 278, 'runtime_snapshot_json' => json_encode($snapshot), 'runtime_content_hash' => $hash]; $observer = (new ReflectionClass(ResultsV2Observer::class))->newInstanceWithoutConstructor(); $property = new ReflectionProperty(ResultsV2Observer::class, 'store'); $property->setAccessible(true); $property->setValue($observer, new FakeResultsStore()); $observed = $observer->observeIndicator( $state, 'supertrend', ['atr_length' => 20, 'candle_size' => 15, 'multiplier' => (float)16], 123.4, ['lower_band' => 120.0, 'upper_band' => 125.0, 'direction' => 'up'], 'signal_evaluation', 'price', 'instrument_price', ['indicator' => 'supertrend'] ); $observer->persistTerminal($state, 'end_reached', true); echo json_encode([$observed, FakeResultsStore::$summary['data_completeness'], ResultsV2TelemetryBuffer::$reasons, FakeResultsStore::$indicatorPoints]); """ result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) observed, completeness, reasons, points = json.loads(result.stdout) self.assertTrue(observed) self.assertEqual("complete", completeness) self.assertEqual([], reasons) self.assertEqual(1, len(points)) self.assertEqual("supertrend", points[0]["series_type"]) self.assertEqual(278, points[0]["strategy_revision_id"]) self.assertEqual(64, len(points[0]["parameters_hash"])) def test_supertrend_provenance_failures_use_stable_codes_and_remain_fail_closed(self): snapshot = { "phases": [{ "phase": 1, "setup": { "indicator_control_enabled": 1, "indicator_parameters": { "supertrend": { "supertrend_atr_period": 20, "supertrend_candle_size": 15, "supertrend_multiplier": 16, } }, }, }] } source = f""" class ResultsV2TelemetryBuffer {{ public static $reasons = []; function __construct($id) {{}} function addIncompleteReason($reason) {{ if (!in_array($reason, self::$reasons, true)) self::$reasons[] = $reason; return true; }} }} class FakeResultsStore {{}} require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; $snapshot = json_decode({json.dumps(json.dumps(snapshot))}, true); $hash = ResultsV2::strategySnapshotHash($snapshot); $state = ['id' => 42, 'phase' => 1, 'strategy_revision_id' => 278, 'runtime_snapshot_json' => json_encode($snapshot), 'runtime_content_hash' => $hash]; $observer = (new ReflectionClass(ResultsV2Observer::class))->newInstanceWithoutConstructor(); $property = new ReflectionProperty(ResultsV2Observer::class, 'store'); $property->setAccessible(true); $property->setValue($observer, new FakeResultsStore()); $bad = $observer->observeIndicator( $state, 'supertrend', ['atr_length' => 20, 'candle_size' => 15, 'multiplier' => (float)17], 123.4, [], 'signal_evaluation', 'price', 'instrument_price', ['indicator' => 'supertrend'] ); $unknown = $observer->observeIndicator( $state, 'supertrend', ['atr_length' => 20, 'candle_size' => 15, 'multiplier' => (float)16, 'unexpected' => 1], 123.4, [], 'signal_evaluation', 'price', 'instrument_price', ['indicator' => 'supertrend'] ); $missingState = $state; $missingState['runtime_content_hash'] = ''; $missing = $observer->observeIndicator( $missingState, 'supertrend', ['atr_length' => 20, 'candle_size' => 15, 'multiplier' => (float)16], 123.4, [], 'signal_evaluation', 'price', 'instrument_price', ['indicator' => 'supertrend'] ); $mismatchCode = 0; try {{ ResultsV2::indicatorSnapshot(42, 'supertrend', 'supertrend', ['atr_length' => 20, 'candle_size' => 15, 'multiplier' => (float)17], 278, $hash, $snapshot, 1, 'signal_evaluation', 'price', 'instrument_price', ['indicator' => 'supertrend'], [['timestamp' => 1, 'value' => 1]]); }} catch (InvalidArgumentException $error) {{ $mismatchCode = $error->getCode(); }} $missingCode = 0; try {{ ResultsV2::indicatorSnapshot(42, 'supertrend', 'supertrend', ['atr_length' => 20, 'candle_size' => 15, 'multiplier' => (float)16], 278, '', $snapshot, 1, 'signal_evaluation', 'price', 'instrument_price', ['indicator' => 'supertrend'], [['timestamp' => 1, 'value' => 1]]); }} catch (InvalidArgumentException $error) {{ $missingCode = $error->getCode(); }} echo json_encode([$bad, $unknown, $missing, ResultsV2TelemetryBuffer::$reasons, $mismatchCode, $missingCode]); """ result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual( [False, False, False, ["indicator_provenance_mismatch", "indicator_provenance_missing"], 1002, 1001], json.loads(result.stdout), ) def test_indicator_provenance_is_bound_to_the_exact_enabled_frozen_phase(self): snapshot = { "phases": [ {"phase": 1, "setup": {"order_dist_use_ind": 1, "indicator_parameters": { "supertrend": {"supertrend_atr_period": 20, "supertrend_candle_size": 15, "supertrend_multiplier": 16}}}}, {"phase": 2, "setup": {"order_dist_use_ind": 0, "indicator_parameters": { "supertrend": {"supertrend_atr_period": 20, "supertrend_candle_size": 15, "supertrend_multiplier": 16}}}}, {"phase": 3, "setup": {"grid_type_use_ind": 1, "indicator_parameters": { "sma": {"period": 20, "candle_size": 15}}}}, {"phase": 4, "setup": {"indicator_parameters": {"by_id": []}}}, {"phase": 5, "setup": {"indicator_parameters": {"by_id": []}}}, ] } source = f""" require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; $snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); $hash=ResultsV2::strategySnapshotHash($snapshot); $parameters=['atr_length'=>20,'candle_size'=>15,'multiplier'=>(float)16]; $call=function($phase,$type,$parameters,$hash) use ($snapshot) {{ try {{ return count(ResultsV2::indicatorSnapshot(42,$type,$type,$parameters,278,$hash,$snapshot,$phase, 'signal_evaluation','price','instrument_price',['indicator'=>$type], [['timestamp'=>1,'value'=>1]])); }} catch (InvalidArgumentException $error) {{ return $error->getCode(); }} }}; echo json_encode([ $call(1,'supertrend',$parameters,$hash), $call(null,'supertrend',$parameters,$hash), $call(0,'supertrend',$parameters,$hash), $call(6,'supertrend',$parameters,$hash), $call(2,'supertrend',$parameters,$hash), $call(3,'supertrend',$parameters,$hash), $call(1,'supertrend',['atr_length'=>20,'candle_size'=>15,'multiplier'=>(float)17],$hash), $call(1,'supertrend',$parameters,str_repeat('0',64)) ]); """ result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([1, 1001, 1001, 1001, 1002, 1002, 1002, 1002], json.loads(result.stdout)) def test_indicator_endpoints_require_phase_and_attach_only_request_local_provenance(self): for endpoint in ("get_sma.php", "get_ema.php", "get_supertrend.php"): with self.subTest(endpoint=endpoint): source = (ROOT / "api" / "ind" / endpoint).read_text(encoding="utf-8") self.assertIn("FILTER_VALIDATE_INT", source) self.assertIn("'min_range' => 1, 'max_range' => 5", source) self.assertIn("indicator_provenance_missing", source) self.assertIn("$test_data['phase'] = (int)$phase;", source) self.assertNotIn("UPDATE `bee_tests`", source) def test_real_supertrend_endpoint_transports_phase_to_the_observer(self): endpoint = ROOT / "api" / "ind" / "get_supertrend.php" init = """42,'timestamp'=>1500,'start_date'=>60,'instrument_id'=>92]; } } class Candle { function __construct($id,$size) {} function getInPeriod($instrument,$from,$to) { $rows=[]; for($i=0;$i<25;$i++) $rows[]=['time'=>60*($i+1),'high'=>110,'low'=>90,'close'=>100]; return $rows; } public static function checkMissingCandles($candles,$from,$to,$size) { return false; } function load_from_exchange($instrument,$start,$end,$required) {} function getATR($candles,$length) { return array_fill(0,25,1); } } class ResultsV2TelemetryBuffer { public static $reasons=[]; function __construct($id) {} function addIncompleteReason($reason) { self::$reasons[]=$reason; } } class ResultsV2Observer { public static $calls=[]; public static function fromEnvironment($id) { return new self(); } function observeIndicator(...$args) { self::$calls[]=$args; return true; } } """ request = { "instance_id": "42", "atr_length": "20", "candle_size": "15", "multiplier": "16", "prevLowerBand": "0", "prevUpperBand": "0", "prevSuperTrend": "0", "prevTimestamp": "0", "prevDirection": "", "phase": "1", } with tempfile.TemporaryDirectory() as directory: pathlib.Path(directory, "init.inc.php").write_text(init, encoding="utf-8") source = ( f"$_SERVER['DOCUMENT_ROOT']={json.dumps(directory)}; $_SERVER['HTTP_HOST']='bee.test'; " f"$_POST=json_decode({json.dumps(json.dumps(request))},true); " f"require {json.dumps(str(endpoint))}; echo PHP_EOL.json_encode(ResultsV2Observer::$calls);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) response, calls = result.stdout.strip().splitlines() self.assertEqual(200, json.loads(response)["code"]) call = json.loads(calls)[0] self.assertEqual(1, call[0]["phase"]) self.assertEqual("supertrend", call[1]) self.assertEqual({"atr_length": 20, "candle_size": 15, "multiplier": 16}, call[2]) for invalid in (None, "1.0", "0", "-1", "6"): with self.subTest(phase=invalid): invalid_request = dict(request) if invalid is None: invalid_request.pop("phase") else: invalid_request["phase"] = invalid invalid_source = ( f"$_SERVER['DOCUMENT_ROOT']={json.dumps(directory)}; $_SERVER['HTTP_HOST']='bee.test'; " f"$_POST=json_decode({json.dumps(json.dumps(invalid_request))},true); " f"require {json.dumps(str(endpoint))};" ) invalid_result = subprocess.run(["php", "-r", invalid_source], capture_output=True, text=True) self.assertEqual(0, invalid_result.returncode, invalid_result.stderr) invalid_response = json.loads(invalid_result.stdout) self.assertEqual(422, invalid_response["code"]) self.assertEqual("indicator_provenance_missing", invalid_response["incomplete_reason"]) def test_indicator_store_failure_remains_fail_closed(self): snapshot = {"phases": [{"phase": 1, "setup": {"grid_type_use_ind": 1, "indicator_parameters": {"sma": {"period": 20, "candle_size": 15}}}}]} source = f""" class ResultsV2TelemetryBuffer {{ public static $reasons=[]; public static $evidence=[]; function __construct($id) {{}} function addIncompleteReason($reason) {{ self::$reasons[]=$reason; return true; }} function markEvidence($kind) {{ self::$evidence[$kind]=true; return true; }} }} class FakeResultsStore {{ public static $durable=[]; function persistIndicatorPoints($points) {{ return false; }} }} require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; $snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); $state=['id'=>42,'timestamp'=>120,'phase'=>1,'strategy_revision_id'=>278, 'runtime_snapshot_json'=>json_encode($snapshot), 'runtime_content_hash'=>ResultsV2::strategySnapshotHash($snapshot)]; $observer=(new ReflectionClass(ResultsV2Observer::class))->newInstanceWithoutConstructor(); $property=new ReflectionProperty(ResultsV2Observer::class,'store'); $property->setAccessible(true); $property->setValue($observer,new FakeResultsStore()); $ok=$observer->observeIndicator($state,'sma',['period'=>20,'candle_size'=>15],101.5,[], 'signal_evaluation','price','instrument_price',['indicator'=>'sma']); echo json_encode([$ok,FakeResultsStore::$durable,ResultsV2TelemetryBuffer::$evidence,ResultsV2TelemetryBuffer::$reasons]); """ result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([False, [], [], ["indicator_write_failed"]], json.loads(result.stdout)) def test_indicator_provenance_failures_require_missing_evidence_at_terminal(self): snapshot = { "phases": [{ "phase": 1, "setup": { "indicator_control_enabled": 1, "indicator_parameters": { "supertrend": { "supertrend_atr_period": 20, "supertrend_candle_size": 15, "supertrend_multiplier": 16, } }, }, }] } cases = { "changed_multiplier": "$parameters['multiplier'] = (float)17;", "unknown_parameter": "$parameters['unexpected'] = 1;", "missing_hash": "$state['runtime_content_hash'] = '';", "missing_snapshot": "unset($state['runtime_snapshot_json']);", "missing_phase": "unset($state['phase']);", "missing_purpose": "$purpose = '';", "missing_pane": "$pane = '';", "missing_unit": "$unit = '';", "missing_decision_context": "$decisionContext = [];", } for name, mutation in cases.items(): with self.subTest(name=name): source = f""" class ResultsV2TelemetryBuffer {{ public static $reasons = []; public static $evidence = []; function __construct($id) {{}} function addIncompleteReason($reason) {{ if (!in_array($reason, self::$reasons, true)) self::$reasons[] = $reason; return true; }} function markEvidence($kind) {{ self::$evidence[$kind] = true; return true; }} function hasEvidence($kind) {{ return !empty(self::$evidence[$kind]); }} function transactions() {{ return [['id' => 'tx-1', 'timestamp' => 120, 'position' => 0, 'pnl' => 0]]; }} function ordersCount() {{ return 1; }} function terminalRuntimeSeconds() {{ return 1; }} function incompleteReasons() {{ return self::$reasons; }} }} class FakeResultsStore {{ public static $summary; public static $indicatorPoints = []; function persistIndicatorPoints($points) {{ self::$indicatorPoints = $points; return true; }} function persistEquityPoints($id, $points) {{ return true; }} function getTerminalEquityPoints($id, $initial = 0) {{ return ['points' => [], 'truncated' => false, 'point_count' => 1, 'final_equity' => 100.0, 'terminal_timestamp' => 120, 'max_drawdown' => 0.0, 'peak_timestamp' => null, 'trough_timestamp' => null]; }} function getTerminalTransactions($id) {{ return ['transactions' => [], 'truncated' => false]; }} function persistSummary($summary) {{ self::$summary = $summary; return true; }} }} require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; $snapshot = json_decode({json.dumps(json.dumps(snapshot))}, true); $hash = ResultsV2::strategySnapshotHash($snapshot); $state = ['id' => 42, 'start_equity' => 100, 'timestamp' => 120, 'phase' => 1, 'strategy_revision_id' => 278, 'runtime_snapshot_json' => json_encode($snapshot), 'runtime_content_hash' => $hash]; $parameters = ['atr_length' => 20, 'candle_size' => 15, 'multiplier' => (float)16]; $purpose = 'signal_evaluation'; $pane = 'price'; $unit = 'instrument_price'; $decisionContext = ['indicator' => 'supertrend']; {mutation} $observer = (new ReflectionClass(ResultsV2Observer::class))->newInstanceWithoutConstructor(); $property = new ReflectionProperty(ResultsV2Observer::class, 'store'); $property->setAccessible(true); $property->setValue($observer, new FakeResultsStore()); $observed = $observer->observeIndicator($state, 'supertrend', $parameters, 123.4, [], $purpose, $pane, $unit, $decisionContext); $observer->persistTerminal($state, 'end_reached', true); echo json_encode([$observed, count(FakeResultsStore::$indicatorPoints), FakeResultsStore::$summary['data_completeness'], ResultsV2TelemetryBuffer::$reasons]); """ result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) observed, point_count, completeness, reasons = json.loads(result.stdout) self.assertFalse(observed) self.assertEqual(0, point_count) self.assertEqual("partial", completeness) expected_provenance = ( "indicator_provenance_mismatch" if name in {"changed_multiplier", "unknown_parameter"} else "indicator_provenance_missing" ) self.assertIn(expected_provenance, reasons) self.assertIn("indicator_evidence_missing", reasons) def test_indicator_evidence_requirement_follows_frozen_snapshot_and_allows_explicit_empty_indicators(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " "$with = ['strategy_revision_id' => 7, 'strategy_snapshot' => ['indicator' => ['period' => 20]]]; $with['strategy_snapshot_hash'] = ResultsV2::snapshotHash($with['strategy_snapshot']); " "$empty = ['strategy_revision_id' => 7, 'strategy_snapshot' => ['indicators' => []]]; $empty['strategy_snapshot_hash'] = ResultsV2::snapshotHash($empty['strategy_snapshot']); " "$multiphase = ['strategy_revision_id' => 7, 'strategy_snapshot' => ['instrument_id' => 92, 'phases' => [['phase' => 1, 'setup' => ['win_use_ind' => 1, 'indicator_parameters' => ['by_id' => ['4:64' => ['period' => 20]], 'sma' => ['period' => 20], 'scs' => ['period' => 20]]]]]]]; $multiphase['strategy_snapshot_hash'] = ResultsV2::snapshotHash($multiphase['strategy_snapshot']); " "echo json_encode([ResultsV2Observer::indicatorEvidenceRequired($with), ResultsV2Observer::indicatorEvidenceRequired($empty), ResultsV2::frozenIndicatorSeries($multiphase['strategy_snapshot'])]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([True, False, ['sma', 'scs']], json.loads(result.stdout)) observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") self.assertIn("indicator_evidence_missing", observer) def test_empty_by_id_indicator_registry_does_not_require_indicator_evidence(self): disabled_setup = { "grid_type_use_ind": 0, "amount_use_ind": 0, "step_use_ind": 0, "take_profit_use_ind": 0, "stop_loss_use_ind": 0, "first_order_dist_use_ind": 0, "order_dist_use_ind": 0, "win1_use_ind": 0, "win2_use_ind": 0, "win3_use_ind": 0, "win_use_ind": 0, "indicator_control_enabled": 0, "indicator_id": None, "indicator_type": None, "indicator_parameters": {"by_id": []}, "sma_period": 14, "ema_period": 20, "supertrend_atr_period": 10, "supertrend_multiplier": 3, } snapshot = { "phases": [ {"phase": phase, "setup": disabled_setup} for phase in range(1, 6) ] } supported_enabled = { "phases": [{ "setup": { "indicator_control_enabled": 1, "indicator_parameters": { "sma": {"period": 20}, "ema": {"period": 12}, "supertrend": {"atr_period": 10, "multiplier": 3}, } } }] } state = { "id": 42, "strategy_revision_id": 215, "runtime_snapshot_json": json.dumps(snapshot), } source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); " f"$supported=json_decode({json.dumps(json.dumps(supported_enabled))}, true); " f"$state=json_decode({json.dumps(json.dumps(state))}, true); " "$state['runtime_content_hash']=ResultsV2::strategySnapshotHash($snapshot); " "$supportedState=['strategy_revision_id'=>215,'runtime_snapshot_json'=>json_encode($supported),'runtime_content_hash'=>ResultsV2::strategySnapshotHash($supported)]; " "$disabledWithParams=[]; foreach (['sma'=>['period'=>20],'ema'=>['period'=>12],'supertrend'=>['atr_period'=>10]] as $type=>$params) { $disabledSnapshot=['phases'=>[['setup'=>['indicator_control_enabled'=>0,'grid_type_use_ind'=>0,'indicator_parameters'=>[$type=>$params]]]]]; $disabledState=['strategy_revision_id'=>215,'runtime_snapshot_json'=>json_encode($disabledSnapshot),'runtime_content_hash'=>ResultsV2::strategySnapshotHash($disabledSnapshot)]; $disabledWithParams[$type]=[ResultsV2::frozenIndicatorSeries($disabledSnapshot),ResultsV2Observer::indicatorEvidenceRequired($disabledState)]; } " "$summary=ResultsV2Observer::terminalSummary($state, [], [], true, 'end_reached', true, 0, 0, []); " "echo json_encode([ResultsV2::frozenIndicatorSeries($snapshot), ResultsV2Observer::indicatorEvidenceRequired($state), $disabledWithParams, ResultsV2::frozenIndicatorSeries($supported), ResultsV2Observer::indicatorEvidenceRequired($supportedState), $summary['data_completeness'], $summary['incomplete_reasons']]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) series, required, disabled_with_params, supported_series, supported_required, completeness, reasons = json.loads(result.stdout) self.assertEqual([], series) self.assertFalse(required) self.assertEqual( {"sma": [[], False], "ema": [[], False], "supertrend": [[], False]}, disabled_with_params, ) self.assertEqual(["sma", "ema", "supertrend"], supported_series) self.assertTrue(supported_required) self.assertEqual("complete", completeness) self.assertEqual([], reasons) def test_terminal_metrics_consume_full_series_without_1000_point_truncation(self): state = { "id": 42, "start_equity": 100.0, "equity": 120.0, "fee": 0.1, "position": 0.0, "average_price": 0.0, "timestamp": 1499 * 60, "strategy_revision_id": 7, "strategy_snapshot": {"instrument_id": 92, "phases": []}, } source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Store.php'))}; " "class FakeChunkedDb { " "public $points = []; public $queries = []; public $failAfter = null; public $fakeCount = null; " "public function mysqli_fetch($sql, $params = []) { " "if ($this->failAfter !== null && count($this->queries) >= $this->failAfter) return false; " "$this->queries[] = $sql; " "if (str_contains($sql, 'COUNT(*)')) return ['point_count' => $this->fakeCount !== null ? $this->fakeCount : count($this->points)]; " "return false; } " "public function mysqli_fetch_all($sql, $params = []) { " "if ($this->failAfter !== null && count($this->queries) >= $this->failAfter) return false; " "$this->queries[] = $sql; " "if (str_contains($sql, 'ORDER BY `timestamp` ASC LIMIT')) { " "$lastTimestamp = $params['values'][1]; preg_match('/LIMIT\\s+(\\d+)/', $sql, $m); " "$limit = isset($m[1]) ? (int)$m[1] : 1000; $filtered = []; " "foreach ($this->points as $p) { if ($p['timestamp'] > $lastTimestamp) { " "$filtered[] = $p; " "if (count($filtered) >= $limit) break; } } " "return $filtered; } return false; } } " "$fakeDb = new FakeChunkedDb(); " "$points = []; for ($i = 0; $i < 1500; $i++) { " "$equity = 100.0; if ($i === 500) $equity = 200.0; elseif ($i === 750) $equity = 50.0; elseif ($i > 750) $equity = 120.0; " "$fakeDb->points[] = ['timestamp' => $i * 60, 'wallet_balance' => $equity, 'unrealized_pnl' => 0.0]; } " "$store = (new ReflectionClass(ResultsV2Store::class))->newInstanceWithoutConstructor(); " "$prop = new ReflectionProperty(ResultsV2Store::class, 'resultsDb'); $prop->setAccessible(true); $prop->setValue($store, $fakeDb); " "$terminal = $store->getTerminalEquityPoints(42, 100.0, 500); " f"$state=json_decode({json.dumps(json.dumps(state))}, true); " "$state['strategy_snapshot_hash']=ResultsV2::strategySnapshotHash($state['strategy_snapshot']); " "$summary=ResultsV2Observer::terminalSummary($state, $terminal, [], true, 'end_reached', true, 0, 0, []); " "$failingDb = new FakeChunkedDb(); $failingDb->points = $fakeDb->points; $failingDb->failAfter = 2; " "$failingStore = (new ReflectionClass(ResultsV2Store::class))->newInstanceWithoutConstructor(); $prop->setValue($failingStore, $failingDb); " "$failedTerminal = $failingStore->getTerminalEquityPoints(42, 100.0, 500); " "$mismatchDb = new FakeChunkedDb(); $mismatchDb->points = array_slice($fakeDb->points, 0, 1200); $mismatchDb->fakeCount = 1500; " "$mismatchStore = (new ReflectionClass(ResultsV2Store::class))->newInstanceWithoutConstructor(); $prop->setValue($mismatchStore, $mismatchDb); " "$mismatchTerminal = $mismatchStore->getTerminalEquityPoints(42, 100.0, 500); " "echo json_encode([count($fakeDb->queries), $terminal['point_count'], $terminal['truncated'], $summary['max_drawdown'], $summary['peak_timestamp'], $summary['trough_timestamp'], $summary['final_equity'], $summary['data_completeness'], $summary['incomplete_reasons'], $failedTerminal['truncated'], $mismatchTerminal['truncated']]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) queries_count, count, truncated, max_drawdown, peak_time, trough_time, final_equity, completeness, reasons, failed_truncated, mismatch_truncated = json.loads(result.stdout) self.assertEqual(4, queries_count) self.assertEqual(1500, count) self.assertFalse(truncated) self.assertEqual(75.0, max_drawdown) self.assertEqual(30000, peak_time) self.assertEqual(45000, trough_time) self.assertEqual(120.0, final_equity) self.assertEqual("complete", completeness) self.assertNotIn("equity_points_truncated", reasons) self.assertTrue(failed_truncated) self.assertTrue(mismatch_truncated) def test_terminal_summary_is_idempotent_and_never_claims_complete_without_a_transaction_buffer(self): state = { "id": 42, "start_equity": 10.0, "equity": 11.0, "fee": 0.1, "position": 0.0, "average_price": 0.0, "timestamp": 180, } points = [{"timestamp": 60, "wallet_balance": 10.0, "unrealized_pnl": 0.0}, {"timestamp": 180, "wallet_balance": 11.0, "unrealized_pnl": 0.0}] transactions = [{"timestamp": 60, "position": 1.0, "pnl": 0.0}, {"timestamp": 180, "position": 0.0, "pnl": 1.0}] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$state = json_decode({json.dumps(json.dumps(state))}, true); " "$state['strategy_revision_id'] = 7; $state['strategy_snapshot'] = ['indicator' => ['period' => 20]]; $state['strategy_snapshot_hash'] = ResultsV2::snapshotHash($state['strategy_snapshot']); " f"$points = json_decode({json.dumps(json.dumps(points))}, true); " f"$transactions = json_decode({json.dumps(json.dumps(transactions))}, true); " "$complete = ResultsV2Observer::terminalSummary($state, $points, $transactions, true, 'end_reached', true); " "$repeat = ResultsV2Observer::terminalSummary($state, $points, $transactions, true, 'end_reached', true); " "$partial = ResultsV2Observer::terminalSummary($state, $points, [], false, 'end_reached', true); " "echo json_encode([$complete, $repeat, $partial]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) complete, repeat, partial = json.loads(result.stdout) self.assertEqual(complete, repeat) self.assertEqual("complete", complete["data_completeness"]) self.assertEqual(1, complete["closed_trades"]) self.assertEqual("partial", partial["data_completeness"]) self.assertEqual(0, partial["closed_trades"]) transaction_log = (ROOT / "classes" / "TransactionLog.php").read_text(encoding="utf-8") self.assertIn("ResultsV2TelemetryBuffer", transaction_log) self.assertIn("->record($transaction_data)", transaction_log) self.assertIn("if (!$this->telemetryBuffer->record($transaction_data))", transaction_log) self.assertIn("$.results_v2_telemetry_incomplete", transaction_log) observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") self.assertIn("public function observe(array $testData, bool $markTelemetry = true)", observer) self.assertIn("$this->observe($testData, false);", observer) self.assertIn("empty($testData['results_v2_telemetry_incomplete'])", observer) buffer = (ROOT / "classes" / "ResultsV2TelemetryBuffer.php").read_text(encoding="utf-8") self.assertIn("if ($this->redis->exists($this->incompleteKey())) return null;", buffer) self.assertIn("$this->markIncomplete();", buffer) def test_real_order_count_and_wall_clock_terminal_runtime_are_buffered_idempotently(self): order = (ROOT / "classes" / "Order.php").read_text(encoding="utf-8") start = (ROOT / "api" / "private" / "start.php").read_text(encoding="utf-8") observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") buffer = (ROOT / "classes" / "ResultsV2TelemetryBuffer.php").read_text(encoding="utf-8") self.assertIn("->recordOrder($order_id)", order) self.assertIn("ResultsV2TelemetryBuffer($test_id)", start) self.assertIn("->mark();", start) self.assertIn("terminalRuntimeSeconds", buffer) self.assertIn("ordersCount", buffer) self.assertIn("'orders_count' => $ordersCount", observer) self.assertIn("'runtime_seconds' => $runtimeSeconds", observer) self.assertNotIn("end_time'] ?? 0) - (int)($testData['start_time']", observer) def test_terminal_observer_bounds_equity_and_cannot_block_the_stop_lifecycle(self): observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") store = (ROOT / "classes" / "ResultsV2Store.php").read_text(encoding="utf-8") self.assertIn("getTerminalEquityPoints($testId", observer) self.assertIn("public function getTerminalEquityPoints", store) self.assertIn("public function getTerminalTransactions", store) self.assertIn("equity_points_truncated", observer) self.assertIn("transactions_truncated", observer) source = ( "class ResultsV2TelemetryBuffer { " "public static $reasons = []; " "function __construct($testId) {} function mark() {} function markEvidence($name) {} " "function addIncompleteReason($reason) { self::$reasons[] = $reason; } " "function transactions() { return []; } function ordersCount() { return 0; } " "function terminalRuntimeSeconds() { return 0; } function hasEvidence($name) { return true; } " "function incompleteReasons() { return self::$reasons; } " "} " "class FailingTerminalStore { " "function persistEquityPoints($testId, $points) { return true; } " "function getTerminalEquityPoints($testId) { throw new RuntimeException('simulated results failure'); } " "} " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " "$observer = (new ReflectionClass('ResultsV2Observer'))->newInstanceWithoutConstructor(); " "$property = (new ReflectionClass('ResultsV2Observer'))->getProperty('store'); " "$property->setAccessible(true); $property->setValue($observer, new FailingTerminalStore()); " "$saved = $observer->persistTerminal(['id' => 42, 'equity' => 100, 'timestamp' => 1], 'stopped', false); " "echo json_encode([$saved, ResultsV2TelemetryBuffer::$reasons]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) saved, reasons = json.loads(result.stdout) self.assertFalse(saved) self.assertIn("terminal_summary_failed", reasons) def test_evidence_write_failures_have_durable_reasons_and_force_partial(self): observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") buffer = (ROOT / "classes" / "ResultsV2TelemetryBuffer.php").read_text(encoding="utf-8") for reason in ( "equity_point_write_failed", "indicator_write_failed", "indicator_provenance_missing", "transaction_write_failed", "terminal_summary_write_failed", "results_store_unavailable", ): self.assertIn(reason, observer + buffer) self.assertIn("incompleteReasons", buffer) self.assertIn("'incomplete_reasons' => $incompleteReasons", observer) def test_failed_transaction_buffer_write_produces_partial_terminal_evidence(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2TelemetryBuffer.php'))}; " "class FakeRedis { " "public $keys = []; public $sets = []; " "function set($key, $value) { $this->keys[$key] = $value; return 'OK'; } " "function exists($key) { return isset($this->keys[$key]) || isset($this->sets[$key]); } " "function sadd($key, $value) { $new = !isset($this->sets[$key][$value]); $this->sets[$key][$value] = true; return $new ? 1 : 0; } " "function srem($key, $value) { unset($this->sets[$key][$value]); return 1; } " "function smembers($key) { return array_keys($this->sets[$key] ?? []); } " "function jsonset($key, $path, $value) { return false; } " "function jsonget($key) { return null; } " "function jsondel($key, $path) { return 1; } " "function del($key) { unset($this->keys[$key], $this->sets[$key]); return 1; } " "} " "$buffer = new ResultsV2TelemetryBuffer(42, new FakeRedis()); " "$saved = $buffer->record(['id' => '42-1', 'timestamp' => 60]); " "echo json_encode([$saved, $buffer->transactions()]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) saved, transactions = json.loads(result.stdout) self.assertFalse(saved) self.assertIsNone(transactions) def test_order_count_runtime_and_incomplete_recovery_are_idempotent(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2TelemetryBuffer.php'))}; " "class RuntimeRedis { public $keys = []; public $sets = []; " "function set($k,$v){$this->keys[$k]=$v;return 'OK';} function get($k){return $this->keys[$k]??null;} " "function exists($k){return isset($this->keys[$k])||isset($this->sets[$k]);} " "function sadd($k,$v){$n=!isset($this->sets[$k][$v]);$this->sets[$k][$v]=true;return $n?1:0;} " "function smembers($k){return array_keys($this->sets[$k]??[]);} function srem($k,$v){unset($this->sets[$k][$v]);return 1;} " "function jsonset($k,$p,$v){$this->keys[$k]=$v;return 'OK';} function jsonget($k){return $this->keys[$k]??null;} " "function jsondel($k,$p){unset($this->keys[$k]);return 1;} function del($k){unset($this->keys[$k],$this->sets[$k]);return 1;} } " "$redis=new RuntimeRedis();$buffer=new ResultsV2TelemetryBuffer(42,$redis);$buffer->mark();" "$buffer->recordOrder('o1');$buffer->recordOrder('o1');$buffer->recordOrder('o2');" "$runtime1=$buffer->terminalRuntimeSeconds();$runtime2=$buffer->terminalRuntimeSeconds();" "$buffer->addIncompleteReason('indicator_provenance_missing');$buffer->record(['id'=>'t1','timestamp'=>1]);" "echo json_encode([$buffer->ordersCount(),$runtime1,$runtime2,$buffer->incompleteReasons(),$buffer->transactions()]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) orders, runtime1, runtime2, reasons, transactions = json.loads(result.stdout) self.assertEqual(2, orders) self.assertEqual(runtime1, runtime2) self.assertEqual(["indicator_provenance_missing"], reasons) self.assertEqual(1, len(transactions)) def test_order_and_evidence_redis_failures_are_not_masked_and_duplicates_are_idempotent(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2TelemetryBuffer.php'))}; " "class FakeRedis { public $keys=[]; public $sets=[]; public $fail=false; " "function exists($k){return isset($this->keys[$k])||isset($this->sets[$k]);} function set($k,$v){$this->keys[$k]=$v;return 'OK';} function get($k){return $this->keys[$k]??null;} " "function sadd($k,$v){if($this->fail)return 0; $new=!isset($this->sets[$k][$v]);$this->sets[$k][$v]=true;return $new?1:0;} function smembers($k){return array_keys($this->sets[$k]??[]);} " "function jsonset($k,$p,$v){return 'OK';} function srem($k,$v){unset($this->sets[$k][$v]);return 1;} function del($k){unset($this->keys[$k],$this->sets[$k]);return 1;} " "} $r=new FakeRedis(); $b=new ResultsV2TelemetryBuffer(42,$r); $first=$b->recordOrder('o1'); $dup=$b->recordOrder('o1'); $r->fail=true; $failed=$b->recordOrder('o2'); $evidence=$b->markEvidence('equity'); echo json_encode([$first,$dup,$failed,$evidence,$b->ordersCount(),$b->incompleteReasons()]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) first, duplicate, failed, evidence, count, reasons = json.loads(result.stdout) self.assertTrue(first) self.assertTrue(duplicate) self.assertFalse(failed) self.assertFalse(evidence) self.assertEqual(1, count) self.assertIn("order_write_failed", reasons) self.assertIn("equity_evidence_write_failed", reasons) if __name__ == "__main__": unittest.main()