105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from quant60.domain import Side
|
|
from quant60.execution import (
|
|
ExecutionBucket,
|
|
ParentOrderIntent,
|
|
ResidualPolicy,
|
|
market_data_is_fresh,
|
|
plan_guarded_twap_pov,
|
|
)
|
|
|
|
|
|
UTC8 = timezone(timedelta(hours=8))
|
|
|
|
|
|
class ExecutionPlannerTests(unittest.TestCase):
|
|
def _parent(self, quantity=1000):
|
|
return ParentOrderIntent(
|
|
decision_id="D-20260726",
|
|
symbol="600000.SH",
|
|
side=Side.BUY,
|
|
quantity=quantity,
|
|
limit_price=10.0,
|
|
window_start=datetime(2026, 7, 27, 9, 35, tzinfo=UTC8),
|
|
window_end=datetime(2026, 7, 27, 10, 0, tzinfo=UTC8),
|
|
market_data_as_of=datetime(2026, 7, 27, 9, 34, 59, tzinfo=UTC8),
|
|
max_participation=0.10,
|
|
max_child_quantity=400,
|
|
max_child_notional=3000,
|
|
residual_policy=ResidualPolicy.DEFER,
|
|
)
|
|
|
|
def test_plan_conserves_quantity_and_honors_all_caps(self):
|
|
parent = self._parent()
|
|
buckets = [
|
|
ExecutionBucket(
|
|
datetime(2026, 7, 27, 9, 35 + index * 5, tzinfo=UTC8),
|
|
2000,
|
|
)
|
|
for index in range(5)
|
|
]
|
|
plan = plan_guarded_twap_pov(parent, buckets)
|
|
self.assertEqual(
|
|
plan.planned_quantity + plan.residual_quantity,
|
|
parent.quantity,
|
|
)
|
|
self.assertTrue(plan.children)
|
|
for child in plan.children:
|
|
self.assertLessEqual(child.quantity, 200)
|
|
self.assertLessEqual(child.quantity * child.limit_price, 3000)
|
|
self.assertLessEqual(
|
|
child.quantity,
|
|
child.participation_cap_quantity,
|
|
)
|
|
self.assertEqual(
|
|
len({child.child_id for child in plan.children}),
|
|
len(plan.children),
|
|
)
|
|
|
|
def test_zero_volume_buckets_leave_explicit_residual(self):
|
|
parent = self._parent(quantity=500)
|
|
buckets = [
|
|
ExecutionBucket(
|
|
datetime(2026, 7, 27, 9, 35, tzinfo=UTC8),
|
|
0,
|
|
)
|
|
]
|
|
plan = plan_guarded_twap_pov(parent, buckets)
|
|
self.assertEqual(plan.planned_quantity, 0)
|
|
self.assertEqual(plan.residual_quantity, 500)
|
|
self.assertEqual(plan.residual_policy, ResidualPolicy.DEFER)
|
|
|
|
def test_market_data_freshness_rejects_old_and_future_quotes(self):
|
|
now = datetime(2026, 7, 27, 9, 35, tzinfo=UTC8)
|
|
self.assertTrue(
|
|
market_data_is_fresh(
|
|
as_of=now - timedelta(seconds=2),
|
|
now=now,
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
market_data_is_fresh(
|
|
as_of=now - timedelta(seconds=6),
|
|
now=now,
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
market_data_is_fresh(
|
|
as_of=now + timedelta(seconds=2),
|
|
now=now,
|
|
)
|
|
)
|
|
|
|
def test_parent_identity_is_decision_symbol_side(self):
|
|
parent = self._parent()
|
|
self.assertEqual(
|
|
parent.idempotency_key,
|
|
"D-20260726:600000.XSHG:BUY",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|