Add small-account cash substitution notes

Author: Pigbibi

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "quant-platform-kit"
-version = "0.7.33"
+version = "0.7.34"
 description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies."
 readme = "README.md"
 requires-python = ">=3.9"

src/quant_platform_kit/common/small_account_compatibility.py

@@ -3,15 +3,54 @@
 from __future__ import annotations
 
 from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
 
 
-__all__ = ["project_unbuyable_value_targets_to_cash"]
+__all__ = [
+    "SmallAccountCashCompatibilityResult",
+    "apply_small_account_cash_compatibility",
+    "format_small_account_cash_substitution_notes",
+    "project_unbuyable_value_targets_to_cash",
+]
+
+
+@dataclass(frozen=True)
+class SmallAccountCashCompatibilityResult:
+    targets: dict[str, float]
+    whole_share_substituted_symbols: tuple[str, ...]
+    safe_haven_cash_substituted_symbols: tuple[str, ...]
+    cash_substitution_notes: tuple[dict[str, object], ...]
 
 
 def _normalize_symbol(value: object) -> str:
     return str(value or "").strip().upper()
 
 
+def _positive_target_total(targets: Mapping[str, object]) -> float:
+    total = 0.0
+    for value in dict(targets or {}).values():
+        try:
+            total += max(0.0, float(value or 0.0))
+        except (TypeError, ValueError):
+            continue
+    return total
+
+
+def _normalize_prices(prices: Mapping[str, object]) -> dict[str, float]:
+    return {
+        _normalize_symbol(symbol): float(price or 0.0)
+        for symbol, price in dict(prices or {}).items()
+    }
+
+
+def _format_symbol(symbol: str, *, suffix: str) -> str:
+    normalized = _normalize_symbol(symbol)
+    normalized_suffix = str(suffix or "").strip()
+    if normalized_suffix and not normalized.endswith(normalized_suffix.upper()):
+        return f"{normalized}{normalized_suffix}"
+    return normalized
+
+
 def project_unbuyable_value_targets_to_cash(
     target_values: Mapping[str, object],
     prices: Mapping[str, object],
@@ -40,10 +79,7 @@ def project_unbuyable_value_targets_to_cash(
         candidate_symbols = tuple(dict.fromkeys(_normalize_symbol(symbol) for symbol in symbols))
 
     substituted: list[str] = []
-    normalized_prices = {
-        _normalize_symbol(symbol): float(price or 0.0)
-        for symbol, price in dict(prices or {}).items()
-    }
+    normalized_prices = _normalize_prices(prices)
     for symbol in candidate_symbols:
         if not symbol:
             continue
@@ -56,3 +92,140 @@ def project_unbuyable_value_targets_to_cash(
             substituted.append(symbol)
 
     return adjusted, tuple(dict.fromkeys(substituted))
+
+
+def apply_small_account_cash_compatibility(
+    target_values: Mapping[str, object],
+    prices: Mapping[str, object],
+    *,
+    candidate_symbols: Iterable[str] | None = None,
+    safe_haven_cash_symbols: Iterable[str] = (),
+    quantity_step: float = 1.0,
+    cash_substitute_limit_usd: float = 2000.0,
+) -> SmallAccountCashCompatibilityResult:
+    """Apply whole-share small-account projection and cash-safe-haven fallback.
+
+    If every risk/income target that remains positive is below one tradable unit,
+    and the remaining positive safe-haven/cash-sweep sleeve is still small, the
+    safe-haven target is also projected to cash. The returned notes preserve the
+    original target and price so platform notifications can explain why no risk
+    or safe-haven rebuy was submitted.
+    """
+
+    adjusted_targets, substituted = project_unbuyable_value_targets_to_cash(
+        target_values,
+        prices,
+        symbols=candidate_symbols,
+        quantity_step=quantity_step,
+    )
+    normalized_candidates = (
+        tuple(adjusted_targets)
+        if candidate_symbols is None
+        else tuple(dict.fromkeys(_normalize_symbol(symbol) for symbol in candidate_symbols))
+    )
+    remaining_non_safe_targets = [
+        symbol
+        for symbol in normalized_candidates
+        if float(adjusted_targets.get(_normalize_symbol(symbol), 0.0) or 0.0) > 0.0
+    ]
+    safe_haven_symbols = tuple(
+        dict.fromkeys(
+            _normalize_symbol(symbol)
+            for symbol in safe_haven_cash_symbols
+            if _normalize_symbol(symbol)
+        )
+    )
+    safe_haven_substituted: list[str] = []
+    if (
+        substituted
+        and not remaining_non_safe_targets
+        and _positive_target_total(adjusted_targets) <= max(0.0, float(cash_substitute_limit_usd or 0.0))
+    ):
+        for symbol in safe_haven_symbols:
+            if float(adjusted_targets.get(symbol, 0.0) or 0.0) > 0.0:
+                adjusted_targets[symbol] = 0.0
+                safe_haven_substituted.append(symbol)
+
+    notes: list[dict[str, object]] = []
+    if safe_haven_substituted:
+        normalized_targets = {
+            _normalize_symbol(symbol): float(value or 0.0)
+            for symbol, value in dict(target_values or {}).items()
+        }
+        normalized_prices = _normalize_prices(prices)
+        for symbol in substituted:
+            target_value = max(0.0, float(normalized_targets.get(symbol, 0.0) or 0.0))
+            price = max(0.0, float(normalized_prices.get(symbol, 0.0) or 0.0))
+            if target_value <= 0.0 or price <= 0.0:
+                continue
+            notes.append(
+                {
+                    "symbol": symbol,
+                    "target_value": target_value,
+                    "price": price,
+                    "cash_symbols": tuple(safe_haven_substituted),
+                }
+            )
+
+    return SmallAccountCashCompatibilityResult(
+        targets=adjusted_targets,
+        whole_share_substituted_symbols=substituted,
+        safe_haven_cash_substituted_symbols=tuple(safe_haven_substituted),
+        cash_substitution_notes=tuple(notes),
+    )
+
+
+def format_small_account_cash_substitution_notes(
+    notes: Iterable[Mapping[str, object]],
+    *,
+    translator,
+    wrapper_key: str = "buy_deferred",
+    detail_key: str = "buy_deferred_small_account_cash_substitution",
+    cash_label_key: str = "cash_label",
+    symbol_suffix: str = ".US",
+) -> tuple[str, ...]:
+    """Render small-account cash substitution notes through platform i18n."""
+
+    messages: list[str] = []
+    seen_keys: set[tuple[str, str, str]] = set()
+    for note in tuple(notes or ()):
+        if not isinstance(note, Mapping):
+            continue
+        symbol = _normalize_symbol(note.get("symbol"))
+        if not symbol:
+            continue
+        target_value = max(0.0, float(note.get("target_value") or 0.0))
+        price = max(0.0, float(note.get("price") or 0.0))
+        if target_value <= 0.0 or price <= 0.0:
+            continue
+        cash_symbols = tuple(
+            dict.fromkeys(
+                _normalize_symbol(cash_symbol)
+                for cash_symbol in tuple(note.get("cash_symbols") or ())
+                if _normalize_symbol(cash_symbol)
+            )
+        )
+        cash_symbols_text = ", ".join(
+            _format_symbol(cash_symbol, suffix=symbol_suffix)
+            for cash_symbol in cash_symbols
+        )
+        if not cash_symbols_text:
+            cash_symbols_text = str(translator(cash_label_key)).strip()
+            if not cash_symbols_text or cash_symbols_text == cash_label_key:
+                cash_symbols_text = "cash"
+        note_key = (symbol, f"{target_value:.2f}", cash_symbols_text)
+        if note_key in seen_keys:
+            continue
+        seen_keys.add(note_key)
+        detail = translator(
+            detail_key,
+            symbol=_format_symbol(symbol, suffix=symbol_suffix),
+            diff=f"{target_value:.2f}",
+            price=f"{price:.2f}",
+            cash_symbols=cash_symbols_text,
+        )
+        message = translator(wrapper_key, detail=detail)
+        if not message or message == wrapper_key:
+            message = detail
+        messages.append(message)
+    return tuple(messages)

tests/test_small_account_compatibility.py

@@ -1,6 +1,8 @@
 import unittest
 
 from quant_platform_kit.common.small_account_compatibility import (
+    apply_small_account_cash_compatibility,
+    format_small_account_cash_substitution_notes,
     project_unbuyable_value_targets_to_cash,
 )
 
@@ -28,6 +30,74 @@ def test_keeps_targets_that_can_buy_one_quantity_step(self):
         self.assertEqual(adjusted["BBB"], 0.0)
         self.assertEqual(substituted, ("BBB",))
 
+    def test_projects_safe_haven_to_cash_when_only_risk_target_is_unbuyable(self):
+        result = apply_small_account_cash_compatibility(
+            {"SOXX": 163.14, "BOXX": 1224.46},
+            {"SOXX": 504.60, "BOXX": 116.59},
+            candidate_symbols=("SOXX",),
+            safe_haven_cash_symbols=("BOXX",),
+            cash_substitute_limit_usd=2000.0,
+        )
+
+        self.assertEqual(result.targets["SOXX"], 0.0)
+        self.assertEqual(result.targets["BOXX"], 0.0)
+        self.assertEqual(result.whole_share_substituted_symbols, ("SOXX",))
+        self.assertEqual(result.safe_haven_cash_substituted_symbols, ("BOXX",))
+        self.assertEqual(
+            result.cash_substitution_notes,
+            (
+                {
+                    "symbol": "SOXX",
+                    "target_value": 163.14,
+                    "price": 504.60,
+                    "cash_symbols": ("BOXX",),
+                },
+            ),
+        )
+
+    def test_keeps_safe_haven_when_cash_projection_exceeds_small_account_limit(self):
+        result = apply_small_account_cash_compatibility(
+            {"SOXX": 163.14, "BOXX": 5000.0},
+            {"SOXX": 504.60, "BOXX": 116.59},
+            candidate_symbols=("SOXX",),
+            safe_haven_cash_symbols=("BOXX",),
+            cash_substitute_limit_usd=2000.0,
+        )
+
+        self.assertEqual(result.targets["SOXX"], 0.0)
+        self.assertEqual(result.targets["BOXX"], 5000.0)
+        self.assertEqual(result.whole_share_substituted_symbols, ("SOXX",))
+        self.assertEqual(result.safe_haven_cash_substituted_symbols, ())
+        self.assertEqual(result.cash_substitution_notes, ())
+
+    def test_formats_cash_substitution_notes_through_i18n(self):
+        messages = format_small_account_cash_substitution_notes(
+            (
+                {
+                    "symbol": "SOXX",
+                    "target_value": 163.14,
+                    "price": 504.60,
+                    "cash_symbols": ("BOXX",),
+                },
+            ),
+            translator=lambda key, **kwargs: {
+                "cash_label": "现金",
+                "buy_deferred": "ℹ️ [买入说明] {detail}",
+                "buy_deferred_small_account_cash_substitution": (
+                    "{symbol} 目标金额 ${diff} 低于 1 股价格 ${price}；"
+                    "为避免超过目标仓位，小账户本轮保留现金，不回补 {cash_symbols}"
+                ),
+            }.get(key, key).format(**kwargs),
+        )
+
+        self.assertEqual(
+            messages,
+            (
+                "ℹ️ [买入说明] SOXX.US 目标金额 $163.14 低于 1 股价格 $504.60；"
+                "为避免超过目标仓位，小账户本轮保留现金，不回补 BOXX.US",
+            ),
+        )
+
 
 if __name__ == "__main__":
     unittest.main()