2180 lines
82 KiB
Python
2180 lines
82 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from quant60.backtest import _contract_manifest, _engine_source_manifest
|
|
from quant60.qmt_shadow import (
|
|
DECISION_EQUITY_ABSOLUTE_TOLERANCE,
|
|
DECISION_EQUITY_RELATIVE_TOLERANCE,
|
|
MAX_BROKER_ASSET_ABSOLUTE_TOLERANCE,
|
|
MAX_SHADOW_QUERY_DURATION_SECONDS,
|
|
QmtShadowPlanError,
|
|
_authenticated_evidence_envelope,
|
|
_authenticated_evidence_hmac,
|
|
build_qmt_shadow_plan,
|
|
verify_qmt_shadow_plan,
|
|
write_qmt_shadow_plan,
|
|
)
|
|
from quant60.ledger import canonical_json
|
|
from quant60.snapshot_pipeline import (
|
|
first_executable_window,
|
|
write_snapshot_decision,
|
|
)
|
|
|
|
|
|
UTC8 = timezone(timedelta(hours=8))
|
|
ACCOUNT_HASH_KEY = b"k" * 32
|
|
|
|
|
|
def fixture_account_hash(account_id: str = "account-fixture") -> str:
|
|
return hmac.new(
|
|
ACCOUNT_HASH_KEY,
|
|
("qmt-account-v1\0" + account_id).encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
|
|
def decision_artifact(
|
|
root: Path,
|
|
as_of: datetime,
|
|
*,
|
|
bind_account: bool = True,
|
|
broker_account_hash: str | None = None,
|
|
broker_observed_at: datetime | None = None,
|
|
next_session: date = date(2026, 7, 27),
|
|
decision_equity: float | None = 100_000,
|
|
first_target_quantity: int = 1000,
|
|
) -> str:
|
|
run_id = "q60d-shadow-fixture"
|
|
decision_id = "D-shadow-fixture"
|
|
data_version = "fixture-data-v1"
|
|
targets = [
|
|
{
|
|
"schema_version": "1.0",
|
|
"run_id": run_id,
|
|
"decision_id": decision_id,
|
|
"as_of": as_of.isoformat(),
|
|
"symbol": "000001.XSHE",
|
|
"target_weight": 0.10,
|
|
"target_quantity": first_target_quantity,
|
|
"lot_size": 100,
|
|
"signal_id": "S-000001",
|
|
"constraint_state": "FEASIBLE",
|
|
"config_hash": "fixture-config",
|
|
"metadata": {},
|
|
},
|
|
{
|
|
"schema_version": "1.0",
|
|
"run_id": run_id,
|
|
"decision_id": decision_id,
|
|
"as_of": as_of.isoformat(),
|
|
"symbol": "600000.XSHG",
|
|
"target_weight": 0.05,
|
|
"target_quantity": 500,
|
|
"lot_size": 100,
|
|
"signal_id": "S-600000",
|
|
"constraint_state": "FEASIBLE",
|
|
"config_hash": "fixture-config",
|
|
"metadata": {},
|
|
},
|
|
]
|
|
engine_hash, engine_sources = _engine_source_manifest()
|
|
contract_hash, contracts = _contract_manifest()
|
|
executable_window = first_executable_window(next_session)
|
|
default_observation = datetime(
|
|
next_session.year,
|
|
next_session.month,
|
|
next_session.day,
|
|
9,
|
|
0,
|
|
tzinfo=UTC8,
|
|
)
|
|
input_data_manifest = {
|
|
"data_version": data_version,
|
|
"next_trading_session": next_session.isoformat(),
|
|
"query": {
|
|
"trading_sessions": [
|
|
as_of.date().isoformat(),
|
|
next_session.isoformat(),
|
|
]
|
|
},
|
|
}
|
|
input_manifest_sha256 = hashlib.sha256(
|
|
(
|
|
json.dumps(
|
|
input_data_manifest,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
+ "\n"
|
|
).encode("utf-8")
|
|
).hexdigest()
|
|
result = {
|
|
"signals": [],
|
|
"targets": targets,
|
|
"decision": {
|
|
"schema_version": "1.0",
|
|
"run_id": run_id,
|
|
"decision_id": decision_id,
|
|
"as_of": as_of.isoformat(),
|
|
"signal_as_of": as_of.isoformat(),
|
|
"next_trading_session": next_session.isoformat(),
|
|
"first_executable_window": executable_window,
|
|
"evidence_class": "provider_snapshot_decision",
|
|
"provider": "fixture",
|
|
"provider_version": "1",
|
|
"data_version": data_version,
|
|
"input_manifest_sha256": input_manifest_sha256,
|
|
"config_hash": "fixture-config",
|
|
"model_version": "portable-momentum-v1",
|
|
"equity": decision_equity,
|
|
"weights": {
|
|
"000001.XSHE": 0.10,
|
|
"600000.XSHG": 0.05,
|
|
},
|
|
"targets": {
|
|
"000001.XSHE": first_target_quantity,
|
|
"600000.XSHG": 500,
|
|
},
|
|
"orders": {},
|
|
"universe": {},
|
|
"risk": {
|
|
"approved": True,
|
|
"gross_weight": 0.15,
|
|
"one_way_turnover": 0.10,
|
|
"violations": [],
|
|
},
|
|
"broker_snapshot": (
|
|
{
|
|
"snapshot_id": "QMT-SHADOW-fixture",
|
|
"signal_as_of": as_of.isoformat(),
|
|
"next_trading_session": next_session.isoformat(),
|
|
"first_executable_window": executable_window,
|
|
"observation_as_of": (
|
|
broker_observed_at or default_observation
|
|
).isoformat(),
|
|
"as_of": (
|
|
broker_observed_at or default_observation
|
|
).isoformat(),
|
|
"source_time": (
|
|
broker_observed_at or default_observation
|
|
).isoformat(),
|
|
"broker": "QMT_XTTRADER",
|
|
"account_hash": (
|
|
broker_account_hash or fixture_account_hash()
|
|
),
|
|
"snapshot_sha256": "fixture-snapshot-sha256",
|
|
}
|
|
if bind_account
|
|
else None
|
|
),
|
|
"investment_value_claim": False,
|
|
"real_platform_backtest": False,
|
|
},
|
|
"input_data_manifest": input_data_manifest,
|
|
"manifest": {
|
|
"schema_version": "1.0",
|
|
"run_id": run_id,
|
|
"evidence_class": "provider_snapshot_decision",
|
|
"data_version": data_version,
|
|
"input_manifest_sha256": input_manifest_sha256,
|
|
"config_hash": "fixture-config",
|
|
"engine_source_sha256": engine_hash,
|
|
"engine_sources": engine_sources,
|
|
"contracts_sha256": contract_hash,
|
|
"contract_schemas": contracts,
|
|
"model_version": "portable-momentum-v1",
|
|
"signal_as_of": as_of.isoformat(),
|
|
"signal_clock": "T_CLOSE_COMPLETE",
|
|
"first_executable_clock": "T_PLUS_1_OPEN",
|
|
"next_trading_session": next_session.isoformat(),
|
|
"first_executable_window": executable_window,
|
|
"price_adjustment": "PIT_FRONT_FROM_RAW_AND_FACTOR",
|
|
"deterministic": True,
|
|
},
|
|
}
|
|
return write_snapshot_decision(result, root)["manifest"]
|
|
|
|
|
|
class FakeReadOnlyAdapter:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
orders=None,
|
|
second_orders=None,
|
|
trades=None,
|
|
asset=None,
|
|
positions=None,
|
|
errors=None,
|
|
live_authorization_audit=None,
|
|
flip_live_after_query=False,
|
|
flip_state_after_query=False,
|
|
):
|
|
self.state = "READY"
|
|
self.allow_live_orders = False
|
|
self.errors = list(errors or [])
|
|
self.live_authorization_audit = list(
|
|
live_authorization_audit or []
|
|
)
|
|
self.calls = []
|
|
self.mutation_calls = []
|
|
self._orders = list(orders or [])
|
|
self._trades = list(trades or [])
|
|
self._positions = list(
|
|
positions
|
|
if positions is not None
|
|
else [
|
|
{
|
|
"account_id": "account-fixture",
|
|
"symbol": "600000.XSHG",
|
|
"volume": 1000,
|
|
"sellable": 1000,
|
|
"market_value": 10_000,
|
|
"avg_price": 10,
|
|
}
|
|
]
|
|
)
|
|
self._asset = dict(
|
|
asset
|
|
or {
|
|
"account_id": "account-fixture",
|
|
"cash": 90_000,
|
|
"market_value": 10_000,
|
|
"total_asset": 100_000,
|
|
}
|
|
)
|
|
self._second_orders = (
|
|
list(second_orders) if second_orders is not None else None
|
|
)
|
|
self._order_queries = 0
|
|
self._flip_live_after_query = bool(flip_live_after_query)
|
|
self._flip_state_after_query = bool(flip_state_after_query)
|
|
|
|
def query_asset(self):
|
|
self.calls.append("asset")
|
|
return dict(self._asset)
|
|
|
|
def query_positions(self):
|
|
self.calls.append("positions")
|
|
return list(self._positions)
|
|
|
|
def query_orders(self, cancelable_only=False):
|
|
self.calls.append(("orders", cancelable_only))
|
|
self._order_queries += 1
|
|
if self._order_queries == 2 and self._flip_live_after_query:
|
|
self.allow_live_orders = True
|
|
if self._order_queries == 2 and self._flip_state_after_query:
|
|
self.state = "DEGRADED"
|
|
if self._order_queries == 2 and self._second_orders is not None:
|
|
return list(self._second_orders)
|
|
return list(self._orders)
|
|
|
|
def query_trades(self):
|
|
self.calls.append("trades")
|
|
return list(self._trades)
|
|
|
|
def submit_order(self, *args, **kwargs):
|
|
self.mutation_calls.append(("submit", args, kwargs))
|
|
raise AssertionError("shadow planner must never submit")
|
|
|
|
def cancel_order(self, *args, **kwargs):
|
|
self.mutation_calls.append(("cancel", args, kwargs))
|
|
raise AssertionError("shadow planner must never cancel")
|
|
|
|
|
|
def fixed_clock(moment: datetime):
|
|
values = iter((moment, moment))
|
|
return lambda: next(values)
|
|
|
|
|
|
_UNCHANGED = object()
|
|
|
|
|
|
def payload_sha256(payload) -> str:
|
|
return hashlib.sha256(
|
|
canonical_json(payload).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def reseal_shadow_artifacts(
|
|
paths: dict[str, str],
|
|
plan: dict,
|
|
*,
|
|
broker_observation=_UNCHANGED,
|
|
broker_snapshot=_UNCHANGED,
|
|
evidence_hmac_key: bytes | None = None,
|
|
manifest_updates: dict | None = None,
|
|
) -> None:
|
|
if broker_observation is not _UNCHANGED:
|
|
Path(paths["broker_observation"]).write_text(
|
|
json.dumps(
|
|
broker_observation,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
if broker_snapshot is not _UNCHANGED:
|
|
Path(paths["broker_snapshot"]).write_text(
|
|
json.dumps(
|
|
broker_snapshot,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
plan["plan_id"] = ""
|
|
plan["plan_id"] = "QSP-" + payload_sha256(plan)[:20]
|
|
plan_path = Path(paths["plan"])
|
|
plan_path.write_text(
|
|
json.dumps(
|
|
plan,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
manifest_path = Path(paths["manifest"])
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["plan_id"] = plan["plan_id"]
|
|
if manifest_updates:
|
|
manifest.update(manifest_updates)
|
|
if evidence_hmac_key is not None:
|
|
if broker_observation is _UNCHANGED:
|
|
broker_observation = json.loads(
|
|
Path(paths["broker_observation"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
if broker_snapshot is _UNCHANGED:
|
|
broker_snapshot = json.loads(
|
|
Path(paths["broker_snapshot"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
manifest["authenticated_evidence_hmac_sha256"] = (
|
|
_authenticated_evidence_hmac(
|
|
_authenticated_evidence_envelope(
|
|
plan=plan,
|
|
broker_observation=broker_observation,
|
|
broker_snapshot=broker_snapshot,
|
|
decision_manifest_sha256=manifest[
|
|
"decision_manifest_sha256"
|
|
],
|
|
planner_source_sha256=manifest[
|
|
"planner_source_sha256"
|
|
],
|
|
engine_source_sha256=manifest[
|
|
"engine_source_sha256"
|
|
],
|
|
published_artifact_sha256={
|
|
name: hashlib.sha256(
|
|
Path(paths[path_key]).read_bytes()
|
|
).hexdigest()
|
|
for name, path_key in (
|
|
("shadow_plan.json", "plan"),
|
|
(
|
|
"broker_snapshot.json",
|
|
"broker_snapshot",
|
|
),
|
|
(
|
|
"broker_observation.json",
|
|
"broker_observation",
|
|
),
|
|
)
|
|
},
|
|
),
|
|
evidence_hmac_key,
|
|
)
|
|
)
|
|
for artifact_name, path_key in (
|
|
("shadow_plan.json", "plan"),
|
|
("broker_snapshot.json", "broker_snapshot"),
|
|
("broker_observation.json", "broker_observation"),
|
|
):
|
|
manifest["artifact_sha256"][artifact_name] = hashlib.sha256(
|
|
Path(paths[path_key]).read_bytes()
|
|
).hexdigest()
|
|
manifest_path.write_text(
|
|
json.dumps(
|
|
manifest,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def reseal_shadow_plan(paths: dict[str, str], plan: dict) -> None:
|
|
reseal_shadow_artifacts(
|
|
paths,
|
|
plan,
|
|
evidence_hmac_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
|
|
def force_ready_projection(plan: dict) -> None:
|
|
plan["blockers"] = []
|
|
plan["status"] = "SHADOW_READY"
|
|
plan["safety"]["ready_for_manual_review"] = True
|
|
for row in plan["target_diffs"]:
|
|
delta = int(row["board_lot_feasible_delta"])
|
|
row["proposed_delta"] = delta
|
|
row["proposed_side"] = (
|
|
"BUY" if delta > 0 else "SELL" if delta < 0 else None
|
|
)
|
|
row["proposed_quantity"] = abs(delta)
|
|
|
|
|
|
class QmtShadowPlanTests(unittest.TestCase):
|
|
def test_ready_plan_is_deterministic_read_only_and_verifiable(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, 5, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
bound_observation = observed - timedelta(seconds=5)
|
|
decision = decision_artifact(
|
|
root / "decision",
|
|
decision_time,
|
|
broker_observed_at=bound_observation,
|
|
)
|
|
first_adapter = FakeReadOnlyAdapter()
|
|
second_adapter = FakeReadOnlyAdapter()
|
|
first = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=first_adapter,
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
second = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=second_adapter,
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(first["plan"]["status"], "SHADOW_READY")
|
|
self.assertFalse(first["plan"]["blockers"])
|
|
equity_contract = first["plan"]["portfolio"][
|
|
"equity_consistency"
|
|
]
|
|
self.assertTrue(equity_contract["within_tolerance"])
|
|
self.assertEqual(
|
|
equity_contract["absolute_tolerance"],
|
|
DECISION_EQUITY_ABSOLUTE_TOLERANCE,
|
|
)
|
|
self.assertEqual(
|
|
equity_contract["relative_tolerance"],
|
|
DECISION_EQUITY_RELATIVE_TOLERANCE,
|
|
)
|
|
self.assertEqual(
|
|
first["plan"]["decision"]["signal_as_of"],
|
|
decision_time.isoformat(),
|
|
)
|
|
self.assertEqual(
|
|
first["plan"]["decision"][
|
|
"bound_broker_observation_as_of"
|
|
],
|
|
bound_observation.isoformat(),
|
|
)
|
|
rows = {
|
|
row["symbol"]: row
|
|
for row in first["plan"]["target_diffs"]
|
|
}
|
|
self.assertEqual(rows["000001.XSHE"]["proposed_delta"], 1000)
|
|
self.assertEqual(rows["600000.XSHG"]["proposed_delta"], -500)
|
|
self.assertAlmostEqual(
|
|
rows["600000.XSHG"]["current_weight"], 0.10
|
|
)
|
|
self.assertFalse(
|
|
first["plan"]["safety"]["ready_for_live_submission"]
|
|
)
|
|
self.assertEqual(first_adapter.mutation_calls, [])
|
|
self.assertEqual(
|
|
first_adapter.calls,
|
|
[
|
|
("orders", False),
|
|
"asset",
|
|
"positions",
|
|
"trades",
|
|
("orders", False),
|
|
],
|
|
)
|
|
paths = write_qmt_shadow_plan(first, root / "shadow")
|
|
verified = verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertTrue(verified["ok"])
|
|
self.assertTrue(verified["read_only"])
|
|
self.assertEqual(verified["status"], "SHADOW_READY")
|
|
observation = json.loads(
|
|
Path(paths["broker_observation"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
self.assertNotIn("account-fixture", json.dumps(observation))
|
|
self.assertEqual(
|
|
observation["asset"]["account_hash"],
|
|
fixture_account_hash(),
|
|
)
|
|
broker_snapshot = json.loads(
|
|
Path(paths["broker_snapshot"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
broker_snapshot["signal_as_of"],
|
|
decision_time.isoformat(),
|
|
)
|
|
self.assertEqual(
|
|
broker_snapshot["observation_as_of"],
|
|
observed.isoformat(),
|
|
)
|
|
self.assertEqual(
|
|
broker_snapshot["next_trading_session"],
|
|
"2026-07-27",
|
|
)
|
|
self.assertEqual(
|
|
broker_snapshot["first_executable_window"]["boundary"],
|
|
"[start,end)",
|
|
)
|
|
|
|
def test_low_and_high_current_equity_drift_fail_closed(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
assets = (
|
|
{
|
|
"account_id": "account-fixture",
|
|
"cash": 10_000,
|
|
"market_value": 10_000,
|
|
"total_asset": 20_000,
|
|
},
|
|
{
|
|
"account_id": "account-fixture",
|
|
"cash": 990_000,
|
|
"market_value": 10_000,
|
|
"total_asset": 1_000_000,
|
|
},
|
|
)
|
|
for asset in assets:
|
|
with self.subTest(total_asset=asset["total_asset"]), tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(asset=asset),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {
|
|
item["code"] for item in result["plan"]["blockers"]
|
|
}
|
|
self.assertIn("DECISION_EQUITY_DRIFT", codes)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
self.assertFalse(
|
|
result["plan"]["portfolio"]["equity_consistency"][
|
|
"within_tolerance"
|
|
]
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
row["proposed_delta"] == 0
|
|
for row in result["plan"]["target_diffs"]
|
|
)
|
|
)
|
|
# The current observation can still be trustworthy and is
|
|
# deliberately emitted so the operator can rebuild targets.
|
|
self.assertIsNotNone(result["broker_snapshot"])
|
|
|
|
def test_equity_tolerance_boundary_is_inclusive_and_fixed(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
cases = (
|
|
(100_010.0, True),
|
|
(100_010.01, False),
|
|
)
|
|
for total_asset, expected_ready in cases:
|
|
with self.subTest(total_asset=total_asset), tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(
|
|
asset={
|
|
"account_id": "account-fixture",
|
|
"cash": total_asset - 10_000,
|
|
"market_value": 10_000,
|
|
"total_asset": total_asset,
|
|
}
|
|
),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
contract = result["plan"]["portfolio"][
|
|
"equity_consistency"
|
|
]
|
|
self.assertEqual(contract["allowed_difference"], 10.0)
|
|
self.assertIs(
|
|
contract["within_tolerance"],
|
|
expected_ready,
|
|
)
|
|
self.assertEqual(
|
|
result["plan"]["status"],
|
|
"SHADOW_READY" if expected_ready else "BLOCKED",
|
|
)
|
|
codes = {
|
|
item["code"] for item in result["plan"]["blockers"]
|
|
}
|
|
self.assertIs(
|
|
"DECISION_EQUITY_DRIFT" not in codes,
|
|
expected_ready,
|
|
)
|
|
|
|
def test_invalid_decision_or_current_equity_fails_closed(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
for decision_equity in (None, -1.0, 0.0):
|
|
with self.subTest(decision_equity=decision_equity), tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
decision_equity=decision_equity,
|
|
)
|
|
adapter = FakeReadOnlyAdapter()
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"decision equity",
|
|
):
|
|
build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=adapter,
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertEqual(adapter.calls, [])
|
|
|
|
invalid_current_values = (None, -1.0, float("nan"))
|
|
for current_total_asset in invalid_current_values:
|
|
with self.subTest(current_total_asset=current_total_asset), tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(
|
|
asset={
|
|
"account_id": "account-fixture",
|
|
"cash": 90_000,
|
|
"market_value": 10_000,
|
|
"total_asset": current_total_asset,
|
|
}
|
|
),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {
|
|
item["code"] for item in result["plan"]["blockers"]
|
|
}
|
|
self.assertIn("INVALID_ASSET_FACT", codes)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
contract = result["plan"]["portfolio"][
|
|
"equity_consistency"
|
|
]
|
|
self.assertFalse(contract["evaluated"])
|
|
self.assertIsNone(contract["current_total_asset"])
|
|
self.assertIsNone(result["broker_snapshot"])
|
|
|
|
def test_position_changes_are_rediffed_from_current_sellable_facts(self):
|
|
class ChangedPositionAdapter(FakeReadOnlyAdapter):
|
|
def query_positions(self):
|
|
self.calls.append("positions")
|
|
return [
|
|
{
|
|
"account_id": "account-fixture",
|
|
"symbol": "600000.XSHG",
|
|
"volume": 500,
|
|
"sellable": 300,
|
|
"market_value": 10_000,
|
|
"avg_price": 10,
|
|
}
|
|
]
|
|
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=ChangedPositionAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
rows = {
|
|
row["symbol"]: row
|
|
for row in result["plan"]["target_diffs"]
|
|
}
|
|
self.assertEqual(result["plan"]["status"], "SHADOW_READY")
|
|
self.assertEqual(
|
|
rows["600000.XSHG"]["current_quantity"],
|
|
500,
|
|
)
|
|
self.assertEqual(
|
|
rows["600000.XSHG"]["sellable_quantity"],
|
|
300,
|
|
)
|
|
self.assertEqual(
|
|
rows["600000.XSHG"]["proposed_delta"],
|
|
0,
|
|
)
|
|
self.assertEqual(
|
|
result["plan"]["safety"]["position_diff_basis"],
|
|
"CURRENT_BROKER_POSITIONS_AND_SELLABLE_QUANTITY",
|
|
)
|
|
|
|
def test_zero_volume_position_is_counted_dropped_and_verifiable(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
zero_position = {
|
|
"account_id": "account-fixture",
|
|
"symbol": "600000.XSHG",
|
|
"volume": 0,
|
|
"sellable": 0,
|
|
"market_value": 0,
|
|
"avg_price": 0,
|
|
}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(
|
|
positions=[zero_position],
|
|
asset={
|
|
"account_id": "account-fixture",
|
|
"cash": 100_000,
|
|
"market_value": 0,
|
|
"total_asset": 100_000,
|
|
},
|
|
),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertEqual(result["plan"]["status"], "SHADOW_READY")
|
|
self.assertEqual(
|
|
result["broker_observation"]["semantic_inputs"][
|
|
"raw_position_count"
|
|
],
|
|
1,
|
|
)
|
|
self.assertEqual(
|
|
result["broker_observation"]["semantic_inputs"][
|
|
"dropped_zero_position_count"
|
|
],
|
|
1,
|
|
)
|
|
self.assertEqual(
|
|
result["broker_observation"]["positions"],
|
|
[],
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
verified = verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertEqual(verified["status"], "SHADOW_READY")
|
|
|
|
def test_only_unique_next_session_pre_open_window_is_accepted(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
cases = (
|
|
(
|
|
"weekend",
|
|
datetime(2026, 7, 25, 9, 0, tzinfo=UTC8),
|
|
"QUERY_WRONG_TRADING_SESSION",
|
|
),
|
|
(
|
|
"same_day_after_close",
|
|
datetime(2026, 7, 24, 15, 1, tzinfo=UTC8),
|
|
"QUERY_WRONG_TRADING_SESSION",
|
|
),
|
|
(
|
|
"before_window",
|
|
datetime(2026, 7, 27, 8, 59, 59, tzinfo=UTC8),
|
|
"QUERY_BEFORE_EXECUTABLE_WINDOW",
|
|
),
|
|
(
|
|
"exclusive_end",
|
|
datetime(2026, 7, 27, 9, 30, tzinfo=UTC8),
|
|
"QUERY_MISSED_EXECUTABLE_WINDOW",
|
|
),
|
|
(
|
|
"late_multiple_days",
|
|
datetime(2026, 7, 30, 9, 0, tzinfo=UTC8),
|
|
"QUERY_WRONG_TRADING_SESSION",
|
|
),
|
|
)
|
|
for name, observed, expected_code in cases:
|
|
with self.subTest(name=name), tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {
|
|
item["code"] for item in result["plan"]["blockers"]
|
|
}
|
|
self.assertIn(expected_code, codes)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
self.assertIsNone(result["broker_snapshot"])
|
|
self.assertTrue(
|
|
all(
|
|
row["proposed_delta"] == 0
|
|
for row in result["plan"]["target_diffs"]
|
|
)
|
|
)
|
|
|
|
def test_national_day_gap_uses_declared_calendar_not_day_count(self):
|
|
decision_time = datetime(2024, 9, 30, 15, 0, tzinfo=UTC8)
|
|
next_session = date(2024, 10, 8)
|
|
observed = datetime(2024, 10, 8, 9, 29, 59, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
next_session=next_session,
|
|
broker_observed_at=datetime(
|
|
2024,
|
|
10,
|
|
8,
|
|
9,
|
|
0,
|
|
tzinfo=UTC8,
|
|
),
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertEqual(result["plan"]["status"], "SHADOW_READY")
|
|
self.assertEqual(
|
|
result["plan"]["decision"]["next_trading_session"],
|
|
"2024-10-08",
|
|
)
|
|
self.assertIsNotNone(result["broker_snapshot"])
|
|
|
|
def test_unfinished_or_unknown_order_blocks_every_proposed_delta(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
for state, code in (
|
|
("ACCEPTED", "UNFINISHED_ORDER"),
|
|
("UNKNOWN", "UNKNOWN_ORDER_STATE"),
|
|
):
|
|
with self.subTest(state=state), tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision", decision_time
|
|
)
|
|
order = {
|
|
"account_id": "account-fixture",
|
|
"broker_order_id": 88,
|
|
"client_order_id": "decision-001",
|
|
"symbol": "600000.XSHG",
|
|
"side": "BUY",
|
|
"quantity": 100,
|
|
"filled_quantity": 0,
|
|
"limit_price": 10.5,
|
|
"state": state,
|
|
}
|
|
adapter = FakeReadOnlyAdapter(orders=[order])
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=adapter,
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
self.assertIn(
|
|
code,
|
|
{item["code"] for item in result["plan"]["blockers"]},
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
row["proposed_delta"] == 0
|
|
for row in result["plan"]["target_diffs"]
|
|
)
|
|
)
|
|
self.assertEqual(adapter.mutation_calls, [])
|
|
|
|
def test_order_change_and_wrong_next_session_fail_closed(self):
|
|
decision_time = datetime(2026, 7, 1, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 26, 9, 0, tzinfo=UTC8)
|
|
changed = {
|
|
"account_id": "account-fixture",
|
|
"broker_order_id": 99,
|
|
"client_order_id": None,
|
|
"symbol": "600000.XSHG",
|
|
"side": "SELL",
|
|
"quantity": 100,
|
|
"filled_quantity": 0,
|
|
"limit_price": 10,
|
|
"state": "CANCELLED",
|
|
}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
next_session=date(2026, 7, 2),
|
|
)
|
|
adapter = FakeReadOnlyAdapter(
|
|
orders=[],
|
|
second_orders=[changed],
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=adapter,
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {item["code"] for item in result["plan"]["blockers"]}
|
|
self.assertIn("ORDERS_CHANGED_DURING_QUERY", codes)
|
|
self.assertIn("QUERY_WRONG_TRADING_SESSION", codes)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
self.assertIsNone(result["broker_snapshot"])
|
|
self.assertEqual(adapter.mutation_calls, [])
|
|
|
|
def test_bound_broker_observation_after_current_query_is_blocked(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
broker_observed_at=observed + timedelta(seconds=6),
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {item["code"] for item in result["plan"]["blockers"]}
|
|
self.assertIn("DECISION_OBSERVATION_FROM_FUTURE", codes)
|
|
self.assertIn("DECISION_SOURCE_FROM_FUTURE", codes)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
|
|
def test_live_enabled_or_degraded_adapter_is_rejected_before_query(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision", decision_time
|
|
)
|
|
for state, live, message in (
|
|
("READY", True, "allow_live_orders"),
|
|
("DEGRADED", False, "READY state"),
|
|
):
|
|
with self.subTest(state=state, live=live):
|
|
adapter = FakeReadOnlyAdapter()
|
|
adapter.state = state
|
|
adapter.allow_live_orders = live
|
|
with self.assertRaisesRegex(QmtShadowPlanError, message):
|
|
build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=adapter,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertEqual(adapter.calls, [])
|
|
self.assertEqual(adapter.mutation_calls, [])
|
|
|
|
def test_policy_thresholds_cannot_be_widened(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
cases = (
|
|
(
|
|
{
|
|
"max_query_duration_seconds": (
|
|
MAX_SHADOW_QUERY_DURATION_SECONDS + 1
|
|
)
|
|
},
|
|
"max_query_duration_seconds cannot exceed",
|
|
),
|
|
(
|
|
{
|
|
"market_value_tolerance": (
|
|
MAX_BROKER_ASSET_ABSOLUTE_TOLERANCE + 0.01
|
|
)
|
|
},
|
|
"market_value_tolerance cannot exceed",
|
|
),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision", decision_time
|
|
)
|
|
for kwargs, expected in cases:
|
|
with self.subTest(kwargs=kwargs):
|
|
adapter = FakeReadOnlyAdapter()
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError, expected
|
|
):
|
|
build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=adapter,
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
**kwargs,
|
|
)
|
|
self.assertEqual(adapter.calls, [])
|
|
|
|
def test_resealed_semantic_policy_widening_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
for field, widened, expected in (
|
|
(
|
|
"max_query_duration_seconds",
|
|
MAX_SHADOW_QUERY_DURATION_SECONDS + 1,
|
|
"query duration policy",
|
|
),
|
|
(
|
|
"market_value_tolerance",
|
|
MAX_BROKER_ASSET_ABSOLUTE_TOLERANCE + 0.01,
|
|
"asset tolerance",
|
|
),
|
|
):
|
|
with self.subTest(field=field), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(
|
|
result, root / "shadow"
|
|
)
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
broker_observation = json.loads(
|
|
Path(paths["broker_observation"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_snapshot = json.loads(
|
|
Path(paths["broker_snapshot"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_observation["semantic_inputs"][field] = widened
|
|
observation_hash = payload_sha256(broker_observation)
|
|
plan["observation"]["observed_facts_sha256"] = (
|
|
observation_hash
|
|
)
|
|
broker_snapshot["metadata"][
|
|
"observed_facts_sha256"
|
|
] = observation_hash
|
|
broker_snapshot["snapshot_id"] = ""
|
|
broker_snapshot["snapshot_id"] = (
|
|
"QMT-SHADOW-"
|
|
+ payload_sha256(broker_snapshot)[:16]
|
|
)
|
|
plan["broker_snapshot_sha256"] = payload_sha256(
|
|
broker_snapshot
|
|
)
|
|
reseal_shadow_artifacts(
|
|
paths,
|
|
plan,
|
|
broker_observation=broker_observation,
|
|
broker_snapshot=broker_snapshot,
|
|
evidence_hmac_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError, expected
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_tampered_shadow_plan_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(root / "decision", decision_time)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan_path = Path(paths["plan"])
|
|
plan = json.loads(plan_path.read_text(encoding="utf-8"))
|
|
plan["safety"]["ready_for_live_submission"] = True
|
|
plan_path.write_text(json.dumps(plan), encoding="utf-8")
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError, "artifact hash mismatch"
|
|
):
|
|
verify_qmt_shadow_plan(paths["manifest"])
|
|
|
|
def test_authenticated_evidence_rejects_wrong_key(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"authenticated shadow evidence HMAC",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=b"x" * 32,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"account_hash_key is required",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
)
|
|
|
|
def test_authenticated_evidence_rejects_unsigned_observation_rewrite(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
broker_observation = json.loads(
|
|
Path(paths["broker_observation"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_snapshot = json.loads(
|
|
Path(paths["broker_snapshot"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_observation["semantic_inputs"][
|
|
"max_query_duration_seconds"
|
|
] = 9.0
|
|
observation_hash = payload_sha256(broker_observation)
|
|
plan["observation"]["observed_facts_sha256"] = (
|
|
observation_hash
|
|
)
|
|
broker_snapshot["metadata"][
|
|
"observed_facts_sha256"
|
|
] = observation_hash
|
|
broker_snapshot["snapshot_id"] = ""
|
|
broker_snapshot["snapshot_id"] = (
|
|
"QMT-SHADOW-"
|
|
+ payload_sha256(broker_snapshot)[:16]
|
|
)
|
|
plan["broker_snapshot_sha256"] = payload_sha256(
|
|
broker_snapshot
|
|
)
|
|
reseal_shadow_artifacts(
|
|
paths,
|
|
plan,
|
|
broker_observation=broker_observation,
|
|
broker_snapshot=broker_snapshot,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"authenticated shadow evidence HMAC",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_authenticated_evidence_binds_original_decision(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
original_decision = decision_artifact(
|
|
root / "original-decision", decision_time
|
|
)
|
|
substituted_decision = decision_artifact(
|
|
root / "substituted-decision",
|
|
decision_time,
|
|
first_target_quantity=2000,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=original_decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
substituted_manifest_sha256 = hashlib.sha256(
|
|
Path(substituted_decision).read_bytes()
|
|
).hexdigest()
|
|
plan["decision"][
|
|
"decision_manifest_sha256"
|
|
] = substituted_manifest_sha256
|
|
row = next(
|
|
item
|
|
for item in plan["target_diffs"]
|
|
if item["symbol"] == "000001.XSHE"
|
|
)
|
|
row.update(
|
|
{
|
|
"target_quantity": 2000,
|
|
"desired_delta": 2000,
|
|
"board_lot_feasible_delta": 2000,
|
|
"proposed_delta": 2000,
|
|
"proposed_side": "BUY",
|
|
"proposed_quantity": 2000,
|
|
}
|
|
)
|
|
reseal_shadow_artifacts(
|
|
paths,
|
|
plan,
|
|
manifest_updates={
|
|
"decision_manifest_sha256": (
|
|
substituted_manifest_sha256
|
|
)
|
|
},
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"authenticated shadow evidence HMAC",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=substituted_decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_payload_reformat_with_resealed_byte_hash_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan_path = Path(paths["plan"])
|
|
plan = json.loads(plan_path.read_text(encoding="utf-8"))
|
|
plan_path.write_text(
|
|
json.dumps(
|
|
plan,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
manifest_path = Path(paths["manifest"])
|
|
manifest = json.loads(
|
|
manifest_path.read_text(encoding="utf-8")
|
|
)
|
|
manifest["artifact_sha256"]["shadow_plan.json"] = (
|
|
hashlib.sha256(plan_path.read_bytes()).hexdigest()
|
|
)
|
|
manifest_path.write_text(
|
|
json.dumps(
|
|
manifest,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"shadow_plan.json is not in deterministic published encoding",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_manifest_reformat_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
manifest_path = Path(paths["manifest"])
|
|
manifest = json.loads(
|
|
manifest_path.read_text(encoding="utf-8")
|
|
)
|
|
manifest_path.write_text(
|
|
json.dumps(
|
|
manifest,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"manifest is not in deterministic published encoding",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_operator_cli_is_bound_into_planner_source_manifest(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertIn(
|
|
"tools/qmt_shadow_plan.py",
|
|
result["manifest"]["planner_sources"],
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
tampered_sources = dict(
|
|
result["manifest"]["planner_sources"]
|
|
)
|
|
tampered_sources["tools/qmt_shadow_plan.py"] = "0" * 64
|
|
tampered_source_sha256 = payload_sha256(
|
|
tampered_sources
|
|
)
|
|
with patch(
|
|
"quant60.qmt_shadow._source_manifest",
|
|
return_value=(
|
|
tampered_source_sha256,
|
|
tampered_sources,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"current shadow planner sources have changed",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_resealed_equity_contract_tampering_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
for tamper in ("result", "tolerance"):
|
|
with self.subTest(tamper=tamper), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(
|
|
result,
|
|
root / "shadow",
|
|
)
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
contract = plan["portfolio"]["equity_consistency"]
|
|
if tamper == "result":
|
|
contract["within_tolerance"] = False
|
|
expected = "tolerance result is inconsistent"
|
|
else:
|
|
contract["absolute_tolerance"] = 100.0
|
|
contract["allowed_difference"] = 100.0
|
|
expected = "tolerance constants/rule were changed"
|
|
reseal_shadow_plan(paths, plan)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
expected,
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_resealed_target_and_feasible_tampering_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
for tamper in (
|
|
"target_quantity",
|
|
"target_weight",
|
|
"lot_size",
|
|
"feasible_delta",
|
|
):
|
|
with self.subTest(tamper=tamper), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(
|
|
result, root / "shadow"
|
|
)
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
row = next(
|
|
item
|
|
for item in plan["target_diffs"]
|
|
if item["symbol"] == "000001.XSHE"
|
|
)
|
|
if tamper == "target_quantity":
|
|
row["target_quantity"] = 100_000
|
|
row["desired_delta"] = 100_000
|
|
row["board_lot_feasible_delta"] = 100_000
|
|
row["proposed_delta"] = 100_000
|
|
row["proposed_side"] = "BUY"
|
|
row["proposed_quantity"] = 100_000
|
|
elif tamper == "target_weight":
|
|
row["target_weight"] = 0.90
|
|
plan["portfolio"]["target_gross_weight"] = 0.95
|
|
elif tamper == "lot_size":
|
|
row["lot_size"] = 200
|
|
else:
|
|
row["board_lot_feasible_delta"] = 1_100
|
|
row["proposed_delta"] = 1_100
|
|
row["proposed_side"] = "BUY"
|
|
row["proposed_quantity"] = 1_100
|
|
reseal_shadow_plan(paths, plan)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"target/order projection|cash/portfolio projection",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_resealed_cross_account_blocker_removal_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision",
|
|
decision_time,
|
|
broker_account_hash="0" * 64,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertIn(
|
|
"DECISION_ACCOUNT_MISMATCH",
|
|
{
|
|
item["code"]
|
|
for item in result["plan"]["blockers"]
|
|
},
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
force_ready_projection(plan)
|
|
reseal_shadow_plan(paths, plan)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"blockers do not match semantic replay",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_resealed_blocker_removal_matrix_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
accepted = {
|
|
"account_id": "account-fixture",
|
|
"broker_order_id": 88,
|
|
"client_order_id": "decision-001",
|
|
"symbol": "600000.XSHG",
|
|
"side": "BUY",
|
|
"quantity": 100,
|
|
"filled_quantity": 0,
|
|
"limit_price": 10,
|
|
"state": "ACCEPTED",
|
|
}
|
|
unknown = dict(accepted, state="UNKNOWN")
|
|
cancelled = dict(accepted, state="CANCELLED")
|
|
filled = dict(
|
|
accepted,
|
|
state="FILLED",
|
|
filled_quantity=100,
|
|
)
|
|
mismatched_trade = {
|
|
"account_id": "account-fixture",
|
|
"trade_id": "trade-88",
|
|
"broker_order_id": 88,
|
|
"symbol": "000001.XSHE",
|
|
"side": "BUY",
|
|
"quantity": 100,
|
|
"price": 10,
|
|
"amount": 1000,
|
|
}
|
|
cases = (
|
|
(
|
|
"unfinished_order",
|
|
"UNFINISHED_ORDER",
|
|
lambda: FakeReadOnlyAdapter(orders=[accepted]),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"unknown_order",
|
|
"UNKNOWN_ORDER_STATE",
|
|
lambda: FakeReadOnlyAdapter(orders=[unknown]),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"orders_changed",
|
|
"ORDERS_CHANGED_DURING_QUERY",
|
|
lambda: FakeReadOnlyAdapter(
|
|
orders=[], second_orders=[cancelled]
|
|
),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"live_mode_changed",
|
|
"LIVE_MODE_CHANGED_DURING_QUERY",
|
|
lambda: FakeReadOnlyAdapter(
|
|
flip_live_after_query=True
|
|
),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"adapter_state_changed",
|
|
"ADAPTER_NOT_READY_AFTER_QUERY",
|
|
lambda: FakeReadOnlyAdapter(
|
|
flip_state_after_query=True
|
|
),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"adapter_errors",
|
|
"ADAPTER_ERRORS_PRESENT",
|
|
lambda: FakeReadOnlyAdapter(errors=["callback error"]),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"live_authorization_history",
|
|
"LIVE_AUTHORIZATION_HISTORY_PRESENT",
|
|
lambda: FakeReadOnlyAdapter(
|
|
live_authorization_audit=[{"event": "attempt"}]
|
|
),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"query_window_too_long",
|
|
"QUERY_WINDOW_TOO_LONG",
|
|
lambda: FakeReadOnlyAdapter(),
|
|
lambda: iter(
|
|
(observed, observed + timedelta(seconds=11))
|
|
).__next__,
|
|
),
|
|
(
|
|
"position_market_value",
|
|
"POSITION_MARKET_VALUE_MISMATCH",
|
|
lambda: FakeReadOnlyAdapter(
|
|
asset={
|
|
"account_id": "account-fixture",
|
|
"cash": 89_000,
|
|
"market_value": 11_000,
|
|
"total_asset": 100_000,
|
|
}
|
|
),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"asset_identity",
|
|
"ASSET_IDENTITY_MISMATCH",
|
|
lambda: FakeReadOnlyAdapter(
|
|
asset={
|
|
"account_id": "account-fixture",
|
|
"cash": 80_000,
|
|
"market_value": 10_000,
|
|
"total_asset": 100_000,
|
|
}
|
|
),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
(
|
|
"trade_order_identity",
|
|
"TRADE_ORDER_IDENTITY_MISMATCH",
|
|
lambda: FakeReadOnlyAdapter(
|
|
orders=[filled],
|
|
trades=[mismatched_trade],
|
|
),
|
|
lambda: fixed_clock(observed),
|
|
),
|
|
)
|
|
for name, expected_code, adapter_factory, clock_factory in cases:
|
|
with self.subTest(name=name), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=adapter_factory(),
|
|
clock=clock_factory(),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
self.assertIn(
|
|
expected_code,
|
|
{
|
|
item["code"]
|
|
for item in result["plan"]["blockers"]
|
|
},
|
|
)
|
|
paths = write_qmt_shadow_plan(
|
|
result, root / "shadow"
|
|
)
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
force_ready_projection(plan)
|
|
reseal_shadow_plan(paths, plan)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"blockers do not match semantic replay",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_resealed_ready_plan_cannot_drop_broker_snapshot(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
plan["broker_snapshot_sha256"] = None
|
|
plan["observation"]["broker_snapshot_valid"] = False
|
|
reseal_shadow_artifacts(
|
|
paths,
|
|
plan,
|
|
broker_snapshot=None,
|
|
evidence_hmac_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"observation projection|broker snapshot",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_resealed_order_observation_injection_is_replayed(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
broker_observation = json.loads(
|
|
Path(paths["broker_observation"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_snapshot = json.loads(
|
|
Path(paths["broker_snapshot"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_observation["orders_after"].append(
|
|
{
|
|
"broker_order_id": "88",
|
|
"client_order_id": "decision-001",
|
|
"symbol": "600000.XSHG",
|
|
"side": "BUY",
|
|
"quantity": 100,
|
|
"filled_quantity": 0,
|
|
"limit_price": 10.0,
|
|
"state": "ACCEPTED",
|
|
}
|
|
)
|
|
broker_observation["semantic_inputs"][
|
|
"raw_order_after_count"
|
|
] = 1
|
|
observation_hash = payload_sha256(broker_observation)
|
|
plan["observation"]["observed_facts_sha256"] = (
|
|
observation_hash
|
|
)
|
|
plan["observation"]["order_count"] = 1
|
|
broker_snapshot["metadata"][
|
|
"observed_facts_sha256"
|
|
] = observation_hash
|
|
broker_snapshot["snapshot_id"] = ""
|
|
broker_snapshot["snapshot_id"] = (
|
|
"QMT-SHADOW-"
|
|
+ payload_sha256(broker_snapshot)[:16]
|
|
)
|
|
plan["broker_snapshot_sha256"] = payload_sha256(
|
|
broker_snapshot
|
|
)
|
|
reseal_shadow_artifacts(
|
|
paths,
|
|
plan,
|
|
broker_observation=broker_observation,
|
|
broker_snapshot=broker_snapshot,
|
|
evidence_hmac_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"blockers do not match semantic replay",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_resealed_invalid_order_state_fill_is_rejected(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
broker_observation = json.loads(
|
|
Path(paths["broker_observation"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_snapshot = json.loads(
|
|
Path(paths["broker_snapshot"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
impossible_filled_order = {
|
|
"broker_order_id": "88",
|
|
"client_order_id": "decision-001",
|
|
"symbol": "600000.XSHG",
|
|
"side": "BUY",
|
|
"quantity": 100,
|
|
"filled_quantity": 0,
|
|
"limit_price": 10.0,
|
|
"state": "FILLED",
|
|
}
|
|
broker_observation["orders_before"] = [
|
|
impossible_filled_order
|
|
]
|
|
broker_observation["orders_after"] = [
|
|
impossible_filled_order
|
|
]
|
|
broker_observation["semantic_inputs"].update(
|
|
{
|
|
"raw_order_before_count": 1,
|
|
"raw_order_after_count": 1,
|
|
}
|
|
)
|
|
observation_hash = payload_sha256(broker_observation)
|
|
plan["observation"].update(
|
|
{
|
|
"observed_facts_sha256": observation_hash,
|
|
"order_count": 1,
|
|
}
|
|
)
|
|
broker_snapshot["metadata"].update(
|
|
{
|
|
"observed_facts_sha256": observation_hash,
|
|
"terminal_order_count": 1,
|
|
}
|
|
)
|
|
broker_snapshot["snapshot_id"] = ""
|
|
broker_snapshot["snapshot_id"] = (
|
|
"QMT-SHADOW-"
|
|
+ payload_sha256(broker_snapshot)[:16]
|
|
)
|
|
plan["broker_snapshot_sha256"] = payload_sha256(
|
|
broker_snapshot
|
|
)
|
|
reseal_shadow_artifacts(
|
|
paths,
|
|
plan,
|
|
broker_observation=broker_observation,
|
|
broker_snapshot=broker_snapshot,
|
|
evidence_hmac_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"filled order 88 is incomplete",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_resealed_asset_observation_fuzz_is_replayed(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
decision = decision_artifact(
|
|
root / "decision", decision_time
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
paths = write_qmt_shadow_plan(result, root / "shadow")
|
|
plan = json.loads(
|
|
Path(paths["plan"]).read_text(encoding="utf-8")
|
|
)
|
|
broker_observation = json.loads(
|
|
Path(paths["broker_observation"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_snapshot = json.loads(
|
|
Path(paths["broker_snapshot"]).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
broker_observation["asset"].update(
|
|
{
|
|
"cash": 0.0,
|
|
"market_value": 100_000.0,
|
|
"total_asset": 100_000.0,
|
|
}
|
|
)
|
|
observation_hash = payload_sha256(broker_observation)
|
|
plan["observation"]["observed_facts_sha256"] = (
|
|
observation_hash
|
|
)
|
|
broker_snapshot["metadata"][
|
|
"observed_facts_sha256"
|
|
] = observation_hash
|
|
broker_snapshot["snapshot_id"] = ""
|
|
broker_snapshot["snapshot_id"] = (
|
|
"QMT-SHADOW-"
|
|
+ payload_sha256(broker_snapshot)[:16]
|
|
)
|
|
plan["broker_snapshot_sha256"] = payload_sha256(
|
|
broker_snapshot
|
|
)
|
|
reseal_shadow_artifacts(
|
|
paths,
|
|
plan,
|
|
broker_observation=broker_observation,
|
|
broker_snapshot=broker_snapshot,
|
|
evidence_hmac_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
QmtShadowPlanError,
|
|
"blockers do not match semantic replay|cash/portfolio",
|
|
):
|
|
verify_qmt_shadow_plan(
|
|
paths["manifest"],
|
|
decision_manifest=decision,
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
|
|
def test_unbound_decision_emits_bootstrap_snapshot_and_blocks(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
observed = datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
bind_account=False,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(observed),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {item["code"] for item in result["plan"]["blockers"]}
|
|
self.assertIn("DECISION_NOT_ACCOUNT_BOUND", codes)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
self.assertIsNotNone(result["broker_snapshot"])
|
|
self.assertEqual(
|
|
result["broker_snapshot"]["account_hash"],
|
|
fixture_account_hash(),
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
row["proposed_delta"] == 0
|
|
for row in result["plan"]["target_diffs"]
|
|
)
|
|
)
|
|
|
|
def test_cross_account_decision_is_blocked(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
broker_account_hash="0" * 64,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(),
|
|
clock=fixed_clock(
|
|
datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {item["code"] for item in result["plan"]["blockers"]}
|
|
self.assertIn("DECISION_ACCOUNT_MISMATCH", codes)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
self.assertTrue(
|
|
all(
|
|
row["proposed_delta"] == 0
|
|
for row in result["plan"]["target_diffs"]
|
|
)
|
|
)
|
|
|
|
def test_trade_order_identity_mismatch_invalidates_broker_snapshot(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
order = {
|
|
"account_id": "account-fixture",
|
|
"broker_order_id": 88,
|
|
"client_order_id": "decision-001",
|
|
"symbol": "600000.XSHG",
|
|
"side": "BUY",
|
|
"quantity": 100,
|
|
"filled_quantity": 100,
|
|
"limit_price": 10,
|
|
"state": "FILLED",
|
|
}
|
|
trade = {
|
|
"account_id": "account-fixture",
|
|
"trade_id": "trade-88",
|
|
"broker_order_id": 88,
|
|
"symbol": "000001.XSHE",
|
|
"side": "BUY",
|
|
"quantity": 100,
|
|
"price": 10,
|
|
"amount": 1000,
|
|
}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(
|
|
orders=[order],
|
|
trades=[trade],
|
|
),
|
|
clock=fixed_clock(
|
|
datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {item["code"] for item in result["plan"]["blockers"]}
|
|
self.assertIn("TRADE_ORDER_IDENTITY_MISMATCH", codes)
|
|
self.assertEqual(result["plan"]["status"], "BLOCKED")
|
|
self.assertIsNone(result["broker_snapshot"])
|
|
|
|
def test_asset_identity_mismatch_invalidates_broker_snapshot(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(
|
|
asset={
|
|
"account_id": "account-fixture",
|
|
"cash": 80_000,
|
|
"market_value": 20_000,
|
|
"total_asset": 100_000,
|
|
}
|
|
),
|
|
clock=fixed_clock(
|
|
datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {item["code"] for item in result["plan"]["blockers"]}
|
|
self.assertIn("POSITION_MARKET_VALUE_MISMATCH", codes)
|
|
self.assertIsNone(result["broker_snapshot"])
|
|
|
|
def test_live_mode_change_during_queries_blocks_and_invalidates_snapshot(self):
|
|
decision_time = datetime(2026, 7, 24, 15, 0, tzinfo=UTC8)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
decision = decision_artifact(
|
|
Path(directory) / "decision",
|
|
decision_time,
|
|
)
|
|
result = build_qmt_shadow_plan(
|
|
decision_manifest=decision,
|
|
adapter=FakeReadOnlyAdapter(
|
|
flip_live_after_query=True,
|
|
),
|
|
clock=fixed_clock(
|
|
datetime(2026, 7, 27, 9, 0, tzinfo=UTC8)
|
|
),
|
|
account_hash_key=ACCOUNT_HASH_KEY,
|
|
)
|
|
codes = {item["code"] for item in result["plan"]["blockers"]}
|
|
self.assertIn("LIVE_MODE_CHANGED_DURING_QUERY", codes)
|
|
self.assertIsNone(result["broker_snapshot"])
|
|
self.assertEqual(
|
|
result["plan"]["observation"][
|
|
"adapter_allow_live_orders_after_query"
|
|
],
|
|
True,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|