Add fractional/notional DCA execution helpers and broker paths. (#114)

Introduce shared execution capability gating, Schwab DOLLARS orders,
LongBridge fractional buys, IBKR notional rejection, and verification script.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: Pigbibi

pyproject.toml

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

scripts/verify_fractional_dca_execution.py

@@ -3,7 +3,7 @@
 
 setup(
     name="quant-platform-kit",
-    version="0.7.40",
+    version="0.7.41",
     description="Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies.",
     package_dir={"": "src"},
     packages=find_packages(where="src"),

setup.py

@@ -4,7 +4,7 @@
 used by older strategy repositories.
 """
 
-__version__ = "0.7.40"
+__version__ = "0.7.41"
 
 from .common.models import (
     ExecutionReport,

src/quant_platform_kit/__init__.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from .strategies import PlatformCapabilityMatrix, StrategyCatalog, StrategyDefinition, normalize_profile_name
+
+FRACTIONAL_SHARE_EXECUTION_CAPABILITY = "fractional_share_execution"
+FRACTIONAL_SHARE_EXECUTION_SKIP_REASON = "fractional_share_execution_required"
+
+
+def definition_requires_fractional_share_execution(definition: StrategyDefinition) -> bool:
+    return FRACTIONAL_SHARE_EXECUTION_CAPABILITY in frozenset(definition.compatible_capabilities)
+
+
+def platform_supports_fractional_share_execution(*, capability_matrix: PlatformCapabilityMatrix) -> bool:
+    return FRACTIONAL_SHARE_EXECUTION_CAPABILITY in frozenset(capability_matrix.supported_capabilities)
+
+
+def fractional_share_execution_unsupported_reason(
+    profile: str,
+    *,
+    strategy_catalog: StrategyCatalog,
+    capability_matrix: PlatformCapabilityMatrix,
+) -> str | None:
+    normalized_profile = normalize_profile_name(profile)
+    definition = strategy_catalog.definitions.get(normalized_profile)
+    if definition is None:
+        return None
+    if definition_requires_fractional_share_execution(definition):
+        if not platform_supports_fractional_share_execution(capability_matrix=capability_matrix):
+            return FRACTIONAL_SHARE_EXECUTION_SKIP_REASON
+    return None

src/quant_platform_kit/common/execution_capabilities.py

@@ -201,6 +201,26 @@ def submit_order_intent(
     limit_order_factory: Callable[..., Any] | None = None,
 ) -> ExecutionReport:
     metadata = dict(order_intent.metadata or {})
+    notional_usd = metadata.get("notional_usd")
+    if (
+        notional_usd is not None
+        and not _is_option_intent(order_intent)
+        and not _is_combo_option_intent(order_intent)
+    ):
+        return ExecutionReport(
+            symbol=order_intent.symbol,
+            side=order_intent.side.lower(),
+            quantity=float(notional_usd),
+            status="rejected",
+            raw_payload={
+                "detail": (
+                    "IBKR TWS API does not support fractional or notional equity orders "
+                    f"(notional_usd={float(notional_usd):.2f})."
+                ),
+                "skip_reason": "ibkr_fractional_equity_api_unsupported",
+            },
+        )
+
     if _is_combo_option_intent(order_intent):
         contract = _build_option_combo_contract(
             ib,

src/quant_platform_kit/ibkr/execution.py

@@ -5,23 +5,30 @@
 
 from quant_platform_kit.common.models import ExecutionReport
 
+LONGBRIDGE_FRACTIONAL_QUANTITY_STEP = Decimal("0.0001")
+LONGBRIDGE_MIN_FRACTIONAL_BUY_QUANTITY = Decimal("0.0001")
+
 
 def estimate_max_purchase_quantity(
     t_ctx: Any,
     symbol: str,
     *,
     order_kind: str,
     ref_price: float,
+    fractional_shares: bool = False,
 ) -> float:
     from longport.openapi import OrderSide, OrderType
 
     order_type = OrderType.LO if order_kind == "limit" else OrderType.MO
-    response = t_ctx.estimate_max_purchase_quantity(
-        symbol=symbol,
-        order_type=order_type,
-        side=OrderSide.Buy,
-        price=Decimal(str(ref_price)),
-    )
+    estimate_kwargs: dict[str, Any] = {
+        "symbol": symbol,
+        "order_type": order_type,
+        "side": OrderSide.Buy,
+        "price": Decimal(str(ref_price)),
+    }
+    if fractional_shares:
+        estimate_kwargs["fractional_shares"] = True
+    response = t_ctx.estimate_max_purchase_quantity(**estimate_kwargs)
     cash_max_qty = getattr(response, "cash_max_qty", 0)
     return max(0.0, float(Decimal(str(cash_max_qty or "0"))))
 
@@ -34,13 +41,34 @@ def submit_order(
     side: str,
     quantity: float,
     submitted_price: float | None = None,
+    allow_fractional_shares: bool = False,
+    quantity_step: float = 1.0,
 ) -> ExecutionReport:
     from longport.openapi import OrderSide, OrderType, TimeInForceType
 
     order_type = OrderType.LO if order_kind == "limit" else OrderType.MO
     order_side = OrderSide.Buy if side == "buy" else OrderSide.Sell
     submitted_quantity = Decimal(str(quantity))
-    if submitted_quantity < Decimal("1"):
+    if side == "buy" and allow_fractional_shares:
+        min_buy_quantity = max(
+            LONGBRIDGE_MIN_FRACTIONAL_BUY_QUANTITY,
+            Decimal(str(quantity_step)),
+        )
+        if submitted_quantity < min_buy_quantity:
+            return ExecutionReport(
+                symbol=symbol.split(".")[0],
+                side=side,
+                quantity=float(quantity),
+                status="rejected",
+                raw_payload={
+                    "detail": (
+                        "LongBridge fractional buy submitted_quantity must be at least "
+                        f"{min_buy_quantity}; got {submitted_quantity}."
+                    ),
+                    "order_kind": order_kind,
+                },
+            )
+    elif submitted_quantity < Decimal("1"):
         return ExecutionReport(
             symbol=symbol.split(".")[0],
             side=side,
@@ -54,7 +82,12 @@ def submit_order(
                 "order_kind": order_kind,
             },
         )
-    if order_kind == "limit" and side == "buy" and submitted_quantity != submitted_quantity.to_integral_value():
+    if (
+        not allow_fractional_shares
+        and order_kind == "limit"
+        and side == "buy"
+        and submitted_quantity != submitted_quantity.to_integral_value()
+    ):
         return ExecutionReport(
             symbol=symbol.split(".")[0],
             side=side,

src/quant_platform_kit/longbridge/execution.py

@@ -4,25 +4,64 @@
 
 from quant_platform_kit.common.models import ExecutionReport, OrderIntent
 
+MIN_DOLLAR_BUY_NOTIONAL_USD = 1.0
+
+
+def build_equity_dollar_buy_market_order(symbol: str, notional_usd: float) -> dict[str, Any]:
+    notional = round(float(notional_usd), 2)
+    if notional < MIN_DOLLAR_BUY_NOTIONAL_USD:
+        raise ValueError(
+            f"Schwab dollar buy notional_usd must be at least {MIN_DOLLAR_BUY_NOTIONAL_USD:.2f}; got {notional:.2f}."
+        )
+    normalized_symbol = str(symbol or "").strip().upper()
+    if not normalized_symbol:
+        raise ValueError("Schwab dollar buy requires a non-empty symbol.")
+    return {
+        "orderType": "MARKET",
+        "session": "NORMAL",
+        "duration": "DAY",
+        "orderStrategyType": "SINGLE",
+        "orderLegCollection": [
+            {
+                "instruction": "BUY",
+                "quantity": notional,
+                "quantityType": "DOLLARS",
+                "instrument": {
+                    "symbol": normalized_symbol,
+                    "assetType": "EQUITY",
+                },
+            }
+        ],
+    }
 
-def submit_equity_order(api_client: Any, account_hash: str, order_intent: OrderIntent) -> ExecutionReport:
-    from schwab.orders.equities import equity_buy_limit, equity_buy_market, equity_sell_market
 
+def submit_equity_order(api_client: Any, account_hash: str, order_intent: OrderIntent) -> ExecutionReport:
     side = order_intent.side.lower()
     order_type = order_intent.order_type.lower()
+    metadata = dict(getattr(order_intent, "metadata", {}) or {})
+    notional_usd = metadata.get("notional_usd")
 
-    if side == "sell" and order_type == "market":
-        order = equity_sell_market(order_intent.symbol, order_intent.quantity)
-    elif side == "buy" and order_type == "market":
-        order = equity_buy_market(order_intent.symbol, order_intent.quantity)
-    elif side == "buy" and order_type == "limit":
-        if order_intent.limit_price is None:
-            raise ValueError("Limit buy orders require OrderIntent.limit_price.")
-        order = equity_buy_limit(order_intent.symbol, order_intent.quantity, f"{order_intent.limit_price:.2f}")
+    if side == "buy" and notional_usd is not None:
+        order = build_equity_dollar_buy_market_order(order_intent.symbol, float(notional_usd))
+        reported_quantity = float(notional_usd)
     else:
-        raise ValueError(
-            f"Unsupported Schwab order intent: side={order_intent.side!r}, order_type={order_intent.order_type!r}"
-        )
+        from schwab.orders.equities import equity_buy_limit, equity_buy_market, equity_sell_market
+
+        if side == "sell" and order_type == "market":
+            order = equity_sell_market(order_intent.symbol, order_intent.quantity)
+            reported_quantity = float(order_intent.quantity)
+        elif side == "buy" and order_type == "market":
+            order = equity_buy_market(order_intent.symbol, order_intent.quantity)
+            reported_quantity = float(order_intent.quantity)
+        elif side == "buy" and order_type == "limit":
+            if order_intent.limit_price is None:
+                raise ValueError("Limit buy orders require OrderIntent.limit_price.")
+            order = equity_buy_limit(order_intent.symbol, order_intent.quantity, f"{order_intent.limit_price:.2f}")
+            reported_quantity = float(order_intent.quantity)
+        else:
+            raise ValueError(
+                f"Unsupported Schwab order intent: side={order_intent.side!r}, order_type={order_intent.order_type!r}"
+            )
 
     response = api_client.place_order(account_hash, order)
     if response.status_code in (200, 201):
@@ -31,7 +70,7 @@ def submit_equity_order(api_client: Any, account_hash: str, order_intent: OrderI
         return ExecutionReport(
             symbol=order_intent.symbol,
             side=side,
-            quantity=float(order_intent.quantity),
+            quantity=reported_quantity,
             status="accepted",
             broker_order_id=order_id,
             raw_payload={"status_code": response.status_code},
@@ -40,7 +79,7 @@ def submit_equity_order(api_client: Any, account_hash: str, order_intent: OrderI
     return ExecutionReport(
         symbol=order_intent.symbol,
         side=side,
-        quantity=float(order_intent.quantity),
+        quantity=reported_quantity,
         status="rejected",
         raw_payload={
             "status_code": response.status_code,

src/quant_platform_kit/schwab/execution.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import unittest
+
+from quant_platform_kit.common.execution_capabilities import (
+    FRACTIONAL_SHARE_EXECUTION_CAPABILITY,
+    FRACTIONAL_SHARE_EXECUTION_SKIP_REASON,
+    fractional_share_execution_unsupported_reason,
+)
+from quant_platform_kit.common.strategies import (
+    PlatformCapabilityMatrix,
+    StrategyCatalog,
+    StrategyDefinition,
+    US_EQUITY_DOMAIN,
+    derive_eligible_profiles_for_platform,
+)
+from quant_platform_kit.common.strategy_contracts import StrategyRuntimeAdapter
+
+
+class ExecutionCapabilitiesTests(unittest.TestCase):
+    def test_fractional_share_execution_unsupported_reason(self) -> None:
+        catalog = StrategyCatalog(
+            definitions={
+                "ibit_smart_dca": StrategyDefinition(
+                    profile="ibit_smart_dca",
+                    domain=US_EQUITY_DOMAIN,
+                    supported_platforms=frozenset({"schwab"}),
+                    required_inputs=frozenset({"portfolio_snapshot"}),
+                    target_mode="value",
+                    compatible_capabilities=frozenset({FRACTIONAL_SHARE_EXECUTION_CAPABILITY}),
+                ),
+                "tqqq_growth_income": StrategyDefinition(
+                    profile="tqqq_growth_income",
+                    domain=US_EQUITY_DOMAIN,
+                    supported_platforms=frozenset({"schwab"}),
+                    required_inputs=frozenset({"portfolio_snapshot"}),
+                    target_mode="value",
+                ),
+            }
+        )
+        whole_share_matrix = PlatformCapabilityMatrix(
+            platform_id="schwab",
+            supported_domains=frozenset({US_EQUITY_DOMAIN}),
+            supported_target_modes=frozenset({"value"}),
+            supported_inputs=frozenset({"portfolio_snapshot"}),
+            supported_capabilities=frozenset(),
+        )
+        fractional_matrix = PlatformCapabilityMatrix(
+            platform_id="schwab",
+            supported_domains=frozenset({US_EQUITY_DOMAIN}),
+            supported_target_modes=frozenset({"value"}),
+            supported_inputs=frozenset({"portfolio_snapshot"}),
+            supported_capabilities=frozenset({FRACTIONAL_SHARE_EXECUTION_CAPABILITY}),
+        )
+
+        self.assertEqual(
+            fractional_share_execution_unsupported_reason(
+                "ibit_smart_dca",
+                strategy_catalog=catalog,
+                capability_matrix=whole_share_matrix,
+            ),
+            FRACTIONAL_SHARE_EXECUTION_SKIP_REASON,
+        )
+        self.assertIsNone(
+            fractional_share_execution_unsupported_reason(
+                "ibit_smart_dca",
+                strategy_catalog=catalog,
+                capability_matrix=fractional_matrix,
+            )
+        )
+        self.assertIsNone(
+            fractional_share_execution_unsupported_reason(
+                "tqqq_growth_income",
+                strategy_catalog=catalog,
+                capability_matrix=whole_share_matrix,
+            )
+        )
+
+    def test_capability_matrix_excludes_fractional_dca_profiles(self) -> None:
+        catalog = StrategyCatalog(
+            definitions={
+                "ibit_smart_dca": StrategyDefinition(
+                    profile="ibit_smart_dca",
+                    domain=US_EQUITY_DOMAIN,
+                    supported_platforms=frozenset({"schwab"}),
+                    required_inputs=frozenset({"portfolio_snapshot"}),
+                    target_mode="value",
+                    compatible_capabilities=frozenset({FRACTIONAL_SHARE_EXECUTION_CAPABILITY}),
+                ),
+            }
+        )
+        matrix = PlatformCapabilityMatrix(
+            platform_id="schwab",
+            supported_domains=frozenset({US_EQUITY_DOMAIN}),
+            supported_target_modes=frozenset({"value"}),
+            supported_inputs=frozenset({"portfolio_snapshot"}),
+            supported_capabilities=frozenset(),
+        )
+        adapters = {
+            "ibit_smart_dca": StrategyRuntimeAdapter(
+                available_inputs=frozenset({"portfolio_snapshot"}),
+                available_capabilities=frozenset({FRACTIONAL_SHARE_EXECUTION_CAPABILITY}),
+            ),
+        }
+
+        eligible = derive_eligible_profiles_for_platform(
+            catalog,
+            capability_matrix=matrix,
+            runtime_adapter_loader=lambda profile: adapters[profile],
+        )
+
+        self.assertEqual(eligible, frozenset())
+
+
+if __name__ == "__main__":
+    unittest.main()