Add small-account allocation drift notes

Author: Pigbibi

src/quant_platform_kit/common/small_account_compatibility.py

@@ -9,6 +9,8 @@
 __all__ = [
     "SmallAccountCashCompatibilityResult",
     "apply_small_account_cash_compatibility",
+    "build_small_account_allocation_drift_notes",
+    "format_small_account_allocation_drift_notes",
     "format_small_account_cash_substitution_notes",
     "project_unbuyable_value_targets_to_cash",
 ]
@@ -26,6 +28,14 @@ def _normalize_symbol(value: object) -> str:
     return str(value or "").strip().upper()
 
 
+def _normalize_trade_symbol(value: object, *, symbol_suffix: str = ".US") -> str:
+    symbol = _normalize_symbol(value)
+    suffix = str(symbol_suffix or "").strip().upper()
+    if suffix and symbol.endswith(suffix):
+        return symbol[: -len(suffix)]
+    return symbol
+
+
 def _positive_target_total(targets: Mapping[str, object]) -> float:
     total = 0.0
     for value in dict(targets or {}).values():
@@ -51,6 +61,189 @@ def _format_symbol(symbol: str, *, suffix: str) -> str:
     return normalized
 
 
+def _coerce_float(value: object, default: float = 0.0) -> float:
+    try:
+        return float(value or 0.0)
+    except (TypeError, ValueError):
+        return default
+
+
+def _coerce_order_price(order: Mapping[str, object], prices: Mapping[str, float], symbol: str) -> float:
+    for key in ("average_fill_price", "filled_price", "limit_price", "price", "submitted_price"):
+        price = _coerce_float(order.get(key), 0.0)
+        if price > 0.0:
+            return price
+    return max(0.0, float(prices.get(symbol, 0.0) or 0.0))
+
+
+def _format_weight(value: float) -> str:
+    return f"{float(value or 0.0):.1%}"
+
+
+def _format_weight_drift(value: float) -> str:
+    return f"{float(value or 0.0) * 100:+.1f}pp"
+
+
+def build_small_account_allocation_drift_notes(
+    *,
+    target_values: Mapping[str, object] | None = None,
+    target_weights: Mapping[str, object] | None = None,
+    current_values: Mapping[str, object] | None = None,
+    current_quantities: Mapping[str, object] | None = None,
+    prices: Mapping[str, object] | None = None,
+    submitted_orders: Iterable[Mapping[str, object]] = (),
+    total_value: float | None = None,
+    cash_value: float = 0.0,
+    symbol_suffix: str = ".US",
+    min_abs_weight_drift: float = 0.005,
+    small_account_max_total_value: float = 5000.0,
+    max_notes: int = 5,
+) -> tuple[dict[str, object], ...]:
+    """Estimate target drift after whole-share orders fully fill.
+
+    The estimate is intentionally simple and side-effect free: it uses current
+    values/quantities, order quantities, and reference prices to explain the
+    integer-share gap a small account may see if the submitted orders all fill.
+    """
+
+    normalized_prices = {
+        _normalize_trade_symbol(symbol, symbol_suffix=symbol_suffix): _coerce_float(price)
+        for symbol, price in dict(prices or {}).items()
+    }
+    normalized_current_values = {
+        _normalize_trade_symbol(symbol, symbol_suffix=symbol_suffix): _coerce_float(value)
+        for symbol, value in dict(current_values or {}).items()
+    }
+    normalized_current_quantities = {
+        _normalize_trade_symbol(symbol, symbol_suffix=symbol_suffix): _coerce_float(quantity)
+        for symbol, quantity in dict(current_quantities or {}).items()
+    }
+    for symbol, quantity in normalized_current_quantities.items():
+        if symbol not in normalized_current_values:
+            normalized_current_values[symbol] = quantity * max(0.0, normalized_prices.get(symbol, 0.0))
+
+    denominator = _coerce_float(total_value, 0.0)
+    if denominator <= 0.0:
+        denominator = sum(max(0.0, value) for value in normalized_current_values.values()) + max(
+            0.0,
+            _coerce_float(cash_value, 0.0),
+        )
+    if denominator <= 0.0:
+        return ()
+    if denominator > max(0.0, _coerce_float(small_account_max_total_value, 0.0)):
+        return ()
+
+    normalized_target_values: dict[str, float] = {}
+    if target_values:
+        normalized_target_values.update(
+            {
+                _normalize_trade_symbol(symbol, symbol_suffix=symbol_suffix): _coerce_float(value)
+                for symbol, value in dict(target_values or {}).items()
+            }
+        )
+    if target_weights:
+        for symbol, weight in dict(target_weights or {}).items():
+            normalized_target_values[_normalize_trade_symbol(symbol, symbol_suffix=symbol_suffix)] = (
+                denominator * _coerce_float(weight)
+            )
+
+    if not normalized_target_values:
+        return ()
+
+    projected_values = dict(normalized_current_values)
+    projected_quantities = dict(normalized_current_quantities)
+    for raw_order in tuple(submitted_orders or ()):
+        if not isinstance(raw_order, Mapping):
+            continue
+        symbol = _normalize_trade_symbol(raw_order.get("symbol"), symbol_suffix=symbol_suffix)
+        if not symbol:
+            continue
+        side = str(raw_order.get("side") or "").strip().lower()
+        quantity = _coerce_float(raw_order.get("quantity"), 0.0)
+        if quantity <= 0.0 or side not in {"buy", "sell"}:
+            continue
+        price = _coerce_order_price(raw_order, normalized_prices, symbol)
+        if price <= 0.0:
+            continue
+        signed_quantity = quantity if side == "buy" else -quantity
+        projected_quantities[symbol] = projected_quantities.get(symbol, 0.0) + signed_quantity
+        projected_values[symbol] = max(0.0, projected_values.get(symbol, 0.0) + signed_quantity * price)
+        normalized_prices.setdefault(symbol, price)
+
+    notes: list[dict[str, object]] = []
+    symbols = sorted(set(normalized_target_values))
+    for symbol in symbols:
+        target_value = max(0.0, normalized_target_values.get(symbol, 0.0))
+        projected_value = max(0.0, projected_values.get(symbol, 0.0))
+        if target_value <= 0.0 and projected_value <= 0.0:
+            continue
+        target_weight = target_value / denominator
+        projected_weight = projected_value / denominator
+        drift_weight = projected_weight - target_weight
+        if abs(drift_weight) < max(0.0, _coerce_float(min_abs_weight_drift, 0.0)):
+            continue
+        notes.append(
+            {
+                "kind": "small_account_allocation_drift",
+                "symbol": symbol,
+                "target_value": round(target_value, 2),
+                "projected_value": round(projected_value, 2),
+                "target_weight": target_weight,
+                "projected_weight": projected_weight,
+                "drift_weight": drift_weight,
+                "drift_value": round(projected_value - target_value, 2),
+                "projected_quantity": projected_quantities.get(symbol),
+            }
+        )
+
+    notes.sort(key=lambda note: abs(float(note.get("drift_weight") or 0.0)), reverse=True)
+    return tuple(notes[: max(0, int(max_notes or 0))])
+
+
+def format_small_account_allocation_drift_notes(
+    notes: Iterable[Mapping[str, object]],
+    *,
+    translator,
+    wrapper_key: str = "small_account_allocation_drift",
+    detail_key: str = "small_account_allocation_drift_detail",
+    symbol_suffix: str = ".US",
+) -> tuple[str, ...]:
+    """Render small-account projected allocation drift notes."""
+
+    details: list[str] = []
+    seen_symbols: set[str] = set()
+    for note in tuple(notes or ()):
+        if not isinstance(note, Mapping):
+            continue
+        if str(note.get("kind") or "") != "small_account_allocation_drift":
+            continue
+        symbol = _normalize_symbol(note.get("symbol"))
+        if not symbol or symbol in seen_symbols:
+            continue
+        seen_symbols.add(symbol)
+        detail = translator(
+            detail_key,
+            symbol=_format_symbol(symbol, suffix=symbol_suffix),
+            projected_weight=_format_weight(_coerce_float(note.get("projected_weight"))),
+            target_weight=_format_weight(_coerce_float(note.get("target_weight"))),
+            drift_weight=_format_weight_drift(_coerce_float(note.get("drift_weight"))),
+        )
+        if not detail or detail == detail_key:
+            detail = (
+                f"{_format_symbol(symbol, suffix=symbol_suffix)} projected "
+                f"{_format_weight(_coerce_float(note.get('projected_weight')))} vs target "
+                f"{_format_weight(_coerce_float(note.get('target_weight')))} "
+                f"({_format_weight_drift(_coerce_float(note.get('drift_weight')))})"
+            )
+        details.append(str(detail))
+    if not details:
+        return ()
+    message = translator(wrapper_key, details="; ".join(details))
+    if not message or message == wrapper_key:
+        message = f"Small-account integer-share drift: {'; '.join(details)}"
+    return (message,)
+
+
 def project_unbuyable_value_targets_to_cash(
     target_values: Mapping[str, object],
     prices: Mapping[str, object],

tests/test_small_account_compatibility.py

@@ -2,6 +2,8 @@
 
 from quant_platform_kit.common.small_account_compatibility import (
     apply_small_account_cash_compatibility,
+    build_small_account_allocation_drift_notes,
+    format_small_account_allocation_drift_notes,
     format_small_account_cash_substitution_notes,
     project_unbuyable_value_targets_to_cash,
 )
@@ -108,6 +110,74 @@ def test_formats_cash_substitution_notes_through_i18n(self):
             ),
         )
 
+    def test_builds_projected_allocation_drift_notes_from_submitted_orders(self):
+        notes = build_small_account_allocation_drift_notes(
+            target_values={"SOXL": 218.19, "SOXX": 342.86},
+            current_values={"SOXL": 0.0, "SOXX": 0.0},
+            current_quantities={"SOXL": 0.0, "SOXX": 0.0},
+            prices={"SOXL": 229.73, "SOXX": 603.0},
+            submitted_orders=(
+                {"symbol": "SOXL.US", "side": "buy", "quantity": 1, "limit_price": 233.18},
+            ),
+            total_value=623.39,
+            min_abs_weight_drift=0.005,
+        )
+
+        self.assertEqual(notes[0]["symbol"], "SOXX")
+        self.assertAlmostEqual(notes[0]["target_weight"], 342.86 / 623.39)
+        self.assertAlmostEqual(notes[0]["projected_weight"], 0.0)
+        self.assertEqual(notes[1]["symbol"], "SOXL")
+        self.assertAlmostEqual(notes[1]["projected_weight"], 233.18 / 623.39)
+
+    def test_skips_drift_notes_for_larger_accounts_by_default(self):
+        notes = build_small_account_allocation_drift_notes(
+            target_values={"AAA": 5000.0},
+            current_values={"AAA": 0.0},
+            prices={"AAA": 100.0},
+            submitted_orders=({"symbol": "AAA", "side": "buy", "quantity": 49, "limit_price": 100.0},),
+            total_value=50_000.0,
+        )
+
+        self.assertEqual(notes, ())
+
+    def test_drift_notes_ignore_symbols_outside_reference_targets(self):
+        notes = build_small_account_allocation_drift_notes(
+            target_values={"SOXL": 500.0},
+            current_values={"SOXL": 0.0, "BOXX": 1000.0},
+            current_quantities={"SOXL": 0.0, "BOXX": 10.0},
+            prices={"SOXL": 100.0, "BOXX": 100.0},
+            submitted_orders=(
+                {"symbol": "BOXX.US", "side": "buy", "quantity": 1, "limit_price": 100.0},
+            ),
+            total_value=1000.0,
+        )
+
+        self.assertEqual([note["symbol"] for note in notes], ["SOXL"])
+
+    def test_formats_projected_allocation_drift_notes_through_i18n(self):
+        messages = format_small_account_allocation_drift_notes(
+            (
+                {
+                    "kind": "small_account_allocation_drift",
+                    "symbol": "SOXL",
+                    "projected_weight": 0.3740,
+                    "target_weight": 0.3500,
+                    "drift_weight": 0.0240,
+                },
+            ),
+            translator=lambda key, **kwargs: {
+                "small_account_allocation_drift": "📏 整数股偏离：若本轮订单全部成交，{details}",
+                "small_account_allocation_drift_detail": (
+                    "{symbol} 预计 {projected_weight} vs 目标 {target_weight}（{drift_weight}）"
+                ),
+            }.get(key, key).format(**kwargs),
+        )
+
+        self.assertEqual(
+            messages,
+            ("📏 整数股偏离：若本轮订单全部成交，SOXL.US 预计 37.4% vs 目标 35.0%（+2.4pp）",),
+        )
+
 
 if __name__ == "__main__":
     unittest.main()