diff --git a/src/quant_platform_kit/common/feature_snapshot.py b/src/quant_platform_kit/common/feature_snapshot.py index 2f5006e..ee034d3 100644 --- a/src/quant_platform_kit/common/feature_snapshot.py +++ b/src/quant_platform_kit/common/feature_snapshot.py @@ -4,6 +4,7 @@ import hashlib import json +import re import shutil import tempfile from dataclasses import dataclass @@ -27,6 +28,11 @@ DEFAULT_FEATURE_SNAPSHOT_FALLBACK_CACHE_DIR = ( DEFAULT_ARTIFACT_CACHE_DIR / "last_valid_feature_snapshots" ) +_CURRENT_GENERATION_SCHEMA = "current_generation.v1" +_CURRENT_GENERATION_OBJECT_NAMES = ("snapshot", "manifest", "ranking", "release_summary") +_CURRENT_GENERATION_POINTER_FILENAME = "current_generation.json" +_CURRENT_GENERATION_DIGEST_RE = re.compile(r"[0-9a-f]{64}\Z") +_CURRENT_GENERATION_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z") _MANIFEST_DIAGNOSTIC_FIELDS = ( "price_as_of", "universe_as_of", @@ -172,9 +178,253 @@ def _download_remote_object(uri: str, destination: Path) -> None: destination.write_bytes(get_object_store().read_bytes(uri)) +def _is_current_generation_pointer_reference(reference: str) -> bool: + raw_reference = str(reference or "").strip() + return _is_cloud_uri(raw_reference) and raw_reference.endswith( + f"/{_CURRENT_GENERATION_POINTER_FILENAME}" + ) + + +def _reject_duplicate_json_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate pointer field") + result[key] = value + return result + + +def _reject_nonfinite_json_constant(value: str) -> None: + raise ValueError(f"invalid pointer JSON constant: {value}") + + +def _is_safe_generation_basename(value: object) -> bool: + if not isinstance(value, str) or not value or "\x00" in value: + return False + return ( + "/" not in value + and "\\" not in value + and value not in {".", ".."} + and Path(value).name == value + ) + + +def _validate_current_generation_pointer_uri(pointer_uri: str) -> str: + bucket, object_name = _parse_cloud_uri(pointer_uri) + if ( + object_name.split("/")[-1] != _CURRENT_GENERATION_POINTER_FILENAME + or any(segment in {"", ".", ".."} for segment in object_name.split("/")) + or any(character.isspace() or character == "\\" for character in object_name) + ): + raise ValueError("unsafe current generation pointer URI") + parent = object_name.rsplit("/", 1)[0] if "/" in object_name else "" + return f"{pointer_uri[:5]}{bucket}/{parent}" if parent else f"{pointer_uri[:5]}{bucket}" + + +def _read_current_generation_pointer(pointer_bytes: bytes) -> dict[str, object]: + try: + text = pointer_bytes.decode("utf-8") + payload = json.loads( + text, + object_pairs_hook=_reject_duplicate_json_keys, + parse_constant=_reject_nonfinite_json_constant, + ) + except (UnicodeDecodeError, json.JSONDecodeError, ValueError, TypeError): + raise ValueError("invalid current generation pointer") from None + + if not isinstance(payload, dict): + raise ValueError("invalid current generation pointer") + if set(payload) != { + "schema", + "profile", + "generation_id", + "immutable_prefix", + "snapshot_as_of", + "objects", + }: + raise ValueError("invalid current generation pointer fields") + + try: + canonical_bytes = json.dumps( + payload, + ensure_ascii=True, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + b"\n" + except (TypeError, ValueError): + raise ValueError("invalid current generation pointer") from None + if pointer_bytes != canonical_bytes: + raise ValueError("non-canonical current generation pointer") + + if payload["schema"] != _CURRENT_GENERATION_SCHEMA: + raise ValueError("unsupported current generation pointer schema") + if ( + not isinstance(payload["profile"], str) + or not payload["profile"] + or payload["profile"] != payload["profile"].strip() + ): + raise ValueError("invalid current generation pointer profile") + generation_id = payload["generation_id"] + if not isinstance(generation_id, str) or _CURRENT_GENERATION_ID_RE.fullmatch(generation_id) is None: + raise ValueError("invalid current generation pointer generation_id") + snapshot_as_of = payload["snapshot_as_of"] + if ( + not isinstance(snapshot_as_of, str) + or pd.Timestamp(snapshot_as_of).strftime("%Y-%m-%d") != snapshot_as_of + ): + raise ValueError("invalid current generation pointer snapshot_as_of") + + objects = payload["objects"] + if not isinstance(objects, dict) or set(objects) != set(_CURRENT_GENERATION_OBJECT_NAMES): + raise ValueError("invalid current generation pointer objects") + basenames: set[str] = set() + for name in _CURRENT_GENERATION_OBJECT_NAMES: + item = objects[name] + if not isinstance(item, dict) or set(item) != {"basename", "sha256"}: + raise ValueError("invalid current generation pointer object fields") + basename = item["basename"] + digest = item["sha256"] + if not _is_safe_generation_basename(basename) or basename in basenames: + raise ValueError("invalid current generation pointer basename") + if not isinstance(digest, str) or _CURRENT_GENERATION_DIGEST_RE.fullmatch(digest) is None: + raise ValueError("invalid current generation pointer sha256") + basenames.add(basename) + + return payload + + +def _load_current_generation_feature_snapshot_guarded( + pointer_uri: str, + *, + run_as_of, + required_columns: Iterable[str] | None, + snapshot_date_columns: Iterable[str], + max_snapshot_month_lag: int, + expected_strategy_profile: str | None, + expected_config_name: str | None, + expected_config_path: str | None, + expected_contract_version: str | None, +) -> FeatureSnapshotGuardResult: + pointer_metadata = { + "feature_snapshot_pointer_uri": pointer_uri, + "feature_snapshot_generation_id": None, + "feature_snapshot_immutable_prefix": None, + "feature_snapshot_object_digests": None, + } + try: + prefix_root = _validate_current_generation_pointer_uri(pointer_uri) + with tempfile.TemporaryDirectory(prefix="feature-snapshot-generation-") as temporary_dir: + pointer_path = Path(temporary_dir) / _CURRENT_GENERATION_POINTER_FILENAME + _download_gcs_object(pointer_uri, pointer_path) + payload = _read_current_generation_pointer(pointer_path.read_bytes()) + pointer_profile = str(payload["profile"]) + if expected_strategy_profile and _normalize_strategy_profile_label( + pointer_profile + ) != _normalize_strategy_profile_label(expected_strategy_profile): + raise ValueError("current generation pointer profile mismatch") + generation_id = payload["generation_id"] + immutable_prefix = payload["immutable_prefix"] + expected_prefix = f"{prefix_root}/generations/{generation_id}" + if immutable_prefix != expected_prefix: + raise ValueError("current generation pointer prefix mismatch") + + objects = payload["objects"] + object_digests = { + name: objects[name]["sha256"] for name in _CURRENT_GENERATION_OBJECT_NAMES + } + pointer_metadata.update( + { + "feature_snapshot_generation_id": generation_id, + "feature_snapshot_immutable_prefix": immutable_prefix, + "feature_snapshot_object_digests": dict(object_digests), + } + ) + local_paths: dict[str, Path] = {} + for name in _CURRENT_GENERATION_OBJECT_NAMES: + basename = objects[name]["basename"] + object_uri = f"{immutable_prefix}/{basename}" + local_path = Path(temporary_dir) / f"{name}-{basename}" + _download_gcs_object(object_uri, local_path) + if _sha256_file(local_path) != object_digests[name]: + return FeatureSnapshotGuardResult( + frame=None, + metadata=_build_guard_metadata( + snapshot_path=pointer_uri, + decision="fail_closed", + snapshot_exists=False, + **pointer_metadata, + fail_reason="feature_snapshot_pointer_object_digest_mismatch", + ), + ) + local_paths[name] = local_path + + result = _load_feature_snapshot_guarded_without_fallback( + str(local_paths["snapshot"]), + run_as_of=run_as_of, + required_columns=required_columns, + snapshot_date_columns=snapshot_date_columns, + max_snapshot_month_lag=max_snapshot_month_lag, + manifest_path=str(local_paths["manifest"]), + require_manifest=True, + expected_strategy_profile=expected_strategy_profile or pointer_profile, + expected_config_name=expected_config_name, + expected_config_path=expected_config_path, + expected_contract_version=expected_contract_version, + ) + metadata = dict(result.metadata) + metadata.update(pointer_metadata) + metadata.update( + { + "feature_snapshot_path": pointer_uri, + "snapshot_path": pointer_uri, + "snapshot_source_uri": f"{immutable_prefix}/{objects['snapshot']['basename']}", + "snapshot_manifest_path": f"{immutable_prefix}/{objects['manifest']['basename']}", + "snapshot_manifest_source_uri": f"{immutable_prefix}/{objects['manifest']['basename']}", + "snapshot_local_path": None, + "snapshot_manifest_local_path": None, + } + ) + if result.metadata.get("snapshot_guard_decision") != "proceed": + metadata = _build_guard_metadata( + snapshot_path=pointer_uri, + decision="fail_closed", + snapshot_exists=False, + **pointer_metadata, + fail_reason="feature_snapshot_pointer_guard_failed", + ) + return FeatureSnapshotGuardResult(frame=None, metadata=metadata) + if result.metadata.get("snapshot_guard_decision") == "proceed" and str( + result.metadata.get("snapshot_as_of") + )[:10] != payload["snapshot_as_of"]: + return FeatureSnapshotGuardResult( + frame=None, + metadata={ + **metadata, + "snapshot_guard_decision": "fail_closed", + "fail_reason": "feature_snapshot_pointer_snapshot_as_of_mismatch", + }, + ) + return FeatureSnapshotGuardResult(frame=result.frame, metadata=metadata) + except Exception: + return FeatureSnapshotGuardResult( + frame=None, + metadata=_build_guard_metadata( + snapshot_path=pointer_uri, + decision="fail_closed", + snapshot_exists=False, + **pointer_metadata, + fail_reason="feature_snapshot_pointer_read_failed", + ), + ) + + # Backward-compatible aliases _parse_gcs_uri = _parse_cloud_uri -_download_gcs_object = _download_remote_object + + +def _download_gcs_object(uri: str, destination: Path) -> None: + _download_remote_object(uri, destination) def _cache_path_for_remote_artifact(reference: str) -> Path: @@ -264,6 +514,34 @@ def load_feature_snapshot_guarded( ) -> FeatureSnapshotGuardResult: """Load a guarded snapshot, optionally falling back to the last valid artifact.""" + raw_path = str(path or "").strip() + if _is_current_generation_pointer_reference(raw_path): + if str(manifest_path or "").strip() not in {"", raw_path}: + return FeatureSnapshotGuardResult( + frame=None, + metadata=_build_guard_metadata( + snapshot_path=raw_path, + decision="fail_closed", + snapshot_exists=False, + feature_snapshot_pointer_uri=raw_path, + feature_snapshot_generation_id=None, + feature_snapshot_immutable_prefix=None, + feature_snapshot_object_digests=None, + fail_reason="feature_snapshot_pointer_manifest_mismatch", + ), + ) + return _load_current_generation_feature_snapshot_guarded( + raw_path, + run_as_of=run_as_of, + required_columns=required_columns, + snapshot_date_columns=snapshot_date_columns, + max_snapshot_month_lag=max_snapshot_month_lag, + expected_strategy_profile=expected_strategy_profile, + expected_config_name=expected_config_name, + expected_config_path=expected_config_path, + expected_contract_version=expected_contract_version, + ) + fallback_context = _feature_snapshot_fallback_context( path=path, manifest_path=manifest_path, @@ -339,6 +617,33 @@ def _load_feature_snapshot_guarded_without_fallback( ), ) + if _is_current_generation_pointer_reference(raw_path): + if str(manifest_path or "").strip() not in {"", raw_path}: + return FeatureSnapshotGuardResult( + frame=None, + metadata=_build_guard_metadata( + snapshot_path=raw_path, + decision="fail_closed", + snapshot_exists=False, + feature_snapshot_pointer_uri=raw_path, + feature_snapshot_generation_id=None, + feature_snapshot_immutable_prefix=None, + feature_snapshot_object_digests=None, + fail_reason="feature_snapshot_pointer_manifest_mismatch", + ), + ) + return _load_current_generation_feature_snapshot_guarded( + raw_path, + run_as_of=run_as_of, + required_columns=required_columns, + snapshot_date_columns=snapshot_date_columns, + max_snapshot_month_lag=max_snapshot_month_lag, + expected_strategy_profile=expected_strategy_profile, + expected_config_name=expected_config_name, + expected_config_path=expected_config_path, + expected_contract_version=expected_contract_version, + ) + manifest_reference = _resolve_manifest_reference(raw_path, manifest_path) if _is_cloud_uri(raw_path) or _is_cloud_uri(manifest_reference): try: diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index d57f88e..6445094 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -45,6 +45,21 @@ _TQQQ_ETF_ONLY_FACTORS = {"TQQQ": 3, "BOXX": 1} _TQQQ_ETF_ONLY_NOMINAL_CAPS = {"TQQQ": 0.15, "BOXX": 0.50} _TQQQ_ETF_ONLY_EFFECTIVE_CAPS = {"TQQQ": 0.45, "BOXX": 0.50} +_TQQQ_EVIDENCE_MANDATE_SCHEMA = "qsl.tqqq-evidence-risk-mandate.v1" +_TQQQ_EVIDENCE_PORTFOLIO_SCHEMA = "qsl.tqqq-evidence-portfolio-snapshot.v1" +_TQQQ_EVIDENCE_RISK_STATE_SCHEMA = "qsl.tqqq-evidence-risk-state.v1" +_TQQQ_EVIDENCE_MANDATE_ID = "tqqq_core_parity_v1" +_TQQQ_EVIDENCE_PURPOSE = "TQQQ_CANDIDATE_RESEARCH_EVIDENCE_ONLY" +_TQQQ_EVIDENCE_FACTORS = {"TQQQ": 3, "QQQM": 1, "BOXX": 1} +_TQQQ_EVIDENCE_NOMINAL_CAPS = {"TQQQ": 0.15, "QQQM": 0.50, "BOXX": 0.50} +_TQQQ_EVIDENCE_EFFECTIVE_CAPS = {"TQQQ": 0.45, "QQQM": 0.50, "BOXX": 0.50} +_TQQQ_EVIDENCE_DRAWDOWN_SCALARS = { + "at_or_below_0_05": 1.0, + "above_0_05_to_0_10": 0.5, + "above_0_10": 0.0, +} +_TQQQ_EVIDENCE_MAX_AGE_SECONDS = 300.0 +_TQQQ_EVIDENCE_STRESS_LOSS_DISTANCE = 0.05 _RETIRED_GLOBAL_ETF_RESEARCH_MANDATE = ( "global_etf_rotation_etf_only_research_v1" ) @@ -78,6 +93,15 @@ def _canonical_digest(value: Mapping[str, Any]) -> str: return hashlib.sha256(encoded).hexdigest() +def _capital_base_snapshot_commitment(snapshot: CapitalBaseSnapshot) -> str: + return _canonical_digest( + { + **snapshot.to_safe_dict(), + "target_equity": snapshot.target_equity, + } + ) + + def _finite_number(value: Any) -> float | None: if type(value) not in (int, float): return None @@ -222,6 +246,54 @@ def _parse_utc_timestamp(value: Any) -> datetime | None: return parsed.astimezone(timezone.utc) if parsed.tzinfo is not None else None +def _parse_utc_whole_second(value: Any) -> datetime | None: + if type(value) is datetime: + try: + offset = value.utcoffset() + except Exception: + return None + if offset is None or offset.total_seconds() != 0.0 or value.microsecond != 0: + return None + return value.astimezone(timezone.utc) + timestamp, valid = _canonical_string(value) + if not valid or timestamp is None or not timestamp.endswith("Z"): + return None + try: + parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) + except (OverflowError, TypeError, ValueError): + return None + if parsed.microsecond != 0 or _utc_timestamp(parsed) != timestamp: + return None + return parsed.astimezone(timezone.utc) + + +def _exact_mapping(value: Any, fields: frozenset[str]) -> Mapping[str, Any] | None: + if not isinstance(value, Mapping) or len(value) != len(fields): + return None + try: + return value if set(value) == fields else None + except Exception: + return None + + +def _logical_evaluation_time( + value: datetime | None, + *, + wall_time: datetime, +) -> tuple[datetime, set[str]]: + if value is None: + return wall_time, set() + logical_time = _parse_utc_whole_second(value) + if logical_time is None or type(value) is not datetime: + return wall_time, {"invalid_logical_evaluation_time"} + age_seconds = (wall_time - logical_time).total_seconds() + if age_seconds < 0.0: + return wall_time, {"future_logical_evaluation_time"} + if age_seconds > _TQQQ_EVIDENCE_MAX_AGE_SECONDS: + return wall_time, {"stale_logical_evaluation_time"} + return logical_time, set() + + def _sha256(value: Any) -> str | None: normalized, valid = _canonical_string(value) if not valid or normalized is None or len(normalized) != 64: @@ -351,6 +423,7 @@ def _snapshot_metrics( *, now: datetime, max_snapshot_age_seconds: float | None, + require_utc_whole_second: bool = False, ) -> tuple[dict[str, Any], float | None, float | None, set[str]]: if isinstance(portfolio_snapshot, Mapping): as_of_value = portfolio_snapshot.get("as_of") @@ -363,7 +436,11 @@ def _snapshot_metrics( total_equity_value = portfolio_snapshot.total_equity else: return {}, None, None, {"invalid_portfolio_snapshot"} - as_of = _parse_utc_timestamp(as_of_value) + as_of = ( + _parse_utc_whole_second(as_of_value) + if require_utc_whole_second + else _parse_utc_timestamp(as_of_value) + ) observed = _finite_number(observed_value) total_equity = _finite_number(total_equity_value) if ( @@ -551,6 +628,355 @@ def _exact_tqqq_mandate_errors( return {"invalid_tqqq_research_mandate"} if invalid else set() +def _tqqq_evidence_mandate_fields( + mandate_provenance: Mapping[str, Any], + *, + now: datetime, +) -> tuple[dict[str, Any], set[str]]: + invalid = {"invalid_tqqq_evidence_risk_mandate"} + top_level = _exact_mapping( + mandate_provenance, + frozenset( + { + "schema_version", + "mandate_id", + "mandate_version", + "purpose", + "candidate_binding", + "validity", + "portfolio_policy", + "capital_binding", + "portfolio_binding", + "risk_state_binding", + "authority", + } + ), + ) + if top_level is None: + return {}, invalid + candidate = _exact_mapping( + top_level.get("candidate_binding"), + frozenset( + { + "strategy_profile", + "account_mode", + "strategy_revision", + "runner_revision", + "config_sha256", + "input_manifest_sha256", + "candidate_identity_sha256", + } + ), + ) + validity = _exact_mapping( + top_level.get("validity"), + frozenset( + { + "effective_at", + "expires_at", + "snapshot_max_age_seconds", + "single_consumption", + } + ), + ) + portfolio_policy = _exact_mapping( + top_level.get("portfolio_policy"), + frozenset( + { + "allowed_nonzero_assets", + "benchmark_only_assets", + "product_leverage_factors", + "max_nonzero_assets", + "effective_exposure_cap", + "nominal_caps", + "product_effective_caps", + "loss_budget", + "loss_budget_equity_reference", + "modeled_stress_loss_distance", + "stress_loss_is_model_assumption", + "drawdown_scalars", + "broker_margin_factor", + "margin_stacking", + "borrowing", + "shorting", + } + ), + ) + capital_binding = _exact_mapping( + top_level.get("capital_binding"), + frozenset( + { + "schema_version", + "snapshot_digest_sha256", + "as_of", + "account_mode", + "capital_scope", + "valuation_basis", + "target_currency", + "max_age_seconds", + "fx_conversion_allowed", + } + ), + ) + portfolio_binding = _exact_mapping( + top_level.get("portfolio_binding"), + frozenset( + { + "schema_version", + "snapshot_digest_sha256", + "as_of", + "source_identity_sha256", + "max_age_seconds", + } + ), + ) + risk_state_binding = _exact_mapping( + top_level.get("risk_state_binding"), + frozenset( + { + "schema_version", + "snapshot_digest_sha256", + "as_of", + "max_age_seconds", + } + ), + ) + authority = _exact_mapping( + top_level.get("authority"), + frozenset( + { + "authority_scope", + "authority_receipt_sha256", + "source_revision", + "runner_is_authority", + "no_order", + "no_paper", + "no_shadow", + "no_live", + "no_promotion_authority", + } + ), + ) + if any( + section is None + for section in ( + candidate, + validity, + portfolio_policy, + capital_binding, + portfolio_binding, + risk_state_binding, + authority, + ) + ): + return {}, invalid + assert candidate is not None + assert validity is not None + assert portfolio_policy is not None + assert capital_binding is not None + assert portfolio_binding is not None + assert risk_state_binding is not None + assert authority is not None + + strategy_profile, valid_strategy_profile = _canonical_string( + candidate.get("strategy_profile") + ) + account_mode, valid_account_mode = _canonical_string( + candidate.get("account_mode") + ) + strategy_revision = _git_revision(candidate.get("strategy_revision")) + runner_revision = _git_revision(candidate.get("runner_revision")) + config_sha256 = _sha256(candidate.get("config_sha256")) + input_manifest_sha256 = _sha256(candidate.get("input_manifest_sha256")) + candidate_identity_sha256 = _sha256( + candidate.get("candidate_identity_sha256") + ) + receipt_sha256 = _sha256(authority.get("authority_receipt_sha256")) + source_revision = _git_revision(authority.get("source_revision")) + effective_at = _parse_utc_whole_second(validity.get("effective_at")) + expires_at = _parse_utc_whole_second(validity.get("expires_at")) + capital_as_of = _parse_utc_whole_second(capital_binding.get("as_of")) + portfolio_as_of = _parse_utc_whole_second(portfolio_binding.get("as_of")) + risk_state_as_of = _parse_utc_whole_second(risk_state_binding.get("as_of")) + capital_digest = _sha256(capital_binding.get("snapshot_digest_sha256")) + portfolio_digest = _sha256(portfolio_binding.get("snapshot_digest_sha256")) + portfolio_source_identity = _sha256( + portfolio_binding.get("source_identity_sha256") + ) + risk_state_digest = _sha256( + risk_state_binding.get("snapshot_digest_sha256") + ) + snapshot_max_age = _finite_number( + validity.get("snapshot_max_age_seconds") + ) + capital_max_age = _finite_number(capital_binding.get("max_age_seconds")) + portfolio_max_age = _finite_number(portfolio_binding.get("max_age_seconds")) + risk_state_max_age = _finite_number( + risk_state_binding.get("max_age_seconds") + ) + allowed_assets = _canonical_string_list( + portfolio_policy.get("allowed_nonzero_assets") + ) + benchmark_assets = _canonical_string_list( + portfolio_policy.get("benchmark_only_assets") + ) + factors = _canonical_numeric_mapping( + portfolio_policy.get("product_leverage_factors"), + integer=True, + minimum=1.0, + ) + max_nonzero_assets = _bounded_nonnegative_int( + portfolio_policy.get("max_nonzero_assets") + ) + effective_exposure_cap = _finite_number( + portfolio_policy.get("effective_exposure_cap") + ) + loss_budget = _finite_number(portfolio_policy.get("loss_budget")) + + fixed_values_valid = ( + top_level.get("schema_version") == _TQQQ_EVIDENCE_MANDATE_SCHEMA + and top_level.get("mandate_id") == _TQQQ_EVIDENCE_MANDATE_ID + and top_level.get("mandate_version") == "v1" + and top_level.get("purpose") == _TQQQ_EVIDENCE_PURPOSE + and valid_strategy_profile + and strategy_profile == _TQQQ_EVIDENCE_MANDATE_ID + and valid_account_mode + and account_mode == "single_strategy_account_v1" + and strategy_revision is not None + and runner_revision is not None + and config_sha256 is not None + and input_manifest_sha256 is not None + and candidate_identity_sha256 is not None + and receipt_sha256 is not None + and source_revision is not None + and effective_at is not None + and expires_at is not None + and capital_as_of is not None + and portfolio_as_of is not None + and risk_state_as_of is not None + and capital_digest is not None + and portfolio_digest is not None + and portfolio_source_identity is not None + and risk_state_digest is not None + and snapshot_max_age == _TQQQ_EVIDENCE_MAX_AGE_SECONDS + and capital_max_age == _TQQQ_EVIDENCE_MAX_AGE_SECONDS + and portfolio_max_age == _TQQQ_EVIDENCE_MAX_AGE_SECONDS + and risk_state_max_age == _TQQQ_EVIDENCE_MAX_AGE_SECONDS + and validity.get("single_consumption") is True + and type(portfolio_policy.get("allowed_nonzero_assets")) is list + and allowed_assets == ["TQQQ", "QQQM", "BOXX"] + and type(portfolio_policy.get("benchmark_only_assets")) is list + and benchmark_assets == ["QQQ"] + and factors == _TQQQ_EVIDENCE_FACTORS + and max_nonzero_assets == 3 + and effective_exposure_cap == 0.50 + and _exact_numeric_mapping( + portfolio_policy.get("nominal_caps"), + _TQQQ_EVIDENCE_NOMINAL_CAPS, + ) + and _exact_numeric_mapping( + portfolio_policy.get("product_effective_caps"), + _TQQQ_EVIDENCE_EFFECTIVE_CAPS, + ) + and loss_budget == 0.01 + and portfolio_policy.get("loss_budget_equity_reference") + == "completed_session_equity" + and _finite_number( + portfolio_policy.get("modeled_stress_loss_distance") + ) + == _TQQQ_EVIDENCE_STRESS_LOSS_DISTANCE + and portfolio_policy.get("stress_loss_is_model_assumption") is True + and _exact_numeric_mapping( + portfolio_policy.get("drawdown_scalars"), + _TQQQ_EVIDENCE_DRAWDOWN_SCALARS, + ) + and _bounded_nonnegative_int( + portfolio_policy.get("broker_margin_factor") + ) + == 1 + and portfolio_policy.get("margin_stacking") is False + and portfolio_policy.get("borrowing") is False + and portfolio_policy.get("shorting") is False + and capital_binding.get("schema_version") == "qpk.capital_base.v2" + and capital_binding.get("account_mode") == account_mode + and capital_binding.get("capital_scope") == "allocated_sleeve" + and capital_binding.get("valuation_basis") == "allocated_sleeve_ledger" + and capital_binding.get("target_currency") == "USD" + and capital_binding.get("fx_conversion_allowed") is False + and portfolio_binding.get("schema_version") + == _TQQQ_EVIDENCE_PORTFOLIO_SCHEMA + and risk_state_binding.get("schema_version") + == _TQQQ_EVIDENCE_RISK_STATE_SCHEMA + and authority.get("authority_scope") == "RESEARCH_ONLY" + and authority.get("runner_is_authority") is False + and authority.get("no_order") is True + and authority.get("no_paper") is True + and authority.get("no_shadow") is True + and authority.get("no_live") is True + and authority.get("no_promotion_authority") is True + ) + if ( + not fixed_values_valid + or expires_at <= effective_at + or (expires_at - effective_at).total_seconds() + > _TQQQ_EVIDENCE_MAX_AGE_SECONDS + ): + return {}, invalid + if effective_at > now or expires_at < now: + return {}, {"expired_mandate"} + + return { + "schema_version": _TQQQ_EVIDENCE_MANDATE_SCHEMA, + "mandate_id": _TQQQ_EVIDENCE_MANDATE_ID, + "mandate_version": "v1", + "authority_receipt_sha256": receipt_sha256, + "authority_scope": "RESEARCH_ONLY", + "source_revision": source_revision, + "strategy_profile": strategy_profile, + "account_mode": account_mode, + "strategy_revision": strategy_revision, + "runner_revision": runner_revision, + "config_sha256": config_sha256, + "input_manifest_sha256": input_manifest_sha256, + "candidate_identity_sha256": candidate_identity_sha256, + "effective_exposure_cap": effective_exposure_cap, + "max_snapshot_age_seconds": snapshot_max_age, + "loss_budget": loss_budget, + "product_leverage_factors": factors, + "product_caps": dict(_TQQQ_EVIDENCE_NOMINAL_CAPS), + "nominal_caps": dict(_TQQQ_EVIDENCE_NOMINAL_CAPS), + "product_effective_caps": dict(_TQQQ_EVIDENCE_EFFECTIVE_CAPS), + "allowed_nonzero_assets": set(allowed_assets or ()), + "benchmark_only_assets": set(benchmark_assets or ()), + "max_nonzero_assets": max_nonzero_assets, + "modeled_stress_loss_distance": _TQQQ_EVIDENCE_STRESS_LOSS_DISTANCE, + "capital_binding": { + "schema_version": "qpk.capital_base.v2", + "snapshot_digest_sha256": capital_digest, + "as_of": _utc_timestamp(capital_as_of), + "account_mode": account_mode, + "capital_scope": "allocated_sleeve", + "valuation_basis": "allocated_sleeve_ledger", + "target_currency": "USD", + "max_age_seconds": capital_max_age, + "fx_conversion_allowed": False, + }, + "portfolio_binding": { + "schema_version": _TQQQ_EVIDENCE_PORTFOLIO_SCHEMA, + "snapshot_digest_sha256": portfolio_digest, + "as_of": _utc_timestamp(portfolio_as_of), + "source_identity_sha256": portfolio_source_identity, + "max_age_seconds": portfolio_max_age, + }, + "risk_state_binding": { + "schema_version": _TQQQ_EVIDENCE_RISK_STATE_SCHEMA, + "snapshot_digest_sha256": risk_state_digest, + "as_of": _utc_timestamp(risk_state_as_of), + "max_age_seconds": risk_state_max_age, + }, + }, set() + + def _mandate_fields( mandate_provenance: Mapping[str, Any] | None, *, @@ -571,6 +997,14 @@ def _mandate_fields( }, {"missing_mandate"} if not isinstance(mandate_provenance, Mapping): return {}, {"invalid_mandate"} + schema_version = mandate_provenance.get("schema_version") + if schema_version is not None: + if schema_version != _TQQQ_EVIDENCE_MANDATE_SCHEMA: + return {}, {"unsupported_mandate_schema"} + return _tqqq_evidence_mandate_fields( + mandate_provenance, + now=now, + ) if mandate_provenance.get("mandate_id") == _RETIRED_GLOBAL_ETF_RESEARCH_MANDATE: return {}, {"retired_global_etf_research_mandate"} @@ -813,6 +1247,11 @@ def _budget_authority_errors( decision: StrategyDecision, mandate: Mapping[str, Any], ) -> set[str]: + if ( + mandate.get("schema_version") == _TQQQ_EVIDENCE_MANDATE_SCHEMA + and decision.budgets + ): + return {"unsupported_evidence_budget"} requested_budget = 0.0 for budget in decision.budgets or (): amount = _finite_number(getattr(budget, "amount", None)) @@ -829,6 +1268,212 @@ def _budget_authority_errors( return set() +def _tqqq_evidence_capital_errors( + validation: CapitalBaseValidation, + *, + mandate: Mapping[str, Any], +) -> set[str]: + if not validation.is_valid: + return set() + snapshot = validation.snapshot + binding = validation.binding + expected = mandate.get("capital_binding") + if snapshot is None or binding is None or not isinstance(expected, Mapping): + return {"capital_base_mandate_mismatch"} + errors: set[str] = set() + if _capital_base_snapshot_commitment(snapshot) != expected.get( + "snapshot_digest_sha256" + ): + errors.add("capital_base_digest_mismatch") + if _parse_utc_whole_second(snapshot.as_of) is None: + errors.add("invalid_capital_base_timestamp") + elif _utc_timestamp(snapshot.as_of) != expected.get("as_of"): + errors.add("capital_base_as_of_mismatch") + capital_scope = ( + None if snapshot.capital_scope is None else snapshot.capital_scope.value + ) + valuation_basis = ( + None if snapshot.valuation_basis is None else snapshot.valuation_basis.value + ) + binding_capital_scope = ( + None if binding.capital_scope is None else binding.capital_scope.value + ) + binding_valuation_basis = ( + None if binding.valuation_basis is None else binding.valuation_basis.value + ) + if ( + snapshot.contract_version != expected.get("schema_version") + or capital_scope != expected.get("capital_scope") + or valuation_basis != expected.get("valuation_basis") + or binding_capital_scope != expected.get("capital_scope") + or binding_valuation_basis != expected.get("valuation_basis") + or snapshot.target_currency != expected.get("target_currency") + or snapshot.reported_currency != expected.get("target_currency") + or binding.target_currency != expected.get("target_currency") + or snapshot.fx_rate_to_target != 1.0 + or snapshot.fx_source_digest_sha256 is not None + or binding.max_age_seconds != expected.get("max_age_seconds") + or binding.strategy_scope != mandate.get("strategy_profile") + ): + errors.add("capital_base_mandate_mismatch") + return errors + + +def _tqqq_evidence_portfolio_fields( + portfolio_snapshot: Any, + *, + mandate: Mapping[str, Any], +) -> tuple[dict[str, Any], set[str]]: + snapshot = _exact_mapping( + portfolio_snapshot, + frozenset( + { + "schema_version", + "as_of", + "observed_effective_exposure", + "total_equity", + "source_identity_sha256", + } + ), + ) + binding = mandate.get("portfolio_binding") + if snapshot is None or not isinstance(binding, Mapping): + return {}, {"invalid_portfolio_snapshot"} + schema_version, valid_schema = _canonical_string(snapshot.get("schema_version")) + as_of = _parse_utc_whole_second(snapshot.get("as_of")) + observed = _finite_number(snapshot.get("observed_effective_exposure")) + total_equity = _finite_number(snapshot.get("total_equity")) + source_identity = _sha256(snapshot.get("source_identity_sha256")) + if ( + not valid_schema + or schema_version != _TQQQ_EVIDENCE_PORTFOLIO_SCHEMA + or as_of is None + or observed is None + or observed < 0.0 + or total_equity is None + or total_equity <= 0.0 + or source_identity is None + ): + return {}, {"invalid_portfolio_snapshot"} + payload = { + "schema_version": schema_version, + "as_of": _utc_timestamp(as_of), + "observed_effective_exposure": observed, + "total_equity": total_equity, + "source_identity_sha256": source_identity, + } + errors: set[str] = set() + if binding.get("schema_version") != schema_version: + errors.add("portfolio_snapshot_schema_mismatch") + if binding.get("as_of") != payload["as_of"]: + errors.add("portfolio_snapshot_as_of_mismatch") + if binding.get("source_identity_sha256") != source_identity: + errors.add("portfolio_snapshot_source_mismatch") + if binding.get("snapshot_digest_sha256") != _canonical_digest(payload): + errors.add("portfolio_snapshot_digest_mismatch") + return payload, errors + + +def _tqqq_evidence_risk_control_fields( + risk_control_state: Mapping[str, Any] | None, + *, + mandate: Mapping[str, Any], + now: datetime, +) -> tuple[dict[str, Any], set[str]]: + empty = { + "stop_loss_distance": None, + "stop_intent_ready": None, + "strategy_breaker_triggered": False, + "account_breaker_triggered": False, + "account_drawdown_fraction": None, + "drawdown_scalar": None, + "modeled_stress_loss_distance": None, + "risk_control_state_digest_sha256": None, + } + state = _exact_mapping( + risk_control_state, + frozenset( + { + "schema_version", + "as_of", + "mandate_id", + "candidate_identity_sha256", + "modeled_stress_loss_distance", + "account_drawdown_fraction", + "drawdown_scalar", + } + ), + ) + binding = mandate.get("risk_state_binding") + if state is None or not isinstance(binding, Mapping): + return empty, {"invalid_risk_control_state"} + + errors: set[str] = set() + schema_version, valid_schema = _canonical_string(state.get("schema_version")) + as_of = _parse_utc_whole_second(state.get("as_of")) + mandate_id, valid_mandate_id = _canonical_string(state.get("mandate_id")) + candidate_digest = _sha256(state.get("candidate_identity_sha256")) + stress_distance = _finite_number(state.get("modeled_stress_loss_distance")) + account_drawdown = _finite_number(state.get("account_drawdown_fraction")) + drawdown_scalar = _finite_number(state.get("drawdown_scalar")) + payload = { + "schema_version": schema_version, + "as_of": _utc_timestamp(as_of) if as_of is not None else None, + "mandate_id": mandate_id, + "candidate_identity_sha256": candidate_digest, + "modeled_stress_loss_distance": stress_distance, + "account_drawdown_fraction": account_drawdown, + "drawdown_scalar": drawdown_scalar, + } + digest = _canonical_digest(payload) + + if not valid_schema or schema_version != _TQQQ_EVIDENCE_RISK_STATE_SCHEMA: + errors.add("invalid_risk_control_state") + if as_of is None: + errors.add("invalid_risk_control_state") + else: + max_age = _finite_number(binding.get("max_age_seconds")) + age_seconds = (now - as_of).total_seconds() + if max_age is None or age_seconds < 0.0 or age_seconds > max_age: + errors.add("stale_risk_control_state") + if mandate_id != mandate.get("mandate_id"): + errors.add("risk_control_mandate_mismatch") + if candidate_digest != mandate.get("candidate_identity_sha256"): + errors.add("risk_control_candidate_mismatch") + if stress_distance != mandate.get("modeled_stress_loss_distance"): + errors.add("invalid_modeled_stress_loss_distance") + if binding.get("schema_version") != schema_version: + errors.add("risk_control_state_schema_mismatch") + if binding.get("as_of") != payload["as_of"]: + errors.add("risk_control_state_as_of_mismatch") + if binding.get("snapshot_digest_sha256") != digest: + errors.add("risk_control_state_digest_mismatch") + + expected_scalar: float | None = None + if account_drawdown is None or not 0.0 <= account_drawdown <= 1.0: + errors.add("invalid_account_drawdown") + elif account_drawdown <= 0.05: + expected_scalar = 1.0 + elif account_drawdown <= 0.10: + expected_scalar = 0.50 + else: + expected_scalar = 0.0 + if expected_scalar is None or drawdown_scalar != expected_scalar: + errors.add("drawdown_scalar_mismatch") + account_breaker = account_drawdown is not None and account_drawdown > 0.10 + if account_breaker: + errors.add("account_breaker_triggered") + + return { + **empty, + "account_breaker_triggered": account_breaker, + "account_drawdown_fraction": account_drawdown, + "drawdown_scalar": drawdown_scalar, + "modeled_stress_loss_distance": stress_distance, + "risk_control_state_digest_sha256": digest, + }, errors + + def _risk_control_fields( risk_control_state: Mapping[str, Any] | None, *, @@ -843,8 +1488,15 @@ def _risk_control_fields( "account_breaker_triggered": None, "account_drawdown_fraction": None, "drawdown_scalar": None, + "modeled_stress_loss_distance": None, "risk_control_state_digest_sha256": None, } + if mandate.get("schema_version") == _TQQQ_EVIDENCE_MANDATE_SCHEMA: + return _tqqq_evidence_risk_control_fields( + risk_control_state, + mandate=mandate, + now=now, + ) if mandate.get("mandate_id") != _TQQQ_ETF_ONLY_RESEARCH_MANDATE: return empty, set() if ( @@ -999,33 +1651,64 @@ def _assess_with_evidence_static( candidate_identity, ) ) + is_tqqq_evidence_mandate = ( + mandate.get("schema_version") == _TQQQ_EVIDENCE_MANDATE_SCHEMA + ) cap = mandate.get("effective_exposure_cap") snapshot_payload, observed, total_equity, snapshot_errors = _snapshot_metrics( portfolio_snapshot, now=now, max_snapshot_age_seconds=mandate.get("max_snapshot_age_seconds"), + require_utc_whole_second=is_tqqq_evidence_mandate, ) reason_codes.update(snapshot_errors) + if is_tqqq_evidence_mandate: + portfolio_fields, portfolio_errors = _tqqq_evidence_portfolio_fields( + portfolio_snapshot, + mandate=mandate, + ) + reason_codes.update(portfolio_errors) + if portfolio_fields: + snapshot_payload.update(portfolio_fields) value_targets_present = any( type(position) is PositionTarget and position.target_value is not None for position in getattr(decision, "positions", ()) ) - if value_targets_present: + if value_targets_present or is_tqqq_evidence_mandate: capital_base_validation = validate_capital_base( capital_base, binding=capital_base_binding, now=now, ) - snapshot_payload["capital_base"] = capital_base_validation.to_safe_dict() + capital_base_payload = capital_base_validation.to_safe_dict() + if ( + is_tqqq_evidence_mandate + and capital_base_validation.is_valid + and capital_base_validation.snapshot is not None + ): + capital_base_payload["snapshot_commitment_sha256"] = ( + _capital_base_snapshot_commitment(capital_base_validation.snapshot) + ) + snapshot_payload["capital_base"] = capital_base_payload if not capital_base_validation.is_valid: reason_codes.update(capital_base_validation.findings) - elif ( - type(candidate_identity) is CandidateRiskIdentity - and capital_base_validation.binding is not None - and capital_base_validation.binding.strategy_scope - != candidate_identity.strategy_profile - ): - reason_codes.add("capital_base_strategy_candidate_mismatch") + else: + if ( + type(candidate_identity) is CandidateRiskIdentity + and capital_base_validation.binding is not None + and capital_base_validation.binding.strategy_scope + != candidate_identity.strategy_profile + ): + reason_codes.add("capital_base_strategy_candidate_mismatch") + if is_tqqq_evidence_mandate: + reason_codes.update( + _tqqq_evidence_capital_errors( + capital_base_validation, + mandate=mandate, + ) + ) + if total_equity != capital_base_validation.target_equity: + reason_codes.add("capital_base_portfolio_equity_mismatch") total_equity = capital_base_validation.target_equity decision_payload, active_positions, decision_errors = _decision_metrics( decision, @@ -1058,6 +1741,26 @@ def _assess_with_evidence_static( if can_evaluate_policy: factors = mandate["product_leverage_factors"] allowed_assets = mandate["allowed_nonzero_assets"] + benchmark_only_assets = mandate.get("benchmark_only_assets", set()) + if is_tqqq_evidence_mandate: + for position in decision_payload["positions"]: + symbol = position["symbol"] + if symbol in benchmark_only_assets: + reason_codes.add("benchmark_only_asset") + elif symbol is not None and symbol not in allowed_assets: + reason_codes.add("asset_not_authorized") + target_weights: dict[str, float] = {} + for symbol, weight in active_positions: + combined_weight = target_weights.get(symbol, 0.0) + weight + if not math.isfinite(combined_weight): + reason_codes.add("invalid_risk_metadata") + continue + target_weights[symbol] = combined_weight + policy_positions = ( + list(target_weights.items()) + if is_tqqq_evidence_mandate + else active_positions + ) weighted_exposure = 0.0 if mandate_provenance is None and len(active_positions) > 1: reason_codes.add("fallback_position_count") @@ -1066,7 +1769,15 @@ def _assess_with_evidence_static( and len(active_positions) > mandate["max_nonzero_assets"] ): reason_codes.add("single_strategy_position_count") - for symbol, weight in active_positions: + if ( + is_tqqq_evidence_mandate + and len(target_weights) > mandate["max_nonzero_assets"] + ): + reason_codes.add("max_nonzero_assets") + for symbol, weight in policy_positions: + if symbol in benchmark_only_assets: + reason_codes.add("benchmark_only_asset") + continue if allowed_assets is not None and symbol not in allowed_assets: reason_codes.add("asset_not_authorized") continue @@ -1124,13 +1835,19 @@ def _assess_with_evidence_static( reason_codes.add("invalid_risk_metadata") weighted_exposure = 0.0 - target_weights: dict[str, float] = {} - for symbol, weight in active_positions: - combined_weight = target_weights.get(symbol, 0.0) + weight - if not math.isfinite(combined_weight): - reason_codes.add("invalid_risk_metadata") - continue - target_weights[symbol] = combined_weight + if is_tqqq_evidence_mandate: + stress_distance = control_fields["modeled_stress_loss_distance"] + drawdown_scalar = control_fields["drawdown_scalar"] + loss_budget = mandate.get("loss_budget") + total_nominal_weight = sum(target_weights.values()) + if ( + stress_distance is not None + and drawdown_scalar is not None + and loss_budget is not None + and total_nominal_weight * stress_distance + > loss_budget * drawdown_scalar + 1e-9 + ): + reason_codes.add("risk_budget_exposure_cap") valid_normalization = False if normalization_origin_weights is not None: normalized_origin_material = _canonical_numeric_mapping( @@ -1246,15 +1963,16 @@ def _invalid_assessment_result( mandate_provenance: Mapping[str, Any] | None, candidate_identity: CandidateRiskIdentity | None, now: datetime, + reason_codes: set[str] | None = None, ) -> RiskGateResult: - reason_codes = {"invalid_risk_metadata"} + normalized_reason_codes = set(reason_codes or {"invalid_risk_metadata"}) try: if ( isinstance(mandate_provenance, Mapping) and mandate_provenance.get("mandate_id") == _RETIRED_GLOBAL_ETF_RESEARCH_MANDATE ): - reason_codes.add("retired_global_etf_research_mandate") + normalized_reason_codes.add("retired_global_etf_research_mandate") except Exception: pass normalized_scope, valid_scope = _canonical_string(scope) @@ -1289,7 +2007,7 @@ def _invalid_assessment_result( observed_effective_exposure=None, proposed_effective_exposure=None, outcome="REJECT", - reason_codes=tuple(sorted(reason_codes)), + reason_codes=tuple(sorted(normalized_reason_codes)), execution_authorized=False, ) return RiskGateResult( @@ -1315,13 +2033,16 @@ def assess_with_evidence( risk_control_state: Mapping[str, Any] | None = None, capital_base: CapitalBaseSnapshot | Mapping[str, Any] | None = None, capital_base_binding: CapitalBaseBinding | Mapping[str, Any] | None = None, + logical_evaluation_time: datetime | None = None, ) -> RiskGateResult: """Run the sole promotion/evidence-grade risk assessment API. ``mandate_provenance`` and a matching ``candidate_identity`` are required for an evidence-grade approval. Missing or invalid authority is rejected; the legacy :meth:`RiskEngine.assess` approval is never sufficient on its - own. The returned receipt is redacted and canonical. + own. An explicit ``logical_evaluation_time`` must be a fresh UTC + whole-second datetime; omitting it preserves wall-clock evaluation. The + returned receipt is redacted and canonical. """ try: risk_action = build_risk_engine().assess( @@ -1334,7 +2055,20 @@ def assess_with_evidence( risk_engine_failed = True else: risk_engine_failed = False - now = _utc_now() + wall_time = _utc_now() + now, logical_time_errors = _logical_evaluation_time( + logical_evaluation_time, + wall_time=wall_time, + ) + if logical_time_errors: + return _invalid_assessment_result( + decision, + scope=scope, + mandate_provenance=mandate_provenance, + candidate_identity=candidate_identity, + now=wall_time, + reason_codes=logical_time_errors, + ) try: return _assess_with_evidence_static( decision, diff --git a/tests/test_feature_snapshot.py b/tests/test_feature_snapshot.py index 61d02c9..b5890ae 100644 --- a/tests/test_feature_snapshot.py +++ b/tests/test_feature_snapshot.py @@ -5,8 +5,350 @@ import tempfile import unittest from pathlib import Path +from unittest.mock import patch -from quant_platform_kit.common.feature_snapshot import load_feature_snapshot_guarded +from quant_platform_kit.common.feature_snapshot import ( + load_feature_snapshot_guarded, +) + + +POINTER_URI = "gs://bucket/feature/current_generation.json" + + +def _current_generation_fixture() -> tuple[dict[str, bytes], bytes]: + snapshot = b"as_of,symbol,close\n2026-04-01,QQQ,500\n" + manifest = json.dumps( + { + "snapshot_as_of": "2026-04-01", + "strategy_profile": "feature_snapshot_strategy", + "config_name": "feature_snapshot_strategy", + "contract_version": "feature_snapshot_strategy.feature_snapshot.v1", + "snapshot_sha256": hashlib.sha256(snapshot).hexdigest(), + "config_sha256": "a" * 64, + } + ).encode("utf-8") + objects = { + "snapshot": snapshot, + "manifest": manifest, + "ranking": b"rank,symbol\n1,QQQ\n", + "release_summary": b'{"release_status":"ready"}\n', + } + immutable_prefix = "gs://bucket/feature/generations/g-1" + payload = { + "schema": "current_generation.v1", + "profile": "feature_snapshot_strategy", + "generation_id": "g-1", + "immutable_prefix": immutable_prefix, + "snapshot_as_of": "2026-04-01", + "objects": { + name: { + "basename": f"{name}.json", + "sha256": hashlib.sha256(data).hexdigest(), + } + for name, data in objects.items() + }, + } + payload["objects"]["snapshot"]["basename"] = "feature.csv" + payload["objects"]["manifest"]["basename"] = "feature.manifest.json" + payload["objects"]["ranking"]["basename"] = "ranking.csv" + payload["objects"]["release_summary"]["basename"] = "release_status_summary.json" + pointer = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + b"\n" + remote_objects = { + POINTER_URI: pointer, + **{ + f"{immutable_prefix}/{item['basename']}": objects[name] + for name, item in payload["objects"].items() + }, + } + return remote_objects, pointer + + +class FeatureSnapshotCurrentGenerationReaderTests(unittest.TestCase): + def test_reader_dereferences_and_verifies_all_objects_before_guarded_load(self) -> None: + remote_objects, _ = _current_generation_fixture() + downloads: list[str] = [] + + def download(uri: str, destination: Path) -> None: + downloads.append(uri) + destination.write_bytes(remote_objects[uri]) + + with patch( + "quant_platform_kit.common.feature_snapshot._download_remote_object", + side_effect=download, + ), patch( + "quant_platform_kit.common.feature_snapshot._download_gcs_object", + side_effect=download, + ): + result = load_feature_snapshot_guarded( + POINTER_URI, + run_as_of="2026-04-02", + required_columns=("as_of", "symbol", "close"), + manifest_path=POINTER_URI, + require_manifest=True, + expected_strategy_profile="feature_snapshot_strategy", + expected_config_name="feature_snapshot_strategy", + ) + + self.assertIsNotNone(result.frame) + self.assertEqual( + downloads, + [ + POINTER_URI, + "gs://bucket/feature/generations/g-1/feature.csv", + "gs://bucket/feature/generations/g-1/feature.manifest.json", + "gs://bucket/feature/generations/g-1/ranking.csv", + "gs://bucket/feature/generations/g-1/release_status_summary.json", + ], + ) + self.assertEqual(result.metadata["feature_snapshot_pointer_uri"], POINTER_URI) + self.assertEqual(result.metadata["feature_snapshot_generation_id"], "g-1") + self.assertEqual( + result.metadata["feature_snapshot_immutable_prefix"], + "gs://bucket/feature/generations/g-1", + ) + self.assertEqual( + result.metadata["feature_snapshot_object_digests"], + { + name: hashlib.sha256(data).hexdigest() + for name, data in { + "snapshot": remote_objects[ + "gs://bucket/feature/generations/g-1/feature.csv" + ], + "manifest": remote_objects[ + "gs://bucket/feature/generations/g-1/feature.manifest.json" + ], + "ranking": remote_objects[ + "gs://bucket/feature/generations/g-1/ranking.csv" + ], + "release_summary": remote_objects[ + "gs://bucket/feature/generations/g-1/release_status_summary.json" + ], + }.items() + }, + ) + + def test_reader_rejects_non_pointer_manifest_path_without_download(self) -> None: + remote_objects, _ = _current_generation_fixture() + downloads: list[str] = [] + + def download(uri: str, destination: Path) -> None: + downloads.append(uri) + destination.write_bytes(remote_objects[uri]) + + with patch( + "quant_platform_kit.common.feature_snapshot._download_remote_object", + side_effect=download, + ), patch( + "quant_platform_kit.common.feature_snapshot._download_gcs_object", + side_effect=download, + ): + result = load_feature_snapshot_guarded( + POINTER_URI, + run_as_of="2026-04-02", + manifest_path="gs://bucket/feature/other.manifest.json", + ) + + self.assertIsNone(result.frame) + self.assertEqual(result.metadata["snapshot_guard_decision"], "fail_closed") + self.assertEqual(result.metadata["fail_reason"], "feature_snapshot_pointer_manifest_mismatch") + self.assertEqual(downloads, []) + + def test_reader_rejects_noncanonical_pointer_and_bad_contract_without_object_download(self) -> None: + remote_objects, canonical = _current_generation_fixture() + payload = json.loads(canonical) + bad_digest_payload = json.loads(canonical) + bad_digest_payload["objects"]["snapshot"]["sha256"] = "0" * 64 + cases = { + "noncanonical": json.dumps(payload).encode(), + "unknown_field": canonical.replace(b'"schema"', b'"extra":1,"schema"'), + "path_traversal": canonical.replace(b'feature.csv', b'../feature.csv'), + "absolute_path": canonical.replace(b'feature.csv', b'/tmp/feature.csv'), + "wrong_prefix": canonical.replace( + b"gs://bucket/feature/generations/g-1", + b"gs://bucket/other/generations/g-1", + ), + "wrong_digest": json.dumps( + bad_digest_payload, sort_keys=True, separators=(",", ":") + ).encode() + + b"\n", + } + for name, pointer in cases.items(): + with self.subTest(name=name): + downloads: list[str] = [] + remote_objects[POINTER_URI] = pointer + + def download(uri: str, destination: Path) -> None: + downloads.append(uri) + destination.write_bytes(remote_objects[uri]) + + with patch( + "quant_platform_kit.common.feature_snapshot._download_remote_object", + side_effect=download, + ), patch( + "quant_platform_kit.common.feature_snapshot._download_gcs_object", + side_effect=download, + ): + result = load_feature_snapshot_guarded( + POINTER_URI, + run_as_of="2026-04-02", + manifest_path=POINTER_URI, + require_manifest=True, + ) + + self.assertIsNone(result.frame) + self.assertEqual(result.metadata["snapshot_guard_decision"], "fail_closed") + expected_downloads = [POINTER_URI] + if name == "wrong_digest": + expected_downloads.append( + "gs://bucket/feature/generations/g-1/feature.csv" + ) + self.assertEqual(downloads, expected_downloads) + remote_objects[POINTER_URI] = canonical + + def test_reader_rejects_downloaded_object_digest_mismatch_without_guarded_load(self) -> None: + remote_objects, _ = _current_generation_fixture() + remote_objects["gs://bucket/feature/generations/g-1/feature.csv"] = b"tampered" + downloads: list[str] = [] + + def download(uri: str, destination: Path) -> None: + downloads.append(uri) + destination.write_bytes(remote_objects[uri]) + + with patch( + "quant_platform_kit.common.feature_snapshot._download_remote_object", + side_effect=download, + ): + result = load_feature_snapshot_guarded( + POINTER_URI, + run_as_of="2026-04-02", + manifest_path=POINTER_URI, + require_manifest=True, + ) + + self.assertIsNone(result.frame) + self.assertEqual(result.metadata["snapshot_guard_decision"], "fail_closed") + self.assertEqual( + result.metadata["fail_reason"], + "feature_snapshot_pointer_object_digest_mismatch", + ) + self.assertEqual( + downloads, + [ + POINTER_URI, + "gs://bucket/feature/generations/g-1/feature.csv", + ], + ) + + def test_reader_binds_pointer_profile_to_expected_profile(self) -> None: + remote_objects, canonical = _current_generation_fixture() + payload = json.loads(canonical) + payload["profile"] = "other_profile" + remote_objects[POINTER_URI] = ( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + b"\n" + ) + downloads: list[str] = [] + + def download(uri: str, destination: Path) -> None: + downloads.append(uri) + destination.write_bytes(remote_objects[uri]) + + with patch( + "quant_platform_kit.common.feature_snapshot._download_remote_object", + side_effect=download, + ): + result = load_feature_snapshot_guarded( + POINTER_URI, + run_as_of="2026-04-02", + manifest_path=POINTER_URI, + expected_strategy_profile="feature_snapshot_strategy", + ) + + self.assertIsNone(result.frame) + self.assertEqual(result.metadata["snapshot_guard_decision"], "fail_closed") + self.assertEqual(result.metadata["fail_reason"], "feature_snapshot_pointer_read_failed") + self.assertEqual(downloads, [POINTER_URI]) + + def test_reader_binds_pointer_profile_to_manifest_without_caller_expectation(self) -> None: + remote_objects, canonical = _current_generation_fixture() + payload = json.loads(canonical) + payload["profile"] = "other_profile" + remote_objects[POINTER_URI] = ( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + b"\n" + ) + + def download(uri: str, destination: Path) -> None: + destination.write_bytes(remote_objects[uri]) + + with patch( + "quant_platform_kit.common.feature_snapshot._download_gcs_object", + side_effect=download, + ): + result = load_feature_snapshot_guarded( + POINTER_URI, + run_as_of="2026-04-02", + manifest_path=POINTER_URI, + require_manifest=True, + ) + + self.assertIsNone(result.frame) + self.assertEqual(result.metadata["snapshot_guard_decision"], "fail_closed") + self.assertEqual(result.metadata["fail_reason"], "feature_snapshot_pointer_guard_failed") + + def test_pointer_failure_bypasses_last_valid_fallback(self) -> None: + remote_objects, _ = _current_generation_fixture() + remote_objects[POINTER_URI] = b"{}\n" + + def download(uri: str, destination: Path) -> None: + destination.write_bytes(remote_objects[uri]) + + with patch( + "quant_platform_kit.common.feature_snapshot._download_remote_object", + side_effect=download, + ), patch( + "quant_platform_kit.common.feature_snapshot._feature_snapshot_fallback_context", + side_effect=AssertionError("pointer must bypass fallback context"), + ), patch( + "quant_platform_kit.common.feature_snapshot._load_feature_snapshot_last_valid", + side_effect=AssertionError("pointer must bypass fallback read"), + ): + result = load_feature_snapshot_guarded( + POINTER_URI, + run_as_of="2026-04-02", + manifest_path=POINTER_URI, + fallback_mode="last_valid", + ) + + self.assertIsNone(result.frame) + self.assertEqual(result.metadata["snapshot_guard_decision"], "fail_closed") + + def test_pointer_success_does_not_write_last_valid_fallback(self) -> None: + remote_objects, _ = _current_generation_fixture() + + def download(uri: str, destination: Path) -> None: + destination.write_bytes(remote_objects[uri]) + + with patch( + "quant_platform_kit.common.feature_snapshot._download_remote_object", + side_effect=download, + ), patch( + "quant_platform_kit.common.feature_snapshot._feature_snapshot_fallback_context", + side_effect=AssertionError("pointer must bypass fallback context"), + ), patch( + "quant_platform_kit.common.feature_snapshot._write_feature_snapshot_last_valid", + side_effect=AssertionError("pointer must not write fallback"), + ), patch( + "quant_platform_kit.common.feature_snapshot._load_feature_snapshot_last_valid", + side_effect=AssertionError("pointer must not read fallback"), + ): + result = load_feature_snapshot_guarded( + POINTER_URI, + run_as_of="2026-04-02", + manifest_path=POINTER_URI, + require_manifest=True, + fallback_mode="last_valid", + ) + + self.assertIsNotNone(result.frame) class FeatureSnapshotGuardAliasTests(unittest.TestCase): diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index c691a6c..567b75a 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -1,5 +1,7 @@ from __future__ import annotations +import hashlib +import json import unittest from collections.abc import Iterator, Mapping from datetime import datetime, timedelta, timezone @@ -2096,6 +2098,727 @@ def test_inactive_tqqq_stop_identities_are_still_required(self) -> None: engine.assess.assert_called_once() +class TqqqEvidenceRiskMandateTests(unittest.TestCase): + _NOW = datetime(2026, 9, 2, 6, 0, tzinfo=timezone.utc) + _SCHEMA_VERSION = "qsl.tqqq-evidence-risk-mandate.v1" + _RISK_STATE_SCHEMA_VERSION = "qsl.tqqq-evidence-risk-state.v1" + _MANDATE_ID = "tqqq_core_parity_v1" + _STRATEGY_PROFILE = "tqqq_core_parity_v1" + _DEFAULT = object() + + @staticmethod + def _digest(value: object) -> str: + encoded = json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + @classmethod + def _candidate(cls) -> CandidateRiskIdentity: + return CandidateRiskIdentity( + strategy_profile=cls._STRATEGY_PROFILE, + account_mode="single_strategy_account_v1", + strategy_revision="b" * 40, + runner_revision="c" * 40, + config_sha256="d" * 64, + input_manifest_sha256="e" * 64, + authority_receipt_sha256="a" * 64, + ) + + @classmethod + def _capital_base(cls, *, as_of: datetime | None = None) -> CapitalBaseSnapshot: + return _capital_base( + as_of=as_of or cls._NOW - timedelta(seconds=1), + strategy_scope=cls._STRATEGY_PROFILE, + capital_scope=CapitalScope.ALLOCATED_SLEEVE, + valuation_basis=CapitalValuationBasis.ALLOCATED_SLEEVE_LEDGER, + allocation_scope="tqqq-candidate-research-sleeve", + component_coverage_digest_sha256="2" * 64, + ) + + @classmethod + def _capital_base_binding(cls, **overrides: object) -> CapitalBaseBinding: + values: dict[str, object] = { + "strategy_scope": cls._STRATEGY_PROFILE, + "capital_scope": CapitalScope.ALLOCATED_SLEEVE, + "valuation_basis": CapitalValuationBasis.ALLOCATED_SLEEVE_LEDGER, + "allocation_scope": "tqqq-candidate-research-sleeve", + "max_age_seconds": 300, + } + values.update(overrides) + return _capital_base_binding(**values) + + @classmethod + def _risk_state(cls, **overrides: object) -> dict[str, object]: + state: dict[str, object] = { + "schema_version": cls._RISK_STATE_SCHEMA_VERSION, + "as_of": "2026-09-02T05:59:59Z", + "mandate_id": cls._MANDATE_ID, + "candidate_identity_sha256": cls._candidate().candidate_sha256, + "modeled_stress_loss_distance": 0.05, + "account_drawdown_fraction": 0.05, + "drawdown_scalar": 1.0, + } + state.update(overrides) + return state + + @classmethod + def _mandate( + cls, + *, + capital_base: CapitalBaseSnapshot | None = None, + portfolio_snapshot: Mapping[str, object] | None = None, + risk_state: Mapping[str, object] | None = None, + ) -> dict[str, object]: + candidate = cls._candidate() + capital_base = capital_base or cls._capital_base() + portfolio_snapshot = portfolio_snapshot or cls._snapshot() + risk_state = risk_state or cls._risk_state() + return { + "schema_version": cls._SCHEMA_VERSION, + "mandate_id": cls._MANDATE_ID, + "mandate_version": "v1", + "purpose": "TQQQ_CANDIDATE_RESEARCH_EVIDENCE_ONLY", + "candidate_binding": { + "strategy_profile": candidate.strategy_profile, + "account_mode": candidate.account_mode, + "strategy_revision": candidate.strategy_revision, + "runner_revision": candidate.runner_revision, + "config_sha256": candidate.config_sha256, + "input_manifest_sha256": candidate.input_manifest_sha256, + "candidate_identity_sha256": candidate.candidate_sha256, + }, + "validity": { + "effective_at": "2026-09-02T05:59:30Z", + "expires_at": "2026-09-02T06:04:30Z", + "snapshot_max_age_seconds": 300, + "single_consumption": True, + }, + "portfolio_policy": { + "allowed_nonzero_assets": ["TQQQ", "QQQM", "BOXX"], + "benchmark_only_assets": ["QQQ"], + "product_leverage_factors": {"TQQQ": 3, "QQQM": 1, "BOXX": 1}, + "max_nonzero_assets": 3, + "effective_exposure_cap": 0.50, + "nominal_caps": {"TQQQ": 0.15, "QQQM": 0.50, "BOXX": 0.50}, + "product_effective_caps": {"TQQQ": 0.45, "QQQM": 0.50, "BOXX": 0.50}, + "loss_budget": 0.01, + "loss_budget_equity_reference": "completed_session_equity", + "modeled_stress_loss_distance": 0.05, + "stress_loss_is_model_assumption": True, + "drawdown_scalars": { + "at_or_below_0_05": 1.0, + "above_0_05_to_0_10": 0.5, + "above_0_10": 0.0, + }, + "broker_margin_factor": 1, + "margin_stacking": False, + "borrowing": False, + "shorting": False, + }, + "capital_binding": { + "schema_version": "qpk.capital_base.v2", + "snapshot_digest_sha256": cls._digest( + { + **capital_base.to_safe_dict(), + "target_equity": capital_base.target_equity, + } + ), + "as_of": capital_base.to_safe_dict()["as_of"], + "account_mode": "single_strategy_account_v1", + "capital_scope": "allocated_sleeve", + "valuation_basis": "allocated_sleeve_ledger", + "target_currency": "USD", + "max_age_seconds": 300, + "fx_conversion_allowed": False, + }, + "portfolio_binding": { + "schema_version": "qsl.tqqq-evidence-portfolio-snapshot.v1", + "snapshot_digest_sha256": cls._digest(portfolio_snapshot), + "as_of": portfolio_snapshot["as_of"], + "source_identity_sha256": portfolio_snapshot[ + "source_identity_sha256" + ], + "max_age_seconds": 300, + }, + "risk_state_binding": { + "schema_version": cls._RISK_STATE_SCHEMA_VERSION, + "snapshot_digest_sha256": cls._digest(risk_state), + "as_of": risk_state["as_of"], + "max_age_seconds": 300, + }, + "authority": { + "authority_scope": "RESEARCH_ONLY", + "authority_receipt_sha256": candidate.authority_receipt_sha256, + "source_revision": "f" * 40, + "runner_is_authority": False, + "no_order": True, + "no_paper": True, + "no_shadow": True, + "no_live": True, + "no_promotion_authority": True, + }, + } + + @staticmethod + def _snapshot(**overrides: object) -> dict[str, object]: + snapshot: dict[str, object] = { + "schema_version": "qsl.tqqq-evidence-portfolio-snapshot.v1", + "as_of": "2026-09-02T05:59:59Z", + "observed_effective_exposure": 0.0, + "total_equity": 100_000.0, + "source_identity_sha256": "3" * 64, + } + snapshot.update(overrides) + return snapshot + + def _assess( + self, + decision: StrategyDecision, + *, + mandate: Mapping[str, object] | None = None, + risk_state: object = _DEFAULT, + capital_base: object = _DEFAULT, + capital_base_binding: object = _DEFAULT, + snapshot: object = _DEFAULT, + logical_evaluation_time: datetime | None = _NOW, + wall_time: datetime | None = None, + ) -> tuple[object, Mock]: + resolved_risk_state = ( + self._risk_state() if risk_state is self._DEFAULT else risk_state + ) + resolved_capital_base = ( + self._capital_base() if capital_base is self._DEFAULT else capital_base + ) + resolved_binding = ( + self._capital_base_binding() + if capital_base_binding is self._DEFAULT + else capital_base_binding + ) + resolved_snapshot = self._snapshot() if snapshot is self._DEFAULT else snapshot + resolved_mandate = mandate or self._mandate( + capital_base=( + resolved_capital_base + if isinstance(resolved_capital_base, CapitalBaseSnapshot) + else self._capital_base() + ), + risk_state=( + resolved_risk_state + if isinstance(resolved_risk_state, Mapping) + else self._risk_state() + ), + ) + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + with ( + patch( + "quant_platform_kit.risk.gate._utc_now", + return_value=wall_time or self._NOW, + ), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = assess_with_evidence( + decision, + resolved_snapshot, + scope="MEMBER", + mandate_provenance=resolved_mandate, + market_data={}, + candidate_identity=self._candidate(), + risk_control_state=resolved_risk_state, + capital_base=resolved_capital_base, + capital_base_binding=resolved_binding, + logical_evaluation_time=logical_evaluation_time, + ) + return result, engine + + def test_valid_weight_targets_approve_evidence_but_never_execution(self) -> None: + decision = _decision( + positions=( + PositionTarget(symbol="TQQQ", target_weight=0.10), + PositionTarget(symbol="QQQM", target_weight=0.05), + PositionTarget(symbol="BOXX", target_weight=0.05), + ) + ) + + result, engine = self._assess(decision) + + self.assertEqual(result.assessment.outcome, "APPROVE") + self.assertEqual(result.assessment.evaluated_at, "2026-09-02T06:00:00Z") + self.assertAlmostEqual(result.assessment.proposed_effective_exposure, 0.40) + self.assertEqual(len(result.assessment.risk_control_state_digest_sha256), 64) + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, decision.positions) + engine.assess.assert_called_once_with(decision, self._snapshot(), market_data={}) + + def test_new_schema_validation_is_exact_and_closed(self) -> None: + base = self._mandate() + invalid_mandates = [] + for section in ( + None, + "candidate_binding", + "validity", + "portfolio_policy", + "capital_binding", + "portfolio_binding", + "risk_state_binding", + "authority", + ): + mutated = dict(base) + if section is None: + mutated["unknown"] = True + else: + mutated[section] = {**base[section], "unknown": True} + invalid_mandates.append(mutated) + invalid_mandates.extend( + ( + {**base, "schema_version": "qsl.tqqq-evidence-risk-mandate.v2"}, + { + **base, + "validity": {**base["validity"], "single_consumption": False}, + }, + { + **base, + "validity": { + **base["validity"], + "expires_at": "2026-09-02T06:04:31Z", + }, + }, + { + **base, + "portfolio_policy": { + **base["portfolio_policy"], + "stress_loss_is_model_assumption": False, + }, + }, + { + **base, + "authority": {**base["authority"], "runner_is_authority": True}, + }, + ) + ) + for field in ( + "no_order", + "no_paper", + "no_shadow", + "no_live", + "no_promotion_authority", + ): + invalid_mandates.append( + { + **base, + "authority": {**base["authority"], field: False}, + } + ) + + for mandate in invalid_mandates: + with self.subTest(mandate=mandate): + result, engine = self._assess(_decision(), mandate=mandate) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + engine.assess.assert_called_once() + + def test_weight_targets_require_exact_capital_base_binding(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),) + ) + stale = self._capital_base(as_of=self._NOW - timedelta(seconds=301)) + future = self._capital_base(as_of=self._NOW + timedelta(seconds=1)) + wrong_scope = _capital_base( + as_of=self._NOW - timedelta(seconds=1), + strategy_scope=self._STRATEGY_PROFILE, + ) + wrong_scope_binding = _capital_base_binding( + strategy_scope=self._STRATEGY_PROFILE, + ) + digest_mismatch = self._mandate() + digest_mismatch["capital_binding"] = { + **digest_mismatch["capital_binding"], + "snapshot_digest_sha256": "9" * 64, + } + cases = ( + (None, self._capital_base_binding(), self._mandate(), "missing_capital_base"), + ( + stale, + self._capital_base_binding(), + self._mandate(capital_base=stale), + "stale_capital_base", + ), + ( + future, + self._capital_base_binding(), + self._mandate(capital_base=future), + "future_capital_base", + ), + ( + wrong_scope, + wrong_scope_binding, + self._mandate(capital_base=wrong_scope), + "capital_base_mandate_mismatch", + ), + ( + self._capital_base(), + self._capital_base_binding(), + digest_mismatch, + "capital_base_digest_mismatch", + ), + ) + for capital_base, binding, mandate, reason in cases: + with self.subTest(reason=reason): + result, engine = self._assess( + decision, + mandate=mandate, + capital_base=capital_base, + capital_base_binding=binding, + ) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn(reason, result.assessment.reason_codes) + self.assertEqual(result.decision.positions, ()) + engine.assess.assert_called_once() + + def test_capital_snapshot_digest_binds_target_equity_denominator(self) -> None: + authorized_capital = self._capital_base() + substituted_capital = _capital_base( + reported_equity=25_000.0, + as_of=self._NOW - timedelta(seconds=1), + strategy_scope=self._STRATEGY_PROFILE, + capital_scope=CapitalScope.ALLOCATED_SLEEVE, + valuation_basis=CapitalValuationBasis.ALLOCATED_SLEEVE_LEDGER, + allocation_scope="tqqq-candidate-research-sleeve", + component_coverage_digest_sha256="2" * 64, + ) + result, engine = self._assess( + _decision( + positions=(PositionTarget(symbol="TQQQ", target_value=10_000.0),) + ), + mandate=self._mandate(capital_base=authorized_capital), + capital_base=substituted_capital, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("capital_base_digest_mismatch", result.assessment.reason_codes) + self.assertEqual(result.decision.positions, ()) + engine.assess.assert_called_once() + + def test_portfolio_snapshot_is_authority_digest_bound(self) -> None: + authorized_snapshot = self._snapshot(observed_effective_exposure=0.60) + substituted_snapshot = self._snapshot(observed_effective_exposure=0.0) + result, engine = self._assess( + _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),) + ), + mandate=self._mandate(portfolio_snapshot=authorized_snapshot), + snapshot=substituted_snapshot, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn( + "portfolio_snapshot_digest_mismatch", + result.assessment.reason_codes, + ) + self.assertEqual(result.decision.positions, ()) + engine.assess.assert_called_once() + + def test_assessment_snapshot_digest_commits_to_target_equity(self) -> None: + first_capital = self._capital_base() + first_snapshot = self._snapshot() + second_capital = _capital_base( + reported_equity=25_000.0, + as_of=self._NOW - timedelta(seconds=1), + strategy_scope=self._STRATEGY_PROFILE, + capital_scope=CapitalScope.ALLOCATED_SLEEVE, + valuation_basis=CapitalValuationBasis.ALLOCATED_SLEEVE_LEDGER, + allocation_scope="tqqq-candidate-research-sleeve", + component_coverage_digest_sha256="2" * 64, + ) + second_snapshot = self._snapshot(total_equity=25_000.0) + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),) + ) + first, first_engine = self._assess( + decision, + mandate=self._mandate( + capital_base=first_capital, + portfolio_snapshot=first_snapshot, + ), + capital_base=first_capital, + snapshot=first_snapshot, + ) + second, second_engine = self._assess( + decision, + mandate=self._mandate( + capital_base=second_capital, + portfolio_snapshot=second_snapshot, + ), + capital_base=second_capital, + snapshot=second_snapshot, + ) + + self.assertEqual(first.assessment.outcome, "APPROVE") + self.assertEqual(second.assessment.outcome, "APPROVE") + self.assertNotEqual( + first.assessment.portfolio_snapshot_digest_sha256, + second.assessment.portfolio_snapshot_digest_sha256, + ) + first_engine.assess.assert_called_once() + second_engine.assess.assert_called_once() + + def test_mandate_and_portfolio_snapshot_freshness_are_logical_time_bound(self) -> None: + base = self._mandate() + expired = { + **base, + "validity": { + **base["validity"], + "effective_at": "2026-09-02T05:55:00Z", + "expires_at": "2026-09-02T05:59:59Z", + }, + } + future = { + **base, + "validity": { + **base["validity"], + "effective_at": "2026-09-02T06:00:01Z", + "expires_at": "2026-09-02T06:05:01Z", + }, + } + cases = ( + (expired, self._snapshot(), "expired_mandate"), + (future, self._snapshot(), "expired_mandate"), + (base, self._snapshot(as_of="2026-09-02T05:54:59Z"), "stale_portfolio_snapshot"), + (base, self._snapshot(as_of="2026-09-02T06:00:01Z"), "stale_portfolio_snapshot"), + (base, self._snapshot(as_of="2026-09-02T14:00:00+08:00"), "invalid_portfolio_snapshot"), + ) + for mandate, snapshot, reason in cases: + with self.subTest(reason=reason): + result, engine = self._assess( + _decision(), + mandate=mandate, + snapshot=snapshot, + ) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn(reason, result.assessment.reason_codes) + self.assertFalse(result.assessment.execution_authorized) + engine.assess.assert_called_once() + + def test_risk_state_is_required_fresh_candidate_bound_and_digest_bound(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),) + ) + stale = self._risk_state(as_of="2026-09-02T05:54:59Z") + future = self._risk_state(as_of="2026-09-02T06:00:01Z") + digest_mismatch = self._risk_state(account_drawdown_fraction=0.04) + cases = ( + ({}, self._mandate(), "invalid_risk_control_state"), + (stale, self._mandate(risk_state=stale), "stale_risk_control_state"), + (future, self._mandate(risk_state=future), "stale_risk_control_state"), + (digest_mismatch, self._mandate(), "risk_control_state_digest_mismatch"), + ( + self._risk_state(candidate_identity_sha256="0" * 64), + self._mandate( + risk_state=self._risk_state(candidate_identity_sha256="0" * 64) + ), + "risk_control_candidate_mismatch", + ), + ) + for risk_state, mandate, reason in cases: + with self.subTest(reason=reason): + result, engine = self._assess( + decision, + mandate=mandate, + risk_state=risk_state, + ) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn(reason, result.assessment.reason_codes) + self.assertEqual(result.decision.positions, ()) + engine.assess.assert_called_once() + + def test_assets_caps_effective_exposure_and_loss_budget_fail_closed(self) -> None: + cases = ( + ( + _decision(positions=(PositionTarget(symbol="QQQ", target_weight=0.10),)), + "benchmark_only_asset", + ), + ( + _decision(positions=(PositionTarget(symbol="QQQ", target_weight=0.0),)), + "benchmark_only_asset", + ), + ( + _decision(positions=(PositionTarget(symbol="QQQ", target_value=0.0),)), + "benchmark_only_asset", + ), + ( + _decision(positions=(PositionTarget(symbol="SPY", target_weight=0.10),)), + "asset_not_authorized", + ), + ( + _decision(positions=(PositionTarget(symbol="SPY", target_weight=0.0),)), + "asset_not_authorized", + ), + ( + _decision(positions=(PositionTarget(symbol="SPY", target_value=0.0),)), + "asset_not_authorized", + ), + ( + _decision( + positions=( + PositionTarget(symbol="TQQQ", target_weight=0.10), + PositionTarget(symbol="TQQQ", target_weight=0.06), + ) + ), + "product_exposure_cap", + ), + ( + _decision( + positions=( + PositionTarget(symbol="TQQQ", target_weight=0.15), + PositionTarget(symbol="QQQM", target_weight=0.06), + ) + ), + "effective_exposure_cap", + ), + ( + _decision( + positions=( + PositionTarget(symbol="TQQQ", target_weight=0.10), + PositionTarget(symbol="QQQM", target_weight=0.11), + ) + ), + "risk_budget_exposure_cap", + ), + ( + StrategyDecision(budgets=(BudgetIntent(name="loss", amount=0.011),)), + "unsupported_evidence_budget", + ), + ) + for decision, reason in cases: + with self.subTest(reason=reason): + result, engine = self._assess(decision) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn(reason, result.assessment.reason_codes) + self.assertFalse(result.assessment.execution_authorized) + engine.assess.assert_called_once() + + def test_evidence_mandate_rejects_undefined_budget_units(self) -> None: + result, engine = self._assess( + StrategyDecision( + budgets=( + BudgetIntent( + name="unrelated", + amount=0.005, + unit="arbitrary_unit", + ), + ) + ) + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("unsupported_evidence_budget", result.assessment.reason_codes) + self.assertEqual(result.decision.budgets, ()) + engine.assess.assert_called_once() + + def test_drawdown_above_ten_percent_parks_as_account_breaker(self) -> None: + risk_state = self._risk_state( + account_drawdown_fraction=0.100001, + drawdown_scalar=0.0, + ) + result, engine = self._assess( + _decision(positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),)), + mandate=self._mandate(risk_state=risk_state), + risk_state=risk_state, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("account_breaker_triggered", result.assessment.reason_codes) + self.assertTrue(result.assessment.account_breaker_triggered) + self.assertFalse(result.assessment.execution_authorized) + engine.assess.assert_called_once() + + def test_drawdown_scalar_boundaries_approve_only_with_exact_loss_budget(self) -> None: + cases = ( + self._risk_state(account_drawdown_fraction=0.05, drawdown_scalar=1.0), + self._risk_state(account_drawdown_fraction=0.050001, drawdown_scalar=0.5), + self._risk_state(account_drawdown_fraction=0.10, drawdown_scalar=0.5), + ) + for risk_state in cases: + with self.subTest(risk_state=risk_state): + result, engine = self._assess( + _decision( + positions=( + PositionTarget(symbol="TQQQ", target_weight=0.10), + ) + ), + mandate=self._mandate(risk_state=risk_state), + risk_state=risk_state, + ) + self.assertEqual(result.assessment.outcome, "APPROVE") + self.assertFalse(result.assessment.execution_authorized) + engine.assess.assert_called_once() + + def test_explicit_logical_time_rejects_future_stale_non_utc_and_subsecond(self) -> None: + cases = ( + ( + self._NOW + timedelta(seconds=1), + "future_logical_evaluation_time", + ), + ( + self._NOW - timedelta(seconds=301), + "stale_logical_evaluation_time", + ), + ( + self._NOW.astimezone(timezone(timedelta(hours=8))), + "invalid_logical_evaluation_time", + ), + ( + self._NOW.replace(microsecond=1), + "invalid_logical_evaluation_time", + ), + ) + for logical_time, reason in cases: + with self.subTest(reason=reason): + result, engine = self._assess( + _decision(), + logical_evaluation_time=logical_time, + ) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertEqual(result.assessment.reason_codes, (reason,)) + self.assertFalse(result.assessment.execution_authorized) + engine.assess.assert_called_once() + + def test_same_logical_time_produces_identical_receipt_bytes(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="TQQQ", target_weight=0.10),) + ) + first, first_engine = self._assess(decision, wall_time=self._NOW) + second, second_engine = self._assess( + decision, + wall_time=self._NOW + timedelta(seconds=100), + ) + + first_bytes = json.dumps( + first.assessment.__dict__, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + second_bytes = json.dumps( + second.assessment.__dict__, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + self.assertEqual(first.assessment.outcome, "APPROVE") + self.assertEqual(first_bytes, second_bytes) + self.assertEqual( + first.assessment.assessment_sha256, + second.assessment.assessment_sha256, + ) + first_engine.assess.assert_called_once() + second_engine.assess.assert_called_once() + + class RetiredGlobalEtfRotationMandateTests(unittest.TestCase): _NOW = datetime(2026, 8, 9, 2, 0, tzinfo=timezone.utc) _MANDATE_ID = "global_etf_rotation_etf_only_research_v1"