diff --git a/application/rebalance_service.py b/application/rebalance_service.py index 01ccbd7..4496f14 100644 --- a/application/rebalance_service.py +++ b/application/rebalance_service.py @@ -111,6 +111,30 @@ def _snapshot_session_date(snapshot) -> str: return str(value or "")[:10] +def _is_completed_session_soxl_command(command) -> bool: + execution = command.intent.get("execution") if isinstance(command.intent, dict) else None + if ( + not isinstance(execution, dict) + or str(execution.get("completed_session_date") or "") != command.signal_date + ): + return False + try: + from application.runtime_strategy_adapters import _next_xnys_session_date + + return _next_xnys_session_date(command.signal_date) == command.effective_date + except (TypeError, ValueError): + return False + + +def _is_completed_session_soxl_plan(*, execution: dict, session_date: str, strategy_profile: str) -> bool: + return ( + strategy_profile == "soxl_soxx_trend_income" + and str(execution.get("completed_session_date") or "") == str(execution.get("signal_date") or "") + and str(execution.get("signal_date") or "") < session_date + and str(execution.get("effective_date") or "") == session_date + ) + + def _physical_account_digest(config: LongBridgeRebalanceConfig) -> str: return hashlib.sha256(_resolve_physical_account_id(config=config).encode("utf-8")).hexdigest() @@ -665,47 +689,106 @@ def fetch_replanned_state(): selected_plan = None if live_command_enabled: session_date = _snapshot_session_date(initial_snapshot) - today_commands = tuple(c for c in matching_commands if c.signal_date == session_date) - if not today_commands: - # One current signal is saved for the next session independently of - # consuming yesterday's frozen signal. It never routes directly. - selected_plan = load_plan(current_snapshot=initial_snapshot) - _, _, signal_execution, signal_allocation = selected_plan - if str(signal_execution.get("signal_date") or "") != session_date: - raise RuntimeError("live signal session does not match fresh account session") - if not str(signal_execution.get("effective_date") or "") > session_date: - raise RuntimeError("durable live signal must target a future session") - produced = enqueue_live_execution_command( - enabled=True, dry_run_only=config.dry_run_only, - store=config.execution_command_store, platform="longbridge", - account_scope=str(config.execution_state_account_scope or "unknown"), - strategy_profile=str(config.strategy_profile or "unknown"), - physical_account_id=_resolve_physical_account_id(config=config), - runtime_identity_digest=config.durable_execution_runtime_identity_digest, - execution=signal_execution, allocation=signal_allocation, - strategy_release=getattr(config, "expected_strategy_release", None), + terminal = {ExecutionCommandState.FILLED, ExecutionCommandState.CANCELLED, ExecutionCommandState.REJECTED} + states = {c.command_id: config.execution_command_store.current_state(c) for c in matching_commands} + unresolved_before_enqueue = tuple(c for c in matching_commands if states[c.command_id] not in terminal) + completed_session_path = str(config.strategy_profile or "") == "soxl_soxx_trend_income" + if completed_session_path: + # A same-day consumer is safe only for commands minted with the + # completed-session marker. Old future-dated commands remain + # immutable and are blocked rather than reinterpreted as fresh. + current_session_commands = tuple( + c for c in matching_commands if c.effective_date == session_date + ) + terminal_today = tuple(c for c in current_session_commands if states[c.command_id] in terminal) + unsafe_due = tuple( + c for c in unresolved_before_enqueue + if c.effective_date == session_date and not _is_completed_session_soxl_command(c) ) - command, created = produced - matching_commands = (*matching_commands, command) - today_commands = (command,) - live_command_observation = { - "command_id": command.command_id, "status": "QUEUED" if created else "ALREADY_QUEUED", - "effective_date": command.effective_date, - } + today_commands = tuple( + c for c in unresolved_before_enqueue + if c.effective_date == session_date and _is_completed_session_soxl_command(c) + ) + if terminal_today: + # A command already ended in this session remains its own + # immutable record. Do not evaluate a changed target or mint a + # replacement decision for the same session. + live_command = terminal_today[0] + live_command_blocked = True + elif unsafe_due: + live_command = unsafe_due[0] + live_command_blocked = True + elif not today_commands: + selected_plan = load_plan(current_snapshot=initial_snapshot) + _, _, signal_execution, signal_allocation = selected_plan + if not _is_completed_session_soxl_plan( + execution=signal_execution, + session_date=session_date, + strategy_profile=str(config.strategy_profile or ""), + ): + raise RuntimeError("production SOXL live signal requires a completed session proof") + produced = enqueue_live_execution_command( + enabled=True, dry_run_only=config.dry_run_only, + store=config.execution_command_store, platform="longbridge", + account_scope=str(config.execution_state_account_scope or "unknown"), + strategy_profile=str(config.strategy_profile or "unknown"), + physical_account_id=_resolve_physical_account_id(config=config), + runtime_identity_digest=config.durable_execution_runtime_identity_digest, + execution=signal_execution, allocation=signal_allocation, + strategy_release=getattr(config, "expected_strategy_release", None), + ) + command, created = produced + matching_commands = (*matching_commands, command) + today_commands = (command,) + live_command_observation = { + "command_id": command.command_id, "status": "QUEUED" if created else "ALREADY_QUEUED", + "effective_date": command.effective_date, + } + else: + live_command_observation = { + "command_id": today_commands[0].command_id, "status": "ALREADY_QUEUED", + "effective_date": today_commands[0].effective_date, + } else: - live_command_observation = { - "command_id": today_commands[0].command_id, "status": "ALREADY_QUEUED", - "effective_date": today_commands[0].effective_date, - } - terminal = {ExecutionCommandState.FILLED, ExecutionCommandState.CANCELLED, ExecutionCommandState.REJECTED} + # Existing non-SOXL and V7 queues retain their next-session flow. + today_commands = tuple(c for c in matching_commands if c.signal_date == session_date) + if not today_commands: + selected_plan = load_plan(current_snapshot=initial_snapshot) + _, _, signal_execution, signal_allocation = selected_plan + if str(signal_execution.get("signal_date") or "") != session_date: + raise RuntimeError("live signal session does not match fresh account session") + if not str(signal_execution.get("effective_date") or "") > session_date: + raise RuntimeError("durable live signal must target a future session") + produced = enqueue_live_execution_command( + enabled=True, dry_run_only=config.dry_run_only, + store=config.execution_command_store, platform="longbridge", + account_scope=str(config.execution_state_account_scope or "unknown"), + strategy_profile=str(config.strategy_profile or "unknown"), + physical_account_id=_resolve_physical_account_id(config=config), + runtime_identity_digest=config.durable_execution_runtime_identity_digest, + execution=signal_execution, allocation=signal_allocation, + strategy_release=getattr(config, "expected_strategy_release", None), + ) + command, created = produced + matching_commands = (*matching_commands, command) + today_commands = (command,) + live_command_observation = { + "command_id": command.command_id, "status": "QUEUED" if created else "ALREADY_QUEUED", + "effective_date": command.effective_date, + } + else: + live_command_observation = { + "command_id": today_commands[0].command_id, "status": "ALREADY_QUEUED", + "effective_date": today_commands[0].effective_date, + } states = {c.command_id: config.execution_command_store.current_state(c) for c in matching_commands} unresolved = tuple(c for c in matching_commands if states[c.command_id] not in terminal) due = tuple(c for c in unresolved if c.is_due_on(session_date)) prior_unresolved = tuple(c for c in unresolved if ( states[c.command_id] is not ExecutionCommandState.QUEUED or c.effective_date < session_date )) - if prior_unresolved or len(due) > 1 or len(today_commands) > 1: - live_command = (prior_unresolved or due or today_commands)[0] + if live_command_blocked or prior_unresolved or len(due) > 1 or len(today_commands) > 1: + live_command = live_command or (prior_unresolved or due or today_commands)[0] live_command_blocked = True elif due: live_command = due[0] diff --git a/application/runtime_strategy_adapters.py b/application/runtime_strategy_adapters.py index b579bef..a3b596e 100644 --- a/application/runtime_strategy_adapters.py +++ b/application/runtime_strategy_adapters.py @@ -4,6 +4,7 @@ from collections.abc import Collection, Mapping, Callable from dataclasses import dataclass, replace +from datetime import datetime, timedelta, timezone from typing import Any from quant_platform_kit.common.strategy_plugins import ( @@ -26,6 +27,31 @@ _V7_PROFILE_NAME = "soxl_soxx_core_only_p2_v7_longterm_compounding_cash_reserve" +def _completed_xnys_session_date(now: datetime | None = None) -> str: + """Resolve the latest completed XNYS session, including early closes.""" + import pandas as pd + import pandas_market_calendars as mcal + + instant = now or datetime.now(timezone.utc) + instant = pd.Timestamp(instant) + instant = instant.tz_localize("UTC") if instant.tzinfo is None else instant.tz_convert("UTC") + schedule = mcal.get_calendar("NYSE").schedule( + start_date=(instant - timedelta(days=10)).date(), end_date=instant.date() + ) + completed = schedule[schedule["market_close"] <= instant] + if completed.empty: + raise RuntimeError("no completed XNYS session is available") + return str(completed.index[-1].date()) + + +def _next_xnys_session_date(session_date: str) -> str: + import pandas_market_calendars as mcal + + schedule = mcal.get_calendar("NYSE").schedule(start_date=session_date, end_date=(datetime.fromisoformat(session_date) + timedelta(days=10)).date()) + sessions = [str(index.date()) for index in schedule.index] + return next(date for date in sessions if date > session_date) + + def _typed_execution_candidate(materials: Mapping[str, Any]) -> CandidateRiskIdentity | None: candidate = materials.get("candidate_risk_identity") if isinstance(candidate, CandidateRiskIdentity): @@ -166,6 +192,12 @@ def calculate_strategy_indicators(self, quote_context): if "benchmark_history" in available_inputs or "qqq_history" in available_inputs: return self.broker_adapters.build_price_history(market_data_port, self.benchmark_symbol) trend_ma_window = int(self.strategy_runtime_config.get("trend_ma_window", 150)) + if self.strategy_profile == "soxl_soxx_trend_income": + return self.calculate_rotation_indicators_fn( + quote_context, + trend_window=trend_ma_window, + completed_session_date=_completed_xnys_session_date(), + ) return self.calculate_rotation_indicators_fn(quote_context, trend_window=trend_ma_window) def _market_history_symbols(self) -> tuple[str, ...]: @@ -267,15 +299,38 @@ def resolve_rebalance_plan(self, *, indicators, snapshot=None, account_state=Non snapshot=resolved_snapshot, ) runtime_metadata = dict(getattr(evaluation, "metadata", None) or {}) + completed = indicators.get("completed_session") if isinstance(indicators, Mapping) else None + if ( + self.strategy_profile == "soxl_soxx_trend_income" + and isinstance(completed, Mapping) + and completed.get("date") + ): + signal_date = str(completed["date"]) + annotations = dict(runtime_metadata.get("execution_annotations") or {}) + annotations.update({"signal_date": signal_date, "effective_date": _next_xnys_session_date(signal_date)}) + runtime_metadata["execution_annotations"] = annotations if self.execution_policy is not None: runtime_metadata["longbridge_execution_policy"] = dict(self.execution_policy) - return self.map_strategy_decision_to_plan_fn( + plan = self.map_strategy_decision_to_plan_fn( decision, account_state=resolved_account_state if "account_state" in available_inputs else None, snapshot=resolved_snapshot, strategy_profile=self.strategy_profile, runtime_metadata=runtime_metadata, ) + # This is deliberately an opt-in production SOXL path. The mapper may + # retain strategy diagnostics over runtime annotations, so stamp the + # final mapped execution payload that becomes the immutable command. + if self.strategy_profile == "soxl_soxx_trend_income" and isinstance(completed, Mapping) and completed.get("date"): + execution = plan.get("execution") if isinstance(plan, Mapping) else None + if isinstance(execution, dict): + execution.update({ + "signal_date": signal_date, + "effective_date": _next_xnys_session_date(signal_date), + "completed_session_date": signal_date, + "generated_at": datetime.now(timezone.utc).isoformat(), + }) + return plan def resolve_frozen_rebalance_plan(self, *, allocation, execution, snapshot): """Re-map one stored target decision against a fresh broker snapshot.""" diff --git a/pyproject.toml b/pyproject.toml index e252b57..522b338 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ dependencies = [ "google-cloud-storage", "google-auth", "longport==3.0.23", - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@d51bb79dea8dc5773384d21b06e726620276558c", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@68c51590da8a5097b7de6d75b4ccb6a175318b48", "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@b83ef4b3ae67c47d132ddd660ba3ccc60d474c85", "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@709e5e1cde7841aed538d94eb26b552b46cb7806", ] @@ -61,5 +61,5 @@ include = [ [tool.uv] override-dependencies = [ - "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@d51bb79dea8dc5773384d21b06e726620276558c", + "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@68c51590da8a5097b7de6d75b4ccb6a175318b48", ] diff --git a/qsl.toml b/qsl.toml index 763e32e..4b12bc4 100644 --- a/qsl.toml +++ b/qsl.toml @@ -5,7 +5,7 @@ upgrade_ring = "ring_d" allow_legacy = false [qsl.requires] -quant_platform_kit = "d51bb79dea8dc5773384d21b06e726620276558c" +quant_platform_kit = "68c51590da8a5097b7de6d75b4ccb6a175318b48" us_equity_strategies = "c31f28484b8489a1ecda8e595a6d2afd3bce8185" hk_equity_strategies = "709e5e1cde7841aed538d94eb26b552b46cb7806" diff --git a/tests/test_rebalance_service.py b/tests/test_rebalance_service.py index cff5e61..09522f0 100644 --- a/tests/test_rebalance_service.py +++ b/tests/test_rebalance_service.py @@ -1635,7 +1635,7 @@ def frozen_plan(*, allocation, execution, snapshot): config = LongBridgeRebalanceConfig( limit_sell_discount=0.995, limit_buy_premium=1.005, separator="-", translator=build_translator("en"), with_prefix=lambda message: message, - strategy_profile="soxl_soxx_trend_income", execution_state_account_scope="SG", + strategy_profile="russell_top50_leader_rotation", execution_state_account_scope="SG", physical_account_id="lb-sg-001", dry_run_only=False, notify_no_trade_cycles=False, execution_dedup_enabled=True, execution_state_store=marker_store, durable_execution_command_live_enabled=True, execution_command_store=command_store, @@ -1719,7 +1719,7 @@ def test_terminal_stale_live_command_is_ignored_before_current_cycle(self): ) command_store = ExecutionCommandStore(local_dir=self.enterContext(TemporaryDirectory())) stale = build_live_execution_command( - platform="longbridge", account_scope="SG", strategy_profile="soxl_soxx_trend_income", + platform="longbridge", account_scope="SG", strategy_profile="russell_top50_leader_rotation", physical_account_id="lb-sg-001", runtime_identity_digest="b" * 64, execution={**plan["execution"], "signal_date": "2026-04-20", "effective_date": "2026-04-21"}, allocation=plan["allocation"], @@ -1742,7 +1742,7 @@ def test_terminal_stale_live_command_is_ignored_before_current_cycle(self): config = LongBridgeRebalanceConfig( limit_sell_discount=0.995, limit_buy_premium=1.005, separator="-", translator=build_translator("en"), with_prefix=lambda message: message, - strategy_profile="soxl_soxx_trend_income", execution_state_account_scope="SG", + strategy_profile="russell_top50_leader_rotation", execution_state_account_scope="SG", physical_account_id="lb-sg-001", dry_run_only=False, notify_no_trade_cycles=False, durable_execution_command_live_enabled=True, execution_command_store=command_store, durable_live_execution_session_authorized=True, @@ -1772,7 +1772,7 @@ def test_same_day_terminal_live_command_prevents_duplicate_command_or_submit(sel ) command_store = ExecutionCommandStore(local_dir=self.enterContext(TemporaryDirectory())) terminal = build_live_execution_command( - platform="longbridge", account_scope="SG", strategy_profile="soxl_soxx_trend_income", + platform="longbridge", account_scope="SG", strategy_profile="russell_top50_leader_rotation", physical_account_id="lb-sg-001", runtime_identity_digest="b" * 64, execution=plan["execution"], allocation=plan["allocation"], ) @@ -1794,7 +1794,7 @@ def test_same_day_terminal_live_command_prevents_duplicate_command_or_submit(sel config = LongBridgeRebalanceConfig( limit_sell_discount=0.995, limit_buy_premium=1.005, separator="-", translator=build_translator("en"), with_prefix=lambda message: message, - strategy_profile="soxl_soxx_trend_income", execution_state_account_scope="SG", + strategy_profile="russell_top50_leader_rotation", execution_state_account_scope="SG", physical_account_id="lb-sg-001", dry_run_only=False, notify_no_trade_cycles=False, durable_execution_command_live_enabled=True, execution_command_store=command_store, durable_live_execution_session_authorized=True, @@ -1824,12 +1824,12 @@ def test_queued_stale_live_command_blocks_without_store_or_broker_writes(self): ) command_store = ExecutionCommandStore(local_dir=self.enterContext(TemporaryDirectory())) stale = build_live_execution_command( - platform="longbridge", account_scope="SG", strategy_profile="soxl_soxx_trend_income", + platform="longbridge", account_scope="SG", strategy_profile="russell_top50_leader_rotation", physical_account_id="lb-sg-001", runtime_identity_digest="b" * 64, execution=plan["execution"], allocation=plan["allocation"], ) valid = build_live_execution_command( - platform="longbridge", account_scope="SG", strategy_profile="soxl_soxx_trend_income", + platform="longbridge", account_scope="SG", strategy_profile="russell_top50_leader_rotation", physical_account_id="lb-sg-001", runtime_identity_digest="a" * 64, execution=plan["execution"], allocation=plan["allocation"], ) @@ -1852,7 +1852,7 @@ def test_queued_stale_live_command_blocks_without_store_or_broker_writes(self): config = LongBridgeRebalanceConfig( limit_sell_discount=0.995, limit_buy_premium=1.005, separator="-", translator=build_translator("en"), with_prefix=lambda message: message, - strategy_profile="soxl_soxx_trend_income", execution_state_account_scope="SG", + strategy_profile="russell_top50_leader_rotation", execution_state_account_scope="SG", physical_account_id="lb-sg-001", dry_run_only=False, notify_no_trade_cycles=False, durable_execution_command_live_enabled=True, execution_command_store=command_store, durable_live_execution_session_authorized=True, @@ -1876,6 +1876,103 @@ def test_queued_stale_live_command_blocks_without_store_or_broker_writes(self): self.assertEqual(command_store.events(stale), ()) self.assertEqual(len(rebalance_service.list_live_execution_commands(command_store)), 2) + def test_completed_session_soxl_consumes_today_once_and_blocks_unmarked_old_queue(self): + from application.durable_execution_commands import build_live_execution_command + + plan = _build_plan( + strategy_symbols=("SOXL",), risk_symbols=("SOXL",), + targets={"SOXL": 400.0}, market_values={"SOXL": 0.0}, + sellable_quantities={"SOXL": 0}, quantities={"SOXL": 0}, + current_min_trade=10.0, trade_threshold_value=10.0, + investable_cash=500.0, available_cash=500.0, total_strategy_equity=500.0, + market_status="Risk on", deploy_ratio_text="70.0%", income_ratio_text="0.0%", + income_locked_ratio_text="0.0%", signal_message="SOXL target", + portfolio_rows=(("SOXL",),), signal_date="2026-07-02", effective_date="2026-07-06", + ) + plan["execution"].update(completed_session_date="2026-07-02", generated_at="2026-07-06T19:45:00+00:00") + store = ExecutionCommandStore(local_dir=self.enterContext(TemporaryDirectory())) + markers = ExecutionMarkerStore(local_dir=self.enterContext(TemporaryDirectory())) + orders, plans = [], [] + runtime = LongBridgeRebalanceRuntime( + bootstrap=lambda: ("quote", "trade", {"trend": "ok"}), + resolve_rebalance_plan=lambda **_kwargs: (plans.append(True), plan)[1], + resolve_frozen_rebalance_plan=lambda *, allocation, execution, snapshot: { + **plan, "allocation": dict(allocation), "execution": dict(execution), + }, + market_data_port_factory=lambda _context: CallableMarketDataPort( + quote_loader=lambda symbol: QuoteSnapshot(symbol=symbol, as_of="2026-07-06", last_price=100.0)), + estimate_max_purchase_quantity=lambda *_args, **_kwargs: 5, + notifications=CallableNotificationPort(lambda _message: None), notify_issue=lambda *_args: None, + portfolio_port_factory=lambda *_contexts: CallablePortfolioPort( + lambda: replace(_build_snapshot(plan), as_of="2026-07-06")), + execution_port_factory=lambda _context: CallableExecutionPort( + lambda intent: (orders.append(intent), ExecutionReport( + symbol=intent.symbol, side=intent.side, quantity=intent.quantity, + status="accepted", broker_order_id="same-day"))[1]), + ) + config = LongBridgeRebalanceConfig( + limit_sell_discount=0.995, limit_buy_premium=1.005, separator="-", + translator=build_translator("en"), with_prefix=lambda message: message, + strategy_profile="soxl_soxx_trend_income", execution_state_account_scope="SG", + physical_account_id="lb-sg-001", dry_run_only=False, notify_no_trade_cycles=False, + execution_dedup_enabled=True, execution_state_store=markers, + durable_execution_command_live_enabled=True, execution_command_store=store, + durable_live_execution_session_authorized=True, durable_execution_runtime_identity_digest="a" * 64, + ) + + first = rebalance_service.run_strategy(runtime=runtime, config=config) + second = rebalance_service.run_strategy(runtime=runtime, config=config) + + self.assertTrue(first.action_done) + self.assertFalse(second.action_done) + self.assertEqual(len(plans), 1) + self.assertEqual(len(orders), 1) + command = rebalance_service.list_live_execution_commands(store)[0] + self.assertEqual((command.signal_date, command.effective_date), ("2026-07-02", "2026-07-06")) + + terminal_store = ExecutionCommandStore(local_dir=self.enterContext(TemporaryDirectory())) + terminal = build_live_execution_command( + platform="longbridge", account_scope="SG", strategy_profile="soxl_soxx_trend_income", + physical_account_id="lb-sg-001", runtime_identity_digest="a" * 64, + execution=plan["execution"], allocation=plan["allocation"], + ) + self.assertTrue(terminal_store.enqueue(terminal)) + terminal_store.append_event(terminal, next_state=ExecutionCommandState.CANCELLED) + plan["allocation"]["targets"] = {"SOXL": 999.0} # A changed target must not create a same-day replacement. + terminal_reentry = rebalance_service.run_strategy( + runtime=runtime, config=replace(config, execution_command_store=terminal_store) + ) + self.assertFalse(terminal_reentry.action_done) + self.assertEqual(len(plans), 1) + self.assertEqual(len(orders), 1) + self.assertEqual(len(rebalance_service.list_live_execution_commands(terminal_store)), 1) + self.assertEqual(terminal.intent["allocation"]["targets"], {"SOXL": 400.0}) + self.assertIs(terminal_store.current_state(terminal), ExecutionCommandState.CANCELLED) + + stale_marker = build_live_execution_command( + platform="longbridge", account_scope="SG", strategy_profile="soxl_soxx_trend_income", + physical_account_id="lb-sg-001", runtime_identity_digest="a" * 64, + execution={**terminal.intent["execution"], "effective_date": "2026-07-07"}, + allocation=terminal.intent["allocation"], + ) + self.assertFalse(rebalance_service._is_completed_session_soxl_command(stale_marker)) + + old_store = ExecutionCommandStore(local_dir=self.enterContext(TemporaryDirectory())) + old = build_live_execution_command( + platform="longbridge", account_scope="SG", strategy_profile="soxl_soxx_trend_income", + physical_account_id="lb-sg-001", runtime_identity_digest="a" * 64, + execution={key: value for key, value in plan["execution"].items() if key != "completed_session_date"}, + allocation=plan["allocation"], + ) + self.assertTrue(old_store.enqueue(old)) + blocked = rebalance_service.run_strategy( + runtime=runtime, config=replace(config, execution_command_store=old_store) + ) + self.assertFalse(blocked.action_done) + self.assertTrue(blocked.execution["direct_live_routing_blocked"]) + self.assertEqual(len(plans), 1) + self.assertEqual(len(orders), 1) + def test_live_order_detail_normalizes_real_sdk_enum_and_checks_identity(self): from application.longbridge_execution import fetch_live_order_status from longport.openapi import OrderStatus diff --git a/tests/test_runtime_strategy_adapters.py b/tests/test_runtime_strategy_adapters.py index 1a6dedc..9d93ee7 100644 --- a/tests/test_runtime_strategy_adapters.py +++ b/tests/test_runtime_strategy_adapters.py @@ -121,8 +121,8 @@ def load_market_history(_broker_client, symbol): def test_runtime_strategy_adapters_fall_back_to_rotation_indicators(): observed = {} - def fake_rotation_indicators(quote_context, *, trend_window): - observed["rotation_call"] = (quote_context, trend_window) + def fake_rotation_indicators(quote_context, *, trend_window, completed_session_date): + observed["rotation_call"] = (quote_context, trend_window, completed_session_date) return {"rotation": True} adapters = build_runtime_strategy_adapters( @@ -141,7 +141,8 @@ def fake_rotation_indicators(quote_context, *, trend_window): result = adapters.calculate_strategy_indicators("quote-context") - assert observed["rotation_call"] == ("quote-context", 180) + assert observed["rotation_call"][:2] == ("quote-context", 180) + assert observed["rotation_call"][2] assert result == {"rotation": True} @@ -517,3 +518,125 @@ def test_runtime_strategy_adapters_builds_escalated_plugin_alert_message(): assert "mode=shadow" in alerts[0].body assert "would_trade=" not in alerts[0].body assert "source=" not in alerts[0].body + + +def test_completed_session_soxl_overrides_mapper_annotations_on_final_plan(monkeypatch): + from datetime import datetime, timezone + from decision_mapper import map_strategy_decision_to_plan + from quant_platform_kit.common.strategy_contracts import PositionTarget, StrategyDecision + + class BrokerAdapters: + def build_account_state_from_snapshot(self, _snapshot): + return { + "available_cash": 1_000.0, + "market_values": {"SOXL": 0.0}, + "quantities": {"SOXL": 0.0}, + "sellable_quantities": {"SOXL": 0.0}, + "total_strategy_equity": 1_000.0, + } + + decision = StrategyDecision( + positions=(PositionTarget(symbol="SOXL", target_value=500.0),), + diagnostics={ + "execution_annotations": { + "signal_date": "1999-01-01", + "effective_date": "1999-01-02", + "execution_timing_contract": "next_trading_day", + "trade_threshold_value": 10.0, + "current_min_trade": 10.0, + "investable_cash": 1_000.0, + }, + }, + ) + runtime = SimpleNamespace(evaluate=lambda **_kwargs: SimpleNamespace(decision=decision, metadata={})) + adapters = build_runtime_strategy_adapters( + strategy_runtime=runtime, + strategy_profile="soxl_soxx_trend_income", + strategy_runtime_config={}, + available_inputs=("portfolio_snapshot", "account_state"), + benchmark_symbol="SOXX", + signal_text_fn=str, + translator=lambda key, **_kwargs: key, + broker_adapters=BrokerAdapters(), + calculate_rotation_indicators_fn=lambda *_args, **_kwargs: {}, + build_strategy_evaluation_inputs_fn=lambda **kwargs: kwargs, + map_strategy_decision_to_plan_fn=map_strategy_decision_to_plan, + ) + monkeypatch.setattr( + "application.runtime_strategy_adapters._next_xnys_session_date", + lambda value: "2026-07-06" if value == "2026-07-02" else None, + ) + before = datetime.now(timezone.utc) + plan = adapters.resolve_rebalance_plan( + indicators={"completed_session": {"date": "2026-07-02"}}, + snapshot=object(), + ) + after = datetime.now(timezone.utc) + + assert plan["execution"]["signal_date"] == "2026-07-02" + assert plan["execution"]["effective_date"] == "2026-07-06" + assert plan["execution"]["completed_session_date"] == "2026-07-02" + assert before <= datetime.fromisoformat(plan["execution"]["generated_at"]) <= after + + +def test_completed_session_does_not_change_non_soxl_mapper_output(): + observed = {} + adapters = build_runtime_strategy_adapters( + strategy_runtime=SimpleNamespace(evaluate=lambda **_kwargs: SimpleNamespace(decision="decision", metadata={})), + strategy_profile="russell_top50_leader_rotation", + strategy_runtime_config={}, + available_inputs=("portfolio_snapshot",), + benchmark_symbol="SOXX", + signal_text_fn=str, + translator=lambda key, **_kwargs: key, + broker_adapters=SimpleNamespace(), + calculate_rotation_indicators_fn=lambda *_args, **_kwargs: {}, + build_strategy_evaluation_inputs_fn=lambda **kwargs: kwargs, + map_strategy_decision_to_plan_fn=lambda _decision, **kwargs: observed.setdefault( + "plan", {"execution": {"signal_date": "2026-07-06", "effective_date": "2026-07-07"}} + ), + ) + + plan = adapters.resolve_rebalance_plan( + indicators={"completed_session": {"date": "2026-07-02"}}, snapshot=object() + ) + + assert plan["execution"] == {"signal_date": "2026-07-06", "effective_date": "2026-07-07"} + + +def test_completed_xnys_session_respects_close_holiday_and_weekend(): + from datetime import datetime, timezone + from application.runtime_strategy_adapters import _completed_xnys_session_date + + # 2026-11-27 is the XNYS early-close session after Thanksgiving. + # At the production 15:45 New York schedule on the Monday after the + # July 3 holiday, Friday's completed session remains July 2. + assert _completed_xnys_session_date(datetime(2026, 7, 6, 19, 45, tzinfo=timezone.utc)) == "2026-07-02" + assert _completed_xnys_session_date(datetime(2026, 11, 27, 17, 59, tzinfo=timezone.utc)) == "2026-11-25" + assert _completed_xnys_session_date(datetime(2026, 11, 27, 18, 1, tzinfo=timezone.utc)) == "2026-11-27" + assert _completed_xnys_session_date(datetime(2026, 11, 29, 20, 0, tzinfo=timezone.utc)) == "2026-11-27" + + +def test_v7_derived_indicators_fallback_keeps_legacy_rotation_signature(): + observed = {} + + def legacy_rotation_indicators(quote_context, *, trend_window): + observed["call"] = (quote_context, trend_window) + return {"derived": True} + + adapters = build_runtime_strategy_adapters( + strategy_runtime=SimpleNamespace(evaluate=lambda **_kwargs: None), + strategy_profile=V7_PROFILE, + strategy_runtime_config={"trend_ma_window": 150}, + available_inputs=("derived_indicators", "portfolio_snapshot"), + benchmark_symbol="SOXX", + signal_text_fn=str, + translator=lambda key, **_kwargs: key, + broker_adapters=SimpleNamespace(), + calculate_rotation_indicators_fn=legacy_rotation_indicators, + build_strategy_evaluation_inputs_fn=lambda **kwargs: kwargs, + map_strategy_decision_to_plan_fn=lambda *_args, **_kwargs: {}, + ) + + assert adapters.calculate_strategy_indicators("v7-quote") == {"derived": True} + assert observed["call"] == ("v7-quote", 150) diff --git a/uv.lock b/uv.lock index 2be9688..1ad702a 100644 --- a/uv.lock +++ b/uv.lock @@ -17,7 +17,7 @@ resolution-markers = [ ] [manifest] -overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=d51bb79dea8dc5773384d21b06e726620276558c" }] +overrides = [{ name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=68c51590da8a5097b7de6d75b4ccb6a175318b48" }] [[package]] name = "blinker" @@ -727,7 +727,7 @@ requires-dist = [ { name = "pytest", marker = "extra == 'test'" }, { name = "pytest-cov", marker = "extra == 'test'" }, { name = "pytz" }, - { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=d51bb79dea8dc5773384d21b06e726620276558c" }, + { name = "quant-platform-kit", git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=68c51590da8a5097b7de6d75b4ccb6a175318b48" }, { name = "requests" }, { name = "ruff", marker = "extra == 'test'" }, { name = "us-equity-strategies", git = "https://github.com/QuantStrategyLab/UsEquityStrategies.git?rev=b83ef4b3ae67c47d132ddd660ba3ccc60d474c85" }, @@ -1210,7 +1210,7 @@ wheels = [ [[package]] name = "quant-platform-kit" version = "1.0.0" -source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=d51bb79dea8dc5773384d21b06e726620276558c#d51bb79dea8dc5773384d21b06e726620276558c" } +source = { git = "https://github.com/QuantStrategyLab/QuantPlatformKit.git?rev=68c51590da8a5097b7de6d75b4ccb6a175318b48#68c51590da8a5097b7de6d75b4ccb6a175318b48" } [[package]] name = "requests"