From 78a364ea9069d0c34c61198e42552a181da62beb Mon Sep 17 00:00:00 2001 From: Li SiYi Date: Tue, 13 Oct 2026 12:32:56 +0000 Subject: [PATCH] feat: add opt-in unified net metrics to historical reports --- .github/workflows/ci.yml | 53 ++++ .../c3_self_financing_combo_ledger.zh-CN.md | 39 +++ ...run_c3_batch_a_existing_member_baseline.py | 24 +- .../c3_batch_a_existing_member_baseline.py | 11 +- .../research/tqqq_core_trial_journal.py | 48 +++- ...est_c3_batch_a_existing_member_baseline.py | 255 ++++++++++++++++++ 6 files changed, 425 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 09ec6c27..68adfd73 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -144,3 +144,56 @@ jobs: skipped = sum(int(suite.attrib.get("skipped", 0)) for suite in suites) assert skipped == 0, f"required execution tests skipped: {skipped}" PY + + research-unified-net-consumer: + # Explicit CLI/report acceptance only. Existing production pins and both + # other research overlay environments keep their prior identities. + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + QPK_UNIFIED_NET_REPORT_SHA: b23511cbb28b72743e35512ef4b3c73f1cd43c23 + UES_REQUIRE_UNIFIED_NET_REPORT: "1" + steps: + - name: Checkout + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0 + + - name: Setup Python + uses: actions/setup-python@v6 + with: + python-version: "3.11" + + - name: Install isolated latest report overlay + run: | + set -euo pipefail + python -m pip install --upgrade pip + python -m pip install -e . pytest pandas_market_calendars + python -m pip install --force-reinstall \ + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@${QPK_UNIFIED_NET_REPORT_SHA}" + python -m pip check + python - <<'PY' + import inspect, json, os + from importlib.metadata import distribution + from quant_platform_kit.research_stats.unified_report import build_unified_net_report + installed = json.loads(distribution("quant-platform-kit").read_text("direct_url.json")) + assert installed["vcs_info"]["commit_id"] == os.environ["QPK_UNIFIED_NET_REPORT_SHA"] + assert "include_outcome_diagnostics" in inspect.signature(build_unified_net_report).parameters + report = build_unified_net_report([-.1, 0.0, .2], include_outcome_diagnostics=True) + assert len(report.metrics) == 12 and report.no_order is True + print("Required latest unified/outcome report capability and immutable source verified.") + PY + + - name: Run required actual CLI and report acceptance + run: | + set -euo pipefail + PYTHONPATH=src python -m pytest -q -rs \ + tests/test_c3_batch_a_existing_member_baseline.py \ + --junitxml=unified-net-consumer-tests.xml + python - <<'PY' + import xml.etree.ElementTree as ET + root = ET.parse("unified-net-consumer-tests.xml").getroot() + cases = list(root.iter("testcase")) + assert cases and any("unified_cli_uses_actual_selected_net_returns" in c.attrib["name"] for c in cases) + assert any("unified_hand_math" in c.attrib["name"] for c in cases) + assert not any(list(root.iter(tag)) for tag in ("skipped", "failure", "error")) + print("Actual CLI chain and positive hand-math tests executed with no skipped/failed cases.") + PY diff --git a/docs/research/c3_self_financing_combo_ledger.zh-CN.md b/docs/research/c3_self_financing_combo_ledger.zh-CN.md index a1becc5d..184e54be 100644 --- a/docs/research/c3_self_financing_combo_ledger.zh-CN.md +++ b/docs/research/c3_self_financing_combo_ledger.zh-CN.md @@ -128,3 +128,42 @@ MAR shortfall RMS 的 Sortino。rf 和 MAR 分别是显式年简单小数，除 `NOT_ESTIMATED`：这些单路径和固定预算比较没有完整 trial 集合及真实依赖/ 多重试验假设。报告始终 research/synthetic、report-only、非晋级、不可实盘 或配仓。真实净成本 OOS、联合风险资格与成员资格不由合成测试授予。 + +## Batch A 历史入口的可选 B3 诊断 + +已有 `scripts/run_c3_batch_a_existing_member_baseline.py --historical-account-report` +可显式增加 `--include-unified-net-report`。对应 Python 入口 +`report_historical_batch_a_baselines` 和 `compute_historical_research_metrics` +的参数为 `include_unified_net_report=False`；未请求时保留原字段、算术和 +行为，不导入或要求新 B3 能力。Batch A cloud workflow 默认不增加此 flag， +生产 QPK pin、既有两个研究 overlay、冻结候选/报告/账本和排名保持。 + +显式请求需要 QPK `b23511cbb28b72743e35512ef4b3c73f1cd43c23` 或实际具有 +相同 B3/outcome 合同的研究安装。旧 pin 缺该能力时返回明确的 +`UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE`，不会静默用八项代替十二项。 +CLI 的模式/能力错误在读取私有输入前拒绝；输入绑定、来源重算与整个 +历史报告的已有原子拒绝边界仍保留。新隔离 CI job 核验实际 immutable +安装来源、强制能力，并执行同一 CLI/report 链和数值验收，禁止跳过。 + +每本已有账户报告附独立 `unified_net_report` 及 `unified_net_report_basis`。 +它使用该本已验证 ledger 的精确同窗口净简单收益，不额外扣费，不补零、 +删观察或增加初始 NAV 收益。B3 的 Sharpe 是同频未年化 raw-net、ddof=1； +旧扣 rf 的年化 Sharpe/MAR Sortino 原字段不替换。四项 outcome 为 +`net_win_rate_ex_zero_per_period`、`net_payoff_ratio_per_period`、 +`net_profit_factor_per_period`、`mean_net_return_per_period`：周期剔零胜率、 +平均正负收益比、正负收益总额比及含零观察的平均净收益。它们不表示 +逐笔闭合交易成绩、未来收益或新排名/否决条件。 + +输入没有同路径 gross counterfactual 或完整累计 trial 证据，故成本拖累与 +DSR 分别保留 `UNCOMPUTABLE/missing_gross_returns` 和 +`UNCOMPUTABLE/missing_trial_evidence`、value=null；不反加费用、不拿其他 +成本场景冒充 gross，也不拿旧 ledger 声明48当完整全集。无资产基准时 +比较/对齐计数为 absent/null。报告估计器的既有默认样本数、概率参数不 +构成晋级阈值；无定义与实际零严格区分。仅输出聚合报告，不新增原始 +价格、逐日收益、持仓路径或私有位置。 + +输入仍为 source-seen development、data/oos unqualified，不证明 PIT、许可、 +真实收费、完整插件或实盘采用；原研究/执行拒绝状态保持。修改代码会 +更新报告已有的真实 `implementation_sha256` 及其派生 `evidence_digest`， +不得硬编码旧身份以假称完整字节不变。未开启诊断时除此必要来源身份 +之外，历史报告数值/字段/行为保持；旧 RS artifact 和 trial ledger 不重写。 diff --git a/scripts/run_c3_batch_a_existing_member_baseline.py b/scripts/run_c3_batch_a_existing_member_baseline.py index 4c68ef16..02881629 100644 --- a/scripts/run_c3_batch_a_existing_member_baseline.py +++ b/scripts/run_c3_batch_a_existing_member_baseline.py @@ -121,6 +121,8 @@ def main(argv: Sequence[str] | None = None) -> int: help="Original cloud staging root; historical report revalidates snapshots and the pack.") parser.add_argument("--annual-risk-free-rate", type=float, default=None) parser.add_argument("--annual-minimum-acceptable-return", type=float, default=None) + parser.add_argument("--include-unified-net-report", action="store_true", + help="Explicit research-only B3 sidecar with per-session outcome diagnostics; requires historical-account-report and latest QPK.") parser.add_argument( "--private-root", type=Path, @@ -128,6 +130,25 @@ def main(argv: Sequence[str] | None = None) -> int: help=argparse.SUPPRESS, # legacy flag retained only to emit a clear park reason ) args = parser.parse_args(argv) + if args.include_unified_net_report: + if not args.historical_account_report: + reason = "HISTORICAL_REPORT_OPT_IN_REQUIRED" + else: + from us_equity_strategies.research.tqqq_core_trial_journal import _unified_net_report_builder + try: + _unified_net_report_builder() + except ValueError: + reason = "UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE" + else: + reason = None + if reason is not None: + # Refuse mode/capability errors before loading private inputs. + result = {"status": "PARKED", "reason_codes": [reason], + "research_only": True, "execution_authorized": False, + "promotion_authorized": False, "no_order": True, + "data_qualified": False, "oos_qualified": False, "batch_a_accepted": False} + print(json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False)) + return 2 if args.private_root is not None and args.frozen_member_pack is None: result = { "schema_version": "qsl.c3-batch-a-existing-member-baseline-research.v1", @@ -168,7 +189,8 @@ def main(argv: Sequence[str] | None = None) -> int: args.snapshot_input_root / "tqqq_qqq", identity_path=args.snapshot_input_root / "tqqq_qqq" / "object_identity.json"), annual_risk_free_rate=args.annual_risk_free_rate, - annual_minimum_acceptable_return=args.annual_minimum_acceptable_return) + annual_minimum_acceptable_return=args.annual_minimum_acceptable_return, + include_unified_net_report=args.include_unified_net_report) except (ValueError, TypeError, ArithmeticError): # Never echo private paths, provider bodies or an arbitrary exception. result = {"status": "PARKED", "reason_codes": ["HISTORICAL_ACCOUNT_REPORT_REFUSED"], diff --git a/src/us_equity_strategies/research/c3_batch_a_existing_member_baseline.py b/src/us_equity_strategies/research/c3_batch_a_existing_member_baseline.py index cab26dfe..1628950b 100644 --- a/src/us_equity_strategies/research/c3_batch_a_existing_member_baseline.py +++ b/src/us_equity_strategies/research/c3_batch_a_existing_member_baseline.py @@ -570,6 +570,7 @@ def evaluate_batch_a_existing_member_baselines( def report_historical_batch_a_baselines( *, frozen_member_pack: Mapping[str, object], soxl_snapshot, tqqq_snapshot, annual_risk_free_rate: float, annual_minimum_acceptable_return: float, + include_unified_net_report: bool = False, ) -> dict: """Opt-in source-bound historical control, retaining the old comparison. @@ -582,7 +583,12 @@ def report_historical_batch_a_baselines( from .batch_a_member_pack import materialize_batch_a_member_pack from .c3_capital_path import scale_member_budgets_to_cash from .c3_self_financing_combo_ledger import build_historical_self_financing_ledger - from .tqqq_core_trial_journal import compute_historical_research_metrics + from .tqqq_core_trial_journal import compute_historical_research_metrics, _unified_net_report_builder + + if type(include_unified_net_report) is not bool: + raise ValueError("UNIFIED_NET_REPORT_OPT_IN_INVALID") + if include_unified_net_report: + _unified_net_report_builder() if not all(isinstance(item, PriceSnapshotV2) for item in (soxl_snapshot, tqqq_snapshot)): raise ValueError("HISTORICAL_SNAPSHOTS_REQUIRED") @@ -638,7 +644,8 @@ def book_report(identity, weights, rebalances, fee): "terminal_nav": ledger.days[-1].nav, "terminal_cash": ledger.days[-1].cash, "metric_report": compute_historical_research_metrics( ledger, input_binding=source_binding, annual_risk_free_rate=annual_risk_free_rate, - annual_minimum_acceptable_return=annual_minimum_acceptable_return)} + annual_minimum_acceptable_return=annual_minimum_acceptable_return, + include_unified_net_report=include_unified_net_report)} comparisons = [] by_id = {row["baseline_id"]: row for row in legacy["c3_comparison"]["baselines"]} diff --git a/src/us_equity_strategies/research/tqqq_core_trial_journal.py b/src/us_equity_strategies/research/tqqq_core_trial_journal.py index 3622cf63..dab447b0 100644 --- a/src/us_equity_strategies/research/tqqq_core_trial_journal.py +++ b/src/us_equity_strategies/research/tqqq_core_trial_journal.py @@ -9,6 +9,8 @@ from collections.abc import Mapping from datetime import date, datetime, timezone import hashlib +import importlib +import inspect import json import math from pathlib import Path @@ -98,6 +100,7 @@ def compute_historical_research_metrics( ledger: ResearchDailyLedger, *, input_binding: Mapping, annual_risk_free_rate: float, annual_minimum_acceptable_return: float, start_session: date | None = None, end_session: date | None = None, + include_unified_net_report: bool = False, ) -> dict: """Report historical modeled sleeves; binding is not PIT/OOS qualification. @@ -117,7 +120,8 @@ def compute_historical_research_metrics( ledger, calendar_id="XNYS", contract_version="historical_research_account_metrics_v1", annual_risk_free_rate=annual_risk_free_rate, annual_minimum_acceptable_return=annual_minimum_acceptable_return, - start_session=start_session, end_session=end_session) + start_session=start_session, end_session=end_session, + include_unified_net_report=include_unified_net_report) report.update({"data_qualified": False, "oos_qualified": False, "evidence_scope": "HISTORICAL_SOURCE_SEEN_DEVELOPMENT_MODELED_SLEEVES", "input_binding": json.loads(json.dumps(input_binding, allow_nan=False))}) @@ -131,12 +135,31 @@ def compute_historical_research_metrics( return report +def _unified_net_report_builder(): + """Require the optional latest research capability without eager imports.""" + name = "quant_platform_kit.research_stats.unified_report" + try: + module = importlib.import_module(name) + except ModuleNotFoundError as exc: + if exc.name not in {"quant_platform_kit.research_stats", name}: + raise + raise ValueError("UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE") from None + builder = getattr(module, "build_unified_net_report", None) + if not callable(builder) or "include_outcome_diagnostics" not in inspect.signature(builder).parameters: + raise ValueError("UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE") + return builder + + def _compute_account_metrics( ledger: ResearchDailyLedger, *, calendar_id: str, contract_version: str, annual_risk_free_rate: float, annual_minimum_acceptable_return: float, start_session: date | None, end_session: date | None, + include_unified_net_report: bool = False, ) -> dict: """One arithmetic body shared by the explicit report-only input contracts.""" + if type(include_unified_net_report) is not bool: + raise ValueError("UNIFIED_NET_REPORT_OPT_IN_INVALID") + unified_builder = _unified_net_report_builder() if include_unified_net_report else None if ledger.calendar_id != calendar_id or ledger.periods_per_year != 252: raise ValueError("RESEARCH_BASIS_INVALID") rates = (annual_risk_free_rate, annual_minimum_acceptable_return) @@ -178,7 +201,7 @@ def _compute_account_metrics( "ZERO_DOWNSIDE_DEVIATION" if all( r >= annual_minimum_acceptable_return / 252 for r in returns) else "NONFINITE_RESULT") - return { + report = { "research_only": True, "synthetic": ledger.synthetic, "promotion_eligible": False, "live_ready": False, "size_zero_required": True, "no_order": True, "trial_id": ledger.trial_id, "run_id": ledger.run_id, @@ -222,6 +245,27 @@ def _compute_account_metrics( "trial_matrix_and_dependence_assumptions": "NOT_ESTABLISHED", }, } + if unified_builder is not None: + # The existing ledger has already deducted costs. It has neither a + # matching gross counterfactual nor a complete cumulative trial ledger. + report["unified_net_report"] = unified_builder( + returns, include_outcome_diagnostics=True).to_dict() + report["unified_net_report_basis"] = { + "metric_role": "REPORT_ONLY", "return_basis": "RAW_NET", + "observation_unit": "STRATEGY_ACCOUNT_SESSION_SIMPLE_NET_RETURN", + "return_unit": "SIGNED_DECIMAL_FRACTION", + "frequency": "PER_SESSION_NOT_ANNUALIZED", "volatility_ddof": 1, + "calendar_id": calendar_id, "real_calendar_verified": False, + "start_session": start.isoformat(), "end_session": end.isoformat(), + "zero_return_sessions": "INCLUDED", "initial_nav_is_return_observation": False, + "costs_deducted_again": False, "gross_counterfactual": "NOT_SUPPLIED", + "cumulative_trial_evidence": "NOT_SUPPLIED", + "asset_benchmark": None, "benchmark_alignment_count": None, + "benchmark_dropped_count": None, + "legacy_rf_adjusted_annualized_metrics_replaced": False, + "inference_parameters_role": "ESTIMATOR_DEFAULTS_NOT_ACCEPTANCE_THRESHOLDS", + } + return report def report_journaled_tqqq_core_metrics( diff --git a/tests/test_c3_batch_a_existing_member_baseline.py b/tests/test_c3_batch_a_existing_member_baseline.py index 2aefb81b..1b45e22b 100644 --- a/tests/test_c3_batch_a_existing_member_baseline.py +++ b/tests/test_c3_batch_a_existing_member_baseline.py @@ -2,6 +2,10 @@ import json import math +import importlib +import inspect +import os +import socket from pathlib import Path import pytest @@ -20,6 +24,27 @@ import scripts.run_c3_batch_a_existing_member_baseline as batch_a_cli +@pytest.fixture(autouse=True) +def no_consumer_network(monkeypatch): + attempts = [] + + def denied(*args, **kwargs): + attempts.append(True) + raise AssertionError("LOCAL_CONSUMER_FIXTURE_NETWORK_FORBIDDEN") + + class DeniedSocket(socket.socket): + def __new__(cls, *args, **kwargs): + return denied(*args, **kwargs) + + # Preserve the socket type for stdlib SSL class declarations while refusing + # any actual allocation; replacing it with a function breaks lazy imports. + monkeypatch.setattr(socket, "socket", DeniedSocket) + for name in ("create_connection", "getaddrinfo", "gethostbyname"): + monkeypatch.setattr(socket, name, denied) + yield + assert not attempts, "Report consumer attempted network access" + + def _comp(**overrides: object) -> dict[str, object]: payload = { "as_of": "2026-09-21", @@ -653,3 +678,233 @@ def test_historical_cli_requires_opt_in_and_complete_inputs(historical_sources, assert json.loads(capsys.readouterr().out)["reason_codes"] == ["HISTORICAL_REPORT_OPT_IN_REQUIRED"] assert batch_a_cli.main(["--historical-account-report"]) == 2 assert json.loads(capsys.readouterr().out)["reason_codes"] == ["HISTORICAL_ACCOUNT_REPORT_REFUSED"] + + +def _unified_capability(): + try: + module = importlib.import_module("quant_platform_kit.research_stats.unified_report") + except ModuleNotFoundError as exc: + if exc.name not in {"quant_platform_kit.research_stats", "quant_platform_kit.research_stats.unified_report"}: + raise + builder = None + else: + builder = getattr(module, "build_unified_net_report", None) + if not callable(builder) or "include_outcome_diagnostics" not in inspect.signature(builder).parameters: + builder = None + if os.environ.get("UES_REQUIRE_UNIFIED_NET_REPORT") == "1": + assert builder is not None, "Required latest QPK unified/outcome capability unavailable" + return builder + + +def _unified_cli(sources, capsys): + code = batch_a_cli.main([*_historical_cli_args(sources), "--include-unified-net-report"]) + text = capsys.readouterr().out + return code, text, json.loads(text) + + +def test_unified_flag_requires_historical_mode_before_input_read(tmp_path, capsys, monkeypatch): + def denied(*args, **kwargs): + pytest.fail("wrong opt-in mode read private input") + monkeypatch.setattr(Path, "read_text", denied) + assert batch_a_cli.main(["--include-unified-net-report", "--frozen-member-pack", + str(tmp_path / "private-missing.json")]) == 2 + output = capsys.readouterr().out + assert json.loads(output)["reason_codes"] == ["HISTORICAL_REPORT_OPT_IN_REQUIRED"] + assert str(tmp_path) not in output + + +def test_unified_capability_refused_before_private_input_read(tmp_path, capsys, monkeypatch): + from us_equity_strategies.research import tqqq_core_trial_journal as journal + original = importlib.import_module + + def missing(name, *args, **kwargs): + if name == "quant_platform_kit.research_stats.unified_report": + raise ModuleNotFoundError("private-marker", name=name) + return original(name, *args, **kwargs) + + monkeypatch.setattr(journal.importlib, "import_module", missing) + monkeypatch.setattr(Path, "read_text", lambda *a, **k: pytest.fail("missing capability read input")) + assert batch_a_cli.main(["--historical-account-report", "--include-unified-net-report", + "--frozen-member-pack", str(tmp_path / "private-missing.json")]) == 2 + output = capsys.readouterr().out + assert json.loads(output)["reason_codes"] == ["UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE"] + assert "private-marker" not in output and str(tmp_path) not in output + + +def test_unified_cli_uses_actual_selected_net_returns_and_preserves_legacy(historical_sources, capsys, monkeypatch): + from us_equity_strategies.research import tqqq_core_trial_journal as journal + builder = _unified_capability() + baseline = json.loads(json.dumps(_historical_result(historical_sources))) + before = {p: p.read_bytes() for p in historical_sources[0].rglob("*") if p.is_file()} + if builder is None: + code, _, refused = _unified_cli(historical_sources, capsys) + assert code == 2 and refused["reason_codes"] == ["UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE"] + assert "historical_account_report" not in refused + return + captured = [] + + def actual(net, **kwargs): + captured.append((tuple(net), kwargs)) + return builder(net, **kwargs) + + monkeypatch.setattr(journal, "_unified_net_report_builder", lambda: actual) + code, output, current = _unified_cli(historical_sources, capsys) + assert code == 0 + report = current["historical_account_report"] + metric_reports = [book["metric_report"] for row in report["baselines"] for book in row["books"].values()] + metric_reports += [book["metric_report"] for book in report["standalone"].values()] + # CLI JSON sorts object keys, while source construction order is preserved in captured input. + sidecars = sorted((item["unified_net_report"] for item in metric_reports), + key=lambda item: (item["metrics"][-1]["value"], item["observations"])) + assert len(captured) == len(sidecars) == 31 + for net, kwargs in captured: + assert kwargs == {"include_outcome_diagnostics": True} + assert len(net) == metric_reports[0]["observation_count"] + expected_means = sorted(math.fsum(net) / len(net) for net, _ in captured) + assert [item["metrics"][-1]["value"] for item in sidecars] == pytest.approx(expected_means) + for item in metric_reports: + sidecar = item.pop("unified_net_report") + basis = item.pop("unified_net_report_basis") + assert sidecar["status"] == "DEVELOPMENT_ONLY" + assert sidecar["volatility_ddof"] == 1 + assert sidecar["live_ready"] is False and sidecar["no_order"] is True + assert sidecar["observations"] == item["observation_count"] + assert len(sidecar["metrics"]) == 12 + values = {m["name"]: m for m in sidecar["metrics"]} + assert values["deflated_sharpe_ratio"]["reason_code"] == "missing_trial_evidence" + assert values["cost_log_drag_per_period"]["reason_code"] == "missing_gross_returns" + assert values["deflated_sharpe_ratio"]["value"] is values["cost_log_drag_per_period"]["value"] is None + assert basis["return_basis"] == "RAW_NET" + assert basis["observation_unit"] == "STRATEGY_ACCOUNT_SESSION_SIMPLE_NET_RETURN" + assert basis["metric_role"] == "REPORT_ONLY" + assert basis["asset_benchmark"] is None and basis["benchmark_alignment_count"] is None + assert basis["costs_deducted_again"] is False + assert basis["legacy_rf_adjusted_annualized_metrics_replaced"] is False + current.pop("evidence_digest") + baseline.pop("evidence_digest") + assert current == baseline + assert before == {p: p.read_bytes() for p in historical_sources[0].rglob("*") if p.is_file()} + for forbidden in (str(historical_sources[0]), '"positions"', '"daily_returns"', '"net_returns"'): + assert forbidden not in output + + +def test_unified_missing_outcome_flag_refuses_old_builder_before_read(tmp_path, capsys, monkeypatch): + from types import SimpleNamespace + from us_equity_strategies.research import tqqq_core_trial_journal as journal + original = importlib.import_module + + def old_builder(net_returns): + pytest.fail("older eight-metric builder was substituted") + + def old_module(name, *args, **kwargs): + if name == "quant_platform_kit.research_stats.unified_report": + return SimpleNamespace(build_unified_net_report=old_builder) + return original(name, *args, **kwargs) + + monkeypatch.setattr(journal.importlib, "import_module", old_module) + monkeypatch.setattr(Path, "read_text", lambda *a, **k: pytest.fail("old capability read input")) + assert batch_a_cli.main(["--historical-account-report", "--include-unified-net-report", + "--frozen-member-pack", str(tmp_path / "private.json")]) == 2 + output = capsys.readouterr().out + assert json.loads(output)["reason_codes"] == ["UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE"] + assert str(tmp_path) not in output + + +def test_unified_default_does_not_require_or_import_new_capability(historical_sources, capsys, monkeypatch): + from us_equity_strategies.research import tqqq_core_trial_journal as journal + expected = json.loads(json.dumps(_historical_result(historical_sources))) + monkeypatch.setattr(journal, "_unified_net_report_builder", + lambda: pytest.fail("default report touched new optional capability")) + assert batch_a_cli.main(_historical_cli_args(historical_sources)) == 0 + result = json.loads(capsys.readouterr().out) + assert result == expected + assert "unified_net_report" not in json.dumps(result) + + +def _unified_hand_ledger(navs): + from datetime import date, timedelta + from quant_platform_kit.strategy_lifecycle.contracts import ( + ResearchDailyLedger, ResearchLedgerDay, ResearchPositionMark, + ) + initial = date(2024, 1, 1) + return ResearchDailyLedger( + trial_id="existing-hand-fixture", domain="us_equity", strategy_profile="report-only-hand-fixture", + run_id="fixture-run", param_version=1, input_id="sha256:" + "a" * 64, + calendar_id="XNYS", periods_per_year=252, cost_source="HAND_FIXTURE_NET_MARKS", + cost_inputs={"commission_bps": 0.0, "slippage_bps": 0.0}, + initial_session_date=initial, initial_nav=navs[0], initial_cash=40.0, + initial_positions=(ResearchPositionMark("TQQQ", 1.0, navs[0] - 40.0),), + days=tuple(ResearchLedgerDay( + session_date=initial + timedelta(days=i + 1), cash=40.0, + positions=(ResearchPositionMark("TQQQ", 1.0, end - 40.0),), + trade_net_cashflow=0.0, fees=0.0, nav=end, daily_return=end / start - 1.0, + ) for i, (start, end) in enumerate(zip(navs, navs[1:]))), synthetic=False) + + +@pytest.mark.parametrize("navs,expected", [ + ((100.0, 90.0, 90.0, 108.0), (0.5, 2.0, 2.0, 1.0 / 30)), + ((100.0, 110.0, 121.0), (1.0, "no_loss_observations", "no_loss_observations", 0.1)), + ((100.0, 90.0, 81.0), (0.0, "no_win_observations", 0.0, -0.1)), + ((100.0, 100.0, 100.0), ("no_nonzero_observations", "no_loss_observations", "no_loss_observations", 0.0)), +]) +def test_unified_hand_math_and_known_zero_distinct_from_unknown(navs, expected): + from us_equity_strategies.research.tqqq_core_trial_journal import compute_historical_research_metrics + ledger = _unified_hand_ledger(navs) + kwargs = {"input_binding": {"schema": "qsl.batch-a-historical-input-binding.v1", "pack_digest": "a" * 64}, + "annual_risk_free_rate": 0.252, "annual_minimum_acceptable_return": 0.504, + "include_unified_net_report": True} + if _unified_capability() is None: + with pytest.raises(ValueError, match="^UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE$"): + compute_historical_research_metrics(ledger, **kwargs) + return + report = compute_historical_research_metrics(ledger, **kwargs) + sidecar = report["unified_net_report"] + assert sidecar["observations"] == len(navs) - 1 + values = {item["name"]: item for item in sidecar["metrics"]} + names = ("net_win_rate_ex_zero_per_period", "net_payoff_ratio_per_period", + "net_profit_factor_per_period", "mean_net_return_per_period") + for name, value in zip(names, expected, strict=True): + metric = values[name] + assert metric["details"]["aggregation_level"] == "return_observations_not_trades" + assert metric["details"]["descriptive_only"] is True + if isinstance(value, str): + assert metric["value"] is None and metric["status"] == "UNCOMPUTABLE" + assert metric["reason_code"] == value + else: + assert metric["status"] == "COMPUTED" and metric["value"] == pytest.approx(value) + assert report["evaluation_contract"]["rf"]["annual_simple_decimal"] == 0.252 + assert report["evaluation_contract"]["mar"]["annual_simple_decimal"] == 0.504 + assert report["metrics"]["dsr"] is None and report["metric_status"]["dsr"] == "NOT_COMPUTED" + assert values["deflated_sharpe_ratio"]["reason_code"] == "missing_trial_evidence" + assert values["cost_log_drag_per_period"]["reason_code"] == "missing_gross_returns" + json.dumps(report, allow_nan=False) + + +def test_unified_exact_window_retains_return_count_and_does_not_add_initial_mark(): + from us_equity_strategies.research.tqqq_core_trial_journal import compute_historical_research_metrics + ledger = _unified_hand_ledger((100.0, 90.0, 90.0, 108.0)) + kwargs = {"input_binding": {"schema": "qsl.batch-a-historical-input-binding.v1", "pack_digest": "a" * 64}, + "annual_risk_free_rate": 0.0, "annual_minimum_acceptable_return": 0.0, + "start_session": ledger.days[1].session_date, "end_session": ledger.days[2].session_date, + "include_unified_net_report": True} + if _unified_capability() is None: + with pytest.raises(ValueError, match="^UNIFIED_NET_REPORT_CAPABILITY_UNAVAILABLE$"): + compute_historical_research_metrics(ledger, **kwargs) + return + report = compute_historical_research_metrics(ledger, **kwargs) + values = {item["name"]: item for item in report["unified_net_report"]["metrics"]} + assert report["observation_count"] == report["unified_net_report"]["observations"] == 2 + assert values["mean_net_return_per_period"]["value"] == pytest.approx(0.1) + assert values["net_win_rate_ex_zero_per_period"]["value"] == 1.0 + assert values["net_win_rate_ex_zero_per_period"]["details"]["zero_observations"] == 1 + assert values["max_drawdown"]["value"] == 0.0 + assert report["unified_net_report_basis"]["start_session"] == kwargs["start_session"].isoformat() + + +@pytest.mark.parametrize("value", [None, 1, "true"]) +def test_unified_api_requires_explicit_boolean_before_source_replay(value): + from us_equity_strategies.research.c3_batch_a_existing_member_baseline import report_historical_batch_a_baselines + with pytest.raises(ValueError, match="^UNIFIED_NET_REPORT_OPT_IN_INVALID$"): + report_historical_batch_a_baselines(frozen_member_pack={}, soxl_snapshot=None, tqqq_snapshot=None, + annual_risk_free_rate=0.0, annual_minimum_acceptable_return=0.0, + include_unified_net_report=value)