refactor(notifications): migrate to shared renderer_base (-253 lines)

Replace locally duplicated functions with imports from
quant_platform_kit.notifications.renderer_base.
Keep thin wrappers for IBKR-specific data access patterns
(e.g. execution_annotations flattening, hardcoded cash labels).

Co-Authored-By: Claude <noreply@anthropic.com>

Author: Pigbibi

notifications/renderers.py

@@ -3,52 +3,35 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-import re
 
 from notifications.events import RenderedNotification
 from quant_platform_kit.common.quantity import format_quantity
 from quant_platform_kit.common.notification_localization import (
     localize_notification_text as _base_localize_notification_text,
-    translator_uses_zh as _base_translator_uses_zh,
 )
-
-_PRICE_SOURCE_LABELS = {
-    "longbridge_candlesticks": ("LongBridge 日线K线", "LongBridge daily candlesticks"),
-    "schwab_daily_history_with_live_quote_overlay": ("Schwab 日线历史", "Schwab daily history"),
-    "firstrade_ohlc_with_live_quote_overlay": ("Firstrade OHLC", "Firstrade OHLC"),
-    "market_quote": ("实时行情报价", "market quote"),
-    "mixed_market_quote_snapshot_close": (
-        "实时行情报价 + 快照收盘价回补",
-        "market quote + snapshot close fallback",
-    ),
-    "mixed_market_quote_historical_close": (
-        "实时行情报价 + 历史收盘价回补",
-        "market quote + historical close fallback",
-    ),
-    "snapshot_close": ("快照收盘价", "snapshot close"),
-    "historical_close": ("历史收盘价", "historical close"),
-    "market_data": ("市场数据", "market data"),
-}
-
-try:
-    from quant_platform_kit.common.notification_localization import (
-        localize_price_source_label as _shared_localize_price_source_label,
-    )
-except ImportError:  # pragma: no cover - compatibility with older pinned shared wheels
-    _shared_localize_price_source_label = None
-
-
-def _localize_price_source_label(value, *, translator=None, locale=None):
-    source = str(value or "").strip()
-    use_zh = _base_translator_uses_zh(translator) if translator is not None else str(locale or "").startswith("zh")
-    if not source:
-        return "未知" if use_zh else "unknown"
-    label = _PRICE_SOURCE_LABELS.get(source)
-    if label is not None:
-        return label[0] if use_zh else label[1]
-    if _shared_localize_price_source_label is not None:
-        return _shared_localize_price_source_label(source, translator=translator, locale=locale)
-    return source.replace("_", " ")
+from quant_platform_kit.notifications.renderer_base import (
+    as_float_or_none as _as_float_or_none,
+    build_timing_audit_lines as _build_timing_audit_lines_shared,
+    build_tqqq_risk_control_lines as _build_tqqq_risk_control_lines_shared,
+    compact_dashboard_lines,
+    effective_volatility_delever_threshold as _effective_volatility_delever_threshold,
+    format_percent as _format_percent,
+    format_percentile as _format_percentile,
+    format_sample_count as _format_sample_count,
+    format_signal_snapshot_line as _format_signal_snapshot_line_shared,
+    format_tqqq_volatility_delever_allocation_detail as _format_tqqq_volatility_delever_allocation_detail,
+    format_volatility_delever_threshold_detail as _format_volatility_delever_threshold_detail,
+    is_compact_dashboard_audit_line,
+    is_truthy as _is_truthy,
+    localize_price_source_label as _localize_price_source_label,
+    localize_timing_contract as _localize_timing_contract,
+    present as _present,
+    relabel_dashboard_cash_labels as _relabel_dashboard_cash_labels_shared,
+    resolve_execution,
+    split_detail_segment as _split_detail_segment,
+    split_labeled_text as _split_labeled_text,
+    translator_uses_zh as _translator_uses_zh,
+)
 
 _EXTRA_ZH_REASON_REPLACEMENTS = (
     ("pending_orders_detected", "检测到未完成订单"),
@@ -60,8 +43,6 @@ def _localize_price_source_label(value, *, translator=None, locale=None):
     ("fail_reason=", "失败原因="),
     ("decision=", "决策="),
 )
-_DETAIL_FIELD_SPLIT_RE = re.compile(r"\s+(?=[^\s=:：]+[=:：])")
-
 
 def _format_text(value, *, fallback: str) -> str:
     text = str(value).strip() if value is not None else ""
@@ -79,10 +60,6 @@ def _format_symbol_preview(symbols, *, limit: int = 3) -> str:
     return ",".join(shown)
 
 
-def _translator_uses_zh(translator) -> bool:
-    return _base_translator_uses_zh(translator)
-
-
 def _extra_notification_lines(extra_notification_lines) -> list[str]:
     return [str(line).strip() for line in extra_notification_lines or () if str(line).strip()]
 
@@ -95,42 +72,6 @@ def _localize_notification_text(text: str, *, translator) -> str:
     )
 
 
-def _localize_timing_contract(contract: str, *, translator) -> str:
-    value = str(contract or "").strip()
-    if not value:
-        return ""
-    if value == "same_trading_day":
-        return "当日执行" if _translator_uses_zh(translator) else "same trading day"
-    if value == "next_trading_day":
-        return "次一交易日执行" if _translator_uses_zh(translator) else "next trading day"
-    match = re.fullmatch(r"next_(\d+)_trading_days", value)
-    if match:
-        count = int(match.group(1))
-        if _translator_uses_zh(translator):
-            return f"{count}个交易日后执行"
-        return f"next {count} trading days"
-    return _localize_notification_text(value, translator=translator)
-
-
-def _split_detail_segment(text: str) -> list[str]:
-    value = str(text or "").strip()
-    if not value:
-        return []
-    if "=" not in value and ":" not in value and "：" not in value:
-        return [value]
-    return [part.strip() for part in _DETAIL_FIELD_SPLIT_RE.split(value) if part.strip()]
-
-
-def _split_labeled_text(text: str) -> list[str]:
-    segments = [segment.strip() for segment in str(text or "").split(" | ") if segment.strip()]
-    if not segments:
-        return []
-    lines = [segments[0]]
-    for segment in segments[1:]:
-        lines.extend(_split_detail_segment(segment))
-    return lines
-
-
 def _format_prefixed_text(prefix: str, text: str) -> list[str]:
     parts = _split_labeled_text(text)
     if not parts:
@@ -445,235 +386,41 @@ def _format_dashboard_text(text) -> str:
 
 
 def _relabel_dashboard_cash_labels(text: str, *, cash_only_execution: bool) -> str:
-    value = str(text or "")
-    if cash_only_execution:
-        value = value.replace("总资产（策略净值）", "总资产（策略标的+现金，不含融资额度）")
-        value = value.replace(
-            "Total assets (strategy net liquidation)",
-            "Total assets (strategy symbols + cash, ex-margin)",
-        )
-        value = value.replace("购买力", "可用现金")
-        value = value.replace("Buying power", "Available cash")
-        return value
-    value = value.replace("总资产（策略标的+现金，不含融资额度）", "总资产（策略净值）")
-    value = value.replace("总资产（策略标的+现金）", "总资产（策略净值）")
-    value = value.replace(
-        "Total assets (strategy symbols + cash, ex-margin)",
-        "Total assets (strategy net liquidation)",
-    )
-    value = value.replace(
-        "Total assets (strategy symbols + cash)",
-        "Total assets (strategy net liquidation)",
+    """Delegates to shared renderer_base; IBKR uses hardcoded labels (no translator)."""
+    return _relabel_dashboard_cash_labels_shared(
+        text, cash_only_execution=cash_only_execution, translator=None,
     )
-    value = value.replace("可用现金", "购买力")
-    value = value.replace("Available cash", "Buying power")
-    return value
 
 
 def _format_compact_dashboard_text(text) -> str:
     lines = []
     for line in _format_dashboard_text(text).splitlines():
-        stripped = line.strip()
-        lowered = stripped.lower()
-        if stripped.startswith(("⏱", "🧾", "🛡️", "📊", "🎯")):
-            continue
-        if lowered.startswith(("signal:", "signal：", "market status:")):
-            continue
-        if stripped.startswith(("信号:", "信号：", "市场状态:", "市场状态：")):
+        if not line.strip() or is_compact_dashboard_audit_line(line):
             continue
         lines.append(line)
     return "\n".join(lines)
 
 
 def _build_timing_audit_lines(signal_metadata, *, translator) -> list[str]:
-    metadata = signal_metadata if isinstance(signal_metadata, Mapping) else {}
-    raw_annotations = metadata.get("execution_annotations")
-    annotations = raw_annotations if isinstance(raw_annotations, Mapping) else {}
-    signal_date = str(annotations.get("signal_date") or metadata.get("signal_date") or "").strip()
-    effective_date = str(annotations.get("effective_date") or metadata.get("effective_date") or "").strip()
-    contract = str(
-        annotations.get("execution_timing_contract")
-        or metadata.get("execution_timing_contract")
-        or ""
-    ).strip()
-    if not signal_date and not effective_date and not contract:
-        return []
-    label = "⏱ 执行时点" if _translator_uses_zh(translator) else "⏱ Timing"
-    localized_contract = _localize_timing_contract(contract, translator=translator)
-    if signal_date and effective_date:
-        value = f"{signal_date} -> {effective_date}"
-    else:
-        value = signal_date or effective_date or localized_contract
-    if localized_contract and localized_contract not in value:
-        value = f"{value} ({localized_contract})" if value else localized_contract
-    return [f"{label}: {value}"]
-
-
-def _format_percent(value) -> str:
-    try:
-        return f"{float(value) * 100:.1f}%"
-    except (TypeError, ValueError):
-        return "n/a"
-
-
-def _as_float_or_none(value):
-    try:
-        return float(value)
-    except (TypeError, ValueError):
-        return None
-
-
-def _format_percentile(value) -> str:
-    try:
-        percentile = float(value) * 100
-    except (TypeError, ValueError):
-        return "p?"
-    if float(percentile).is_integer():
-        return f"p{int(percentile)}"
-    return f"p{percentile:.1f}"
-
-
-def _format_sample_count(value) -> str:
-    try:
-        count = float(value)
-    except (TypeError, ValueError):
-        return "n/a"
-    if float(count).is_integer():
-        return str(int(count))
-    return f"{count:.1f}"
-
-
-def _present(value) -> bool:
-    return value not in (None, "")
-
-
-def _is_truthy(value) -> bool:
-    if isinstance(value, bool):
-        return value
-    return str(value or "").strip().lower() in {"1", "true", "yes", "y"}
-
-
-def _effective_volatility_delever_threshold(signal_metadata, *, prefix: str):
-    mode = str(signal_metadata.get(f"{prefix}_threshold_mode") or "").strip().lower()
-    dynamic_threshold = signal_metadata.get(f"{prefix}_dynamic_threshold")
-    if mode == "rolling_percentile" and _present(dynamic_threshold):
-        return dynamic_threshold
-    return signal_metadata.get(f"{prefix}_threshold")
-
-
-def _format_volatility_delever_threshold_detail(signal_metadata, *, prefix: str, translator) -> str:
-    mode = str(signal_metadata.get(f"{prefix}_threshold_mode") or "").strip().lower()
-    fixed_threshold = signal_metadata.get(f"{prefix}_threshold")
-    dynamic_threshold = signal_metadata.get(f"{prefix}_dynamic_threshold")
-    if mode == "rolling_percentile":
-        kwargs = {
-            "percentile": _format_percentile(signal_metadata.get(f"{prefix}_dynamic_percentile")),
-            "lookback": _format_sample_count(signal_metadata.get(f"{prefix}_dynamic_lookback")),
-            "min_periods": _format_sample_count(signal_metadata.get(f"{prefix}_dynamic_min_periods")),
-            "sample_count": _format_sample_count(signal_metadata.get(f"{prefix}_dynamic_sample_count")),
-            "floor": _format_percent(signal_metadata.get(f"{prefix}_dynamic_floor")),
-            "cap": _format_percent(signal_metadata.get(f"{prefix}_dynamic_cap")),
-            "fixed_threshold": _format_percent(fixed_threshold),
-        }
-        if _present(dynamic_threshold):
-            return translator("blend_gate_volatility_threshold_detail_dynamic", **kwargs)
-        return translator("blend_gate_volatility_threshold_detail_dynamic_fallback", **kwargs)
-    return translator(
-        "blend_gate_volatility_threshold_detail_fixed",
-        threshold=_format_percent(fixed_threshold),
-    )
-
-
-def _format_tqqq_volatility_delever_allocation_detail(
-    signal_metadata,
-    *,
-    prefix: str,
-    redirect_symbol: str,
-    translator,
-) -> str:
-    retained_ratio = _as_float_or_none(signal_metadata.get(f"{prefix}_retained_ratio"))
-    redirected_ratio = _as_float_or_none(signal_metadata.get(f"{prefix}_redirected_ratio"))
-    if retained_ratio is None:
-        retained_ratio = _as_float_or_none(signal_metadata.get(f"{prefix}_retention_ratio"))
-    if redirected_ratio is None and retained_ratio is not None:
-        redirected_ratio = max(0.0, min(1.0, 1.0 - retained_ratio))
-    return translator(
-        "tqqq_volatility_delever_allocation_detail",
-        retained_ratio=_format_percent(retained_ratio),
-        redirected_ratio=_format_percent(redirected_ratio),
-        redirect_symbol=redirect_symbol or "QQQ",
+    execution = resolve_execution(
+        signal_metadata if isinstance(signal_metadata, Mapping) else {},
     )
+    return _build_timing_audit_lines_shared(execution, translator=translator)
 
 
 def _build_tqqq_risk_control_lines(signal_metadata, *, translator) -> list[str]:
-    prefix = "dual_drive_volatility_delever"
-    if not _is_truthy(signal_metadata.get(f"{prefix}_applied")):
-        return []
-    redirect_symbol = str(signal_metadata.get(f"{prefix}_redirect_symbol") or "QQQ").strip().upper()
-    window = str(signal_metadata.get(f"{prefix}_window") or "5").strip()
-    threshold = _effective_volatility_delever_threshold(signal_metadata, prefix=prefix)
-    threshold_detail = _format_volatility_delever_threshold_detail(
-        signal_metadata,
-        prefix=prefix,
-        translator=translator,
-    )
-    allocation_detail = _format_tqqq_volatility_delever_allocation_detail(
-        signal_metadata,
-        prefix=prefix,
-        redirect_symbol=redirect_symbol or "QQQ",
+    return _build_tqqq_risk_control_lines_shared(
+        signal_metadata if isinstance(signal_metadata, Mapping) else {},
         translator=translator,
     )
-    if str(signal_metadata.get(f"{prefix}_trigger_reason") or "").strip() == "hysteresis_hold":
-        return [
-            translator(
-                "risk_control_tqqq_volatility_delever_hysteresis_dynamic",
-                window=window,
-                volatility=_format_percent(signal_metadata.get(f"{prefix}_metric")),
-                exit_threshold=_format_percent(signal_metadata.get(f"{prefix}_exit_threshold")),
-                threshold=_format_percent(threshold),
-                threshold_detail=threshold_detail,
-                source_symbol="TQQQ",
-                redirect_symbol=redirect_symbol or "QQQ",
-                allocation_detail=allocation_detail,
-            )
-        ]
-    return [
-        translator(
-            "risk_control_tqqq_volatility_delever_applied_dynamic",
-            window=window,
-            volatility=_format_percent(signal_metadata.get(f"{prefix}_metric")),
-            threshold=_format_percent(threshold),
-            threshold_detail=threshold_detail,
-            source_symbol="TQQQ",
-            redirect_symbol=redirect_symbol or "QQQ",
-            allocation_detail=allocation_detail,
-        )
-    ]
 
 
 def _format_signal_snapshot_line(snapshot, *, translator) -> str:
-    if not isinstance(snapshot, Mapping):
-        return ""
-    market_date = str(snapshot.get("market_date") or snapshot.get("signal_as_of") or "").strip()
-    source = str(snapshot.get("latest_price_source") or "").strip()
-    warning = snapshot.get("data_freshness_warning")
-    if not market_date and not source and warning in (None, "", False):
-        return ""
-    if _translator_uses_zh(translator):
-        parts = [
-            f"日期 {market_date or '未知'}",
-            f"数据源 {_localize_price_source_label(source, translator=translator)}",
-        ]
-        if warning not in (None, "", False):
-            parts.append(f"提示 {_localize_notification_text(warning, translator=translator)}")
-        return "🧾 信号快照: " + " | ".join(parts)
-    parts = [
-        f"date {market_date or 'unknown'}",
-        f"source {_localize_price_source_label(source, translator=translator)}",
-    ]
-    if warning not in (None, "", False):
-        parts.append(f"warning {warning}")
-    return "🧾 Signal snapshot: " + " | ".join(parts)
+    return _format_signal_snapshot_line_shared(
+        snapshot,
+        translator=translator,
+        localize_text=_localize_notification_text,
+    )
 
 
 def _strategy_dashboard_text(signal_metadata, *, translator) -> str: