Add LongBridge fractional order API probe

Author: Pigbibi

src/quant_platform_kit/longbridge/execution.py

@@ -12,6 +12,7 @@ def estimate_max_purchase_quantity(
     *,
     order_kind: str,
     ref_price: float,
+    fractional_shares: bool = False,
 ) -> float:
     from longport.openapi import OrderSide, OrderType
 
@@ -21,6 +22,7 @@ def estimate_max_purchase_quantity(
         order_type=order_type,
         side=OrderSide.Buy,
         price=Decimal(str(ref_price)),
+        fractional_shares=bool(fractional_shares),
     )
     cash_max_qty = getattr(response, "cash_max_qty", 0)
     return max(0.0, float(Decimal(str(cash_max_qty or "0"))))

tests/test_longbridge_execution.py

@@ -53,6 +53,26 @@ def test_estimate_max_purchase_quantity(self) -> None:
             quantity = estimate_max_purchase_quantity(ctx, "SOXL.US", order_kind="limit", ref_price=100.5)
 
         self.assertEqual(quantity, 12)
+        self.assertIs(ctx.estimate_kwargs["fractional_shares"], False)
+
+    def test_estimate_max_purchase_quantity_can_request_fractional_buying_power(self) -> None:
+        longport_module = types.ModuleType("longport")
+        openapi_module = types.ModuleType("longport.openapi")
+        openapi_module.OrderSide = types.SimpleNamespace(Buy="Buy")
+        openapi_module.OrderType = types.SimpleNamespace(LO="LO", MO="MO")
+
+        ctx = FakeTradeContext()
+        with patch.dict(sys.modules, {"longport": longport_module, "longport.openapi": openapi_module}):
+            quantity = estimate_max_purchase_quantity(
+                ctx,
+                "SOXX.US",
+                order_kind="limit",
+                ref_price=495.91,
+                fractional_shares=True,
+            )
+
+        self.assertEqual(quantity, 12)
+        self.assertIs(ctx.estimate_kwargs["fractional_shares"], True)
 
     def test_submit_order(self) -> None:
         longport_module = types.ModuleType("longport")
@@ -68,6 +88,20 @@ def test_submit_order(self) -> None:
         self.assertEqual(report.status, "submitted")
         self.assertEqual(report.broker_order_id, "OID-1")
 
+    def test_submit_order_allows_decimal_quantity_at_or_above_one_share(self) -> None:
+        longport_module = types.ModuleType("longport")
+        openapi_module = types.ModuleType("longport.openapi")
+        openapi_module.OrderSide = types.SimpleNamespace(Buy="Buy", Sell="Sell")
+        openapi_module.OrderType = types.SimpleNamespace(LO="LO", MO="MO")
+        openapi_module.TimeInForceType = types.SimpleNamespace(Day="Day")
+
+        ctx = FakeTradeContext()
+        with patch.dict(sys.modules, {"longport": longport_module, "longport.openapi": openapi_module}):
+            report = submit_order(ctx, "SOXL.US", order_kind="limit", side="buy", quantity=1.5, submitted_price=100.25)
+
+        self.assertEqual(report.status, "submitted")
+        self.assertEqual(str(ctx.submit_args[3]), "1.5")
+
     def test_submit_order_rejects_quantity_below_one_before_api_call(self) -> None:
         longport_module = types.ModuleType("longport")
         openapi_module = types.ModuleType("longport.openapi")

tests/test_longbridge_fractional_order_api_probe.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import os
+import unittest
+from decimal import Decimal
+
+
+def _api_probe_enabled() -> bool:
+    return str(os.getenv("LONGBRIDGE_API_PROBE", "")).strip().lower() in {"1", "true", "yes", "on"}
+
+
+@unittest.skipUnless(
+    _api_probe_enabled(),
+    "Set LONGBRIDGE_API_PROBE=1 with HK simulated LongPort credentials to run live API probes.",
+)
+class LongBridgeFractionalOrderApiProbeTests(unittest.TestCase):
+    """Manual probe for LongBridge API quantity validation against a simulated account.
+
+    These tests intentionally call the broker API. Keep them skipped in normal CI.
+    """
+
+    symbol = os.getenv("LONGBRIDGE_API_PROBE_SYMBOL", "SOXX.US")
+    limit_price = Decimal(os.getenv("LONGBRIDGE_API_PROBE_LIMIT_PRICE", "0.01"))
+
+    def setUp(self) -> None:
+        try:
+            from longport.openapi import Config, TradeContext
+        except ImportError as exc:  # pragma: no cover - only relevant outside probe env
+            raise unittest.SkipTest("longport is required for API probes") from exc
+
+        missing = [
+            name
+            for name in ("LONGPORT_APP_KEY", "LONGPORT_APP_SECRET", "LONGPORT_ACCESS_TOKEN")
+            if not os.getenv(name)
+        ]
+        if missing:
+            raise unittest.SkipTest(f"Missing LongPort credentials: {', '.join(missing)}")
+
+        config = Config(
+            app_key=os.environ["LONGPORT_APP_KEY"],
+            app_secret=os.environ["LONGPORT_APP_SECRET"],
+            access_token=os.environ["LONGPORT_ACCESS_TOKEN"],
+        )
+        self.trade_context = TradeContext(config)
+
+    def _submit_limit_buy(self, quantity: Decimal):
+        from longport.openapi import OrderSide, OrderType, TimeInForceType
+
+        return self.trade_context.submit_order(
+            self.symbol,
+            OrderType.LO,
+            OrderSide.Buy,
+            quantity,
+            TimeInForceType.Day,
+            submitted_price=self.limit_price,
+            remark="qpk-fractional-api-probe",
+        )
+
+    def test_sub_one_fractional_order_is_rejected_by_openapi_quantity_validation(self) -> None:
+        from longport.openapi import OpenApiException
+
+        with self.assertRaises(OpenApiException) as raised:
+            self._submit_limit_buy(Decimal("0.4326"))
+
+        message = str(raised.exception)
+        self.assertIn("SubmittedQuantity", message)
+        self.assertIn("^([1-9]", message)
+
+    def test_fractional_order_at_or_above_one_share_can_be_submitted_then_cancelled(self) -> None:
+        response = self._submit_limit_buy(Decimal("1.5"))
+        order_id = str(getattr(response, "order_id", "") or "").strip()
+        self.assertTrue(order_id, "LongBridge accepted 1.5 quantity but did not return an order_id")
+
+        try:
+            self.trade_context.cancel_order(order_id)
+        except Exception as exc:  # pragma: no cover - preserves the original acceptance assertion
+            self.fail(f"LongBridge accepted 1.5 quantity but cancel failed for order_id={order_id}: {exc}")
+
+
+if __name__ == "__main__":
+    unittest.main()