import json import pathlib import subprocess import unittest ROOT = pathlib.Path(__file__).parents[1] FIXTURE = ROOT / "tests" / "fixtures" / "results_v2_golden.json" class ResultsV2ContractTests(unittest.TestCase): def run_php(self, source: str): result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) return json.loads(result.stdout) def test_golden_summary_uses_wallet_plus_unrealized_and_never_double_subtracts_fees(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$input = json_decode(file_get_contents({json.dumps(str(FIXTURE))}), true); " "echo json_encode(ResultsV2::summarize($input));" ) summary = self.run_php(source) self.assertEqual(9.0, summary["final_equity"]) self.assertEqual(-1.0, summary["net_pnl"]) self.assertEqual(-0.5, summary["gross_pnl"]) self.assertEqual(-0.15, summary["return_percent"]) self.assertEqual(25.0, summary["max_drawdown"]) self.assertEqual(120, summary["peak_timestamp"]) self.assertEqual(180, summary["trough_timestamp"]) self.assertEqual(2, summary["closed_trades"]) self.assertEqual(1, summary["wins"]) self.assertEqual(1, summary["losses"]) self.assertEqual(50.0, summary["win_rate"]) self.assertEqual(2.0, summary["profit_factor"]) self.assertEqual("complete", summary["data_completeness"]) self.assertEqual("v2", summary["methodology_version"]) self.assertEqual({"position": 0.0, "average_price": 0.0}, summary["terminal_position"]) def test_open_terminal_keeps_only_completed_round_trips_and_flat_no_trade_has_null_profit_factor(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " "$open = ResultsV2::summarize(['test_id' => 7, 'initial_equity' => 10, 'fees' => 0, " "'equity_points' => [['timestamp' => 60, 'wallet_balance' => 10, 'unrealized_pnl' => 0]], " "'transactions' => [['timestamp' => 60, 'position' => 1, 'pnl' => 0]], " "'terminal_position' => ['position' => 1, 'average_price' => 100]]); " "$flat = ResultsV2::summarize(['test_id' => 8, 'initial_equity' => 10, 'fees' => 0, " "'equity_points' => [['timestamp' => 60, 'wallet_balance' => 10, 'unrealized_pnl' => 0]], " "'transactions' => [], 'terminal_position' => ['position' => 0, 'average_price' => 0]]); " "echo json_encode([$open, $flat]);" ) open_summary, flat_summary = self.run_php(source) self.assertEqual(0, open_summary["closed_trades"]) self.assertEqual({"position": 1, "average_price": 100}, open_summary["terminal_position"]) self.assertEqual(0, flat_summary["closed_trades"]) self.assertEqual(0, flat_summary["wins"]) self.assertEqual(0, flat_summary["losses"]) self.assertIsNone(flat_summary["profit_factor"]) def test_trade_pnl_accumulates_partial_closes_and_reversal_closes_only_the_prior_trade(self): transactions = [ {"timestamp": 1, "position": 10, "pnl": 0}, {"timestamp": 2, "position": 5, "pnl": 3}, {"timestamp": 3, "position": 0, "pnl": 4}, {"timestamp": 4, "position": 2, "pnl": 0}, {"timestamp": 5, "position": -2, "pnl": -2}, {"timestamp": 6, "position": 0, "pnl": 5}, {"timestamp": 7, "position": 3, "pnl": 0}, {"timestamp": 8, "position": 1, "pnl": 9}, ] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$transactions = json_decode({json.dumps(json.dumps(transactions))}, true); " "echo json_encode(ResultsV2::summarize(['initial_equity' => 10, 'transactions' => $transactions]));" ) summary = self.run_php(source) self.assertEqual(3, summary["closed_trades"]) self.assertEqual(2, summary["wins"]) self.assertEqual(1, summary["losses"]) self.assertEqual(7.0, summary["best_win"]) self.assertEqual(5.0, summary["worst_win"]) self.assertEqual(-2.0, summary["worst_loss"]) def test_zero_initial_equity_has_null_return_and_equity_downsampling_preserves_full_range_edges(self): points = [{"timestamp": i, "wallet_balance": i, "unrealized_pnl": 0} for i in range(1, 21)] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$points = json_decode({json.dumps(json.dumps(points))}, true); " "$summary = ResultsV2::summarize(['initial_equity' => 0, 'equity_points' => $points]); " "$sample = ResultsV2::equityPoints($points, 5); " "echo json_encode([$summary, $sample]);" ) summary, sample = self.run_php(source) self.assertIsNone(summary["return_percent"]) self.assertEqual(5, len(sample)) self.assertEqual(1, sample[0]["timestamp"]) self.assertEqual(20, sample[-1]["timestamp"]) self.assertEqual(sample, self.run_php(source)[1]) def test_orders_count_is_not_inferred_from_transaction_count(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " "echo json_encode(ResultsV2::summarize(['transactions' => [" "['timestamp' => 1, 'position' => 1, 'pnl' => 0]," "['timestamp' => 2, 'position' => 0, 'pnl' => 1]" "]]));" ) summary = self.run_php(source) self.assertEqual(0, summary['orders_count']) self.assertEqual(2, summary['transactions_count']) def test_phase_hits_accept_real_instrument_labels_and_legacy_labels(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " "$input=['test_id'=>1,'initial_equity'=>10,'equity_points'=>[],'transactions'=>[" "['timestamp'=>1,'order_name'=>'grid_order-BTC-PERPETUAL-1']," "['timestamp'=>2,'order_name'=>'grid_order-ETH-PERPETUAL-2']," "['timestamp'=>3,'order_name'=>'grid_order-SOL_USDC-PERPETUAL-3']," "['timestamp'=>4,'order_name'=>'grid_order-4']]]; echo json_encode(ResultsV2::summarize($input)['phase_hits']);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([1, 1, 1, 1, 0], json.loads(result.stdout)) def test_indicator_snapshots_are_immutable_bounded_and_deterministically_downsampled(self): points = [ {"timestamp": index, "value": float(index), "price_context": {"close": index}} for index in range(1, 13) ] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$points = json_decode({json.dumps(json.dumps(points))}, true); " "$materialized = ['phase' => 1, 'indicator' => ['period' => 10, 'multiplier' => 3]]; " "$hash = ResultsV2::snapshotHash($materialized); " "$snapshot = ResultsV2::indicatorSnapshot(42, 'supertrend-main', 'supertrend', " "['period' => 10, 'multiplier' => 3], 7, $hash, $materialized, 1, 'entry', 'price', 'USD', " "['decision' => 'allow_entry'], $points, 5); " "$points[0]['price_context']['close'] = 999; " "echo json_encode($snapshot);" ) snapshot = self.run_php(source) self.assertEqual(5, len(snapshot)) self.assertEqual(1, snapshot[0]["timestamp"]) self.assertEqual(12, snapshot[-1]["timestamp"]) self.assertEqual(1, snapshot[0]["price_context"]["close"]) self.assertEqual("supertrend", snapshot[0]["series_type"]) self.assertEqual("entry", snapshot[0]["purpose"]) self.assertEqual(64, len(snapshot[0]["parameters_hash"])) self.assertEqual(7, snapshot[0]["strategy_revision_id"]) self.assertEqual("allow_entry", snapshot[0]["decision_context"]["decision"]) def test_indicator_provenance_rejects_snapshot_drift(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " "try { ResultsV2::indicatorSnapshot(1, 'sma', 'sma', ['period' => 2], 7, str_repeat('0', 64), " "['period' => 3], null, 'decision', 'price', 'USD', [], [['timestamp' => 1, 'value' => 1]]); " "echo 'no-error'; } catch (InvalidArgumentException $error) { echo $error->getMessage(); }" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual("Strategy snapshot hash mismatch", result.stdout) def test_legacy_result_suppresses_indicator_snapshots_and_true_intra_run_drawdown(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " "$summary = ResultsV2::summarize(['test_id' => 9, 'initial_equity' => 10, 'legacy' => true, " "'equity_points' => [['timestamp' => 60, 'wallet_balance' => 8, 'unrealized_pnl' => 0]]]); " "$points = ResultsV2::indicatorSnapshot(9, 'sma', 'sma', [], 1, str_repeat('0', 64), [], null, " "'entry', 'price', 'USD', [], [['timestamp' => 60, 'value' => 8]], 1000, true); " "echo json_encode([$summary, $points]);" ) summary, points = self.run_php(source) self.assertEqual("legacy_partial", summary["data_completeness"]) self.assertIsNone(summary["max_drawdown"]) self.assertEqual([], points) def test_migration_declares_idempotent_v2_entities_and_unique_keys(self): migration = (ROOT / "deploy" / "migrations" / "20260729_bee_results_v2.sql").read_text(encoding="utf-8") self.assertIn("CREATE TABLE IF NOT EXISTS `bee_test_result_v2`", migration) self.assertIn("CREATE TABLE IF NOT EXISTS `bee_test_equity_point_v2`", migration) self.assertIn("CREATE TABLE IF NOT EXISTS `bee_test_indicator_point_v2`", migration) self.assertIn("UNIQUE KEY `uq_bee_test_equity_point_v2` (`test_id`, `timestamp`)", migration) self.assertIn("UNIQUE KEY `uq_bee_test_indicator_point_v2` (`test_id`, `series_id`, `timestamp`)", migration) self.assertIn("`incomplete_reasons` JSON NOT NULL", migration) self.assertIn("`strategy_revision_id` BIGINT UNSIGNED NOT NULL", migration) self.assertIn("`strategy_snapshot_hash` CHAR(64) NOT NULL", migration) self.assertIn("`decision_context` JSON NOT NULL", migration) self.assertIn("`phase` INT NULL", migration) def test_store_binds_each_terminal_summary_field_with_its_sql_type(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Store.php'))}; " "echo ResultsV2Store::summaryBindTypes();" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual("idddddddiiiiiddddddddiiiiiisisiissss", result.stdout) store = (ROOT / "classes" / "ResultsV2Store.php").read_text(encoding="utf-8") self.assertIn("public function getEquityPoints(int $testId, ?int $maxPoints", store) self.assertIn("public function getIndicatorPoints(int $testId, string $seriesId, int $maxPoints", store) self.assertIn("ResultsV2::downsamplePoints", store) self.assertIn("data_completeness` = 'complete'", store) self.assertIn("VALUES(`data_completeness`)", store) if __name__ == "__main__": unittest.main()