315 lines
11 KiB
Python
315 lines
11 KiB
Python
import copy
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from quant60.target_package import (
|
|
DuplicateTargetDecisionError,
|
|
SHA256_FIELDS,
|
|
TargetPackageError,
|
|
TargetPackageNotFoundError,
|
|
TargetPackageTape,
|
|
TargetPackageV1,
|
|
load_target_package_tape,
|
|
seal_target_package,
|
|
target_package_sha256,
|
|
validate_target_package,
|
|
)
|
|
|
|
|
|
def package_body(
|
|
*,
|
|
package_id: str = "TP-20260724-A",
|
|
decision_id: str = "D-20260724-A",
|
|
signal_as_of: str = "2026-07-24T15:00:00+08:00",
|
|
next_session: str = "2026-07-27",
|
|
release: str = "quant60-baseline-60-v1",
|
|
model_id: str = "alpha158-lightgbm-v1",
|
|
) -> dict:
|
|
return {
|
|
"schema_version": "1.0",
|
|
"package_type": "TargetPackageV1",
|
|
"package_id": package_id,
|
|
"decision_id": decision_id,
|
|
"release": release,
|
|
"evidence_class": "provider-snapshot-local",
|
|
"model_id": model_id,
|
|
"signal_as_of": signal_as_of,
|
|
"next_session": next_session,
|
|
"model_sha256": "1" * 64,
|
|
"data_sha256": "2" * 64,
|
|
"feature_sha256": "3" * 64,
|
|
"risk_sha256": "4" * 64,
|
|
"cost_sha256": "5" * 64,
|
|
"optimizer_sha256": "6" * 64,
|
|
"config_sha256": "7" * 64,
|
|
"source_sha256": "8" * 64,
|
|
"target_weights": {
|
|
"000001.XSHE": 0.30,
|
|
"600000.XSHG": 0.20,
|
|
},
|
|
"universe": {
|
|
"000001.XSHE": {
|
|
"state": "openable",
|
|
"reasons": ["ELIGIBLE"],
|
|
},
|
|
"600000.XSHG": {
|
|
"state": "hold_only",
|
|
"reasons": ["LIQUIDITY_BELOW_FLOOR"],
|
|
},
|
|
},
|
|
"constraints": {
|
|
"constraint_state": "FEASIBLE",
|
|
"long_only": True,
|
|
"max_single_weight": 0.40,
|
|
"max_gross_weight": 0.80,
|
|
"max_one_way_turnover": 0.20,
|
|
"actual_gross_weight": 0.50,
|
|
},
|
|
}
|
|
|
|
|
|
def sealed_package(**overrides) -> dict:
|
|
return seal_target_package(package_body(**overrides))
|
|
|
|
|
|
class TargetPackageTests(unittest.TestCase):
|
|
def test_valid_package_is_deterministic_and_quantity_free(self):
|
|
package = sealed_package()
|
|
self.assertEqual(
|
|
package["package_sha256"],
|
|
target_package_sha256(package),
|
|
)
|
|
self.assertEqual(len(package["package_sha256"]), 64)
|
|
self.assertNotIn("quantity", json.dumps(package))
|
|
self.assertEqual(validate_target_package(package), package)
|
|
wrapped = TargetPackageV1.from_dict(package)
|
|
self.assertEqual(wrapped.package_id, package["package_id"])
|
|
self.assertEqual(wrapped.as_dict(), package)
|
|
|
|
def test_tamper_is_rejected(self):
|
|
package = sealed_package()
|
|
package["target_weights"]["000001.XSHE"] = 0.25
|
|
package["constraints"]["actual_gross_weight"] = 0.45
|
|
with self.assertRaisesRegex(TargetPackageError, "hash mismatch"):
|
|
validate_target_package(package)
|
|
|
|
def test_every_layer_hash_is_required_and_strict(self):
|
|
package = sealed_package()
|
|
for field in SHA256_FIELDS:
|
|
with self.subTest(field=field):
|
|
missing = copy.deepcopy(package)
|
|
missing.pop(field)
|
|
with self.assertRaises(TargetPackageError):
|
|
validate_target_package(missing)
|
|
malformed = copy.deepcopy(package)
|
|
malformed[field] = "A" * 64
|
|
with self.assertRaises(TargetPackageError):
|
|
validate_target_package(malformed)
|
|
placeholder = copy.deepcopy(package)
|
|
placeholder[field] = "0" * 64
|
|
placeholder["package_sha256"] = target_package_sha256(
|
|
placeholder
|
|
)
|
|
with self.assertRaisesRegex(
|
|
TargetPackageError,
|
|
"all-zero placeholder",
|
|
):
|
|
validate_target_package(placeholder)
|
|
|
|
def test_portable_momentum_cannot_masquerade_as_baseline(self):
|
|
body = package_body(model_id="portable-momentum-v1")
|
|
with self.assertRaisesRegex(TargetPackageError, "cannot claim"):
|
|
seal_target_package(body)
|
|
|
|
smoke = package_body(
|
|
release="platform-smoke-v1",
|
|
model_id="portable-momentum-v1",
|
|
)
|
|
self.assertEqual(
|
|
seal_target_package(smoke)["model_id"],
|
|
"portable-momentum-v1",
|
|
)
|
|
|
|
def test_quantities_are_forbidden_at_any_depth(self):
|
|
for mutation in (
|
|
lambda body: body.update({"target_quantity": 100}),
|
|
lambda body: body["constraints"].update({"quantity": 100}),
|
|
lambda body: body["universe"]["000001.XSHE"].update(
|
|
{"sellable_quantity": 100}
|
|
),
|
|
):
|
|
with self.subTest(mutation=mutation):
|
|
body = package_body()
|
|
mutation(body)
|
|
with self.assertRaisesRegex(
|
|
TargetPackageError,
|
|
"do not belong",
|
|
):
|
|
seal_target_package(body)
|
|
|
|
def test_clock_must_be_canonical_close_and_future_weekday(self):
|
|
cases = (
|
|
("2026-07-24T14:59:59+08:00", "2026-07-27"),
|
|
("2026-07-24T07:00:00+00:00", "2026-07-27"),
|
|
("2026-07-24T15:00:00.000000+08:00", "2026-07-27"),
|
|
("2026-07-24T15:00:00+08:00", "2026-07-24"),
|
|
("2026-07-24T15:00:00+08:00", "2026-07-25"),
|
|
)
|
|
for signal_as_of, next_session in cases:
|
|
with self.subTest(
|
|
signal_as_of=signal_as_of,
|
|
next_session=next_session,
|
|
):
|
|
with self.assertRaises(TargetPackageError):
|
|
seal_target_package(
|
|
package_body(
|
|
signal_as_of=signal_as_of,
|
|
next_session=next_session,
|
|
)
|
|
)
|
|
|
|
def test_symbols_weights_and_gross_fail_closed(self):
|
|
noncanonical = package_body()
|
|
noncanonical["target_weights"] = {
|
|
"000001.SZ": 0.30,
|
|
"600000.XSHG": 0.20,
|
|
}
|
|
with self.assertRaisesRegex(TargetPackageError, "canonical"):
|
|
seal_target_package(noncanonical)
|
|
|
|
missing_universe = package_body()
|
|
missing_universe["universe"].pop("000001.XSHE")
|
|
with self.assertRaisesRegex(TargetPackageError, "absent from universe"):
|
|
seal_target_package(missing_universe)
|
|
|
|
bad_gross = package_body()
|
|
bad_gross["constraints"]["actual_gross_weight"] = 0.49
|
|
with self.assertRaisesRegex(TargetPackageError, "does not equal"):
|
|
seal_target_package(bad_gross)
|
|
|
|
gross_breach = package_body()
|
|
gross_breach["constraints"]["max_gross_weight"] = 0.49
|
|
with self.assertRaisesRegex(TargetPackageError, "max_gross"):
|
|
seal_target_package(gross_breach)
|
|
|
|
single_breach = package_body()
|
|
single_breach["constraints"]["max_single_weight"] = 0.25
|
|
with self.assertRaisesRegex(TargetPackageError, "max_single"):
|
|
seal_target_package(single_breach)
|
|
|
|
excluded = package_body()
|
|
excluded["universe"]["000001.XSHE"] = {
|
|
"state": "excluded",
|
|
"reasons": ["NOT_PIT_INDEX_MEMBER"],
|
|
}
|
|
with self.assertRaisesRegex(TargetPackageError, "excluded"):
|
|
seal_target_package(excluded)
|
|
|
|
def test_tape_supports_multiple_decisions_with_exact_lookup(self):
|
|
first = sealed_package()
|
|
second = sealed_package(
|
|
package_id="TP-20260731-B",
|
|
decision_id="D-20260731-B",
|
|
signal_as_of="2026-07-31T15:00:00+08:00",
|
|
next_session="2026-08-03",
|
|
)
|
|
tape = TargetPackageTape([second, first])
|
|
self.assertEqual(len(tape), 2)
|
|
found = tape.lookup_exact(
|
|
signal_as_of=first["signal_as_of"],
|
|
next_session=first["next_session"],
|
|
)
|
|
self.assertEqual(found["decision_id"], first["decision_id"])
|
|
found["release"] = "mutated"
|
|
self.assertEqual(
|
|
tape.lookup_exact(
|
|
signal_as_of=first["signal_as_of"],
|
|
next_session=first["next_session"],
|
|
)["release"],
|
|
first["release"],
|
|
)
|
|
with self.assertRaises(TargetPackageNotFoundError):
|
|
tape.lookup_exact(
|
|
signal_as_of="2026-07-30T15:00:00+08:00",
|
|
next_session="2026-07-31",
|
|
)
|
|
with self.assertRaises(TargetPackageNotFoundError):
|
|
tape.lookup_exact(
|
|
signal_as_of=first["signal_as_of"],
|
|
next_session=first["next_session"],
|
|
decision_id="D-wrong",
|
|
)
|
|
|
|
def test_duplicate_decision_identity_is_rejected(self):
|
|
first = sealed_package()
|
|
duplicate_id = sealed_package(
|
|
package_id="TP-20260731-B",
|
|
decision_id=first["decision_id"],
|
|
signal_as_of="2026-07-31T15:00:00+08:00",
|
|
next_session="2026-08-03",
|
|
)
|
|
with self.assertRaises(DuplicateTargetDecisionError):
|
|
TargetPackageTape([first, duplicate_id])
|
|
|
|
duplicate_clock = sealed_package(
|
|
package_id="TP-20260724-B",
|
|
decision_id="D-20260724-B",
|
|
)
|
|
with self.assertRaises(DuplicateTargetDecisionError):
|
|
TargetPackageTape([first, duplicate_clock])
|
|
|
|
def test_json_and_jsonl_tapes_load_multiple_decisions(self):
|
|
packages = [
|
|
sealed_package(),
|
|
sealed_package(
|
|
package_id="TP-20260731-B",
|
|
decision_id="D-20260731-B",
|
|
signal_as_of="2026-07-31T15:00:00+08:00",
|
|
next_session="2026-08-03",
|
|
),
|
|
]
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
json_path = root / "tape.json"
|
|
json_path.write_text(
|
|
json.dumps({"packages": packages}),
|
|
encoding="utf-8",
|
|
)
|
|
jsonl_path = root / "tape.jsonl"
|
|
jsonl_path.write_text(
|
|
"".join(json.dumps(item) + "\n" for item in packages),
|
|
encoding="utf-8",
|
|
)
|
|
self.assertEqual(len(load_target_package_tape(json_path)), 2)
|
|
self.assertEqual(len(load_target_package_tape(jsonl_path)), 2)
|
|
|
|
def test_single_package_loads_and_strict_json_rejects_ambiguity(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
single = root / "single.json"
|
|
single.write_text(
|
|
json.dumps(sealed_package()),
|
|
encoding="utf-8",
|
|
)
|
|
self.assertEqual(len(load_target_package_tape(single)), 1)
|
|
|
|
duplicate = root / "duplicate.json"
|
|
duplicate.write_text(
|
|
'{"package_type":"TargetPackageV1",'
|
|
'"package_type":"TargetPackageV1"}',
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(TargetPackageError, "duplicate JSON key"):
|
|
load_target_package_tape(duplicate)
|
|
|
|
non_finite = root / "nan.json"
|
|
non_finite.write_text("[NaN]", encoding="utf-8")
|
|
with self.assertRaisesRegex(TargetPackageError, "non-finite"):
|
|
load_target_package_tape(non_finite)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|