Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/quant_platform_kit/longbridge/execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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"))))
Expand Down
34 changes: 34 additions & 0 deletions tests/test_longbridge_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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")
Expand Down
81 changes: 81 additions & 0 deletions tests/test_longbridge_fractional_order_api_probe.py
Original file line number Diff line number Diff line change
@@ -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()