diff --git a/docs/market-regime-composition-candidate.zh-CN.md b/docs/market-regime-composition-candidate.zh-CN.md index 02c5b1a..9672a9d 100644 --- a/docs/market-regime-composition-candidate.zh-CN.md +++ b/docs/market-regime-composition-candidate.zh-CN.md @@ -72,13 +72,38 @@ crisis、硬上限和必需数据门优先约束机会侧。TACO/panic reversal/ ## 三个有界合成验收规格 -以下是待实现测试规格，**本文件未运行新测试或收益优化**。只使用固定 JSON/日期和代数，不读取 provider、运行 AI、回测历史、搜索阈值或连接 broker。旧策略只作 characterization，新政策必须另有入口，不能覆盖旧断言。 +以下为冻结的合成验收规格；2026-10-06 已实现其离线组合比较部分，具体覆盖和未覆盖边界见下节。**未运行收益优化，也未实现跨周期 release consumer**。只使用固定 JSON/日期和代数，不读取 provider、运行 AI、回测历史、搜索阈值或连接 broker。旧策略只作 characterization，新政策必须另有入口，不能覆盖旧断言。 1. **同维优先级与交集对照**：fresh benchmark `(L=.5,R=.5)`、fresh macro `(L=.1,R=.1)`，相同 scope/分母/决策时点。旧输出 source=benchmark、`.5/.5`；候选 `.1/.1`，两维都绑定 macro，仍记录 benchmark。反转输入顺序和重复同身份记录不改变候选 digest/结果。断言不是 `.05`，旧默认输出不变 2. **同源去重与独立维度**：同一个 guard observation 同时给 L=R=.5，并重复同身份约束；baseline 两腿各450，使用上述冻结 legacy adapter。旧与候选 cap 都是 `.5/.5`；候选保留两个维度，代数结果仍为112.5/337.5、总风险450。原因去重不能删 R、删 L、变成总风险225或新增第三个 risk_budget 折扣。这是目标代数检查，不是持仓或实际买入证明 3. **新旧日期与机会冲突**：fresh benchmark `.5/.5`、已过 valid_until 的必需 macro `.1/.1`、同日 opportunity；前一已接受 transition 为受限 episode，且无新策略 intent/release event。旧 arbiter 不在此函数逐源验时，仍优先 benchmark `.5/.5`；候选为 UNKNOWN/BLOCKED，机会不可放宽，过去限制不得因 stale 消失而自动恢复。记录当前有效 `.5` 只作已验约束，不当作 executable full budget；strict consumer 增加风险许可为 false，账户不可信时不提交任何假设减仓 -正式安全测试另需覆盖 policy/input 版本错配、同身份矛盾、多个 scope、不同单位、crisis cap=0、非交换转移顺序、并发/部分成交/挂单、重复 release 与 AI 文案变更；这些要求不冒充本轮额外合成结果。 +本轮纯函数负例覆盖 policy/input 版本错配、同身份矛盾、scope/单位/分母不一致、cap=0 及 AI 文案变更。正式 consumer 安全测试仍需覆盖非交换转移、并发/部分成交/挂单、完整前驱状态、重复 release 和 crash 生命周期；不能把离线合成结果当作这些状态链的证明。 + +## 2026-10-06 研究比较器工程状态与冻结接口 + +消费者状态仍为 `DESIGN_ONLY_NOT_RUNTIME`。独立模块 +`src/quant_strategy_plugins/market_regime_composition_research.py` 提供 +`compose_research_constraints(policy, sources, constraints, expected_policy_sha256=...)`， +仅比较**已映射且冻结的合成约束**，不读取 provider、不从事实推导 cap、不接线任何 runner。 +原 `market_regime_arbiter.v1` 及其旧测试不变；`.5` 优先级 characterization 不是“生产 bug 已修复”。 + +冻结的窄研究接口如下；它是策略侧离线 trial 记录，**不是 V2 envelope 的扩展或可执行政策 DSL**： + +- `policy_id` 固定为本文候选 ID，`policy_version=v1`；`expected_policy_sha256` 校验完整 policy canonical JSON，包括下面的源/约束绑定。未知版本、畸形 policy 或 hash 不符直接抛 `ResearchCompositionError`，不返回任何提案 +- policy 必填 `strategy={candidate_id, revision, config_sha256}`、`baseline={baseline_id, baseline_sha256}`；strategy 和 producer revision 均要求完整 40 位 SHA-1 或 64 位 SHA-256。这里只绑定 frozen baseline 身份，不计算 baseline、目标或股数，也不证明该策略或 baseline 已获批准 +- `scope={market, symbols, calendar, time_zone}` 必须精确相同；`denominators` 为两个既定维度分别指定不同的 frozen baseline 分母身份。仅支持 `measurement=baseline_nominal_ratio.v1`、`unit=ratio` 和有限 `[0,1]` cap；金额、quantity、effective exposure、`risk_budget_scalar` alias 均不能冒充第三维或自动转换 +- policy 的 `requested_as_of`、`decision_at` 及 `validity_policy_version=research.explicit_interval_and_age.v1` 必填。逐源 `inputs` 绑定 `observation_id`、`required`、`applicable`、`factor_id`、`algorithm_version`、`window`、`sampling`、`max_delay_seconds`、`source_sha256` 和 `constraints_sha256`；没有默认来源或默认 TTL +- 每个 source 是 `{observation_id, envelope, received_at}`。已有 V2 validator 原样检查 envelope 的 schema、producer repo/full revision/code/config hash、P1 manifest/root hash 和 payload hash；再验证整个 source 的冻结 hash。因此这些是合成输入身份和完整性验证，不是签名、源真实性、运行时安装或 provider 可用性的证明 +- envelope 的确定性 payload 保留自己的 `observation_id/factor_id/algorithm_version/scope/window/sampling`、`window_complete=true`、`observed_as_of/published_at/effective_from/valid_until`。所有时间须带时区；`observed_as_of <= requested_as_of <= decision_at`，且 `observed_as_of <= published_at <= received_at <= decision_at`。有效区间冻结为 `[effective_from, valid_until)`，年龄以 `decision_at - observed_as_of <= max_delay_seconds` 检查，收到新报告不会刷新观测年龄。日历标识只作精确绑定，本 helper 不计算交易 session +- 每条 constraint 必填 `constraint_id/observation_id/policy_version/scope/dimension/measurement/unit/denominator_id/cap/kind/reason_codes/reason_group_id`。`kind` 只允许 `hard/soft/watch_only/opportunity`。约束和对应 source 必须同时匹配 trial 冻结 hash；`constraints_sha256(...)` 只为冻结试验生成排序、精确去重后的集合 hash，不为实际 cap 映射背书 +- source/constraint 输入顺序和精确重复不改变输出；同一约束身份的冲突、未知来源、必需输入缺失/失效会阻塞。适用且有效的 hard 上限逐维取 min，所有并列 binding 和 non-binding 来源均保留。显式 `applicable=false` 的缺失输入记为 `NOT_APPLICABLE`，不扩散成全市场故障；若实际提供该源，仍校验 source 结构、V2/forbidden 字段、固定身份及 hash，畸形内容直接抛 `ResearchCompositionError`、不产生提案，不以不适用跳过完整性校验。实际提供的非适用约束集合也核验冻结 hash 和同身份冲突；非适用约束集合缺失仍允许。该源的其他市场 scope 或过期不触发目标市场的可用性门 +- 输出是 `READY_RESEARCH` 或 `UNKNOWN/BLOCKED` 的研究记录，含完整 source、各约束选用/未选用原因、policy/input 身份及 canonical `decision_digest`。阻塞时只列其余已验证 cap；某维没有可验 hard cap 则为 `null`，不能解释为满额预算。`executable/risk_increase_allowed/release_authority` 始终为 `false`，没有订单、减仓、恢复或账户决策功能 +- 可选 `ai_narrative` 是明确隔离的人类说明输入，完全不读取、不返回、不纳入 digest；AI 混入确定性 envelope 仍由已有 validator 拒绝。确定性原因码和真实来源变化则改变 digest + +`tests/test_market_regime_composition_research.py` 仅使用固定人工日期、JSON 与代数。三项规格中，第一项验证旧 priority `.5/.5` 与研究 min `.1/.1`；第二项验证 L/R 独立并在测试内保留 `112.5/337.5` 的 legacy 代数；第三项仅验证 stale 必需输入加机会信号时阻塞、有效 `.5` 不变成执行预算。**第三项所需前驱 transition、retained quantity ceiling、新 intent/release 和账户订单检查尚未在本 helper 实现或验收**。本轮没有新增 UES adapter、state store、pin/config 或 provider，也没有 consumer 采用、PIT/OOS 或收益结论。 + +复验命令：`PYTHONPATH=src python -m unittest discover -s tests -p test_market_regime_composition_research.py -v`；完整仓库验证仍沿用 CONTRIBUTING/CI。正式采用依赖后续 UES 强 release 合同/真实 intent issuer、source 与状态链验收、独立 review 和预先冻结的 OOS/shadow，不能因纯函数通过测试而跳过。 ## OOS 消融与准入要求 @@ -96,6 +121,6 @@ crisis、硬上限和必需数据门优先约束机会侧。TACO/panic reversal/ - Nautilus [Execution](https://nautilustrader.io/docs/latest/concepts/execution/) 支持订单级检查、REDUCING 合格 reduce-only 与 HALTED 的区别；这些状态不是自动清仓，也不替本系统证明已执行对应控制 - Cboe [Volatility Trading](https://www.cboe.com/tradable-products/volatility-trading) 将 VIX 定义为 SPX 期权隐含的30日期望波动且非方向预测；跨市场适用性和风险折扣仍须本候选验证，不拿 VIX 下降作恢复批准 -最小集成只新增本设计文档；可在已有 market-regime plan 增加“未生效组合候选”的链接，但不修改旧优先级说明为已生效 min-policy。未来实现先做独立 pure research composition helper/测试及 UES candidate adapter，经独立 review 后再决定接线。本文件不推荐迁移 LEAN/Nautilus、重建统一引擎、增加市场指标或删除现有腿约束。 +当前最小落地为本设计文档和独立 pure research composition helper/测试；不修改旧优先级说明为已生效 min-policy。UES candidate adapter 和强 release/state 合同仍是后续独立工作，经独立 review 后再决定接线。本文件不推荐迁移 LEAN/Nautilus、重建统一引擎、增加市场指标或删除现有腿约束。 AI 仅为人的来源可追溯 narrative。AI 输出、失败、延迟、agree/confidence/摘要变更必须不改变确定性事实、policy 状态、目标 digest 或授权；人的实际决定另有明确 decision record。AI 不能放宽风险、选择更宽 cap、解决权限冲突或批准恢复。 diff --git a/src/quant_strategy_plugins/market_regime_composition_research.py b/src/quant_strategy_plugins/market_regime_composition_research.py new file mode 100644 index 0000000..abaab99 --- /dev/null +++ b/src/quant_strategy_plugins/market_regime_composition_research.py @@ -0,0 +1,288 @@ +"""Offline composition of frozen, already-mapped research constraints. + +DESIGN_ONLY_NOT_RUNTIME: this comparator is not a strategy adapter, release +store, risk gate, source reader, or permission issuer. Source-to-cap mappings +are supplied and hash-bound by a synthetic research trial, not inferred here. +No caller, including a caller supplying permissive caps, receives authority. +""" +from __future__ import annotations + +import math +import re +from collections.abc import Mapping, Sequence +from datetime import datetime, timezone +from typing import Any +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from .plugin_signal_envelope_v2 import ( + SignalEnvelopeValidationError, + canonical_json_bytes, + payload_sha256, + validate_signal_envelope, +) + +POLICY_ID = "market_regime_composition.same_dimension_min.v1" +POLICY_VERSION = "v1" +VALIDITY_POLICY_VERSION = "research.explicit_interval_and_age.v1" +DIMENSIONS = ("leverage_leg_nominal", "total_risk_asset_nominal") +_MEASUREMENT = "baseline_nominal_ratio.v1" +_POLICY_FIELDS = { + "policy_id", "policy_version", "strategy", "baseline", "scope", "denominators", + "requested_as_of", "decision_at", "validity_policy_version", "inputs", +} +_BINDING_FIELDS = { + "observation_id", "required", "applicable", "factor_id", "algorithm_version", "window", "sampling", + "max_delay_seconds", "source_sha256", "constraints_sha256", +} +_CONSTRAINT_FIELDS = { + "constraint_id", "observation_id", "policy_version", "scope", "dimension", "measurement", + "unit", "denominator_id", "cap", "kind", "reason_codes", "reason_group_id", +} + + +class ResearchCompositionError(ValueError): + """A malformed/unknown frozen research contract was rejected, without output.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise ResearchCompositionError(message) + + +def _fields(value: Any, fields: set[str], location: str) -> None: + _require(isinstance(value, dict) and set(value) == fields, f"invalid {location} fields") + + +def _text(value: Any) -> bool: + return isinstance(value, str) and bool(value.strip()) and value == value.strip() + + +def _hash(value: Any) -> bool: + return isinstance(value, str) and re.fullmatch(r"[0-9a-f]{64}", value) is not None + + +def _time(value: Any) -> datetime: + _require(_text(value), "timestamp must be a timezone-aware ISO string") + try: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + _require(result.tzinfo is not None and result.utcoffset() is not None, "timestamp needs a timezone") + return result.astimezone(timezone.utc) + except ValueError as exc: + raise ResearchCompositionError("invalid timestamp") from exc + + +def _scope(scope: Any) -> None: + _fields(scope, {"market", "symbols", "calendar", "time_zone"}, "scope") + _require(all(_text(scope[k]) for k in ("market", "calendar", "time_zone")), "invalid scope") + symbols = scope["symbols"] + _require(isinstance(symbols, list) and bool(symbols) and all(_text(s) for s in symbols), "invalid symbols") + _require(len(symbols) == len(set(symbols)), "duplicate symbols") + try: + ZoneInfo(scope["time_zone"]) + except (ZoneInfoNotFoundError, ValueError) as exc: + raise ResearchCompositionError("unknown scope timezone") from exc + + +def _unique(records: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]: + """Detached canonical exact set; order/receipt counts have no semantics.""" + import json + + _require(isinstance(records, (list, tuple)), "records must be a list or tuple") + _require(all(isinstance(record, Mapping) for record in records), "record must be an object") + try: + return [json.loads(raw) for raw in sorted({canonical_json_bytes(record) for record in records})] + except SignalEnvelopeValidationError as exc: + raise ResearchCompositionError("records must be finite JSON") from exc + + +def constraints_sha256(constraints: Sequence[Mapping[str, Any]]) -> str: + """Hash an exact, canonical constraint set when freezing a research trial.""" + return payload_sha256(_unique(constraints)) + + +def _validate_policy(policy: dict[str, Any], expected_sha256: str) -> None: + _fields(policy, _POLICY_FIELDS, "policy") + _require(_hash(expected_sha256) and payload_sha256(policy) == expected_sha256, "policy hash mismatch") + _require(policy["policy_id"] == POLICY_ID and policy["policy_version"] == POLICY_VERSION, + "unknown composition policy") + _require(policy["validity_policy_version"] == VALIDITY_POLICY_VERSION, "unknown validity policy") + _fields(policy["strategy"], {"candidate_id", "revision", "config_sha256"}, "strategy") + strategy = policy["strategy"] + _require(_text(strategy["candidate_id"]) and isinstance(strategy["revision"], str) + and re.fullmatch(r"(?:[0-9a-f]{40}|[0-9a-f]{64})", strategy["revision"]) is not None + and _hash(strategy["config_sha256"]), "invalid strategy identity") + _fields(policy["baseline"], {"baseline_id", "baseline_sha256"}, "baseline") + _require(_text(policy["baseline"]["baseline_id"]) and _hash(policy["baseline"]["baseline_sha256"]), + "invalid frozen baseline identity") + _scope(policy["scope"]) + _fields(policy["denominators"], set(DIMENSIONS), "denominators") + _require(all(_text(x) for x in policy["denominators"].values()) + and len(set(policy["denominators"].values())) == 2, "dimensions need distinct denominator identities") + _require(_time(policy["requested_as_of"]) <= _time(policy["decision_at"]), "request cutoff after decision") + _require(isinstance(policy["inputs"], list) and bool(policy["inputs"]), "input bindings are required") + seen = set() + for binding in policy["inputs"]: + _fields(binding, _BINDING_FIELDS, "input binding") + _require(all(_text(binding[k]) for k in ( + "observation_id", "factor_id", "algorithm_version", "window", "sampling")), "invalid input identity") + _require(binding["observation_id"] not in seen, "duplicate policy observation identity") + seen.add(binding["observation_id"]) + _require(type(binding["required"]) is bool and type(binding["applicable"]) is bool, "invalid applicability") + _require(type(binding["max_delay_seconds"]) is int and binding["max_delay_seconds"] >= 0, + "invalid maximum delay") + _require(_hash(binding["source_sha256"]) and _hash(binding["constraints_sha256"]), "invalid input hashes") + + +def _validate_constraint(constraint: dict[str, Any], policy: dict[str, Any]) -> None: + _fields(constraint, _CONSTRAINT_FIELDS, "constraint") + _require(all(_text(constraint[k]) for k in ("constraint_id", "observation_id", "reason_group_id")), + "invalid constraint identity") + dimension = constraint["dimension"] + _require(isinstance(dimension, str) and dimension in DIMENSIONS, "unsupported dimension") + _require(constraint["policy_version"] == POLICY_VERSION, "constraint policy mismatch") + _require(constraint["scope"] == policy["scope"], "constraint scope mismatch") + _require(constraint["unit"] == "ratio" and constraint["measurement"] == _MEASUREMENT + and constraint["denominator_id"] == policy["denominators"][dimension], "incomparable constraint") + cap = constraint["cap"] + _require(type(cap) in (int, float) and 0 <= cap <= 1 and math.isfinite(cap), "invalid cap") + _require(constraint["kind"] in ("hard", "soft", "watch_only", "opportunity"), "unsupported constraint kind") + reasons = constraint["reason_codes"] + _require(isinstance(reasons, list) and bool(reasons) and all(_text(x) for x in reasons), "invalid reason codes") + + +def _constraint_identity(constraint: Mapping[str, Any]) -> bytes: + return canonical_json_bytes([constraint.get(k) for k in ( + "constraint_id", "scope", "dimension", "observation_id", "policy_version" + )]) + + +def _source_envelope(source: dict[str, Any], binding: dict[str, Any]) -> dict[str, Any]: + """Provided sources must be valid even when their scope is inapplicable.""" + _fields(source, {"observation_id", "envelope", "received_at"}, "source") + envelope = validate_signal_envelope(source["envelope"]) + _require(re.fullmatch(r"(?:[0-9a-f]{40}|[0-9a-f]{64})", envelope["producer"]["revision"]) is not None, + "producer revision must be a full Git SHA-1 or SHA-256") + _require(payload_sha256(source) == binding["source_sha256"], "source hash mismatch") + _time(source["received_at"]) + payload = envelope["payload"] + for field in ("observation_id", "factor_id", "algorithm_version", "window", "sampling"): + _require(payload.get(field) == binding[field], f"source {field} mismatch") + return envelope + + +def _validate_source(source: dict[str, Any], binding: dict[str, Any], policy: dict[str, Any]) -> list[str]: + payload = _source_envelope(source, binding)["payload"] + _require(payload.get("scope") == policy["scope"], "source scope/calendar mismatch") + _require(payload.get("window_complete") is True, "incomplete observation window") + try: + observed, published, effective, until = ( + _time(payload[field]) for field in ("observed_as_of", "published_at", "effective_from", "valid_until") + ) + except KeyError as exc: + raise ResearchCompositionError("missing input time") from exc + received = _time(source["received_at"]) + decision, requested = _time(policy["decision_at"]), _time(policy["requested_as_of"]) + reasons = [] + if not observed <= published <= received <= decision or observed > requested: + reasons.append("NOT_VISIBLE_AT_CUTOFF") + if not effective <= decision: + reasons.append("NOT_EFFECTIVE") + if not effective < until or decision >= until: + reasons.append("EXPIRED") + if (decision - observed).total_seconds() > binding["max_delay_seconds"]: + reasons.append("STALE_OBSERVATION") + return reasons + + +def compose_research_constraints( + policy: Mapping[str, Any], + sources: Sequence[Mapping[str, Any]], + constraints: Sequence[Mapping[str, Any]], + *, + expected_policy_sha256: str, + ai_narrative: Any = None, +) -> dict[str, Any]: + """Return a non-executable min-cap comparator, or reject a malformed policy. + + Policy content includes exact source/constraint bindings for this trial. + Hash checks are integrity checks, not signatures or runtime-install proofs. + Invalid required observations block the proposal and retain other validated + constraints. Missing dimensions under BLOCKED are null, never a free budget. + ``ai_narrative`` is intentionally not inspected, returned, or digest-bound. + """ + policy = _unique([policy])[0] + _validate_policy(policy, expected_policy_sha256) + sources, constraints = _unique(sources), _unique(constraints) + bindings = {item["observation_id"]: item for item in policy["inputs"]} + for record in sources + constraints: + _require(_text(record.get("observation_id")), "missing observation identity") + unknown = sorted({r["observation_id"] for r in sources + constraints} - bindings.keys()) + reports, valid = [], set() + blocked = bool(unknown) + for obs, binding in sorted(bindings.items()): + observed = [s for s in sources if s["observation_id"] == obs] + mapped = [c for c in constraints if c["observation_id"] == obs] + reasons = [] + if not binding["applicable"]: + _require(len(observed) <= 1, "conflicting inapplicable source identity") + if observed: + try: + _source_envelope(observed[0], binding) + except SignalEnvelopeValidationError as exc: + raise ResearchCompositionError("invalid inapplicable envelope: " + str(exc)) from exc + if mapped: + identities = [_constraint_identity(c) for c in mapped] + _require(len(identities) == len(set(identities)), "conflicting inapplicable constraint identity") + _require(constraints_sha256(mapped) == binding["constraints_sha256"], + "inapplicable constraints hash mismatch") + status = "NOT_APPLICABLE" + else: + if len(observed) != 1: + reasons.append("MISSING_INPUT" if not observed else "CONFLICTING_SOURCE_IDENTITY") + identities = [_constraint_identity(c) for c in mapped] + if len(identities) != len(set(identities)): + reasons.append("CONFLICTING_CONSTRAINT_IDENTITY") + try: + if len(observed) == 1: + reasons.extend(_validate_source(observed[0], binding, policy)) + for item in mapped: + _validate_constraint(item, policy) + _require(constraints_sha256(mapped) == binding["constraints_sha256"], "constraints hash mismatch") + except (ResearchCompositionError, SignalEnvelopeValidationError) as exc: + reasons.append("INVALID_INPUT:" + str(exc)) + status = "UNAVAILABLE" if reasons else "VALID" + if status == "VALID": + valid.add(obs) + elif binding["required"] or any(r.startswith("CONFLICTING_") for r in reasons): + blocked = True + reports.append({"observation_id": obs, "status": status, "reason_codes": sorted(set(reasons))}) + reports.extend({"observation_id": obs, "status": "UNAVAILABLE", "reason_codes": ["UNBOUND_INPUT"]} + for obs in unknown) + caps = {} + for dimension in DIMENSIONS: + values = [c["cap"] for c in constraints if c["observation_id"] in valid + and c["kind"] == "hard" and c["dimension"] == dimension] + caps[dimension] = min(values) if values else (None if blocked else 1.0) + selections = [] + for constraint in constraints: + obs = constraint["observation_id"] + if obs in bindings and not bindings[obs]["applicable"]: + selection = "NOT_APPLICABLE" + elif obs not in valid: + selection = "INPUT_UNAVAILABLE" + elif constraint["kind"] == "hard": + selection = "BINDING" if constraint["cap"] == caps[constraint["dimension"]] else "NON_BINDING" + else: + selection = {"opportunity": "OPPORTUNITY_VETOED", "watch_only": "WATCH_ONLY", "soft": "SOFT"}[ + constraint["kind"] + ] + selections.append({"constraint": constraint, "selection": selection}) + result = { + "mode": "research_only", "policy": policy, "policy_sha256": expected_policy_sha256, + "status": "BLOCKED" if blocked else "READY_RESEARCH", "data_state": "UNKNOWN" if blocked else "KNOWN", + "caps": caps, "inputs": sorted(reports, key=lambda item: item["observation_id"]), + "sources": sources, "constraints": selections, + "executable": False, "risk_increase_allowed": False, "release_authority": False, + } + result["decision_digest"] = payload_sha256(result) + return result diff --git a/tests/test_market_regime_composition_research.py b/tests/test_market_regime_composition_research.py new file mode 100644 index 0000000..689239a --- /dev/null +++ b/tests/test_market_regime_composition_research.py @@ -0,0 +1,395 @@ +"""Synthetic, offline characterizations; these are not trading/release tests.""" +from copy import deepcopy +import unittest + +from quant_strategy_plugins.market_regime_control_plugin import build_market_regime_control_signal +from quant_strategy_plugins.market_regime_composition_research import ( + DIMENSIONS, + POLICY_ID, + POLICY_VERSION, + VALIDITY_POLICY_VERSION, + ResearchCompositionError, + compose_research_constraints, + constraints_sha256, +) +from quant_strategy_plugins.plugin_signal_envelope_v2 import build_signal_envelope, payload_sha256 + + +SCOPE = {"market": "US", "symbols": ["QQQ", "TQQQ"], "calendar": "XNYS", "time_zone": "America/New_York"} +NOW = "2026-10-06T20:00:00Z" + + +def fixture(names=("benchmark", "macro")): + policy = { + "policy_id": POLICY_ID, "policy_version": POLICY_VERSION, + "strategy": {"candidate_id": "synthetic_comparator", "revision": "1" * 40, "config_sha256": "2" * 64}, + "baseline": {"baseline_id": "synthetic-frozen-baseline", "baseline_sha256": "3" * 64}, + "scope": deepcopy(SCOPE), + "denominators": {d: "synthetic-baseline:" + d for d in DIMENSIONS}, + "requested_as_of": NOW, "decision_at": NOW, + "validity_policy_version": VALIDITY_POLICY_VERSION, "inputs": [], + } + sources, constraints = [], [] + for name in names: + obs = name + "-observation-1" + payload = { + "observation_id": obs, "factor_id": name, "algorithm_version": "synthetic.v1", + "scope": deepcopy(SCOPE), "window": "synthetic-20-sessions", "sampling": "session-close", + "observed_as_of": "2026-10-06T19:00:00Z", "published_at": "2026-10-06T19:10:00Z", + "effective_from": "2026-10-06T19:10:00Z", "valid_until": "2026-10-07T20:00:00Z", + "window_complete": True, "synthetic_value": 1, + } + envelope = build_signal_envelope( + plugin_id=name, + producer={"repo": "QuantStrategyLab/QuantStrategyPlugins", "revision": "4" * 40, + "entrypoint": "synthetic:fixture", "code_sha256": "5" * 64, "config_sha256": "6" * 64}, + input_provenance={"p1_manifest_sha256": "7" * 64, "input_root_sha256": "8" * 64, + "date_cutoff": "2026-10-06"}, + payload=payload, + ) + sources.append({"observation_id": obs, "envelope": envelope, "received_at": "2026-10-06T19:20:00Z"}) + policy["inputs"].append({ + "observation_id": obs, "required": True, "applicable": True, + "factor_id": name, "algorithm_version": "synthetic.v1", + "window": payload["window"], "sampling": payload["sampling"], + "max_delay_seconds": 86400, "source_sha256": "0" * 64, "constraints_sha256": "0" * 64, + }) + for dimension in DIMENSIONS: + constraints.append({ + "constraint_id": name + "-cap", "observation_id": obs, "policy_version": POLICY_VERSION, + "scope": deepcopy(SCOPE), "dimension": dimension, "measurement": "baseline_nominal_ratio.v1", + "unit": "ratio", "denominator_id": policy["denominators"][dimension], + "cap": 0.5 if name == "benchmark" else 0.1, "kind": "hard", + "reason_codes": [name + "_synthetic_limit"], "reason_group_id": "synthetic-pressure", + }) + freeze(policy, sources, constraints) + return policy, sources, constraints + + +def freeze(policy, sources, constraints): + """Freeze a new synthetic trial; never repair an untrusted production input.""" + for binding in policy["inputs"]: + obs = binding["observation_id"] + source = next((s for s in sources if s["observation_id"] == obs), None) + if source: + source["envelope"]["payload_sha256"] = payload_sha256(source["envelope"]["payload"]) + binding["source_sha256"] = payload_sha256(source) + binding["constraints_sha256"] = constraints_sha256([c for c in constraints if c["observation_id"] == obs]) + + +def compose(policy, sources, constraints, **kwargs): + return compose_research_constraints( + policy, sources, constraints, expected_policy_sha256=payload_sha256(policy), **kwargs + ) + + +class CompositionTests(unittest.TestCase): + def test_priority_characterization_and_same_dimension_min(self): + legacy = build_market_regime_control_signal({ + "benchmark_guard": {"canonical_route": "risk_reduced", "leverage_scalar": .5, "risk_asset_scalar": .5}, + "macro": {"canonical_route": "delever", "leverage_scalar": .1, "risk_asset_scalar": .1}, + }, as_of="2026-10-06") + self.assertEqual(legacy["arbiter"]["route_source"], "benchmark_guard") + self.assertEqual(legacy["position_control"]["leverage_scalar"], .5) + self.assertEqual(legacy["position_control"]["risk_asset_scalar"], .5) + p, s, c = fixture() + result = compose(p, s, c) + self.assertEqual(result["caps"], dict.fromkeys(DIMENSIONS, .1)) + self.assertEqual(result["status"], "READY_RESEARCH") + self.assertEqual([x["selection"] for x in result["constraints"]].count("BINDING"), 2) + self.assertEqual([x["selection"] for x in result["constraints"]].count("NON_BINDING"), 2) + self.assertNotIn(.05, result["caps"].values()) + self.assertEqual(result, compose(p, list(reversed(s)) + s, list(reversed(c)) + c)) + self.assertFalse(result["executable"]) + self.assertFalse(result["risk_increase_allowed"]) + self.assertFalse(result["release_authority"]) + + def test_same_observation_retains_independent_dimensions_and_legacy_algebra(self): + p, s, c = fixture(("benchmark",)) + result = compose(p, s, c + c) + self.assertEqual(len(result["constraints"]), 2) + leverage, risk = (result["caps"][d] for d in DIMENSIONS) + # The existing transfer-then-R formula is a characterization only. + levered = 450 * leverage + unlevered = 450 + (450 - levered) + self.assertEqual((levered * risk, unlevered * risk), (112.5, 337.5)) + self.assertEqual((levered + unlevered) * risk, 450) + + def test_stale_required_source_and_opportunity_are_blocked_not_sell_all(self): + p, s, c = fixture(("benchmark", "macro", "opportunity")) + s[1]["envelope"]["payload"]["valid_until"] = "2026-10-06T19:59:59Z" + for constraint in c: + if constraint["observation_id"].startswith("opportunity"): + constraint["kind"] = "opportunity" + constraint["cap"] = 1.0 + freeze(p, s, c) + result = compose(p, s, c) + self.assertEqual((result["status"], result["data_state"]), ("BLOCKED", "UNKNOWN")) + self.assertEqual(result["caps"], dict.fromkeys(DIMENSIONS, .5)) + self.assertFalse(result["risk_increase_allowed"]) + self.assertFalse(result["executable"]) + self.assertIn("OPPORTUNITY_VETOED", [x["selection"] for x in result["constraints"]]) + self.assertIn("EXPIRED", result["inputs"][1]["reason_codes"]) + legacy = build_market_regime_control_signal({ + "benchmark_guard": {"canonical_route": "risk_reduced", "leverage_scalar": .5, "risk_asset_scalar": .5}, + "macro": {"canonical_route": "delever", "leverage_scalar": .1, "risk_asset_scalar": .1, + "valid_until": "2026-10-06T19:59:59Z"}, + "taco": {"canonical_route": "taco_rebound", "manual_review_required": True}, + }, as_of="2026-10-06") + self.assertEqual(legacy["position_control"]["risk_asset_scalar"], .5) + + def test_missing_required_has_no_permissive_default(self): + p, s, c = fixture(("macro",)) + result = compose(p, [], []) + self.assertEqual(result["status"], "BLOCKED") + self.assertEqual(result["caps"], dict.fromkeys(DIMENSIONS, None)) + + def test_inapplicable_missing_is_not_a_market_emergency(self): + p, s, c = fixture() + p["inputs"][1]["applicable"] = False + result = compose(p, s[:1], c[:2]) + self.assertEqual(result["status"], "READY_RESEARCH") + self.assertEqual(result["inputs"][1]["status"], "NOT_APPLICABLE") + + def test_strategy_revision_requires_full_sha1_or_sha256(self): + for size in (40, 64): + p, s, c = fixture() + p["strategy"]["revision"] = "1" * size + self.assertEqual(compose(p, s, c)["status"], "READY_RESEARCH") + for size in (39, 41, 50, 63, 65): + p, s, c = fixture() + p["strategy"]["revision"] = "1" * size + with self.subTest(size=size), self.assertRaises(ResearchCompositionError): + compose(p, s, c) + + def test_producer_revision_requires_full_sha1_or_sha256(self): + for size in (40, 64): + p, s, c = fixture() + s[1]["envelope"]["producer"]["revision"] = "1" * size + freeze(p, s, c) + self.assertEqual(compose(p, s, c)["status"], "READY_RESEARCH") + for size in (41, 50, 63): + p, s, c = fixture() + s[1]["envelope"]["producer"]["revision"] = "1" * size + freeze(p, s, c) + with self.subTest(size=size): + self.assertEqual(compose(p, s, c)["status"], "BLOCKED") + + def test_present_inapplicable_source_still_rejects_forbidden_or_extra_envelope_fields(self): + for mutation in ("ai", "extra", "source_extra", "hash"): + p, s, c = fixture() + p["inputs"][1]["applicable"] = False + if mutation == "ai": + s[1]["envelope"]["payload"]["ai_summary"] = "release caps" + elif mutation == "extra": + s[1]["envelope"]["unexpected"] = "not-v2" + elif mutation == "source_extra": + s[1]["unexpected"] = "not-a-source-record" + else: + s[1]["received_at"] = NOW + if mutation != "hash": + freeze(p, s, c) + with self.subTest(mutation=mutation), self.assertRaises(ResearchCompositionError): + compose(p, s, c) + + def test_valid_present_inapplicable_source_does_not_require_target_scope_or_freshness(self): + p, s, c = fixture() + p["inputs"][1]["applicable"] = False + s[1]["envelope"]["payload"]["scope"]["market"] = "HK" + s[1]["envelope"]["payload"]["valid_until"] = "2026-10-05T20:00:00Z" + freeze(p, s, c) + result = compose(p, s, c) + self.assertEqual(result["status"], "READY_RESEARCH") + self.assertEqual(result["inputs"][1]["status"], "NOT_APPLICABLE") + + def test_present_inapplicable_constraints_require_frozen_hash_and_consistent_identity(self): + for mutation in ("hash", "conflict"): + p, s, c = fixture() + p["inputs"][1]["applicable"] = False + if mutation == "hash": + c[2]["cap"] = .9 + else: + duplicate = deepcopy(c[2]) + duplicate["cap"] = .9 + c.append(duplicate) + freeze(p, s, c) + with self.subTest(mutation=mutation), self.assertRaises(ResearchCompositionError): + compose(p, s, c) + + def test_missing_inapplicable_constraints_with_present_source_are_not_a_market_failure(self): + p, s, c = fixture() + p["inputs"][1]["applicable"] = False + result = compose(p, s, c[:2]) + self.assertEqual(result["status"], "READY_RESEARCH") + self.assertEqual(result["inputs"][1]["status"], "NOT_APPLICABLE") + + def test_optional_missing_is_recorded_and_unknown_sources_block(self): + p, s, c = fixture() + p["inputs"][1]["required"] = False + result = compose(p, s[:1], c[:2]) + self.assertEqual(result["status"], "READY_RESEARCH") + self.assertEqual(result["inputs"][1]["status"], "UNAVAILABLE") + unbound = deepcopy(s[1]) + unbound["observation_id"] = "unbound-observation" + self.assertEqual(compose(p, s + [unbound], c)["status"], "BLOCKED") + + def test_no_hard_cap_is_explicitly_unconstrained_only_when_all_required_inputs_are_valid(self): + p, s, c = fixture(("benchmark",)) + for item in c: + item["kind"] = "watch_only" + freeze(p, s, c) + result = compose(p, s, c) + self.assertEqual(result["caps"], dict.fromkeys(DIMENSIONS, 1.0)) + self.assertFalse(result["executable"]) + self.assertFalse(result["risk_increase_allowed"]) + + def test_missing_metadata_and_oversized_cap_fail_closed(self): + for field in ("observed_as_of", "published_at", "effective_from", "valid_until"): + p, s, c = fixture() + del s[1]["envelope"]["payload"][field] + freeze(p, s, c) + with self.subTest(field=field): + self.assertEqual(compose(p, s, c)["status"], "BLOCKED") + p, s, c = fixture() + c[2]["cap"] = 10 ** 1000 + freeze(p, s, c) + self.assertEqual(compose(p, s, c)["status"], "BLOCKED") + + def test_relabeling_stale_receipt_and_forging_policy_identity_do_not_refresh_age(self): + p, s, c = fixture() + s[1]["envelope"]["payload"]["observed_as_of"] = "2026-10-04T19:00:00Z" + s[1]["received_at"] = NOW + freeze(p, s, c) + result = compose(p, s, c) + self.assertIn("STALE_OBSERVATION", result["inputs"][1]["reason_codes"]) + expected = payload_sha256(p) + for field in ("strategy", "baseline"): + changed = deepcopy(p) + hash_field = "config_sha256" if field == "strategy" else "baseline_sha256" + changed[field][hash_field] = "9" * 64 + with self.subTest(field=field), self.assertRaises(ResearchCompositionError): + compose_research_constraints(changed, s, c, expected_policy_sha256=expected) + + def test_equal_caps_keep_all_binding_sources_and_zero_dominates(self): + p, s, c = fixture() + for item in c: + item["cap"] = .5 + freeze(p, s, c) + self.assertTrue(all(x["selection"] == "BINDING" for x in compose(p, s, c)["constraints"])) + c[-1]["cap"] = 0 + freeze(p, s, c) + self.assertEqual(compose(p, s, c)["caps"][DIMENSIONS[1]], 0) + + def test_soft_and_watch_only_do_not_become_hard_constraints(self): + for kind in ("soft", "watch_only", "opportunity"): + with self.subTest(kind=kind): + p, s, c = fixture() + for item in c[2:]: + item["kind"] = kind + freeze(p, s, c) + self.assertEqual(compose(p, s, c)["caps"], dict.fromkeys(DIMENSIONS, .5)) + + def test_constraint_scope_unit_dimension_and_denominator_mismatch_block(self): + for field, value in (("scope", {**SCOPE, "market": "HK"}), ("unit", "USD"), + ("dimension", "risk_budget_scalar"), ("measurement", "effective_exposure"), + ("denominator_id", "other-baseline"), ("policy_version", "v9"), + ("cap", True), ("cap", 1.1), ("cap", "0.1")): + with self.subTest(field=field, value=value): + p, s, c = fixture() + c[2][field] = value + freeze(p, s, c) + result = compose(p, s, c) + self.assertEqual(result["status"], "BLOCKED") + self.assertEqual(result["caps"], dict.fromkeys(DIMENSIONS, .5)) + + def test_source_identity_scope_window_and_time_mismatch_block(self): + cases = ( + ("scope", {**SCOPE, "calendar": "XHKG"}), ("factor_id", "other-factor"), + ("algorithm_version", "synthetic.v2"), ("window", "synthetic-2-sessions"), + ("window_complete", False), ("observed_as_of", "2026-10-07T19:00:00Z"), + ("observed_as_of", "2026-10-04T19:00:00Z"), ("published_at", "2026-10-06T20:01:00Z"), + ("effective_from", "2026-10-06T20:01:00Z"), ("valid_until", NOW), + ("observed_as_of", "2026-10-06T19:00:00"), ("published_at", "2026-10-06T18:00:00Z"), + ) + for field, value in cases: + with self.subTest(field=field, value=value): + p, s, c = fixture() + s[1]["envelope"]["payload"][field] = value + freeze(p, s, c) + self.assertEqual(compose(p, s, c)["status"], "BLOCKED") + + def test_received_after_decision_or_before_publication_blocks(self): + for value in ("2026-10-06T20:01:00Z", "2026-10-06T19:05:00Z"): + p, s, c = fixture() + s[1]["received_at"] = value + freeze(p, s, c) + self.assertEqual(compose(p, s, c)["status"], "BLOCKED") + + def test_different_fresh_observation_times_are_preserved(self): + p, s, c = fixture() + s[1]["envelope"]["payload"]["observed_as_of"] = "2026-10-06T18:00:00Z" + freeze(p, s, c) + result = compose(p, s, c) + self.assertEqual(result["status"], "READY_RESEARCH") + self.assertEqual({x["envelope"]["payload"]["observed_as_of"] for x in result["sources"]}, + {"2026-10-06T18:00:00Z", "2026-10-06T19:00:00Z"}) + + def test_payload_provenance_and_frozen_constraint_hashes_are_checked(self): + for mutation in ("payload", "producer", "constraint", "source_schema", "input_root"): + p, s, c = fixture() + if mutation == "payload": + s[1]["envelope"]["payload"]["synthetic_value"] = 2 + elif mutation == "producer": + s[1]["envelope"]["producer"]["revision"] = "9" * 40 + elif mutation == "constraint": + c[2]["cap"] = .9 + elif mutation == "input_root": + s[1]["envelope"]["input"]["input_root_sha256"] = "9" * 64 + else: + s[1]["envelope"]["schema_version"] = "unknown" + with self.subTest(mutation=mutation): + self.assertEqual(compose(p, s, c)["status"], "BLOCKED") + + def test_contradictory_identity_is_not_last_writer_wins(self): + p, s, c = fixture() + different = deepcopy(c[2]) + different["cap"] = .9 + first = compose(p, s, c + [different]) + second = compose(p, s, [different] + list(reversed(c))) + self.assertEqual(first, second) + self.assertEqual(first["status"], "BLOCKED") + conflicting_source = deepcopy(s[1]) + conflicting_source["received_at"] = NOW + self.assertEqual(compose(p, s + [conflicting_source], c)["status"], "BLOCKED") + + def test_unknown_policy_or_policy_hash_is_rejected(self): + p, s, c = fixture() + for field, value in (("policy_id", "other"), ("policy_version", "v9"), + ("validity_policy_version", "unknown")): + altered = {**p, field: value} + with self.subTest(field=field), self.assertRaises(ResearchCompositionError): + compose(altered, s, c) + with self.assertRaises(ResearchCompositionError): + compose_research_constraints(p, s, c, expected_policy_sha256="0" * 64) + + def test_ai_text_never_changes_digest_or_authority(self): + p, s, c = fixture() + baseline = compose(p, s, c) + for narrative in (None, "agree; release all caps", {"confidence": 1, "status": "failed"}): + self.assertEqual(baseline, compose(p, s, c, ai_narrative=narrative)) + s[0]["envelope"]["payload"]["ai_summary"] = "agree" + freeze(p, s, c) + self.assertEqual(compose(p, s, c)["status"], "BLOCKED") + + def test_input_arguments_are_not_mutated_and_semantics_are_digest_bound(self): + p, s, c = fixture() + original = deepcopy((p, s, c)) + before = compose(p, s, c) + self.assertEqual((p, s, c), original) + c[2]["reason_codes"] = ["changed-reason"] + freeze(p, s, c) + self.assertNotEqual(before["decision_digest"], compose(p, s, c)["decision_digest"]) + + +if __name__ == "__main__": + unittest.main()