diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a32798d..7db4e5b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,44 +24,44 @@ jobs: exit 1 fi - - name: Resolve QuantPlatformKit ref - id: quant-platform-kit-ref + - name: Read locked shared dependency refs + id: locked-shared-refs run: | set -euo pipefail - ref="main" - for candidate in "${GITHUB_HEAD_REF:-}" "${GITHUB_BASE_REF:-}"; do - if [ -n "$candidate" ] && git ls-remote --exit-code --heads https://github.com/QuantStrategyLab/QuantPlatformKit.git "$candidate" >/dev/null 2>&1; then - ref="$candidate" - break - fi - done - echo "ref=${ref}" >> "$GITHUB_OUTPUT" - - - name: Resolve UsEquityStrategies ref - id: us-equity-strategies-ref - run: | - set -euo pipefail - ref="main" - for candidate in "${GITHUB_HEAD_REF:-}" "${GITHUB_BASE_REF:-}"; do - if [ -n "$candidate" ] && git ls-remote --exit-code --heads https://github.com/QuantStrategyLab/UsEquityStrategies.git "$candidate" >/dev/null 2>&1; then - ref="$candidate" - break - fi - done - echo "ref=${ref}" >> "$GITHUB_OUTPUT" + python - <<'PY' >> "$GITHUB_OUTPUT" + import re + import tomllib + from pathlib import Path + + required = {"quant-platform-kit", "us-equity-strategies"} + lock = tomllib.loads(Path("uv.lock").read_text(encoding="utf-8")) + refs = {} + for package in lock.get("package", []): + name = package.get("name") + if name not in required: + continue + source = package.get("source") or {} + match = re.search(r"#([0-9a-f]{40})$", str(source.get("git") or "")) + if match: + refs[name] = match.group(1) + if set(refs) != required: + raise SystemExit("uv.lock lacks a full SHA for a shared dependency") + for name in sorted(refs): + print(f"{name.replace('-', '_')}={refs[name]}") + PY - name: Checkout QuantPlatformKit uses: actions/checkout@v6 with: repository: QuantStrategyLab/QuantPlatformKit - ref: ${{ steps.quant-platform-kit-ref.outputs.ref }} + ref: ${{ steps.locked-shared-refs.outputs.quant_platform_kit }} path: external/QuantPlatformKit - name: Checkout UsEquityStrategies uses: actions/checkout@v6 with: repository: QuantStrategyLab/UsEquityStrategies - ref: ${{ steps.us-equity-strategies-ref.outputs.ref }} + ref: ${{ steps.locked-shared-refs.outputs.us_equity_strategies }} path: external/UsEquityStrategies - name: Setup Python @@ -74,10 +74,30 @@ jobs: set -euo pipefail python -m pip install --upgrade pip uv uv sync --frozen --extra test + - name: Verify shared repository refs + run: | + set -euo pipefail + test "$(git -C external/QuantPlatformKit rev-parse HEAD)" = "${{ steps.locked-shared-refs.outputs.quant_platform_kit }}" + test "$(git -C external/UsEquityStrategies rev-parse HEAD)" = "${{ steps.locked-shared-refs.outputs.us_equity_strategies }}" + - name: Smoke import pinned shared packages run: | set -euo pipefail uv run --no-sync python - <<'PY' + import importlib.metadata + import json + import tomllib + from pathlib import Path + + lock = tomllib.loads(Path("uv.lock").read_text(encoding="utf-8")) + for package in lock["package"]: + if package["name"] not in {"quant-platform-kit", "us-equity-strategies"}: + continue + expected = package["source"]["git"].rsplit("#", 1)[1] + installed = json.loads(importlib.metadata.distribution(package["name"]).read_text("direct_url.json") or "{}") + assert installed.get("vcs_info", {}).get("commit_id") == expected, package["name"] + PY + uv run --no-sync python - <<'PY' from quant_platform_kit.common.port_adapters import CallableNotificationPort, CallablePortfolioPort from us_equity_strategies import resolve_canonical_profile @@ -86,11 +106,6 @@ jobs: assert resolve_canonical_profile("russell_top50_leader_rotation") == "russell_top50_leader_rotation" PY - - name: Install editable shared repositories - run: | - set -euo pipefail - uv pip install --no-deps -e external/QuantPlatformKit -e external/UsEquityStrategies - - name: Verify Python dependencies run: uv pip check @@ -102,9 +117,10 @@ jobs: - name: Check QPK pin consistency run: | set -euo pipefail + printf '%s\n' "${{ steps.locked-shared-refs.outputs.quant_platform_kit }}" > "$RUNNER_TEMP/firstrade-qpk-pin" uv run --no-sync python external/QuantPlatformKit/scripts/check_qpk_pin_consistency.py \ --root . \ - --pin-file external/QuantPlatformKit/QPK_PIN + --pin-file "$RUNNER_TEMP/firstrade-qpk-pin" - name: Ensure uv.lock matches pyproject.toml run: uv lock --check diff --git a/application/execution_service.py b/application/execution_service.py index 481ea80..c894fa5 100644 --- a/application/execution_service.py +++ b/application/execution_service.py @@ -648,7 +648,6 @@ def execute_value_target_plan( fetch_order_status=None, before_live_submission: Callable[[], bool] | None = None, ) -> ExecutionCycleResult: - del dry_run_only # ExecutionPort owns preview vs live submission. plan = substitute_small_safe_haven_targets_with_cash( plan, threshold_usd=safe_haven_cash_substitute_threshold_usd, @@ -721,7 +720,11 @@ def _submission_claim_unavailable(symbol: str) -> ExecutionCycleResult: ) def _may_submit_live_order() -> bool: - return before_live_submission is None or bool(before_live_submission()) + if dry_run_only: + return True + if before_live_submission is None: + raise ValueError("Live submission requires a durable submission claim callback.") + return bool(before_live_submission()) tradable_deltas: list[tuple[str, float, float]] = [] for symbol in sorted(set(targets) | set(market_values)): diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 31d96f7..f104ce3 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -642,6 +642,10 @@ def log_message(message: str) -> None: def acquire_submission_claim() -> bool: nonlocal submission_claim_acquired + if not persist_strategy_runs: + raise ValueError( + "Live submission requires enabled strategy-run persistence and a durable state store." + ) if submission_claim_acquired: return True submission_claim_acquired = claim_live_strategy_run( @@ -668,7 +672,7 @@ def acquire_submission_claim() -> bool: fetch_order_status=lambda broker_order_id: client.get_order_status(account, broker_order_id), before_live_submission=( acquire_submission_claim - if persist_strategy_runs and not settings.dry_run_only + if not settings.dry_run_only else None ), ) diff --git a/tests/test_dependency_pin_guard.py b/tests/test_dependency_pin_guard.py index 234a66d..9e39c77 100644 --- a/tests/test_dependency_pin_guard.py +++ b/tests/test_dependency_pin_guard.py @@ -2,6 +2,7 @@ import importlib.util import sys +import textwrap from pathlib import Path from types import ModuleType @@ -62,5 +63,57 @@ def test_dependency_pin_guard_is_blocking_in_ci() -> None: step = workflow[step_start : next_step if next_step != -1 else len(workflow)] assert "external/QuantPlatformKit/scripts/check_qpk_pin_consistency.py" in step - assert "--pin-file external/QuantPlatformKit/QPK_PIN" in step + assert 'steps.locked-shared-refs.outputs.quant_platform_kit' in step + assert '--pin-file "$RUNNER_TEMP/firstrade-qpk-pin"' in step assert "continue-on-error" not in step + + +def test_ci_shared_checkouts_use_locked_refs() -> None: + workflow = CI_WORKFLOW.read_text(encoding="utf-8") + assert "name: Read locked shared dependency refs" in workflow + for name, directory in ( + ("quant_platform_kit", "QuantPlatformKit"), + ("us_equity_strategies", "UsEquityStrategies"), + ): + assert f"ref: ${{{{ steps.locked-shared-refs.outputs.{name} }}}}" in workflow + assert f'git -C external/{directory} rev-parse HEAD' in workflow + assert "git ls-remote" not in workflow + assert "uv pip install --no-deps -e" not in workflow + assert "uv sync --frozen --extra test" in workflow + assert "uv pip check" in workflow + assert "uv lock --check" in workflow + + +def test_ci_installed_shared_identity_matches_lock(monkeypatch) -> None: + import importlib.metadata + import json + import tomllib + + import pytest + + workflow = CI_WORKFLOW.read_text(encoding="utf-8") + smoke = workflow.split("name: Smoke import pinned shared packages", 1)[1] + code = textwrap.dedent(smoke.split("<<'PY'\n", 1)[1].split("\n PY", 1)[0]) + refs = { + package["name"]: package["source"]["git"].rsplit("#", 1)[1] + for package in tomllib.loads(Path("uv.lock").read_text())["package"] + if package["name"] in {"quant-platform-kit", "us-equity-strategies"} + } + identities = {name: {"vcs_info": {"commit_id": ref}} for name, ref in refs.items()} + + class Distribution: + def __init__(self, name): + self.name = name + + def read_text(self, filename): + assert filename == "direct_url.json" + return json.dumps(identities[self.name]) + + monkeypatch.setattr(importlib.metadata, "distribution", Distribution) + exec(code, {}) + for name in refs: + for invalid in ({"vcs_info": {"commit_id": "0" * 40}}, {"dir_info": {"editable": True}}): + with monkeypatch.context() as context: + context.setitem(identities, name, invalid) + with pytest.raises(AssertionError): + exec(code, {}) diff --git a/tests/test_execution_service.py b/tests/test_execution_service.py index 506a3b9..f9036b6 100644 --- a/tests/test_execution_service.py +++ b/tests/test_execution_service.py @@ -3,6 +3,8 @@ from dataclasses import dataclass from datetime import datetime, timezone +import pytest + from application.execution_service import ( _apply_notional_cash_buffer, execute_value_target_plan, @@ -56,6 +58,44 @@ def submit_order(self, order_intent) -> ExecutionReport: ) +@pytest.mark.parametrize("side,notional", [("sell", False), ("buy", False), ("buy", True)]) +def test_live_order_requires_submission_claim_callback(side, notional): + execution_port = SubmittedExecutionPort() + with pytest.raises(ValueError, match="submission claim"): + execute_value_target_plan( + plan={ + "allocation": {"targets": {"AAA": 0.0 if side == "sell" else 20.0}}, + "portfolio": { + "market_values": {"AAA": 20.0 if side == "sell" else 0.0}, + "sellable_quantities": {"AAA": 2.0}, + "liquid_cash": 100.0, + }, + "execution": {"current_min_trade": 1.0}, + }, + market_data_port=FakeMarketDataPort({"AAA": 10.0}), + execution_port=execution_port, + dry_run_only=False, + notional_buy_execution=notional, + ) + assert execution_port.orders == [] + + +def test_live_noop_does_not_require_submission_claim(): + execution_port = SubmittedExecutionPort() + result = execute_value_target_plan( + plan={ + "allocation": {"targets": {"AAA": 20.0}}, + "portfolio": {"market_values": {"AAA": 20.0}}, + "execution": {"current_min_trade": 1.0}, + }, + market_data_port=FakeMarketDataPort({"AAA": 10.0}), + execution_port=execution_port, + dry_run_only=False, + ) + assert execution_port.orders == [] + assert result.idempotency_blocked is False + + def test_execute_value_target_plan_sells_before_buys_and_caps_order_notional(): execution_port = FakeExecutionPort() result = execute_value_target_plan( @@ -99,6 +139,7 @@ def test_execute_value_target_plan_marks_live_submissions_pending_reconciliation market_data_port=FakeMarketDataPort({"AAA": 10.0}), execution_port=execution_port, dry_run_only=False, + before_live_submission=lambda: True, # Simulate an acquired durable claim. ) assert result.action_done is False @@ -666,6 +707,7 @@ def test_notional_buy_keeps_cash_buffer_when_order_would_use_all_available_cash( market_data_port=FakeMarketDataPort({"IBIT": 35.0}), execution_port=execution_port, dry_run_only=False, + before_live_submission=lambda: True, # Simulate an acquired durable claim. notional_buy_execution=True, ) @@ -711,6 +753,7 @@ def submit_order(self, order_intent) -> ExecutionReport: market_data_port=FakeMarketDataPort({"IBIT": 35.0}), execution_port=execution_port, dry_run_only=False, + before_live_submission=lambda: True, # Simulate an acquired durable claim. notional_buy_execution=True, ) diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index 71176af..285cc6d 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -4,6 +4,8 @@ from datetime import datetime, timezone from types import SimpleNamespace +import pytest + from application.firstrade_client import FirstradeCredentials from application.rebalance_service import ( _publish_cycle_notification, @@ -171,6 +173,116 @@ def _latest_strategy_run_payloads(store: FakeStateStore) -> list[dict]: return [payload for key, payload in store.writes if key.endswith("latest.json")] +@pytest.fixture +def live_cycle(monkeypatch): + client = FakeFirstradeClient(None, live_trading_enabled=True) + monkeypatch.setattr( + "application.rebalance_service.load_strategy_runtime", + lambda *_args, **_kwargs: FakeStrategyRuntime(), + ) + monkeypatch.setattr( + "application.rebalance_service._utcnow", + lambda: datetime(2026, 9, 5, tzinfo=timezone.utc), + ) + + def run(*, store=None, **overrides): + settings = dict( + dry_run_only=False, live_trading_enabled=True, + live_order_ack=True, persist_strategy_runs=True, + ) + settings.update(overrides) + return run_strategy_cycle( + runtime_settings=_runtime_settings_with_persistence(**settings), + credentials=FirstradeCredentials(username="", password=""), + client_factory=lambda *_args, **_kwargs: client, + state_store=store, + env_reader=lambda _name, default=None: default, + send_cycle_notification=False, + dispatch_plugin_alerts=False, + ) + + return run, client + + +@pytest.mark.parametrize("persist,has_store", [(True, False), (False, False), (False, True)]) +def test_live_submission_requires_enabled_persistence_and_store(live_cycle, persist, has_store): + run, client = live_cycle + store = FakeStateStore() if has_store else None + for _ in range(2): + with pytest.raises(ValueError, match="persistence"): + run(store=store, persist_strategy_runs=persist) + assert client.orders == [] + + +def test_live_noop_without_persistence_does_not_acquire_claim(live_cycle, monkeypatch): + run, client = live_cycle + monkeypatch.setattr( + "application.rebalance_service.map_strategy_decision_to_plan", + lambda *_args, **_kwargs: { + "allocation": {"targets": {}}, + "portfolio": {}, + "execution": {"current_min_trade": 1.0}, + }, + ) + result = run(persist_strategy_runs=False) + assert result["ok"] is True + assert result["strategy_run_stage"] == "NO_ACTION" + assert client.orders == [] + + +def test_dry_run_without_persistence_only_previews(live_cycle, monkeypatch): + run, client = live_cycle + + def unexpected_claim(**_kwargs): + pytest.fail("dry-run must not acquire a live claim") + + monkeypatch.setattr("application.rebalance_service.claim_live_strategy_run", unexpected_claim) + result = run(dry_run_only=True, persist_strategy_runs=False) + assert result["strategy_run_stage"] == "DRY_RUN_COMPLETED" + assert client.orders + assert all(dry_run for _request, dry_run, _ack in client.orders) + + +def test_accepted_timeout_retains_claim_and_blocks_repeat(live_cycle, monkeypatch): + run, client = live_cycle + store = FakeStateStore() + place_order = client.place_stock_order + + def accepted_then_timeout(*args, **kwargs): + place_order(*args, **kwargs) + raise TimeoutError("synthetic submission timeout") + + monkeypatch.setattr(client, "place_stock_order", accepted_then_timeout) + with pytest.raises(TimeoutError): + run(store=store) + claims = {key: dict(value) for key, value in store.payloads.items() if "/claims/" in key} + assert len(claims) == 1 + repeated = run(store=store) + assert repeated["submission_claim_blocks_repeat"] is True + assert len(client.orders) == 1 + assert all(store.payloads[key] == value for key, value in claims.items()) + + +def test_completed_write_failure_preserves_submission_and_blocks_repeat(live_cycle): + run, client = live_cycle + + class CompletedWriteFailureStore(FakeStateStore): + def write_json(self, key, payload): + if payload.get("stage") == "PENDING_RECONCILIATION": + raise RuntimeError("synthetic completed-state write failure") + return super().write_json(key, payload) + + store = CompletedWriteFailureStore() + result = run(store=store) + assert result["strategy_run_persisted"] is False + assert result["broker_submission_done"] is True + assert result["submitted_orders"][0]["status"] == "submitted" + assert result["strategy_run_stage"] == "PENDING_RECONCILIATION" + repeated = run(store=store) + assert repeated["submission_claim_blocks_repeat"] is True + assert len(client.orders) == 1 + + def test_notification_i18n_keys_are_aligned(): assert set(I18N["zh"]) == set(I18N["en"]) assert build_translator("zh")("account_label", account="****1234") == "🆔 账户: ****1234"