diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index d120de6..f738db2 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -107,6 +107,7 @@ jobs: IBKR_STRATEGY_PLUGIN_MOUNTS_JSON: ${{ vars.IBKR_STRATEGY_PLUGIN_MOUNTS_JSON }} IBKR_RECONCILIATION_OUTPUT_PATH: ${{ vars.IBKR_RECONCILIATION_OUTPUT_PATH }} IBKR_DRY_RUN_ONLY: ${{ vars.IBKR_DRY_RUN_ONLY }} + IBKR_EXECUTION_DEDUP_ENABLED: ${{ vars.IBKR_EXECUTION_DEDUP_ENABLED }} IBKR_PAPER_LIQUIDATE_ONLY: ${{ vars.IBKR_PAPER_LIQUIDATE_ONLY }} IBKR_MARKET: ${{ vars.IBKR_MARKET }} IBKR_MARKET_CALENDAR: ${{ vars.IBKR_MARKET_CALENDAR }} diff --git a/application/ibkr_market_data.py b/application/ibkr_market_data.py new file mode 100644 index 0000000..b3fafbb --- /dev/null +++ b/application/ibkr_market_data.py @@ -0,0 +1,157 @@ +"""IBKR market-data error handling.""" + +from __future__ import annotations + +import json +import logging +import re +from collections import Counter +from collections.abc import Callable, Iterable +from threading import Lock, RLock +from typing import Any + + +_EXPECTED_ENTITLEMENT_ERROR_CODES = frozenset({10089, 10168}) +_SECONDARY_CANCEL_ERROR_CODE = 300 +_ERROR_CODE_PATTERN = re.compile(r"^(?:Error|Warning) (\d+), reqId (-?\d+)") +_LOCK_CREATION_GUARD = Lock() +_QUOTE_FETCH_LOCK_ATTR = "_ibkr_market_data_quote_fetch_lock" + + +def _quote_fetch_lock(wrapper: Any) -> RLock: + if wrapper is None: + return RLock() + with _LOCK_CREATION_GUARD: + lock = getattr(wrapper, _QUOTE_FETCH_LOCK_ATTR, None) + if lock is None: + lock = RLock() + setattr(wrapper, _QUOTE_FETCH_LOCK_ATTR, lock) + return lock + + +class _ExpectedMarketDataErrorTracker: + def __init__(self, *, wrapper: Any, requested_symbols: set[str]) -> None: + self.wrapper = wrapper + self.requested_symbols = requested_symbols + self.entitlement_request_ids: set[int] = set() + + def _matches_requested_contract(self, request_id: int, contract: Any = None) -> bool: + resolved_contract = contract + if resolved_contract is None: + contracts = getattr(self.wrapper, "_reqId2Contract", {}) + resolved_contract = contracts.get(request_id) if hasattr(contracts, "get") else None + symbol = str(getattr(resolved_contract, "symbol", "") or "").strip().upper() + return bool(symbol and symbol in self.requested_symbols) + + def should_suppress_message(self, message: str) -> bool: + match = _ERROR_CODE_PATTERN.match(message) + if match is None: + return False + error_code = int(match.group(1)) + request_id = int(match.group(2)) + if ( + error_code in _EXPECTED_ENTITLEMENT_ERROR_CODES + and self._matches_requested_contract(request_id) + ): + self.entitlement_request_ids.add(request_id) + return True + if ( + error_code == _SECONDARY_CANCEL_ERROR_CODE + and request_id in self.entitlement_request_ids + ): + return True + return False + + def should_count_event(self, request_id: int, error_code: int, contract: Any) -> bool: + if ( + error_code in _EXPECTED_ENTITLEMENT_ERROR_CODES + and self._matches_requested_contract(request_id, contract) + ): + self.entitlement_request_ids.add(request_id) + return True + return ( + error_code == _SECONDARY_CANCEL_ERROR_CODE + and request_id in self.entitlement_request_ids + ) + + +class _InstanceScopedMarketDataLogger(logging.LoggerAdapter): + def __init__(self, logger: logging.Logger, tracker: _ExpectedMarketDataErrorTracker) -> None: + super().__init__(logger, {}) + self.tracker = tracker + + def log(self, level, msg, *args, **kwargs) -> None: + if not self.isEnabledFor(level): + return + try: + message = str(msg) % args if args else str(msg) + except (TypeError, ValueError): + message = str(msg) + if self.tracker.should_suppress_message(message): + return + super().log(level, msg, *args, **kwargs) + + +def fetch_quote_snapshots_with_expected_error_summary( + ib: Any, + symbols: Iterable[str], + *, + fetch_quote_snapshots: Callable[..., dict], + printer: Callable[[str], None] = print, + **kwargs, +) -> dict: + """Collapse expected entitlement errors while preserving unrelated broker errors.""" + normalized_symbols = tuple(dict.fromkeys(str(symbol).strip().upper() for symbol in symbols)) + error_counts: Counter[int] = Counter() + wrapper = getattr(ib, "wrapper", None) + tracker = _ExpectedMarketDataErrorTracker( + wrapper=wrapper, + requested_symbols=set(normalized_symbols), + ) + + def record_error(req_id, error_code, _error_string, contract) -> None: + try: + code = int(error_code) + request_id = int(req_id) + except (TypeError, ValueError): + return + if tracker.should_count_event(request_id, code, contract): + error_counts[code] += 1 + + error_event = getattr(ib, "errorEvent", None) + original_logger = getattr(wrapper, "_logger", None) + quote_fetch_lock = _quote_fetch_lock(wrapper) + scoped_logger = ( + _InstanceScopedMarketDataLogger(original_logger, tracker) + if isinstance(original_logger, logging.Logger) + else None + ) + + with quote_fetch_lock: + if scoped_logger is not None: + wrapper._logger = scoped_logger + if error_event is not None: + error_event += record_error + try: + snapshots = fetch_quote_snapshots(ib, normalized_symbols, **kwargs) + finally: + if error_event is not None: + error_event -= record_error + if scoped_logger is not None and getattr(wrapper, "_logger", None) is scoped_logger: + wrapper._logger = original_logger + + if error_counts: + printer( + "ibkr_market_data_fallback " + + json.dumps( + { + "error_counts": { + str(code): error_counts[code] for code in sorted(error_counts) + }, + "requested_symbols": len(normalized_symbols), + "resolved_symbols": len(snapshots), + }, + sort_keys=True, + ) + ) + return snapshots diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 7dca436..c83a683 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -700,10 +700,39 @@ def _execution_already_recorded_message(*, config: IBKRRebalanceConfig, signal_m def _should_record_execution_marker(*, trade_logs, execution_summary, config: IBKRRebalanceConfig) -> bool: if not getattr(config, "execution_dedup_enabled", False): return False - if tuple(trade_logs or ()): - return True + if execution_summary is None: + return bool(tuple(trade_logs or ())) summary = dict(execution_summary or {}) - return bool(summary.get("action_done")) or int(summary.get("orders_previewed_count") or 0) > 0 + execution_status = str(summary.get("execution_status") or "").strip().lower() + if execution_status in {"blocked", "error", "failed", "failure"}: + return False + accepted_order_keys = ( + "orders_submitted", + "orders_filled", + "orders_partially_filled", + "option_orders_submitted", + "option_orders_filled", + "option_orders_partially_filled", + ) + has_structured_execution_result = any( + key in summary + for key in ( + "execution_status", + "action_done", + "orders_previewed_count", + *accepted_order_keys, + ) + ) + return ( + execution_status == "executed" + or any(tuple(summary.get(key) or ()) for key in accepted_order_keys) + or bool(summary.get("action_done")) + or int(summary.get("orders_previewed_count") or 0) > 0 + or ( + not has_structured_execution_result + and bool(tuple(trade_logs or ())) + ) + ) def _record_execution_marker( diff --git a/main.py b/main.py index e05f07e..a9d011d 100644 --- a/main.py +++ b/main.py @@ -65,8 +65,9 @@ ensure_event_loop as ibkr_ensure_event_loop, fetch_historical_price_candles, fetch_historical_price_series, - fetch_quote_snapshots, + fetch_quote_snapshots as qpk_fetch_quote_snapshots, ) +from application.ibkr_market_data import fetch_quote_snapshots_with_expected_error_summary from application.ibkr_order_execution import probe_order_write_access, submit_order_intent from application.monitor_dispatcher import ( dispatch_due_monitor_targets, @@ -524,9 +525,10 @@ def fetch_market_fallback_historical_candles(symbol, **kwargs): def fetch_market_quote_snapshots(ib, symbols, **kwargs): - return fetch_quote_snapshots( + return fetch_quote_snapshots_with_expected_error_summary( ib, symbols, + fetch_quote_snapshots=qpk_fetch_quote_snapshots, exchange=MARKET_EXCHANGE, currency=MARKET_CURRENCY, **kwargs, diff --git a/scripts/build_cloud_run_env_sync_plan.py b/scripts/build_cloud_run_env_sync_plan.py index 5a55355..dbf63f8 100644 --- a/scripts/build_cloud_run_env_sync_plan.py +++ b/scripts/build_cloud_run_env_sync_plan.py @@ -69,6 +69,7 @@ def _should_add_local_src(candidate: Path) -> bool: "IBKR_MARKET_TIMEZONE", "IB_GATEWAY_ZONE", "IB_GATEWAY_IP_MODE", + "IBKR_EXECUTION_DEDUP_ENABLED", "EXECUTION_REPORT_GCS_URI", } ) @@ -97,6 +98,7 @@ def _should_add_local_src(candidate: Path) -> bool: "IBKR_STRATEGY_PLUGIN_MOUNTS_JSON", "IBKR_RECONCILIATION_OUTPUT_PATH", "IBKR_DRY_RUN_ONLY", + "IBKR_EXECUTION_DEDUP_ENABLED", "IBKR_PAPER_LIQUIDATE_ONLY", "IBKR_MIN_RESERVED_CASH_USD", "IBKR_RESERVED_CASH_RATIO", @@ -429,10 +431,13 @@ def _build_target_plan( value = strategy_defaults.get(name) # 4. Override from RUNTIME_TARGET_JSON.overrides if value is None and isinstance(overrides, Mapping): - override_value = overrides.get(name) or overrides.get(name.lower()) + override_value = ( + overrides[name] + if name in overrides + else overrides.get(name.lower()) + ) if override_value is not None: value = _coerce_env_value(override_value) - if value is None: remove_env_vars.append(name) else: diff --git a/tests/test_connect_timeout_alert.py b/tests/test_connect_timeout_alert.py index aa60b35..0e54503 100644 --- a/tests/test_connect_timeout_alert.py +++ b/tests/test_connect_timeout_alert.py @@ -137,6 +137,12 @@ def run(self, *args, **kwargs): notify_lang="en", tg_token=None, tg_chat_id="chat-id", + notification_channel="telegram", + wecom_webhook_url=None, + dingtalk_webhook_url=None, + feishu_webhook_url=None, + serverchan_webhook_url=None, + strategy_metadata=None, ibkr_feature_snapshot_manifest_path=None, ibkr_reconciliation_output_path=None, market_hours_source="cloud_run", @@ -194,7 +200,7 @@ def test_handle_request_sends_ibkr_connect_timeout_notification(self): module.is_market_open_now = lambda **_kwargs: True module.run_strategy_core = lambda **_kwargs: (_ for _ in ()).throw( - TimeoutError("IBKR API handshake timed out") + module.IBKRGatewayUnavailableError("IBKR API handshake timed out") ) module.persist_execution_report = lambda report: observed.setdefault("report", dict(report)) or "/tmp/report.json" module.publish_notification = lambda *, detailed_text, compact_text: observed["messages"].append( @@ -205,7 +211,7 @@ def test_handle_request_sends_ibkr_connect_timeout_notification(self): with module.app.test_request_context("/", method="POST"): body, status = module.handle_request() - self.assertEqual(status, 500) + self.assertEqual(status, 503) self.assertEqual(body, "Error") self.assertEqual(observed["report"]["errors"][0]["stage"], "ibkr_connect") self.assertIn("IBKR 连接异常", observed["messages"][0][0]) diff --git a/tests/test_event_loop.py b/tests/test_event_loop.py index ea1b321..1102079 100644 --- a/tests/test_event_loop.py +++ b/tests/test_event_loop.py @@ -1,5 +1,4 @@ import asyncio -import types from concurrent.futures import ThreadPoolExecutor import pytest @@ -111,27 +110,15 @@ def test_get_ib_host_resolves_lazily(strategy_module_factory, monkeypatch): assert module.IB_HOST is None - fake_instance = types.SimpleNamespace( - network_interfaces=[ - types.SimpleNamespace( - access_configs=[types.SimpleNamespace(nat_i_p="35.211.181.174")], - network_i_p="10.0.0.8", - ) - ] - ) - - class FakeInstancesClient: - def get(self, project, zone, instance): - assert project == "test-project" + class FakeComputeDiscovery: + def resolve_instance_ip(self, instance, zone, *, project_id, prefer_internal): + assert project_id == "test-project" assert zone == "us-central1-a" assert instance == "ib-gateway" - return fake_instance + assert prefer_internal is True + return "10.0.0.8" - monkeypatch.setattr( - module, - "compute_v1", - types.SimpleNamespace(InstancesClient=FakeInstancesClient), - ) + monkeypatch.setattr(module, "get_compute_discovery", FakeComputeDiscovery) monkeypatch.setattr(module, "get_project_id", lambda: "test-project") assert module.get_ib_host() == "10.0.0.8" @@ -207,27 +194,15 @@ def test_ib_gateway_mode_is_required(strategy_module_factory): def test_resolve_gce_instance_ip_prefers_internal_by_default(strategy_module, monkeypatch): - fake_instance = types.SimpleNamespace( - network_interfaces=[ - types.SimpleNamespace( - access_configs=[types.SimpleNamespace(nat_i_p="35.211.181.174")], - network_i_p="10.0.0.8", - ) - ] - ) - - class FakeInstancesClient: - def get(self, project, zone, instance): - assert project == "test-project" + class FakeComputeDiscovery: + def resolve_instance_ip(self, instance, zone, *, project_id, prefer_internal): + assert project_id == "test-project" assert zone == "us-central1-a" assert instance == "ib-gateway" - return fake_instance + assert prefer_internal is True + return "10.0.0.8" - monkeypatch.setattr( - strategy_module, - "compute_v1", - types.SimpleNamespace(InstancesClient=FakeInstancesClient), - ) + monkeypatch.setattr(strategy_module, "get_compute_discovery", FakeComputeDiscovery) monkeypatch.setattr(strategy_module, "get_project_id", lambda: "test-project") monkeypatch.delenv("IB_GATEWAY_IP_MODE", raising=False) @@ -237,27 +212,15 @@ def get(self, project, zone, instance): def test_resolve_gce_instance_ip_can_use_external_mode(strategy_module, monkeypatch): - fake_instance = types.SimpleNamespace( - network_interfaces=[ - types.SimpleNamespace( - access_configs=[types.SimpleNamespace(nat_i_p="35.211.181.174")], - network_i_p="10.0.0.8", - ) - ] - ) - - class FakeInstancesClient: - def get(self, project, zone, instance): - assert project == "test-project" + class FakeComputeDiscovery: + def resolve_instance_ip(self, instance, zone, *, project_id, prefer_internal): + assert project_id == "test-project" assert zone == "us-central1-a" assert instance == "ib-gateway" - return fake_instance + assert prefer_internal is False + return "35.211.181.174" - monkeypatch.setattr( - strategy_module, - "compute_v1", - types.SimpleNamespace(InstancesClient=FakeInstancesClient), - ) + monkeypatch.setattr(strategy_module, "get_compute_discovery", FakeComputeDiscovery) monkeypatch.setattr(strategy_module, "get_project_id", lambda: "test-project") monkeypatch.setenv("IB_GATEWAY_IP_MODE", "external") diff --git a/tests/test_ibkr_market_data.py b/tests/test_ibkr_market_data.py new file mode 100644 index 0000000..65f31e1 --- /dev/null +++ b/tests/test_ibkr_market_data.py @@ -0,0 +1,199 @@ +import logging +from concurrent.futures import ThreadPoolExecutor +from threading import Event, Lock + +from application.ibkr_market_data import fetch_quote_snapshots_with_expected_error_summary + + +class FakeErrorEvent: + def __init__(self): + self.handlers = [] + + def __iadd__(self, handler): + self.handlers.append(handler) + return self + + def __isub__(self, handler): + self.handlers.remove(handler) + return self + + def emit(self, *args): + for handler in tuple(self.handlers): + handler(*args) + + +class FakeIB: + def __init__(self): + self.errorEvent = FakeErrorEvent() + self.wrapper = type( + "FakeWrapper", + (), + { + "_logger": logging.getLogger("ib_insync.wrapper"), + "_reqId2Contract": {}, + }, + )() + + +class FakeContract: + def __init__(self, symbol): + self.symbol = symbol + + +def emit_broker_error(ib, request_id, code, message, *, symbol=None): + contract = FakeContract(symbol) if symbol else None + if contract is not None: + ib.wrapper._reqId2Contract[request_id] = contract + ib.wrapper._logger.error("Error %s, reqId %s: %s", code, request_id, message) + ib.errorEvent.emit(request_id, code, message, contract) + + +def test_expected_market_data_errors_are_collapsed_without_hiding_unrelated_errors(caplog): + ib = FakeIB() + summaries = [] + def fake_fetch(_ib, symbols, **_kwargs): + for code in (10168, 10089): + emit_broker_error( + ib, + 1, + code, + "expected market data error", + symbol="AAA", + ) + emit_broker_error(ib, 1, 300, "Can't find EId with tickerId:1") + emit_broker_error(ib, -1, 502, "unrelated gateway failure") + return {symbol: object() for symbol in symbols} + + with caplog.at_level(logging.ERROR, logger="ib_insync.wrapper"): + result = fetch_quote_snapshots_with_expected_error_summary( + ib, + ("AAA",), + fetch_quote_snapshots=fake_fetch, + printer=summaries.append, + ) + + assert set(result) == {"AAA"} + assert "Error 10168" not in caplog.text + assert "Error 10089" not in caplog.text + assert "Error 300" not in caplog.text + assert "Error 502" in caplog.text + assert len(summaries) == 1 + assert '"10089": 1' in summaries[0] + assert '"10168": 1' in summaries[0] + assert '"300": 1' in summaries[0] + + +def test_standalone_error_300_is_not_reported_as_entitlement_fallback(caplog): + ib = FakeIB() + summaries = [] + def fake_fetch(_ib, symbols, **_kwargs): + emit_broker_error( + ib, + 1, + 300, + "Can't find EId with tickerId:1", + symbol="AAA", + ) + return {symbol: object() for symbol in symbols} + + with caplog.at_level(logging.ERROR, logger="ib_insync.wrapper"): + result = fetch_quote_snapshots_with_expected_error_summary( + ib, + ("AAA",), + fetch_quote_snapshots=fake_fetch, + printer=summaries.append, + ) + + assert set(result) == {"AAA"} + assert "Error 300" in caplog.text + assert summaries == [] + + +def test_error_300_for_different_request_is_not_hidden(caplog): + ib = FakeIB() + summaries = [] + def fake_fetch(_ib, symbols, **_kwargs): + emit_broker_error( + ib, + 1, + 10168, + "expected market data error", + symbol="AAA", + ) + emit_broker_error( + ib, + 2, + 300, + "unrelated request cancellation", + symbol="BBB", + ) + return {symbol: object() for symbol in symbols} + + with caplog.at_level(logging.ERROR, logger="ib_insync.wrapper"): + result = fetch_quote_snapshots_with_expected_error_summary( + ib, + ("AAA", "BBB"), + fetch_quote_snapshots=fake_fetch, + printer=summaries.append, + ) + + assert set(result) == {"AAA", "BBB"} + assert "Error 10168" not in caplog.text + assert "Error 300, reqId 2" in caplog.text + assert len(summaries) == 1 + assert '"10168": 1' in summaries[0] + assert '"300"' not in summaries[0] + + +def test_other_ib_instance_logs_are_not_filtered(caplog): + ib = FakeIB() + other_ib = FakeIB() + + def fake_fetch(_ib, symbols, **_kwargs): + emit_broker_error( + other_ib, + 9, + 10168, + "unrelated instance entitlement error", + symbol="OTHER", + ) + return {symbol: object() for symbol in symbols} + + with caplog.at_level(logging.ERROR, logger="ib_insync.wrapper"): + fetch_quote_snapshots_with_expected_error_summary( + ib, + ("AAA",), + fetch_quote_snapshots=fake_fetch, + printer=lambda _message: None, + ) + + assert "unrelated instance entitlement error" in caplog.text + + +def test_different_ib_instances_fetch_concurrently(): + entered = set() + entered_lock = Lock() + both_entered = Event() + + def run_fetch(label): + ib = FakeIB() + + def fake_fetch(_ib, symbols, **_kwargs): + with entered_lock: + entered.add(label) + if len(entered) == 2: + both_entered.set() + assert both_entered.wait(timeout=1.0) + return {symbol: object() for symbol in symbols} + + return fetch_quote_snapshots_with_expected_error_summary( + ib, + (label,), + fetch_quote_snapshots=fake_fetch, + printer=lambda _message: None, + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(run_fetch, ("AAA", "BBB"))) + + assert [set(result) for result in results] == [{"AAA"}, {"BBB"}] diff --git a/tests/test_qsl_metadata.py b/tests/test_qsl_metadata.py index beaf5ae..2869589 100644 --- a/tests/test_qsl_metadata.py +++ b/tests/test_qsl_metadata.py @@ -12,7 +12,7 @@ def test_qsl_metadata_has_runtime_platform_fields() -> None: assert qsl["tier"] == "runtime" assert qsl["upgrade_ring"] == "ring_d" assert qsl.get("repo") == "InteractiveBrokersPlatform" - assert qsl["compat"]["bundle"] == "2026.07.3" + assert qsl["compat"]["bundle"] == "2026.07.4" requires = qsl["requires"] assert "quant_platform_kit" in requires assert "us_equity_strategies" in requires diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index a965024..21efe28 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -1,6 +1,11 @@ import json -from application.rebalance_service import _resolve_reconciliation_mode, _strategy_dashboard_text, run_strategy_core +from application.rebalance_service import ( + _resolve_reconciliation_mode, + _should_record_execution_marker, + _strategy_dashboard_text, + run_strategy_core, +) from application.runtime_dependencies import IBKRRebalanceConfig from notifications.renderers import ( _build_notification_trade_lines, @@ -78,6 +83,86 @@ def translate(key, **kwargs): return translate +def test_execution_marker_is_not_recorded_for_blocked_execution_with_trade_logs(): + config = IBKRRebalanceConfig( + translator=_build_test_translator(), + separator="---", + execution_dedup_enabled=True, + ) + + assert not _should_record_execution_marker( + trade_logs=("broker submission failed",), + execution_summary={ + "execution_status": "blocked", + "orders_skipped": [{"reason": "submit_failed"}], + }, + config=config, + ) + + +def test_execution_marker_is_recorded_after_accepted_order(): + config = IBKRRebalanceConfig( + translator=_build_test_translator(), + separator="---", + execution_dedup_enabled=True, + ) + + assert _should_record_execution_marker( + trade_logs=("order submitted",), + execution_summary={ + "execution_status": "executed", + "orders_submitted": [{"symbol": "AAA", "status": "Submitted"}], + }, + config=config, + ) + + +def test_execution_marker_is_recorded_after_partial_submission_success(): + config = IBKRRebalanceConfig( + translator=_build_test_translator(), + separator="---", + execution_dedup_enabled=True, + ) + + assert _should_record_execution_marker( + trade_logs=("one order submitted", "one order rejected"), + execution_summary={ + "execution_status": "executed", + "orders_submitted": [{"symbol": "AAA", "status": "Submitted"}], + "skipped_reasons": ["submit_failed:BBB:Rejected"], + }, + config=config, + ) + + +def test_execution_marker_preserves_legacy_trade_log_fallback(): + config = IBKRRebalanceConfig( + translator=_build_test_translator(), + separator="---", + execution_dedup_enabled=True, + ) + + assert _should_record_execution_marker( + trade_logs=("legacy order submitted",), + execution_summary=None, + config=config, + ) + + +def test_execution_marker_preserves_legacy_trade_log_fallback_with_minimal_summary(): + config = IBKRRebalanceConfig( + translator=_build_test_translator(), + separator="---", + execution_dedup_enabled=True, + ) + + assert _should_record_execution_marker( + trade_logs=("legacy order submitted",), + execution_summary={"adapter_metadata": {"source": "legacy"}}, + config=config, + ) + + def test_build_dashboard_localizes_strategy_details(): dashboard = build_dashboard( positions={}, diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index d4cfb76..68ed279 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -18,6 +18,7 @@ def test_cloud_run_route_contracts_are_registered(strategy_module): "/probe": ["GET", "POST"], "/monitor-dispatch": ["GET", "POST"], "/health": ["GET"], + "/healthz": ["GET"], "/static/": ["GET"], } @@ -1128,25 +1129,36 @@ def fake_connect_ib(): first = strategy_module.run_strategy_core() second = strategy_module.run_strategy_core() - assert first.result == "OK - heartbeat" - assert second.result == "OK - heartbeat" + assert first.result == "OK - no-op" + assert second.result == "OK - no-op" assert observed["connect_calls"] == 2 assert observed["disconnect_calls"] == 2 + assert observed["messages"] == [] -def test_send_tg_message_logs_non_200_response(strategy_module, monkeypatch, capsys): - class FakeResponse: - status_code = 401 - text = "unauthorized" - +def test_send_tg_message_uses_cycle_channel_sender(strategy_module, monkeypatch): + observed = {} monkeypatch.setattr(strategy_module, "TG_TOKEN", "token") monkeypatch.setattr(strategy_module, "TG_CHAT_ID", "chat-id") - monkeypatch.setattr(strategy_module.requests, "post", lambda *args, **kwargs: FakeResponse()) + monkeypatch.setattr( + "quant_platform_kit.notifications.cycle_channel.build_cycle_sender", + lambda **kwargs: ( + observed.setdefault("sender_config", kwargs), + lambda message: observed.setdefault("message", message), + )[1], + ) strategy_module.send_tg_message("hello") - captured = capsys.readouterr() - assert "Telegram send failed with status 401: unauthorized" in captured.out + assert observed == { + "sender_config": { + "channel": "telegram", + "telegram_token": "token", + "telegram_chat_id": "chat-id", + "webhook_url": None, + }, + "message": "hello", + } def test_global_telegram_chat_id_is_used(strategy_module_factory): diff --git a/tests/test_runtime_config_support.py b/tests/test_runtime_config_support.py index a709008..6f145f0 100644 --- a/tests/test_runtime_config_support.py +++ b/tests/test_runtime_config_support.py @@ -1261,7 +1261,11 @@ def test_build_cloud_run_env_sync_plan_generates_live_gateway_deadlines() -> Non check=True, capture_output=True, text=True, - env={**os.environ, "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps(payload)}, + env={ + **os.environ, + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps(payload), + "IBKR_EXECUTION_DEDUP_ENABLED": "true", + }, ) plan = json.loads(result.stdout) @@ -1269,6 +1273,7 @@ def test_build_cloud_run_env_sync_plan_generates_live_gateway_deadlines() -> Non gateway_accounts = ("u15998061", "u16608560", "u18336562", "u18308207") for account in gateway_accounts: service_name = f"interactive-brokers-quant-live-{account}-service" + assert by_service[service_name]["env"]["IBKR_EXECUTION_DEDUP_ENABLED"] == "true" assert by_service[service_name]["scheduler"] == { "timezone": "America/New_York", "main_time": "45 15 * * 1-5", @@ -1286,6 +1291,66 @@ def test_build_cloud_run_env_sync_plan_generates_live_gateway_deadlines() -> Non assert "attempt_deadline" not in by_service["interactive-brokers-us-combo-shadow-service"]["scheduler"] +def test_build_cloud_run_env_sync_plan_does_not_enable_live_dedup_implicitly() -> None: + payload = _four_gateway_warmup_payload("43 9,15 * * 1-5") + result = subprocess.run( + [sys.executable, str(SYNC_PLAN_SCRIPT_PATH), "--json"], + check=True, + capture_output=True, + text=True, + env={**os.environ, "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps(payload)}, + ) + + plan = json.loads(result.stdout) + for target in plan["targets"]: + assert "IBKR_EXECUTION_DEDUP_ENABLED" not in target["env"] + assert "IBKR_EXECUTION_DEDUP_ENABLED" in target["remove_env_vars"] + + +def test_build_cloud_run_env_sync_plan_honors_explicit_dedup_override() -> None: + payload = _four_gateway_warmup_payload("43 9,15 * * 1-5") + payload["targets"][0]["runtime_target"]["overrides"] = { + "IBKR_EXECUTION_DEDUP_ENABLED": False, + } + result = subprocess.run( + [sys.executable, str(SYNC_PLAN_SCRIPT_PATH), "--json"], + check=True, + capture_output=True, + text=True, + env={**os.environ, "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps(payload)}, + ) + + plan = json.loads(result.stdout) + by_service = {target["service_name"]: target for target in plan["targets"]} + assert ( + by_service["interactive-brokers-quant-live-u15998061-service"]["env"][ + "IBKR_EXECUTION_DEDUP_ENABLED" + ] + == "false" + ) + + +def test_build_cloud_run_env_sync_plan_honors_global_dedup_env() -> None: + payload = _four_gateway_warmup_payload("43 9,15 * * 1-5") + result = subprocess.run( + [sys.executable, str(SYNC_PLAN_SCRIPT_PATH), "--json"], + check=True, + capture_output=True, + text=True, + env={ + **os.environ, + "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps(payload), + "IBKR_EXECUTION_DEDUP_ENABLED": "false", + }, + ) + + plan = json.loads(result.stdout) + assert { + target["env"]["IBKR_EXECUTION_DEDUP_ENABLED"] + for target in plan["targets"] + } == {"false"} + + def test_build_cloud_run_env_sync_plan_accepts_strategy_defined_gateway_schedule() -> None: payload = _four_gateway_warmup_payload("35 9,15 * * 1-5") first_target = payload["targets"][0] diff --git a/tests/test_strategy_runtime.py b/tests/test_strategy_runtime.py index a041ea9..cc46d01 100644 --- a/tests/test_strategy_runtime.py +++ b/tests/test_strategy_runtime.py @@ -1056,7 +1056,13 @@ def evaluate(self, ctx): logger=lambda _message: None, ) - portfolio_snapshot = SimpleNamespace(total_equity=50000.0) + portfolio_snapshot = PortfolioSnapshot( + as_of=strategy_runtime_module.pd.Timestamp("2026-04-01").to_pydatetime(), + total_equity=50000.0, + buying_power=50000.0, + cash_balance=50000.0, + positions=(), + ) monkeypatch.setattr(strategy_runtime_module, "fetch_portfolio_snapshot", lambda _ib: portfolio_snapshot) def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): @@ -1079,7 +1085,8 @@ def fake_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): assert captured["market_data"]["derived_indicators"]["soxl"]["ma_trend"] == 100.0 assert captured["market_data"]["derived_indicators"]["soxx"]["price"] == 200.0 assert "realized_volatility_10" in captured["market_data"]["derived_indicators"]["soxx"] - assert captured["portfolio"] is portfolio_snapshot + assert captured["portfolio"].total_equity == portfolio_snapshot.total_equity + assert captured["portfolio"].metadata["consecutive_losses"] == 0 assert "pacing_sec" not in captured["runtime_config"] assert captured["runtime_config"]["signal_effective_after_trading_days"] == 1 assert result.metadata["portfolio_total_equity"] == 50000.0 @@ -1137,7 +1144,13 @@ def evaluate(self, ctx): logger=lambda _message: None, ) - portfolio_snapshot = SimpleNamespace(total_equity=50000.0) + portfolio_snapshot = PortfolioSnapshot( + as_of=strategy_runtime_module.pd.Timestamp("2026-04-01").to_pydatetime(), + total_equity=50000.0, + buying_power=50000.0, + cash_balance=50000.0, + positions=(), + ) monkeypatch.setattr(strategy_runtime_module, "fetch_portfolio_snapshot", lambda _ib: portfolio_snapshot) def fake_candle_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): @@ -1171,7 +1184,8 @@ def fake_close_loader(_ib, symbol, duration="2 Y", bar_size="1 day"): assert len(captured["market_data"]["benchmark_history"]) == 220 assert captured["market_data"]["benchmark_history"][0]["high"] == 101.0 - assert captured["portfolio"] is portfolio_snapshot + assert captured["portfolio"].total_equity == portfolio_snapshot.total_equity + assert captured["portfolio"].metadata["consecutive_losses"] == 0 assert "pacing_sec" not in captured["runtime_config"] assert captured["runtime_config"]["signal_effective_after_trading_days"] == 1 assert result.metadata["portfolio_total_equity"] == 50000.0 diff --git a/tests/test_uv_dependency_workflow.py b/tests/test_uv_dependency_workflow.py index 6311c0d..b3125d5 100644 --- a/tests/test_uv_dependency_workflow.py +++ b/tests/test_uv_dependency_workflow.py @@ -21,7 +21,7 @@ def test_ci_docker_and_env_sync_use_uv_lock() -> None: assert lockfile.startswith("version = ") assert "uv sync --frozen --extra test" in ci assert "uv run --no-sync ruff check --exclude external ." in ci - assert "uv run --no-sync python scripts/check_qpk_pin_consistency.py" in ci + assert "uv run --no-sync python external/QuantPlatformKit/scripts/check_qpk_pin_consistency.py" in ci assert "uv sync --frozen --no-dev" in env_sync assert "uv run --no-sync python scripts/build_cloud_run_env_sync_plan.py --json" in env_sync assert "COPY . ." in dockerfile