Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@ English | [简体中文](README.zh-CN.md)

> Investing involves risk. This project does not provide investment advice and is for education, research, and engineering review only.

The pinned HK strategy policy has no selectable profiles. New selection of `hk_global_etf_tactical_rotation` and `hk_low_vol_dividend_quality_snapshot` is rejected, including dry-run switch plans. Historical catalog metadata and adapters remain available. US strategy pins and existing deployment configuration are unchanged.

## QSL architecture role

- **Layer**: `runtime-platform`.
Expand Down
2 changes: 2 additions & 0 deletions README.zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@ LongBridge 兼容的券商连接执行,并在真实下单前经过区域化的

> 投资有风险。本项目不构成投资建议,仅用于学习、研究和工程审阅。

固定依赖中的港股策略政策已无可选策略。`hk_global_etf_tactical_rotation` 和 `hk_low_vol_dividend_quality_snapshot` 均拒绝新选择,包括 dry-run 切换计划;历史策略元数据和适配器仍保留。美股策略版本及现有部署配置不变。

## QSL 架构角色

- **层级**:`执行平台`。
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ dependencies = [
"longport==3.0.23",
"quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@350dd38ece0952beef096893a7dcf8ae3871609a",
"us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@8412d8338879f335ef8830678c2733e02e53e7c7",
"hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@cc3045689ae9d76eb97b9e1aa3b8fb6288ee1bcc",
"hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@fe0c32e09d454ca4388c3372bf3874097b1bbaf0",
]

[project.optional-dependencies]
Expand Down
12 changes: 8 additions & 4 deletions scripts/run_hk_notification_canary.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,8 @@
_validate_image_only_source,
)
from application.runtime_target_manifest import load_runtime_target_manifest
from strategy_registry import LONGBRIDGE_PLATFORM, resolve_strategy_definition
from hk_equity_strategies import get_strategy_definition
from strategy_registry import LONGBRIDGE_PLATFORM


BASE_APPLICATION_SHA = "922f338fea7c46391b50bb8316ac88154596916f"
Expand Down Expand Up @@ -486,9 +487,12 @@ def _runtime_target(env: Mapping[str, str], *, service: str) -> Mapping[str, Any
if not isinstance(profile, str) or not profile or profile.strip() != profile:
raise _stop("runtime_target_mismatch")
try:
canonical_profile = resolve_strategy_definition(
profile, platform_id=LONGBRIDGE_PLATFORM
).profile
# Inspect an already-disabled historical target, not a new selection.
# The disabled/paper/dry-run and source/configuration gates still apply.
definition = get_strategy_definition(profile)
if LONGBRIDGE_PLATFORM not in definition.supported_platforms:
raise ValueError("Unsupported historical HK profile")
canonical_profile = definition.profile
except (TypeError, ValueError):
raise _stop("runtime_target_mismatch") from None
selector = target.get("account_selector")
Expand Down
22 changes: 21 additions & 1 deletion tests/test_hk_notification_canary.py
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ def test_preconditions_accept_both_existing_hk_selector_shapes(selector):
]


def test_preconditions_accept_legal_native_profile_when_old_and_service_match():
def test_preconditions_accept_retired_native_profile_when_old_and_service_match():
profile = "hk_low_vol_dividend_quality_snapshot"
selected, contract, _ = canary.validate_canary_preconditions(
service=_service(profile=profile),
Expand Down Expand Up @@ -232,6 +232,8 @@ def test_preconditions_reject_old_revision_profile_or_config_drift(drift):
[
("", ""),
("unsupported_profile", "unsupported_profile"),
("global_etf_rotation", "global_etf_rotation"),
("hk_equity_combo", "hk_equity_combo"),
("hk_low_vol_dividend_quality_snapshot", DEFAULT_PROFILE),
],
)
Expand Down Expand Up @@ -1120,3 +1122,21 @@ def test_pause_failure_does_not_mask_terminal_observation_failure(
assert state["run_count"] == 1
assert state["uri_restores"] == 0
assert state["traffic_restores"] == 0


@pytest.mark.parametrize("profile", [DEFAULT_PROFILE, "hk_low_vol_dividend_quality_snapshot"])
@pytest.mark.parametrize("unsafe_field", ["enabled", "live", "target_not_dry_run", "environment_not_dry_run"])
def test_retired_target_inspection_preserves_disabled_paper_guards(profile, unsafe_field):
env = _runtime_env("HK", profile)
target = json.loads(env["RUNTIME_TARGET_JSON"])
if unsafe_field == "enabled":
env["RUNTIME_TARGET_ENABLED"] = "true"
elif unsafe_field == "live":
target["execution_mode"] = "live"
elif unsafe_field == "target_not_dry_run":
target["dry_run_only"] = False
else:
env["LONGBRIDGE_DRY_RUN_ONLY"] = "false"
env["RUNTIME_TARGET_JSON"] = json.dumps(target)
with pytest.raises(canary.CanaryError, match="runtime_target_mismatch"):
canary._runtime_target(env, service=SERVICE)
183 changes: 26 additions & 157 deletions tests/test_runtime_config_support.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,14 +65,10 @@
}
)
OPTIONAL_LONGBRIDGE_PROFILES = frozenset({"global_etf_confidence_vol_gate"})
HK_RUNTIME_ENABLED_PROFILES = frozenset(
HK_DISABLED_PROFILES = frozenset(
{
"hk_global_etf_tactical_rotation",
"hk_low_vol_dividend_quality_snapshot",
}
)
HK_DISABLED_PROFILES = frozenset(
{
"hk_dividend_gold_defensive_rotation",
"hk_blue_chip_leader_rotation",
"hk_index_mean_reversion",
Expand All @@ -83,7 +79,7 @@

def expected_longbridge_enabled_profiles(actual_profiles) -> frozenset[str]:
actual = frozenset(actual_profiles)
return BASE_LONGBRIDGE_PROFILES | HK_RUNTIME_ENABLED_PROFILES | (OPTIONAL_LONGBRIDGE_PROFILES & actual)
return BASE_LONGBRIDGE_PROFILES | (OPTIONAL_LONGBRIDGE_PROFILES & actual)


def expected_longbridge_profiles(actual_profiles) -> frozenset[str]:
Expand Down Expand Up @@ -758,18 +754,6 @@ def test_platform_profile_status_matrix_matches_current_longbridge_rollout(self)
by_profile["russell_top50_leader_rotation"]["display_name"],
"Russell Top50 Leader Rotation",
)
self.assertEqual(
by_profile["hk_global_etf_tactical_rotation"],
{
"canonical_profile": "hk_global_etf_tactical_rotation",
"display_name": "HK Global ETF Tactical Rotation",
"display_name_zh": "港股ETF战术轮动",
"domain": "hk_equity",
"eligible": True,
"enabled": True,
"platform": "longbridge",
},
)
for profile in HK_DISABLED_PROFILES:
self.assertNotIn(profile, by_profile)

Expand Down Expand Up @@ -805,31 +789,15 @@ def test_rejects_disabled_hk_profiles(self):
with self.assertRaisesRegex(ValueError, "Unsupported STRATEGY_PROFILE"):
load_platform_runtime_settings(project_id_resolver=lambda: "project-1")

def test_accepts_runtime_enabled_hk_profiles(self):
expected_names = {
"hk_global_etf_tactical_rotation": "HK Global ETF Tactical Rotation",
}
for profile, display_name in expected_names.items():
def test_rejects_retired_hk_profiles_without_snapshot_inputs(self):
for profile in ("hk_global_etf_tactical_rotation", "hk_low_vol_dividend_quality_snapshot"):
with self.subTest(profile=profile):
with patch.dict(
os.environ,
{
"RUNTIME_TARGET_JSON": runtime_target_json(profile),
"ACCOUNT_REGION": "HK",
},
clear=True,
):
settings = load_platform_runtime_settings(project_id_resolver=lambda: "project-1")

self.assertEqual(settings.strategy_profile, profile)
self.assertEqual(settings.strategy_display_name, display_name)
self.assertEqual(settings.strategy_domain, "hk_equity")
self.assertIsNotNone(settings.strategy_metadata)
self.assertEqual(settings.market, HK_MARKET)
self.assertEqual(settings.market_calendar, HK_MARKET_CALENDAR)
self.assertEqual(settings.market_timezone, HK_MARKET_TIMEZONE)
self.assertEqual(settings.symbol_suffix, HK_SYMBOL_SUFFIX)
self.assertEqual(settings.trading_currency, HK_TRADING_CURRENCY)
with patch.dict(os.environ, {
"RUNTIME_TARGET_JSON": runtime_target_json(profile),
"ACCOUNT_REGION": "HK",
}, clear=True):
with self.assertRaisesRegex(ValueError, "Unsupported STRATEGY_PROFILE"):
load_platform_runtime_settings(project_id_resolver=lambda: "project-1")

def test_derives_feature_snapshot_paths_from_artifact_root(self):
with TemporaryDirectory() as tmp_dir:
Expand Down Expand Up @@ -905,17 +873,8 @@ def test_print_strategy_profile_status_json_matches_registry(self):
"hk_dividend_gold_defensive_rotation",
):
self.assertNotIn(profile, by_profile)
for profile in HK_RUNTIME_ENABLED_PROFILES - {"hk_low_vol_dividend_quality_snapshot"}:
self.assertEqual(by_profile[profile]["profile_group"], "direct_runtime_inputs")
self.assertEqual(by_profile[profile]["input_mode"], "market_history")
self.assertFalse(by_profile[profile]["requires_snapshot_artifacts"])
self.assertFalse(by_profile[profile]["requires_snapshot_manifest_path"])
self.assertFalse(by_profile[profile]["requires_strategy_config_path"])
self.assertEqual(by_profile["hk_low_vol_dividend_quality_snapshot"]["profile_group"], "snapshot_backed")
self.assertEqual(by_profile["hk_low_vol_dividend_quality_snapshot"]["input_mode"], "feature_snapshot")
self.assertTrue(by_profile["hk_low_vol_dividend_quality_snapshot"]["requires_snapshot_artifacts"])
self.assertTrue(by_profile["hk_low_vol_dividend_quality_snapshot"]["requires_snapshot_manifest_path"])
self.assertFalse(by_profile["hk_low_vol_dividend_quality_snapshot"]["requires_strategy_config_path"])
for profile in HK_DISABLED_PROFILES:
self.assertNotIn(profile, by_profile)

def test_print_strategy_profile_status_table_contains_expected_headers(self):
result = subprocess.run(
Expand All @@ -933,10 +892,10 @@ def test_print_strategy_profile_status_table_contains_expected_headers(self):
self.assertIn("requires_snapshot_artifacts", result.stdout)
self.assertIn("soxl_soxx_trend_income", result.stdout)
self.assertIn("global_etf_rotation", result.stdout)
self.assertIn("hk_global_etf_tactical_rotation", result.stdout)
self.assertNotIn("hk_global_etf_tactical_rotation", result.stdout)
self.assertNotIn("hk_dividend_gold_defensive_rotation", result.stdout)
self.assertIn("Global ETF Rotation", result.stdout)
self.assertIn("HK Global ETF Tactical Rotation", result.stdout)
self.assertNotIn("HK Global ETF Tactical Rotation", result.stdout)
self.assertNotIn("HK Dividend-Gold Defensive Rotation", result.stdout)
self.assertIn("Russell Top50 Leader Rotation", result.stdout)
self.assertIn("罗素Top50领涨", result.stdout)
Expand Down Expand Up @@ -1004,78 +963,14 @@ def test_print_strategy_switch_env_plan_for_global_etf_rotation(self):
self.assertIn("LONGBRIDGE_TRADING_CURRENCY", plan["optional_env"])
self.assertNotIn("LONGBRIDGE_FEATURE_SNAPSHOT_PATH", plan["remove_if_present"])

def test_print_strategy_switch_env_plan_for_hk_global_etf_dry_run(self):
def test_print_strategy_switch_env_plan_rejects_retired_hk_global_etf_dry_run(self):
result = subprocess.run(
[
sys.executable,
str(SWITCH_PLAN_SCRIPT_PATH),
"--profile",
"hk_global_etf_tactical_rotation",
"--account-region",
"hk",
"--dry-run-only",
"--deployment-selector",
"hk-verify",
"--account-scope",
"hk-verify",
"--service-name",
"longbridge-quant-hk-verify-service",
"--json",
],
check=True,
capture_output=True,
text=True,
)

plan = json.loads(result.stdout)
self.assertEqual(plan["platform"], "longbridge")
self.assertEqual(plan["canonical_profile"], "hk_global_etf_tactical_rotation")
self.assertEqual(plan["domain"], HK_EQUITY_DOMAIN)
self.assertEqual(plan["set_env"]["ACCOUNT_REGION"], "HK")
self.assertEqual(plan["set_env"]["ACCOUNT_PREFIX"], "HK")
self.assertEqual(plan["set_env"]["LONGBRIDGE_DRY_RUN_ONLY"], "true")
self.assertEqual(plan["set_env"]["LONGBRIDGE_MARKET"], HK_MARKET)
self.assertEqual(plan["set_env"]["LONGBRIDGE_MARKET_CALENDAR"], HK_MARKET_CALENDAR)
self.assertEqual(plan["set_env"]["LONGBRIDGE_MARKET_TIMEZONE"], HK_MARKET_TIMEZONE)
self.assertEqual(plan["set_env"]["LONGBRIDGE_SYMBOL_SUFFIX"], HK_SYMBOL_SUFFIX)
self.assertEqual(plan["set_env"]["LONGBRIDGE_TRADING_CURRENCY"], HK_TRADING_CURRENCY)
self.assertTrue(plan["runtime_target"]["dry_run_only"])
self.assertEqual(plan["runtime_target"]["execution_mode"], "paper")
self.assertEqual(plan["runtime_target"]["deployment_selector"], "hk-verify")
self.assertEqual(plan["runtime_target"]["account_scope"], "hk-verify")
self.assertEqual(plan["runtime_target"]["service_name"], "longbridge-quant-hk-verify-service")
runtime_target_env = json.loads(plan["set_env"]["RUNTIME_TARGET_JSON"])
self.assertTrue(runtime_target_env["dry_run_only"])
self.assertEqual(runtime_target_env["execution_mode"], "paper")
self.assertEqual(plan["profile_group"], "direct_runtime_inputs")
self.assertEqual(plan["input_mode"], "market_history")
self.assertFalse(plan["requires_snapshot_artifacts"])
self.assertIn("LONGBRIDGE_FEATURE_SNAPSHOT_PATH", plan["remove_if_present"])
self.assertTrue(plan["dry_run_plan"]["dry_run_only"])
self.assertTrue(plan["dry_run_plan"]["verify_only"])
self.assertEqual(
plan["dry_run_plan"]["workflow_dispatch"],
{
"workflow": "sync-cloud-run-env.yml",
"target": "hk-verify",
"cloud_run_service": "longbridge-quant-hk-verify-service",
"deploy_image": True,
"sync_env": True,
},
)
self.assertTrue(any("lot-size" in check for check in plan["dry_run_plan"]["checks"]))
self.assertFalse(
any(
"Cloud Run" in action and "deploy" in action
for action in plan["dry_run_plan"]["blocked_actions"]
)
)
self.assertTrue(
any(
"live" in action and "orders" in action
for action in plan["dry_run_plan"]["blocked_actions"]
)
[sys.executable, str(SWITCH_PLAN_SCRIPT_PATH), "--profile",
"hk_global_etf_tactical_rotation", "--dry-run-only", "--json"],
check=False, capture_output=True, text=True,
)
self.assertNotEqual(result.returncode, 0)
self.assertIn("Unsupported STRATEGY_PROFILE", result.stderr)

def test_print_strategy_switch_env_plan_rejects_retired_russell_defensive(self):
result = subprocess.run(
Expand Down Expand Up @@ -1145,40 +1040,14 @@ def test_print_strategy_switch_env_plan_for_mega_cap_top50_balanced(self):
"russell_top50_leader_rotation_feature_snapshot_latest.csv",
)

def test_print_strategy_switch_env_plan_for_hk_low_vol_dividend_quality_snapshot(self):
def test_print_strategy_switch_env_plan_rejects_retired_hk_low_vol_dividend_quality_snapshot(self):
result = subprocess.run(
[
sys.executable,
str(SWITCH_PLAN_SCRIPT_PATH),
"--profile",
"hk_low_vol_dividend_quality_snapshot",
"--account-region",
"hk",
"--dry-run-only",
"--json",
],
check=True,
capture_output=True,
text=True,
)

plan = json.loads(result.stdout)
self.assertEqual(plan["canonical_profile"], "hk_low_vol_dividend_quality_snapshot")
self.assertTrue(plan["enabled"])
self.assertEqual(plan["profile_group"], "snapshot_backed")
self.assertEqual(plan["input_mode"], "feature_snapshot")
self.assertEqual(plan["snapshot_contract_version"], "hk_low_vol_dividend_quality_snapshot.factor_snapshot.v1")
self.assertEqual(plan["set_env"]["LONGBRIDGE_DRY_RUN_ONLY"], "true")
self.assertEqual(plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_PATH"], "<required>")
self.assertEqual(plan["set_env"]["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"], "<required>")
self.assertEqual(
plan["hints"]["feature_snapshot_filename"],
"hk_low_vol_dividend_quality_snapshot_factor_snapshot_latest.csv",
)
self.assertEqual(
plan["hints"]["feature_snapshot_manifest_filename"],
"hk_low_vol_dividend_quality_snapshot_factor_snapshot_latest.csv.manifest.json",
[sys.executable, str(SWITCH_PLAN_SCRIPT_PATH), "--profile",
"hk_low_vol_dividend_quality_snapshot", "--dry-run-only", "--json"],
check=False, capture_output=True, text=True,
)
self.assertNotEqual(result.returncode, 0)
self.assertIn("Unsupported STRATEGY_PROFILE", result.stderr)

def test_print_strategy_switch_env_plan_rejects_hk_disabled_profiles(self):
for profile in sorted(HK_DISABLED_PROFILES):
Expand Down
36 changes: 36 additions & 0 deletions tests/test_strategy_retirement_policy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
"""Retired profiles remain readable but cannot be selected by the platform."""

import pytest

from hk_equity_strategies import get_platform_runtime_adapter, get_strategy_catalog
from strategy_loader import load_strategy_definition
from strategy_registry import get_supported_profiles_for_platform, resolve_strategy_definition


RETIRED_PROFILES = ('hk_global_etf_tactical_rotation', 'hk_low_vol_dividend_quality_snapshot')


@pytest.mark.parametrize("profile", RETIRED_PROFILES)
def test_retired_profiles_are_not_selectable(profile):
assert profile not in get_supported_profiles_for_platform("longbridge")
with pytest.raises(ValueError, match="Unsupported STRATEGY_PROFILE"):
resolve_strategy_definition(profile, platform_id="longbridge")
with pytest.raises(ValueError, match="Unsupported STRATEGY_PROFILE"):
load_strategy_definition(profile)


@pytest.mark.parametrize("profile", RETIRED_PROFILES)
def test_retired_profile_aliases_are_not_selectable(profile):
catalog = get_strategy_catalog()
for alias, canonical in catalog.profile_aliases.items():
if canonical == profile:
with pytest.raises(ValueError, match="Unsupported STRATEGY_PROFILE"):
resolve_strategy_definition(alias, platform_id="longbridge")


@pytest.mark.parametrize("profile", RETIRED_PROFILES)
def test_retired_profile_history_and_adapters_remain_available(profile):
catalog = get_strategy_catalog()
assert profile in catalog.definitions
assert profile in catalog.metadata
assert get_platform_runtime_adapter(profile, platform_id="longbridge") is not None
Loading
Loading