Skip to content

fix: harden M0 research ledger freshness #306

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
Aug 29, 2026
Merged

fix: harden M0 research ledger freshness #306

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 14 additions & 3 deletions docs/m0_research_ledger_contract.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,12 @@ errors: [安全错误码, ...]
- 不含账户、仓位、权重、订单、路由、平台、运行时、密钥或执行语义。

快照不得把失效研究线索重新标为新信号。`ready` 来源必须提供时间与来源
digest；`unavailable` 来源不得携带 hypothesis。
digest，且 `errors` 必须为空；`unavailable` 来源不得携带 hypothesis。来源、
subject、theme 与 hypothesis 标识采用与 `QuantAdvisorResearch` 完全相同的
字符集，不接受 `=`。M0 v5/v6 的 provenance 也是闭合配对：

- v5 必须为 `model_recommendations.v5` 且 `source_input_digest=null`；
- v6 必须为 `model_recommendations.v6` 且 `source_input_digest` 为 SHA-256。

## 聚合行为

Expand All @@ -43,10 +48,16 @@ digest；`unavailable` 来源不得携带 hypothesis。
3. 同一 subject + source digest 出现不同内容时，作为
`m0_source_subject_collision` 故障闭合剔除。
4. 同一 subject 的不同报告保留为独立观测；若 `primary_horizon` 不同，
标记 `horizon_conflict.status=conflict`。这是研究队列的人工/后续验证信号，
不是交易信号。
仅当**当前 fresh**观测不同才标记 `horizon_conflict.status=conflict`。
已失效观测被独立投影为 `historical_stale_horizon_drift`：只有存在当前
fresh 基准且历史 horizon 不同时才标记 `drift`；完全 stale 的 subject
只标记 `unavailable`，不会伪造当前冲突。这两者都是研究队列的人工/后续
验证信息，不是交易信号。
5. 依据 `now` 与 `expires_at` 产生 `fresh`、`stale` 或 `unknown`；来源自身
为 `stale` 时不会被提升为 fresh。
6. source 的 `generated_at` 或 `computed_at` 晚于聚合传入的 `now` 时，整个
source 以 `m0_source_future_timestamp` 故障闭合剔除；不会以 `unknown`
继续展示或参与去重。

输出台账始终固定：

Expand Down
77 changes: 66 additions & 11 deletions python/scripts/m0_research_ledger.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,10 @@
{"event_driven", "long_horizon_growth", "value_quality", "macro_context", "mixed_research"}
)
_SOURCE_STATUSES = frozenset({"ready", "unavailable", "stale"})
_IDENTIFIER = re.compile(r"^[A-Za-z0-9._:/=-]{1,128}$")
# Keep this byte-for-byte compatible with QuantAdvisorResearch's M0 contract.
# In particular, ``=`` is not a valid subject, theme, source, or hypothesis
# identifier there and must not be accepted by this downstream mirror.
_IDENTIFIER = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$")
_ERROR_CODE = re.compile(r"^[a-z][a-z0-9_.-]{0,63}$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_UTC_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$")
Expand Down Expand Up @@ -285,6 +288,8 @@ def validate_m0_research_source_snapshot(payload: object) -> dict[str, Any]:
source_report_digest is None or generated_at is None or computed_at is None
):
raise M0ResearchLedgerValidationError("ready_source_metadata_invalid")
if snapshot["data_status"] == "ready" and errors:
raise M0ResearchLedgerValidationError("ready_source_errors_invalid")
if snapshot["data_status"] == "unavailable" and hypotheses:
raise M0ResearchLedgerValidationError("unavailable_source_hypotheses_invalid")
if hypotheses and source_report_digest is None:
Expand Down Expand Up @@ -322,6 +327,52 @@ def _source_error(error_set: set[str], code: str) -> None:
error_set.add(code)


def _snapshot_time_is_future(snapshot: Mapping[str, Any], now: dt.datetime) -> bool:
"""Return whether source metadata could have been produced after this ledger.

A future source clock is not merely displayed as an ``unknown`` freshness
state. The entire source is omitted so a clock-skewed or replayed source
cannot become a current research input by accident.
"""

for field, label in (("generated_at", "source_generated_at"), ("computed_at", "source_computed_at")):
value = _parse_timestamp(snapshot[field], label, nullable=True)
if value is not None and value > now:
return True
return False


def _horizon_views(observations: Sequence[Mapping[str, Any]]) -> tuple[dict[str, Any], dict[str, Any]]:
"""Separate present disagreement from non-actionable historical drift."""

fresh_horizons = sorted(
{
observation["research_context"]["primary_horizon"]
for observation in observations
if observation["freshness"]["status"] == "fresh"
}
)
stale_horizons = sorted(
{
observation["research_context"]["primary_horizon"]
for observation in observations
if observation["freshness"]["status"] == "stale"
}
)
current = {
"status": "conflict" if len(fresh_horizons) > 1 else "none",
"primary_horizons": fresh_horizons,
}
if fresh_horizons:
stale_status = "drift" if stale_horizons and stale_horizons != fresh_horizons else "none"
else:
# A stale-only subject has no current primary horizon against which to
# call a drift. It remains visible for audit, but is not an alert.
stale_status = "unavailable" if stale_horizons else "none"
historical_stale = {"status": stale_status, "primary_horizons": stale_horizons}
return current, historical_stale


def aggregate_m0_research_sources(
snapshots: Sequence[object], *, now: str | dt.datetime
) -> dict[str, Any]:
Expand Down Expand Up @@ -352,6 +403,9 @@ def aggregate_m0_research_sources(
except M0ResearchLedgerValidationError:
_source_error(error_set, "m0_source_invalid")
continue
if _snapshot_time_is_future(snapshot, now_at):
_source_error(error_set, "m0_source_future_timestamp")
continue
if snapshot["data_status"] == "unavailable":
_source_error(error_set, "m0_source_unavailable")
continue
Expand Down Expand Up @@ -388,13 +442,15 @@ def aggregate_m0_research_sources(
fresh_count = 0
stale_count = 0
unknown_count = 0
conflict_count = 0
current_conflict_count = 0
historical_stale_drift_count = 0
for (kind, identifier), observations in sorted(subject_entries.items()):
observations.sort(key=lambda entry: (entry["source_report_digest"], entry["source_entry_digest"]))
horizons = sorted({entry["research_context"]["primary_horizon"] for entry in observations})
conflict = len(horizons) > 1
if conflict:
conflict_count += 1
horizon_conflict, historical_stale_horizon_drift = _horizon_views(observations)
if horizon_conflict["status"] == "conflict":
current_conflict_count += 1
if historical_stale_horizon_drift["status"] == "drift":
historical_stale_drift_count += 1
for observation in observations:
status = observation["freshness"]["status"]
if status == "fresh":
Expand All @@ -407,10 +463,8 @@ def aggregate_m0_research_sources(
{
"subject": {"kind": kind, "identifier": identifier},
"observations": observations,
"horizon_conflict": {
"status": "conflict" if conflict else "none",
"primary_horizons": horizons,
},
"horizon_conflict": horizon_conflict,
"historical_stale_horizon_drift": historical_stale_horizon_drift,
}
)

Expand All @@ -427,7 +481,8 @@ def aggregate_m0_research_sources(
"fresh_observation_count": fresh_count,
"stale_observation_count": stale_count,
"unknown_observation_count": unknown_count,
"horizon_conflict_count": conflict_count,
"horizon_conflict_count": current_conflict_count,
"historical_stale_horizon_drift_count": historical_stale_drift_count,
},
"subjects": subjects,
"policy": {
Expand Down
85 changes: 85 additions & 0 deletions python/tests/test_m0_research_ledger.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,17 @@ def test_source_and_ledger_schemas_remain_closed_and_read_only(self):
self.assertFalse(source_schema["additionalProperties"])
self.assertFalse(ledger_schema["additionalProperties"])
self.assertEqual(source_schema["properties"]["schema_version"]["const"], "qsl_m0_research_source_snapshot.v1")
self.assertEqual(source_schema["properties"]["source_id"]["pattern"], "^[A-Za-z0-9._:/-]{1,128}$")
self.assertEqual(source_schema["allOf"][0]["then"]["properties"]["errors"], {"maxItems": 0})
provenance_variants = source_schema["$defs"]["m0Hypothesis"]["properties"]["provenance"]["oneOf"]
self.assertEqual(len(provenance_variants), 2)
self.assertEqual(provenance_variants[0]["properties"]["source_schema_version"], {"const": "5"})
self.assertEqual(provenance_variants[0]["properties"]["source_input_digest"], {"type": "null"})
self.assertEqual(provenance_variants[1]["properties"]["source_schema_version"], {"const": "6"})
self.assertEqual(
provenance_variants[1]["properties"]["source_input_digest"],
{"type": "string", "pattern": "^[0-9a-f]{64}$"},
)
self.assertEqual(ledger_schema["properties"]["policy"]["properties"]["no_order"], {"const": True})
self.assertEqual(
ledger_schema["properties"]["policy"]["properties"]["permitted_next_step"],
Expand Down Expand Up @@ -126,9 +137,11 @@ def test_aggregation_deduplicates_subject_and_source_and_flags_horizon_conflict(
"stale_observation_count": 0,
"unknown_observation_count": 0,
"horizon_conflict_count": 1,
"historical_stale_horizon_drift_count": 0,
})
subject = ledger["subjects"][0]
self.assertEqual(subject["horizon_conflict"], {"status": "conflict", "primary_horizons": ["long", "medium"]})
self.assertEqual(subject["historical_stale_horizon_drift"], {"status": "none", "primary_horizons": []})
self.assertEqual(subject["observations"][0]["source_ids"], ["quant-advisor-research", "research-mirror"])

def test_expired_or_stale_source_is_visible_but_cannot_become_fresh(self):
Expand All @@ -143,11 +156,36 @@ def test_expired_or_stale_source_is_visible_but_cannot_become_fresh(self):
self.assertEqual(ledger["data_status"], "stale")
self.assertEqual(ledger["summary"]["fresh_observation_count"], 0)
self.assertEqual(ledger["summary"]["stale_observation_count"], 2)
self.assertEqual(ledger["summary"]["horizon_conflict_count"], 0)
self.assertEqual(ledger["summary"]["historical_stale_horizon_drift_count"], 0)
self.assertEqual(
{entry["freshness"]["status"] for item in ledger["subjects"] for entry in item["observations"]},
{"stale"},
)

def test_historical_stale_horizon_drift_does_not_create_a_current_conflict(self):
fresh = self._snapshot(self._hypothesis(primary_horizon="medium"), data_status="ready")
historical = self._snapshot(
self._hypothesis(
report_digest="d" * 64,
entry_digest="e" * 64,
hypothesis_id="m0r-long-history",
primary_horizon="long",
),
data_status="stale",
)
ledger = m0_research_ledger.aggregate_m0_research_sources(
[fresh, historical], now="2026-08-21T12:00:00Z"
)
subject = ledger["subjects"][0]
self.assertEqual(subject["horizon_conflict"], {"status": "none", "primary_horizons": ["medium"]})
self.assertEqual(
subject["historical_stale_horizon_drift"],
{"status": "drift", "primary_horizons": ["long"]},
)
self.assertEqual(ledger["summary"]["horizon_conflict_count"], 0)
self.assertEqual(ledger["summary"]["historical_stale_horizon_drift_count"], 1)

def test_m0_authority_execution_escape_and_source_digest_mismatch_fail_closed(self):
for mutate, message in (
(lambda value: value.update(authority="shadow_only"), "authority_invalid"),
Expand All @@ -165,6 +203,53 @@ def test_m0_authority_execution_escape_and_source_digest_mismatch_fail_closed(se
with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "source_report_digest_mismatch"):
m0_research_ledger.validate_m0_research_source_snapshot(snapshot)

def test_ready_source_errors_and_non_qar_identifier_fail_closed(self):
snapshot = self._snapshot(self._hypothesis())
snapshot["errors"] = ["upstream_timeout"]
with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "ready_source_errors_invalid"):
m0_research_ledger.validate_m0_research_source_snapshot(snapshot)

hypothesis = self._hypothesis()
hypothesis["subject"]["identifier"] = "SOXX=leveraged"
with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "subject_identifier_invalid"):
m0_research_ledger.validate_m0_research_hypothesis(hypothesis)

def test_v5_v6_provenance_pairing_matches_the_closed_schema(self):
v5 = self._hypothesis()
v5["provenance"].update(
source_schema_version="5",
source_contract_version="model_recommendations.v5",
source_input_digest=None,
)
m0_research_ledger.validate_m0_research_hypothesis(v5)

invalid_v6 = self._hypothesis()
invalid_v6["provenance"]["source_input_digest"] = None
with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "source_input_digest_invalid"):
m0_research_ledger.validate_m0_research_hypothesis(invalid_v6)

invalid_v5 = copy.deepcopy(v5)
invalid_v5["provenance"]["source_contract_version"] = "model_recommendations.v6"
with self.assertRaisesRegex(m0_research_ledger.M0ResearchLedgerValidationError, "source_contract_version_invalid"):
m0_research_ledger.validate_m0_research_hypothesis(invalid_v5)

def test_future_source_metadata_is_omitted_fail_closed(self):
for field, value in (
("generated_at", "2026-08-22T12:00:00Z"),
("computed_at", "2026-08-22T12:00:00Z"),
):
with self.subTest(field=field):
snapshot = self._snapshot(self._hypothesis())
snapshot[field] = value
if field == "generated_at":
snapshot["computed_at"] = "2026-08-22T12:01:00Z"
ledger = m0_research_ledger.aggregate_m0_research_sources(
[snapshot], now="2026-08-21T12:00:00Z"
)
self.assertEqual(ledger["data_status"], "unavailable")
self.assertEqual(ledger["summary"]["observation_count"], 0)
self.assertEqual(ledger["errors"], ["m0_source_future_timestamp"])

def test_same_subject_and_source_with_different_payloads_is_omitted_fail_closed(self):
first = self._hypothesis()
collision = self._hypothesis(entry_digest="f" * 64, hypothesis_id="m0r-collision")
Expand Down
32 changes: 24 additions & 8 deletions schemas/qsl-m0-research-ledger.v1.schema.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,14 +14,15 @@
"summary": {
"type": "object",
"additionalProperties": false,
"required": ["subject_count", "observation_count", "fresh_observation_count", "stale_observation_count", "unknown_observation_count", "horizon_conflict_count"],
"required": ["subject_count", "observation_count", "fresh_observation_count", "stale_observation_count", "unknown_observation_count", "horizon_conflict_count", "historical_stale_horizon_drift_count"],
"properties": {
"subject_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
"observation_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
"fresh_observation_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
"stale_observation_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
"unknown_observation_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
"horizon_conflict_count": { "type": "integer", "minimum": 0, "maximum": 50000 }
"horizon_conflict_count": { "type": "integer", "minimum": 0, "maximum": 50000 },
"historical_stale_horizon_drift_count": { "type": "integer", "minimum": 0, "maximum": 50000 }
}
},
"subjects": {
Expand All @@ -30,15 +31,15 @@
"items": {
"type": "object",
"additionalProperties": false,
"required": ["subject", "observations", "horizon_conflict"],
"required": ["subject", "observations", "horizon_conflict", "historical_stale_horizon_drift"],
"properties": {
"subject": {
"type": "object",
"additionalProperties": false,
"required": ["kind", "identifier"],
"properties": {
"kind": { "enum": ["asset_idea", "theme_context", "strategy_hypothesis", "risk_context"] },
"identifier": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" }
"identifier": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" }
}
},
"observations": {
Expand All @@ -55,7 +56,22 @@
"status": { "enum": ["none", "conflict"] },
"primary_horizons": {
"type": "array",
"minItems": 1,
"minItems": 0,
"maxItems": 4,
"uniqueItems": true,
"items": { "enum": ["short", "medium", "long", "not_applicable"] }
}
}
},
"historical_stale_horizon_drift": {
"type": "object",
"additionalProperties": false,
"required": ["status", "primary_horizons"],
"properties": {
"status": { "enum": ["none", "drift", "unavailable"] },
"primary_horizons": {
"type": "array",
"minItems": 0,
"maxItems": 4,
"uniqueItems": true,
"items": { "enum": ["short", "medium", "long", "not_applicable"] }
Expand Down Expand Up @@ -93,11 +109,11 @@
"minItems": 1,
"maxItems": 100,
"uniqueItems": true,
"items": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" }
"items": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" }
},
"source_report_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
"source_entry_digest": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
"hypothesis_id": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" },
"hypothesis_id": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" },
"as_of": { "type": "string", "format": "date" },
"generated_at": { "type": "string", "format": "date-time", "maxLength": 64 },
"expires_at": { "type": "string", "format": "date-time", "maxLength": 64 },
Expand All @@ -121,7 +137,7 @@
"type": "array",
"maxItems": 24,
"uniqueItems": true,
"items": { "type": "string", "pattern": "^[A-Za-z0-9._:/=-]{1,128}$" }
"items": { "type": "string", "pattern": "^[A-Za-z0-9._:/-]{1,128}$" }
}
}
},
Expand Down
Loading