139 lines
4.4 KiB
Python
139 lines
4.4 KiB
Python
from inspect import getsource
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from control_plane.trading_studio.management.commands.cohort_qualification_hyperscalper_001 import (
|
|
CONTRACT,
|
|
RUNNER_NAME,
|
|
Command,
|
|
)
|
|
from control_plane.trading_studio.management.commands.specimen_hyperscalper_cohort_001 import (
|
|
Command as SpecimenCommand,
|
|
SCENARIO_NAMES,
|
|
)
|
|
|
|
|
|
def _ledger(net_pnl=1.0):
|
|
return {
|
|
"exit_reason": "TP",
|
|
"take_profit_price": 101.0,
|
|
"entry_execution_price": 100.0,
|
|
"net_pnl": net_pnl,
|
|
"deq": {
|
|
"return_1_bars_bps": 10.0,
|
|
"return_2_bars_bps": None,
|
|
"return_3_bars_bps": None,
|
|
"return_5_bars_bps": None,
|
|
"return_10_bars_bps": None,
|
|
"mfe_bps": 12.0,
|
|
"mae_bps": -1.0,
|
|
"time_to_positive_bars": 1,
|
|
"time_to_25_bps_bars": None,
|
|
"time_to_50_bps_bars": None,
|
|
"max_adverse_before_positive_bps": -1.0,
|
|
"winner_negative_first": True,
|
|
"recovery_bars": 1,
|
|
},
|
|
}
|
|
|
|
|
|
def _legacy_specimen():
|
|
scenario = {
|
|
"qualification_run_id": "run",
|
|
"scenario": "",
|
|
"summary": {},
|
|
"ledger": [],
|
|
"ledger_reconciliation": {},
|
|
"deq": {},
|
|
"rescue": {},
|
|
}
|
|
return {
|
|
"machine": {
|
|
"contract": "Cohort001-first-member-specimen-v1",
|
|
"member_ordinal": 1,
|
|
"fold_count": 4,
|
|
"scenario_count_per_fold": 13,
|
|
"folds": [
|
|
{
|
|
"fold": f"Fold {number}",
|
|
"scenarios": [{**scenario, "scenario": name} for name in SCENARIO_NAMES],
|
|
}
|
|
for number in range(1, 5)
|
|
],
|
|
}
|
|
}
|
|
|
|
|
|
def test_cohort_command_contract_limits_new_execution_to_ordinals_two_through_twenty():
|
|
source = getsource(Command)
|
|
|
|
assert CONTRACT == "Cohort001-qualification-v1"
|
|
assert RUNNER_NAME.endswith(":v1")
|
|
assert len(SCENARIO_NAMES) == 13
|
|
assert "memberships[1:]" in source
|
|
assert "QualificationReplayRun.objects.filter" in source
|
|
assert "mixed runner versions" in source
|
|
assert "CanaryCommand._native_state" in source
|
|
assert "--specimen-report-sha" in source
|
|
assert "externally frozen SHA" in source
|
|
assert "require_protocol_identity=True" in source
|
|
assert "specimen_configuration_sha256" not in getsource(SpecimenCommand._report)
|
|
assert "allow_empty_raw=True" in source
|
|
|
|
|
|
def test_legacy_specimen_validation_requires_contract_four_folds_scenarios_and_ledger_fields():
|
|
report = _legacy_specimen()
|
|
|
|
Command._validate_specimen(report)
|
|
report["machine"]["folds"][0]["scenarios"][0]["ledger"] = [{"net_pnl": 1.0}]
|
|
with pytest.raises(ValueError, match="ledger fields"):
|
|
Command._validate_specimen(report)
|
|
|
|
|
|
def test_zero_trade_fold_reconciles_without_fabricated_ledger_rows():
|
|
summary = {
|
|
"ledger_count": 0,
|
|
"gross_pnl": 0.0,
|
|
"net_pnl": 0.0,
|
|
"rescue": {
|
|
"gross_pnl": 0.0,
|
|
"net_pnl": 0.0,
|
|
"commissions": 0.0,
|
|
"slippage": 0.0,
|
|
"funding": 0.0,
|
|
"other_cost": 0.0,
|
|
},
|
|
}
|
|
|
|
assert SpecimenCommand._reconcile(summary, [])["consistent"] is True
|
|
|
|
|
|
def test_indicator_expansion_reports_required_classes_and_missing_data_transparently():
|
|
expansion = Command._indicator_expansion(
|
|
[{"ledger": [_ledger(), _ledger(net_pnl=-1.0)]}]
|
|
)
|
|
|
|
assert expansion["status"] == "AVAILABLE"
|
|
assert expansion["classes"]["exit_reason"] == {"TP": 2}
|
|
assert expansion["classes"]["net_pnl"] == {"LOSS": 1, "WIN": 1}
|
|
assert expansion["deq_distributions"]["return_2_bars_bps"] == {
|
|
"available_count": 0,
|
|
"missing_count": 2,
|
|
"status": "INSUFFICIENT",
|
|
}
|
|
assert "Required classes" in Command._indicator_brief(expansion)
|
|
|
|
|
|
def test_machine_and_human_report_bind_catalog_and_external_specimen_hashes():
|
|
cohort = SimpleNamespace(id="cohort")
|
|
report = Command._report(cohort, [], "manifest", "dataset", "config", "specimen")
|
|
|
|
assert report["machine"]["protocol_identity"] == {
|
|
"cohort_manifest_sha256": "manifest",
|
|
"dataset_sha256": "dataset",
|
|
"qualification_scenario_catalog_sha256": "config",
|
|
"specimen_report_sha256": "specimen",
|
|
"runner_name": RUNNER_NAME,
|
|
}
|
|
assert report["human"]["indicator_expansion_status"] == "INSUFFICIENT"
|