93 lines
3 KiB
Python
93 lines
3 KiB
Python
from inspect import getsource
|
|
from types import SimpleNamespace
|
|
|
|
from control_plane.trading_studio.management.commands.specimen_hyperscalper_cohort_001 import (
|
|
SCENARIO_NAMES,
|
|
Command,
|
|
)
|
|
|
|
|
|
def _ledger(*, net=8.0):
|
|
return {
|
|
"gross_pnl": 10.0,
|
|
"net_pnl": net,
|
|
"entry_commission": 1.0,
|
|
"exit_commission": 1.0,
|
|
"entry_slippage": 0.25,
|
|
"exit_slippage": 0.25,
|
|
"funding": 0.0,
|
|
"other_cost": 0.0,
|
|
"deq": {
|
|
"return_1_bars_bps": 5.0,
|
|
"return_2_bars_bps": None,
|
|
"return_3_bars_bps": None,
|
|
"return_5_bars_bps": None,
|
|
"return_10_bars_bps": None,
|
|
"mfe_bps": 10.0,
|
|
"mae_bps": -2.0,
|
|
"time_to_positive_bars": 1,
|
|
"time_to_25_bps_bars": None,
|
|
"time_to_50_bps_bars": None,
|
|
"max_adverse_before_positive_bps": -2.0,
|
|
"winner_negative_first": True,
|
|
"recovery_bars": 1,
|
|
},
|
|
}
|
|
|
|
|
|
def test_specimen_contract_is_complete_and_limits_execution_to_first_member():
|
|
source = getsource(Command)
|
|
|
|
assert len(SCENARIO_NAMES) == 13
|
|
assert "TOTAL_COST_6BP" in SCENARIO_NAMES
|
|
assert "STOP_WIDTH_25" in SCENARIO_NAMES
|
|
assert "memberships[0]" in source
|
|
|
|
|
|
def test_deq_report_aggregates_every_primitive_including_missing_values_and_booleans():
|
|
report = Command._deq_aggregate([_ledger()])
|
|
|
|
assert report["trade_count"] == 1
|
|
assert set(report["fields"]) == set(_ledger()["deq"])
|
|
assert report["fields"]["return_2_bars_bps"] == {"available_count": 0, "mean": None}
|
|
assert report["fields"]["winner_negative_first"] == {
|
|
"available_count": 1,
|
|
"true_count": 1,
|
|
"false_count": 0,
|
|
}
|
|
|
|
|
|
def test_ledger_reconciliation_checks_summary_and_rescue_totals():
|
|
ledger = [_ledger()]
|
|
summary = {
|
|
"ledger_count": 1,
|
|
"gross_pnl": 10.0,
|
|
"net_pnl": 8.0,
|
|
"rescue": {
|
|
"gross_pnl": 10.0,
|
|
"net_pnl": 8.0,
|
|
"commissions": 2.0,
|
|
"slippage": 0.5,
|
|
"funding": 0.0,
|
|
"other_cost": 0.0,
|
|
},
|
|
}
|
|
|
|
assert Command._reconcile(summary, ledger)["consistent"] is True
|
|
summary["rescue"]["net_pnl"] = 7.0
|
|
assert Command._reconcile(summary, ledger)["consistent"] is False
|
|
summary["rescue"]["net_pnl"] = 8.0
|
|
ledger[0]["net_pnl"] = 7.0
|
|
assert Command._reconcile(summary, ledger)["row_pnl_consistent"] is False
|
|
|
|
|
|
def test_report_has_machine_and_human_sections_and_refuses_monetization_claim_without_criteria():
|
|
scenario = {"ledger_reconciliation": {"consistent": True}}
|
|
cohort = SimpleNamespace(id="cohort", dataset_version_id="dataset", policy_snapshot={})
|
|
membership = SimpleNamespace(ordinal=1, id="member", strategy_version_id="strategy")
|
|
|
|
report = Command._report(cohort, membership, [{"fold": "Fold 1", "scenarios": [scenario]}])
|
|
|
|
assert report["machine"]["edge_vs_monetization"]["status"] == "INSUFFICIENT"
|
|
assert report["human"]["ledger_reconciled"] is True
|
|
assert report["machine"]["member_ordinal"] == 1
|