Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 4 additions & 0 deletions docs/digest_candidates_producer.zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,10 @@ python3 -m pytest tests/test_project_digest_candidates.py tests/test_emit_digest

上述离线错槽位回归不解释真实报告 run `38339935851` 的四个 `runtime_target_mismatch`。该失败仍需各槽位的实际报告身份与受保护配置逐字段对照;guard 成功不能代替新报告和 producer receipt 的身份验收。

publisher 遇到既有身份失配时保留首行 `skipped:runtime_target_mismatch` 和退出码 1,并追加一行仅含固定 `reason` 与 `mismatched_fields` 的 JSON。字段只允许 `project_id`、`service_name`、`runtime_revision`、`account_scope`、`deployment_selector`、`account_selector` 和 `runtime_release_receipt.runtime_revision`;不会打印期望值、实际值、账户、service 映射或报告路径。异常类型也只记录既有比较的字段名;尚未通过报表结构校验的输入保留原原因码,不增加字段判断。

`mismatched_fields` 仅覆盖原拒绝顺序中第一失败检查段:先比较部署/账户身份,只有通过后才比较 receipt revision;更后的账户快照、时点等检查保持原顺序。该列表不是所有字段的完整差异证明。元数据只添加到身份失配的 `skipped` 结果,不进入成功的 account-facts payload,也不改变接受/拒绝决定;真实根因与恢复仍需对应私有事实及自然运行验收。

## 如何注入 QRS(人工,不自动改生产 Environment)

中央文档:QuantRuntimeSettings `docs/digest-candidates-wiring.zh-CN.md`(PR #586)。
Expand Down
71 changes: 62 additions & 9 deletions scripts/publish_account_facts_from_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,12 +92,39 @@
}
)
_HTTP_ERROR_BODY_LIMIT = 16 * 1024
_RUNTIME_TARGET_MISMATCH_FIELDS = (
"project_id",
"service_name",
"runtime_revision",
"account_scope",
"deployment_selector",
"account_selector",
"runtime_release_receipt.runtime_revision",
)


def _safe_mismatched_fields(value: object) -> list[str]:
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
return []
present = {
field for field in value
if isinstance(field, str) and field in _RUNTIME_TARGET_MISMATCH_FIELDS
}
return [field for field in _RUNTIME_TARGET_MISMATCH_FIELDS if field in present]


class _ProjectionError(ValueError):
def __init__(self, reason: str) -> None:
def __init__(self, reason: str, *, mismatched_fields: Sequence[str] = ()) -> None:
super().__init__(reason)
self.reason = reason
self.mismatched_fields = _safe_mismatched_fields(mismatched_fields)


def _projection_error_result(exc: _ProjectionError) -> dict[str, Any]:
result: dict[str, Any] = {"status": "skipped", "reason": exc.reason}
if exc.reason == "runtime_target_mismatch" and exc.mismatched_fields:
result["mismatched_fields"] = exc.mismatched_fields
return result


def _text(value: object) -> str:
Expand Down Expand Up @@ -498,7 +525,7 @@ def project_ibkr_account_facts_history(
expected_deployment_selector=expected_deployment_selector,
)
except _ProjectionError as exc:
return {"status": "skipped", "reason": exc.reason}
return _projection_error_result(exc)


def publish_ibkr_account_facts_history(
Expand Down Expand Up @@ -610,7 +637,7 @@ def publish_ibkr_account_facts_history(
return _publish_failed(stage="unknown", category="unknown")
return {"status": "published"}
except _ProjectionError as exc:
return {"status": "skipped", "reason": exc.reason}
return _projection_error_result(exc)


def _numeric_http_status(value: object) -> int | None:
Expand Down Expand Up @@ -782,6 +809,25 @@ def _format_cli_result(result: Mapping[str, Any]) -> str:
)


def _format_identity_mismatch_diagnostic(result: Mapping[str, Any]) -> str | None:
if result.get("status") != "skipped" or result.get("reason") != "runtime_target_mismatch":
return None
return json.dumps(
{
"reason": "runtime_target_mismatch",
"mismatched_fields": _safe_mismatched_fields(result.get("mismatched_fields")),
},
sort_keys=True,
)


def _print_cli_result(result: Mapping[str, Any]) -> None:
print(_format_cli_result(result))
diagnostic = _format_identity_mismatch_diagnostic(result)
if diagnostic is not None:
print(diagnostic)


class _NoRedirect(HTTPRedirectHandler):
def redirect_request(self, request: Request, *_args: object, **_kwargs: object) -> None:
return None
Expand Down Expand Up @@ -862,7 +908,7 @@ def main() -> int:
result = diagnose_account_facts_ingress(
sync_token=os.environ.get(IBKR_ACCOUNT_FACTS_SYNC_TOKEN_ENV, "")
)
print(_format_cli_result(result))
_print_cli_result(result)
return 0 if result.get("status") == "verified" else 1
if target in IBKR_ACCOUNT_FACTS_ADDITIONAL_TARGETS:
os.environ.update(additional_target_environment(target))
Expand Down Expand Up @@ -904,10 +950,10 @@ def main() -> int:
sync_token=os.environ.get(IBKR_ACCOUNT_FACTS_SYNC_TOKEN_ENV, ""),
**expected,
)
print(_format_cli_result(result))
_print_cli_result(result)
return 0 if result["status"] in {"published", "unchanged"} else 1
except _ProjectionError as exc:
print(f"skipped:{exc.reason}")
_print_cli_result(_projection_error_result(exc))
return 1
except Exception:
print("skipped:publisher_error")
Expand Down Expand Up @@ -959,11 +1005,18 @@ def _project_ibkr_account_facts_history(
"deployment_selector": _exact_text(target.get("deployment_selector")),
}
actual_selector = _selector(target.get("account_selector"))
if actual_fields != expected_fields or actual_selector != expected_selector:
raise _ProjectionError("runtime_target_mismatch")
mismatched_fields = [
field for field in actual_fields if actual_fields[field] != expected_fields[field]
]
if actual_selector != expected_selector:
mismatched_fields.append("account_selector")
if mismatched_fields:
raise _ProjectionError("runtime_target_mismatch", mismatched_fields=mismatched_fields)
receipt_revision = _text(receipt.get("runtime_revision")) if isinstance(receipt, Mapping) else ""
if receipt_revision and receipt_revision != actual_fields["runtime_revision"]:
raise _ProjectionError("runtime_target_mismatch")
raise _ProjectionError(
"runtime_target_mismatch", mismatched_fields=["runtime_release_receipt.runtime_revision"]
)

summary = report.get("summary")
facts = summary.get("account_facts") if isinstance(summary, Mapping) else None
Expand Down
165 changes: 163 additions & 2 deletions tests/test_publish_ibkr_account_facts.py
Original file line number Diff line number Diff line change
Expand Up @@ -294,7 +294,9 @@ def test_projection_rejects_invalid_observation_provenance_and_duplicate_cash():
def test_projection_requires_revision_and_preserves_null_assets():
missing_revision = _report()
missing_revision["diagnostics"]["runtime_revision"] = None # type: ignore[index]
assert _project(missing_revision) == {"status": "skipped", "reason": "runtime_target_mismatch"}
assert _project(missing_revision) == {
"status": "skipped", "reason": "runtime_target_mismatch", "mismatched_fields": ["runtime_revision"]
}

unavailable_assets = _report()
unavailable_assets["summary"]["account_facts"]["net_assets"] = None # type: ignore[index]
Expand Down Expand Up @@ -323,7 +325,9 @@ def test_projection_rejects_snapshot_observation_outside_report_interval():

padded_selector = _report()
padded_selector["runtime_target"]["account_selector"] = [" U00000001"] # type: ignore[index]
assert _project(padded_selector) == {"status": "skipped", "reason": "runtime_target_mismatch"}
assert _project(padded_selector) == {
"status": "skipped", "reason": "runtime_target_mismatch", "mismatched_fields": ["account_selector"]
}


def test_publisher_posts_once_with_fresh_projected_payload_and_dedicated_token(monkeypatch):
Expand Down Expand Up @@ -1393,3 +1397,160 @@ def test_empty_named_report_preserves_latest_report_selection(monkeypatch, capsy
assert latest == [True]
assert loaded == ["gs://example-private/reports/ibkr/2026-10/20261001T120000Z.json"]
assert capsys.readouterr().out.strip() == "published:unknown"


_IDENTITY_DIAGNOSTIC_CASES = [
(("project_id",), "project_id"),
(("service_name",), "service_name"),
(("diagnostics", "runtime_revision"), "runtime_revision"),
(("account_scope",), "account_scope"),
(("runtime_target", "deployment_selector"), "deployment_selector"),
(("runtime_target", "account_selector"), "account_selector"),
(("runtime_release_receipt", "runtime_revision"), "runtime_release_receipt.runtime_revision"),
]


def _mutate_identity_field(report, path, value):
container = report
for key in path[:-1]:
container = container[key]
container[path[-1]] = value


@pytest.mark.parametrize("path,field", _IDENTITY_DIAGNOSTIC_CASES)
def test_identity_mismatch_reports_only_fixed_field_name(path, field):
report = _report()
marker = "PRIVATE_ACTUAL_IDENTITY_MARKER"
value = [marker] if field == "account_selector" else marker
_mutate_identity_field(report, path, value)

result = _project(report)

assert result == {
"status": "skipped", "reason": "runtime_target_mismatch", "mismatched_fields": [field]
}
assert marker not in json.dumps(result)


@pytest.mark.parametrize("path,field", _IDENTITY_DIAGNOSTIC_CASES[:-1])
@pytest.mark.parametrize("value", [None, {}, ["PRIVATE_TYPE_MARKER"], True, " PRIVATE_TYPE_MARKER"])
def test_identity_mismatch_abnormal_values_keep_original_rejection(path, field, value):
report = _report()
_mutate_identity_field(report, path, value)

result = _project(report)

assert result == {
"status": "skipped", "reason": "runtime_target_mismatch", "mismatched_fields": [field]
}
assert "PRIVATE_TYPE_MARKER" not in json.dumps(result)


def test_identity_diagnostic_stops_at_original_first_failed_check_group():
report = _report()
for path, field in _IDENTITY_DIAGNOSTIC_CASES:
value = ["PRIVATE_MULTI_MARKER"] if field == "account_selector" else "PRIVATE_MULTI_MARKER"
_mutate_identity_field(report, path, value)
report["summary"] = "PRIVATE_LATER_SUMMARY_MARKER"

result = _project(report)

assert result == {
"status": "skipped", "reason": "runtime_target_mismatch",
"mismatched_fields": [field for _, field in _IDENTITY_DIAGNOSTIC_CASES[:-1]],
}
assert "PRIVATE" not in json.dumps(result)


def test_identity_diagnostic_does_not_read_receipt_after_earlier_mismatch():
class UnreadableReceipt(dict):
def get(self, *_args):
pytest.fail("receipt comparison must remain after first identity rejection")

report = _report()
report["project_id"] = "PRIVATE_FIRST_GROUP_MARKER"
report["runtime_release_receipt"] = UnreadableReceipt()
assert _project(report) == {
"status": "skipped", "reason": "runtime_target_mismatch",
"mismatched_fields": ["project_id"],
}


@pytest.mark.parametrize("field,value", [("runtime_target", None), ("diagnostics", []), ("runtime_release_receipt", "PRIVATE_STRUCTURE_MARKER")])
def test_unvalidated_report_keeps_original_reason_without_identity_metadata(field, value):
report = _report()
report[field] = value
report["project_id"] = "PRIVATE_PROJECT_MARKER"
assert _project(report) == {"status": "skipped", "reason": "runtime_report_invalid"}


@pytest.mark.parametrize(
"fields,expected",
[
(None, []),
({"project_id": "PRIVATE_VALUE_MARKER"}, []),
("project_id PRIVATE_FIELD_MARKER", []),
(["account_selector", "PRIVATE_FIELD_MARKER", {}, "project_id", "project_id"], ["project_id", "account_selector"]),
([None, True, 3], []),
],
)
def test_cli_identity_diagnostic_filters_untrusted_metadata(fields, expected):
result = {
"status": "skipped", "reason": "runtime_target_mismatch", "mismatched_fields": fields,
"expected": "PRIVATE_EXPECTED_MARKER", "actual": "PRIVATE_ACTUAL_MARKER",
"source_report_uri": "gs://private-uri-marker/path",
}
output = publisher._format_identity_mismatch_diagnostic(result)
assert json.loads(output) == {
"reason": "runtime_target_mismatch", "mismatched_fields": expected
}
assert "PRIVATE" not in output and "private-uri-marker" not in output


@pytest.mark.parametrize("status,reason", [("published", "runtime_target_mismatch"), ("skipped", "PRIVATE_REASON_MARKER")])
def test_cli_identity_diagnostic_does_not_emit_for_other_outcomes(status, reason):
assert publisher._format_identity_mismatch_diagnostic({
"status": status, "reason": reason, "mismatched_fields": ["project_id"]
}) is None


def test_cli_identity_mismatch_keeps_first_line_exit_and_hides_private_values(monkeypatch, capsys):
now = datetime(2026, 9, 30, 1, 1, tzinfo=timezone.utc)

class FrozenDateTime(datetime):
@classmethod
def now(cls, tz=None):
return now

environment = {
"IBKR_ACCOUNT_FACTS_TARGET": "primary-live",
"IBKR_ACCOUNT_FACTS_REPORT_PREFIX": "gs://private-prefix-marker/reports",
"IBKR_ACCOUNT_FACTS_TARGET_ID": "private-target-marker",
"IBKR_ACCOUNT_FACTS_PROJECT_ID": "PRIVATE_EXPECTED_PROJECT_MARKER",
"IBKR_ACCOUNT_FACTS_SERVICE_NAME": "PRIVATE_EXPECTED_SERVICE_MARKER",
"IBKR_ACCOUNT_FACTS_RUNTIME_REVISION": "PRIVATE_EXPECTED_REVISION_MARKER",
"IBKR_ACCOUNT_FACTS_ACCOUNT_SCOPE": "PRIVATE_EXPECTED_SCOPE_MARKER",
"IBKR_ACCOUNT_FACTS_ACCOUNT_SELECTOR_JSON": '["U00000999"]',
"IBKR_ACCOUNT_FACTS_DEPLOYMENT_SELECTOR": "PRIVATE_EXPECTED_DEPLOYMENT_MARKER",
"IBKR_ACCOUNT_FACTS_SYNC_URL": publisher.IBKR_ACCOUNT_FACTS_SYNC_URL,
"IBKR_ACCOUNT_FACTS_SYNC_TOKEN": "PRIVATE_TOKEN_MARKER",
}
for key, value in environment.items():
monkeypatch.setenv(key, value)
monkeypatch.delenv("IBKR_ACCOUNT_FACTS_REPORT_NAME", raising=False)
monkeypatch.setattr(publisher, "datetime", FrozenDateTime)
monkeypatch.setattr(publisher, "_latest_report_uri", lambda **_kwargs: "gs://private-prefix-marker/reports/2026-09/20260930T010000Z.json")
monkeypatch.setattr(publisher, "_load_gcs_report", lambda *_args, **_kwargs: _report())
monkeypatch.setattr(publisher, "build_opener", lambda *_args: pytest.fail("mismatch must stop before POST"))

assert publisher.main() == 1
output = capsys.readouterr().out
lines = output.splitlines()
assert lines[0] == "skipped:runtime_target_mismatch"
assert len(lines) == 2
assert json.loads(lines[1]) == {
"reason": "runtime_target_mismatch",
"mismatched_fields": [field for _, field in _IDENTITY_DIAGNOSTIC_CASES[:-1]],
}
for marker in ["PRIVATE", "private-prefix-marker", "private-target-marker", "U00000001", "U00000999", "ibkr-primary-service"]:
assert marker not in output
Loading