diff --git a/src/quant_platform_kit/common/small_account_compatibility.py b/src/quant_platform_kit/common/small_account_compatibility.py index 1a7d819..e25edab 100644 --- a/src/quant_platform_kit/common/small_account_compatibility.py +++ b/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], diff --git a/tests/test_small_account_compatibility.py b/tests/test_small_account_compatibility.py index 8879469..a2988bf 100644 --- a/tests/test_small_account_compatibility.py +++ b/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()