diff --git a/src/quant_platform_kit/common/__init__.py b/src/quant_platform_kit/common/__init__.py index 282b457..5157dec 100644 --- a/src/quant_platform_kit/common/__init__.py +++ b/src/quant_platform_kit/common/__init__.py @@ -1,5 +1,10 @@ """Shared domain models, ports, strategy contracts, and plugin helpers.""" +from .notification_localization import ( + COMMON_ZH_NOTIFICATION_REPLACEMENTS, + localize_notification_text, + translator_uses_zh, +) from .strategy_plugins import ( PLUGIN_MODE_ADVISORY, PLUGIN_MODE_LIVE, @@ -17,11 +22,13 @@ ) __all__ = [ + "COMMON_ZH_NOTIFICATION_REPLACEMENTS", "PLUGIN_MODE_ADVISORY", "PLUGIN_MODE_LIVE", "PLUGIN_MODE_PAPER", "PLUGIN_MODE_SHADOW", "SUPPORTED_STRATEGY_PLUGIN_MODES", + "localize_notification_text", "StrategyPluginMountConfig", "StrategyPluginSignal", "build_strategy_plugin_report_payload", @@ -29,5 +36,6 @@ "load_strategy_plugin_signal", "normalize_strategy_plugin_mode", "parse_strategy_plugin_mounts", + "translator_uses_zh", "validate_strategy_plugin_signal_payload", ] diff --git a/src/quant_platform_kit/common/notification_localization.py b/src/quant_platform_kit/common/notification_localization.py new file mode 100644 index 0000000..61cf1e5 --- /dev/null +++ b/src/quant_platform_kit/common/notification_localization.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +from collections.abc import Callable, Sequence + +NotificationTranslator = Callable[..., str] +NotificationReplacement = tuple[str, str] + + +COMMON_ZH_NOTIFICATION_REPLACEMENTS: tuple[NotificationReplacement, ...] = ( + ("feature snapshot guard blocked execution", "特征快照校验阻止执行"), + ("feature snapshot required", "需要特征快照"), + ("feature snapshot compute failed", "特征快照计算失败"), + ("feature_snapshot_download_failed", "特征快照下载失败"), + ("feature_snapshot_compute_failed", "特征快照计算失败"), + ("feature_snapshot_path_missing", "缺少特征快照路径"), + ("feature_snapshot_missing", "特征快照不存在"), + ("feature_snapshot_stale", "特征快照过旧"), + ("feature_snapshot_manifest_missing", "缺少快照清单"), + ("feature_snapshot_profile_mismatch", "快照策略名不匹配"), + ("feature_snapshot_config_name_mismatch", "快照配置名不匹配"), + ("feature_snapshot_config_path_mismatch", "快照配置路径不匹配"), + ("feature_snapshot_contract_version_mismatch", "快照契约版本不匹配"), + ("soxl_soxx_trend_income", "SOXL/SOXX 半导体趋势收益"), + ("tqqq_growth_income", "TQQQ 增长收益"), + ("global_etf_rotation", "全球 ETF 轮动"), + ("russell_1000_multi_factor_defensive", "罗素1000多因子"), + ("tech_communication_pullback_enhancement", "科技通信回调增强"), + ("qqq_tech_enhancement", "科技通信回调增强"), + ("mega_cap_leader_rotation_aggressive", "Mega Cap 激进龙头轮动"), + ("mega_cap_leader_rotation_dynamic_top20", "Mega Cap 动态 Top20 龙头轮动"), + ("mega_cap_leader_rotation_top50_balanced", "Mega Cap Top50 平衡龙头轮动"), + ("dynamic_mega_leveraged_pullback", "Mega Cap 2x 回调策略"), + ("outside_monthly_execution_window", "当前不在月度执行窗口"), + ("no_execution_window_after_snapshot", "快照后没有可用执行窗口"), + ("no-op", "不执行"), + ("monthly snapshot cadence", "月度快照节奏"), + ("waiting inside execution window", "等待进入执行窗口"), + ("small_account_warning=true", "小账户提示=是"), + ("portfolio_equity=", "净值="), + ("min_recommended_equity=", "建议最低净值="), + ( + "integer_shares_min_position_value_may_prevent_backtest_replication", + "整数股和最小仓位限制可能导致实盘无法完全复现回测", + ), + ( + "integer-share minimum position sizing may prevent backtest replication", + "整数股和最小仓位限制可能导致实盘无法完全复现回测", + ), + ("small account warning: portfolio equity", "小账户提示：净值"), + ("small account warning", "小账户提示"), + ("is below the recommended", "低于建议"), + ("is below recommended", "低于建议"), + ("snapshot_as_of=", "快照日期="), + ("snapshot=", "快照日期="), + ("allowed=", "允许日期="), + ("", "未知"), + ("", "无"), + ("RISK-ON", "风险开启"), + ("DE-LEVER", "降杠杆"), + ("regime=hard_defense", "市场阶段=强防御"), + ("regime=soft_defense", "市场阶段=软防御"), + ("regime=risk_on", "市场阶段=进攻"), + ("benchmark_trend=down", "基准趋势=向下"), + ("benchmark_trend=up", "基准趋势=向上"), + ("benchmark=down", "基准趋势=向下"), + ("benchmark=up", "基准趋势=向上"), + ("breadth=", "市场宽度="), + ("target_stock=", "目标股票仓位="), + ("realized_stock=", "实际股票仓位="), + ("stock_exposure=", "股票目标仓位="), + ("safe_haven=", "避险仓位="), + ("selected=", "入选标的数="), + ("top=", "前排标的="), + ("no_selection", "无入选标的"), + ("outside_execution_window", "当前不在执行窗口"), + ("insufficient_buying_power", "购买力不足"), + ("missing_price", "缺少报价"), + ("no_equity", "无净值"), + ("fail_closed", "关闭执行"), + ("reason=", "原因="), +) + + +def translator_uses_zh(translator: NotificationTranslator) -> bool: + sample = str(translator("no_trades")) + return any("\u4e00" <= ch <= "\u9fff" for ch in sample) + + +def localize_notification_text( + text: object, + *, + translator: NotificationTranslator, + extra_replacements: Sequence[NotificationReplacement] = (), +) -> str: + value = str(text or "").strip() + if not value or not translator_uses_zh(translator): + return value + localized = value + for source, target in (*tuple(extra_replacements), *COMMON_ZH_NOTIFICATION_REPLACEMENTS): + localized = localized.replace(source, target) + return localized diff --git a/tests/test_notification_localization.py b/tests/test_notification_localization.py new file mode 100644 index 0000000..979649e --- /dev/null +++ b/tests/test_notification_localization.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import unittest + +from quant_platform_kit.common.notification_localization import ( + COMMON_ZH_NOTIFICATION_REPLACEMENTS, + localize_notification_text, + translator_uses_zh, +) + + +def _translator_factory(no_trades_text: str): + def _translator(key: str, **_kwargs) -> str: + if key == "no_trades": + return no_trades_text + return key + + return _translator + + +class NotificationLocalizationTests(unittest.TestCase): + def test_translator_uses_zh_detects_chinese_output(self): + self.assertTrue(translator_uses_zh(_translator_factory("无需交易"))) + self.assertFalse(translator_uses_zh(_translator_factory("No trades"))) + + def test_localize_notification_text_applies_common_replacements(self): + localized = localize_notification_text( + "no-op | reason=outside_monthly_execution_window | snapshot=2026-04-16", + translator=_translator_factory("无需交易"), + ) + + self.assertEqual( + localized, + "不执行 | 原因=当前不在月度执行窗口 | 快照日期=2026-04-16", + ) + + def test_localize_notification_text_keeps_english_for_non_zh_translator(self): + text = "no-op | reason=outside_monthly_execution_window" + self.assertEqual( + localize_notification_text(text, translator=_translator_factory("No trades")), + text, + ) + + def test_localize_notification_text_applies_extra_replacements_after_common_set(self): + localized = localize_notification_text( + "fail_reason=same_day_execution_locked", + translator=_translator_factory("无需交易"), + extra_replacements=( + ("fail_reason=", "失败原因="), + ("same_day_execution_locked", "当日执行锁已存在"), + ), + ) + + self.assertEqual(localized, "失败原因=当日执行锁已存在") + + def test_common_replacements_include_reason_label(self): + self.assertIn(("reason=", "原因="), COMMON_ZH_NOTIFICATION_REPLACEMENTS) + + +if __name__ == "__main__": + unittest.main()