504 lines
17 KiB
Python
504 lines
17 KiB
Python
import ast
|
|
import copy
|
|
import hashlib
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from quant60.ledger import canonical_json
|
|
from quant60.portable_core import (
|
|
TargetPackageLookupMiss,
|
|
_any_true,
|
|
build_rebalance_plan,
|
|
build_target_weight_plan,
|
|
capped_target_weights,
|
|
lookup_target_package_exact,
|
|
momentum_score,
|
|
normalize_symbol,
|
|
order_deltas,
|
|
target_package_tape_sha256,
|
|
target_quantities,
|
|
verify_target_package,
|
|
)
|
|
from quant60.target_package import seal_target_package
|
|
|
|
|
|
def sealed_target_package(
|
|
package_id="TP-20260724-A",
|
|
decision_id="D-20260724-A",
|
|
signal_as_of="2026-07-24T15:00:00+08:00",
|
|
next_session="2026-07-27",
|
|
):
|
|
return seal_target_package(
|
|
{
|
|
"schema_version": "1.0",
|
|
"package_type": "TargetPackageV1",
|
|
"package_id": package_id,
|
|
"decision_id": decision_id,
|
|
"release": "quant60-baseline-60-v1",
|
|
"evidence_class": "provider-snapshot-local",
|
|
"model_id": "alpha158-lightgbm-v1",
|
|
"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,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
class PortableCoreTests(unittest.TestCase):
|
|
def test_private_any_true_preserves_builtin_semantics(self):
|
|
for values in (
|
|
[],
|
|
[False, 0, None, ""],
|
|
[False, 0, "truthy"],
|
|
[1],
|
|
):
|
|
with self.subTest(values=values):
|
|
self.assertEqual(_any_true(iter(values)), any(values))
|
|
|
|
visited = []
|
|
|
|
def values():
|
|
for value in (False, True, "must-not-be-read"):
|
|
visited.append(value)
|
|
yield value
|
|
|
|
self.assertTrue(_any_true(values()))
|
|
self.assertEqual(visited, [False, True])
|
|
|
|
def test_symbol_round_trip(self):
|
|
inputs = [
|
|
("600000.XSHG", "600000.SH", "SH600000"),
|
|
("000001.XSHE", "000001.SZ", "SZ000001"),
|
|
("830799.XBSE", "830799.BJ", "BJ830799"),
|
|
]
|
|
for canonical, qmt, qlib in inputs:
|
|
self.assertEqual(normalize_symbol(qmt), canonical)
|
|
self.assertEqual(normalize_symbol(qlib), canonical)
|
|
self.assertEqual(normalize_symbol(canonical, "qmt"), qmt)
|
|
self.assertEqual(normalize_symbol(canonical, "qlib"), qlib)
|
|
|
|
def test_momentum_uses_latest_completed_bar(self):
|
|
self.assertAlmostEqual(momentum_score([10, 11, 12], 2, 0), 0.2)
|
|
self.assertAlmostEqual(momentum_score([10, 11, 12], 1, 1), 0.1)
|
|
|
|
def test_caps_and_cash_buffer_are_not_double_counted(self):
|
|
weights = capped_target_weights(
|
|
{"600000.XSHG": 3, "000001.XSHE": 2, "300750.XSHE": 1},
|
|
max_weight=0.4,
|
|
gross_target=0.95,
|
|
)
|
|
self.assertAlmostEqual(sum(weights.values()), 0.95)
|
|
self.assertLessEqual(max(weights.values()), 0.4 + 1e-12)
|
|
quantities = target_quantities(
|
|
weights,
|
|
{symbol: 10 for symbol in weights},
|
|
equity=1_000_000,
|
|
lot_size=100,
|
|
cash_buffer=0.02,
|
|
)
|
|
invested = sum(quantity * 10 for quantity in quantities.values())
|
|
self.assertGreater(invested, 930_000)
|
|
self.assertLessEqual(invested, 980_000)
|
|
|
|
def test_non_finite_or_out_of_range_portfolio_inputs_are_rejected(self):
|
|
for value in (float("nan"), float("inf"), float("-inf")):
|
|
with self.subTest(kind="score", value=value):
|
|
with self.assertRaises(ValueError):
|
|
capped_target_weights({"600000.SH": value})
|
|
with self.subTest(kind="weight", value=value):
|
|
with self.assertRaises(ValueError):
|
|
target_quantities(
|
|
{"600000.SH": value},
|
|
{"600000.SH": 10.0},
|
|
100_000,
|
|
)
|
|
with self.subTest(kind="equity", value=value):
|
|
with self.assertRaises(ValueError):
|
|
target_quantities(
|
|
{"600000.SH": 0.5},
|
|
{"600000.SH": 10.0},
|
|
value,
|
|
)
|
|
for max_weight, gross_target in ((1.1, 0.9), (0.2, 1.1)):
|
|
with self.subTest(max_weight=max_weight, gross_target=gross_target):
|
|
with self.assertRaises(ValueError):
|
|
capped_target_weights(
|
|
{"600000.SH": 1.0},
|
|
max_weight=max_weight,
|
|
gross_target=gross_target,
|
|
)
|
|
|
|
def test_sell_delta_respects_sellable(self):
|
|
deltas = order_deltas(
|
|
{"600000.XSHG": 0},
|
|
{"600000.XSHG": 1000},
|
|
{"600000.XSHG": 300},
|
|
lot_size=100,
|
|
)
|
|
self.assertEqual(deltas, {"600000.XSHG": -300})
|
|
|
|
def test_star_targets_and_partial_orders_respect_200_share_minimum(self):
|
|
self.assertEqual(
|
|
target_quantities(
|
|
{"688301.XSHG": 1.0},
|
|
{"688301.XSHG": 300.0},
|
|
equity=50_000,
|
|
lot_size=100,
|
|
cash_buffer=0.02,
|
|
),
|
|
{"688301.XSHG": 0},
|
|
)
|
|
self.assertEqual(
|
|
target_quantities(
|
|
{"688301.XSHG": 1.0},
|
|
{"688301.XSHG": 200.0},
|
|
equity=50_000,
|
|
lot_size=100,
|
|
cash_buffer=0.02,
|
|
),
|
|
{"688301.XSHG": 200},
|
|
)
|
|
self.assertEqual(
|
|
order_deltas(
|
|
{"688301.XSHG": 300},
|
|
{"688301.XSHG": 200},
|
|
{"688301.XSHG": 200},
|
|
lot_size=100,
|
|
),
|
|
{},
|
|
)
|
|
self.assertEqual(
|
|
order_deltas(
|
|
{"688301.XSHG": 1300},
|
|
{"688301.XSHG": 1400},
|
|
{"688301.XSHG": 1400},
|
|
lot_size=100,
|
|
),
|
|
{},
|
|
)
|
|
|
|
def test_complete_odd_lot_liquidation_is_preserved(self):
|
|
self.assertEqual(
|
|
order_deltas(
|
|
{"600000.XSHG": 0},
|
|
{"600000.XSHG": 1050},
|
|
{"600000.XSHG": 1050},
|
|
lot_size=100,
|
|
),
|
|
{"600000.XSHG": -1050},
|
|
)
|
|
self.assertEqual(
|
|
order_deltas(
|
|
{"688301.XSHG": 0},
|
|
{"688301.XSHG": 150},
|
|
{"688301.XSHG": 150},
|
|
lot_size=100,
|
|
),
|
|
{"688301.XSHG": -150},
|
|
)
|
|
|
|
def test_complete_plan_is_deterministic(self):
|
|
histories = {
|
|
"600000.SH": [10 + index * 0.1 for index in range(25)],
|
|
"000001.SZ": [10 + index * 0.05 for index in range(25)],
|
|
}
|
|
arguments = dict(
|
|
price_history=histories,
|
|
current={},
|
|
sellable={},
|
|
equity=100_000,
|
|
lookback=20,
|
|
skip=0,
|
|
top_n=2,
|
|
max_weight=0.5,
|
|
gross_target=0.95,
|
|
cash_buffer=0.02,
|
|
lot_size=100,
|
|
)
|
|
self.assertEqual(
|
|
build_rebalance_plan(**arguments),
|
|
build_rebalance_plan(**arguments),
|
|
)
|
|
|
|
def test_target_weight_plan_is_causal_and_preserves_lineage(self):
|
|
common = {
|
|
"prices": {
|
|
"600000.SH": 10.0,
|
|
"000001.SZ": 10.0,
|
|
},
|
|
"current": {},
|
|
"sellable": {},
|
|
"equity": 100_000,
|
|
"cash_buffer": 0.02,
|
|
"lot_size": 100,
|
|
"decision_id": "D-20260724-A",
|
|
"package_sha256": "a" * 64,
|
|
}
|
|
first = build_target_weight_plan(
|
|
{
|
|
"600000.XSHG": 0.60,
|
|
"000001.XSHE": 0.20,
|
|
},
|
|
**common
|
|
)
|
|
second = build_target_weight_plan(
|
|
{
|
|
"600000.XSHG": 0.20,
|
|
"000001.XSHE": 0.60,
|
|
},
|
|
**common
|
|
)
|
|
self.assertNotEqual(first["targets"], second["targets"])
|
|
self.assertNotEqual(first["orders"], second["orders"])
|
|
self.assertEqual(first["decision_id"], "D-20260724-A")
|
|
self.assertEqual(first["package_sha256"], "a" * 64)
|
|
self.assertNotIn("scores", first)
|
|
|
|
def test_target_weight_plan_rejects_alias_duplicates_and_bad_weights(self):
|
|
common = {
|
|
"prices": {"600000.XSHG": 10.0},
|
|
"current": {},
|
|
"sellable": {},
|
|
"equity": 100_000,
|
|
}
|
|
with self.assertRaisesRegex(ValueError, "duplicate normalized"):
|
|
build_target_weight_plan(
|
|
{
|
|
"600000.SH": 0.20,
|
|
"600000.XSHG": 0.20,
|
|
},
|
|
**common
|
|
)
|
|
for weights in (
|
|
{"600000.XSHG": -0.1},
|
|
{"600000.XSHG": float("nan")},
|
|
{"600000.XSHG": 1.01},
|
|
):
|
|
with self.subTest(weights=weights):
|
|
with self.assertRaises(ValueError):
|
|
build_target_weight_plan(weights, **common)
|
|
with self.assertRaisesRegex(ValueError, "gross"):
|
|
build_target_weight_plan(
|
|
{
|
|
"600000.XSHG": 0.60,
|
|
"000001.XSHE": 0.50,
|
|
},
|
|
prices={
|
|
"600000.XSHG": 10.0,
|
|
"000001.XSHE": 10.0,
|
|
},
|
|
current={},
|
|
sellable={},
|
|
equity=100_000,
|
|
)
|
|
|
|
def test_target_weight_plan_respects_t1_and_odd_lot_liquidation(self):
|
|
t1_limited = build_target_weight_plan(
|
|
{},
|
|
{"600000.XSHG": 10.0},
|
|
{"600000.XSHG": 1050},
|
|
{"600000.XSHG": 250},
|
|
equity=100_000,
|
|
)
|
|
self.assertEqual(t1_limited["weights"], {"600000.XSHG": 0.0})
|
|
self.assertEqual(t1_limited["targets"], {"600000.XSHG": 0})
|
|
self.assertEqual(t1_limited["orders"], {"600000.XSHG": -200})
|
|
|
|
complete = build_target_weight_plan(
|
|
{},
|
|
{"600000.XSHG": 10.0},
|
|
{"600000.XSHG": 1050},
|
|
{"600000.XSHG": 1050},
|
|
equity=100_000,
|
|
)
|
|
self.assertEqual(complete["orders"], {"600000.XSHG": -1050})
|
|
|
|
unsellable = build_target_weight_plan(
|
|
{},
|
|
{"600000.XSHG": 10.0},
|
|
{"600000.XSHG": 1050},
|
|
{},
|
|
equity=100_000,
|
|
)
|
|
self.assertEqual(unsellable["orders"], {})
|
|
|
|
def test_target_weight_plan_requires_exact_managed_price_coverage(self):
|
|
with self.assertRaisesRegex(KeyError, "missing"):
|
|
build_target_weight_plan(
|
|
{"600000.XSHG": 0.20},
|
|
{"600000.XSHG": 10.0},
|
|
{"000001.XSHE": 100},
|
|
{"000001.XSHE": 100},
|
|
equity=100_000,
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "unmanaged"):
|
|
build_target_weight_plan(
|
|
{"600000.XSHG": 0.20},
|
|
{
|
|
"600000.XSHG": 10.0,
|
|
"000001.XSHE": 10.0,
|
|
},
|
|
{},
|
|
{},
|
|
equity=100_000,
|
|
)
|
|
|
|
def test_portable_target_package_verifier_matches_local_seal(self):
|
|
package = sealed_target_package()
|
|
self.assertEqual(verify_target_package(package), package)
|
|
self.assertEqual(
|
|
verify_target_package(
|
|
package,
|
|
expected_signal_as_of=package["signal_as_of"],
|
|
expected_next_session=package["next_session"],
|
|
expected_decision_id=package["decision_id"],
|
|
expected_package_sha256=package["package_sha256"],
|
|
),
|
|
package,
|
|
)
|
|
|
|
tampered = copy.deepcopy(package)
|
|
tampered["target_weights"]["000001.XSHE"] = 0.29
|
|
tampered["constraints"]["actual_gross_weight"] = 0.49
|
|
with self.assertRaisesRegex(ValueError, "hash mismatch"):
|
|
verify_target_package(tampered)
|
|
|
|
quantity_injected = copy.deepcopy(package)
|
|
quantity_injected["target_quantity"] = 100
|
|
with self.assertRaisesRegex(ValueError, "quantities do not belong"):
|
|
verify_target_package(quantity_injected)
|
|
|
|
noncanonical = copy.deepcopy(package)
|
|
noncanonical["target_weights"]["000001.SZ"] = noncanonical[
|
|
"target_weights"
|
|
].pop("000001.XSHE")
|
|
with self.assertRaisesRegex(ValueError, "must be canonical"):
|
|
verify_target_package(noncanonical)
|
|
|
|
def test_target_package_tape_is_pinned_and_exact_only(self):
|
|
first = sealed_target_package()
|
|
second = sealed_target_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",
|
|
)
|
|
packages = [first, second]
|
|
tape_hash = target_package_tape_sha256(packages)
|
|
self.assertEqual(
|
|
tape_hash,
|
|
hashlib.sha256(
|
|
canonical_json(packages).encode("utf-8")
|
|
).hexdigest(),
|
|
)
|
|
found = lookup_target_package_exact(
|
|
packages,
|
|
signal_as_of=second["signal_as_of"],
|
|
next_session=second["next_session"],
|
|
decision_id=second["decision_id"],
|
|
expected_tape_sha256=tape_hash,
|
|
expected_count=2,
|
|
)
|
|
self.assertEqual(found, second)
|
|
found["release"] = "mutated"
|
|
self.assertEqual(packages[1]["release"], second["release"])
|
|
|
|
with self.assertRaises(TargetPackageLookupMiss):
|
|
lookup_target_package_exact(
|
|
packages,
|
|
signal_as_of="2026-07-30T15:00:00+08:00",
|
|
next_session="2026-07-31",
|
|
expected_tape_sha256=tape_hash,
|
|
expected_count=2,
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "decision_id mismatch"):
|
|
lookup_target_package_exact(
|
|
packages,
|
|
signal_as_of=first["signal_as_of"],
|
|
next_session=first["next_session"],
|
|
decision_id="D-WRONG",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "count mismatch"):
|
|
lookup_target_package_exact(
|
|
packages[:1],
|
|
signal_as_of=first["signal_as_of"],
|
|
next_session=first["next_session"],
|
|
expected_count=2,
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "digest mismatch"):
|
|
lookup_target_package_exact(
|
|
packages[:1],
|
|
signal_as_of=first["signal_as_of"],
|
|
next_session=first["next_session"],
|
|
expected_tape_sha256=tape_hash,
|
|
)
|
|
|
|
def test_target_package_tape_rejects_duplicates_and_unmatched_tamper(self):
|
|
first = sealed_target_package()
|
|
duplicate_clock = sealed_target_package(
|
|
package_id="TP-20260724-B",
|
|
decision_id="D-20260724-B",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "duplicate target package clock"):
|
|
lookup_target_package_exact(
|
|
[first, duplicate_clock],
|
|
signal_as_of=first["signal_as_of"],
|
|
next_session=first["next_session"],
|
|
)
|
|
|
|
later = sealed_target_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",
|
|
)
|
|
later["release"] = "tampered"
|
|
with self.assertRaisesRegex(ValueError, "hash mismatch"):
|
|
lookup_target_package_exact(
|
|
[first, later],
|
|
signal_as_of=first["signal_as_of"],
|
|
next_session=first["next_session"],
|
|
)
|
|
|
|
def test_portable_core_parses_as_python_36(self):
|
|
source = (
|
|
Path(__file__).parents[1]
|
|
/ "src"
|
|
/ "quant60"
|
|
/ "portable_core.py"
|
|
).read_text(encoding="utf-8")
|
|
ast.parse(source, feature_version=(3, 6))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|