Fix shared broker runtime behavior (#40)

Author: Pigbibi

src/quant_platform_kit/common/cash_sweep.py

@@ -0,0 +1,43 @@
+"""Shared helpers for cash sweep execution flows."""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Iterable
+
+
+def estimate_cash_sweep_sale_quantity_to_fund_buy(
+    max_quantity: int,
+    cash_sweep_price: float,
+    base_buying_power: float,
+    funding_needs: Iterable[tuple[float, float]],
+) -> int:
+    """Estimate how much cash sweep symbol to sell to fund the first buy candidate.
+
+    The helper keeps platform-specific data fetching outside of shared logic.
+    Each funding need is provided as ``(underweight_value, ask_price)``.
+    """
+    if max_quantity <= 0:
+        return 0
+    sweep_price = float(cash_sweep_price or 0.0)
+    if sweep_price <= 0.0:
+        return 0
+    current_buying_power = max(0.0, float(base_buying_power or 0.0))
+
+    for underweight_value, ask_price in funding_needs:
+        needed_value = float(underweight_value or 0.0)
+        quote_price = float(ask_price or 0.0)
+        if needed_value <= 0.0 or quote_price <= 0.0:
+            continue
+        max_buy_quantity = int(needed_value // quote_price)
+        if max_buy_quantity <= 0:
+            continue
+        required_buying_power = max_buy_quantity * quote_price
+        if current_buying_power >= required_buying_power:
+            return 0
+        return min(
+            int(max_quantity),
+            max(1, math.ceil((required_buying_power - current_buying_power) / sweep_price)),
+        )
+    return 0
+

src/quant_platform_kit/common/runtime_inputs.py

@@ -8,6 +8,9 @@
 from .models import PortfolioSnapshot, Position
 
 
+DEFAULT_SEMICONDUCTOR_ROTATION_HISTORY_LOOKBACK = 420
+
+
 def _normalize_symbols(strategy_symbols: Iterable[str]) -> tuple[str, ...]:
     return tuple(
         str(symbol).strip().upper()
@@ -204,6 +207,19 @@ def build_semiconductor_rotation_indicators_from_history(
     }
 
 
+def required_semiconductor_rotation_history_lookback(
+    *,
+    trend_ma_window: int = 140,
+    dynamic_rsi_quantile_window: int = 252,
+    minimum_lookback: int = DEFAULT_SEMICONDUCTOR_ROTATION_HISTORY_LOOKBACK,
+) -> int:
+    return max(
+        int(minimum_lookback),
+        int(trend_ma_window) + 20,
+        int(dynamic_rsi_quantile_window) + 28,
+    )
+
+
 def build_semiconductor_rotation_inputs_from_history(
     *,
     soxl_history: Iterable[float],

src/quant_platform_kit/ibkr/runtime_inputs.py

@@ -11,6 +11,7 @@
 )
 from quant_platform_kit.common.runtime_inputs import (
     build_semiconductor_rotation_indicators_from_history,
+    required_semiconductor_rotation_history_lookback,
 )
 
 
@@ -73,13 +74,11 @@ def build_semiconductor_rotation_indicators(
     historical_close_loader: Callable[..., Any],
     *,
     trend_ma_window: int = 140,
-    lookback_buffer: int = 20,
     dynamic_rsi_quantile_window: int = 252,
 ) -> dict[str, dict[str, float]]:
-    effective_lookback = max(
-        420,
-        int(trend_ma_window) + int(lookback_buffer),
-        int(dynamic_rsi_quantile_window) + int(lookback_buffer) + 90,
+    effective_lookback = required_semiconductor_rotation_history_lookback(
+        trend_ma_window=trend_ma_window,
+        dynamic_rsi_quantile_window=dynamic_rsi_quantile_window,
     )
     soxl_history = historical_close_loader(
         ib,
@@ -106,15 +105,13 @@ def build_semiconductor_rotation_inputs(
     historical_close_loader: Callable[..., Any],
     *,
     trend_ma_window: int = 140,
-    lookback_buffer: int = 20,
     dynamic_rsi_quantile_window: int = 252,
 ) -> dict[str, dict[str, dict[str, float]]]:
     return {
         "derived_indicators": build_semiconductor_rotation_indicators(
             ib,
             historical_close_loader,
             trend_ma_window=trend_ma_window,
-            lookback_buffer=lookback_buffer,
             dynamic_rsi_quantile_window=dynamic_rsi_quantile_window,
         )
     }

src/quant_platform_kit/longbridge/market_data.py

@@ -7,6 +7,7 @@
 
 from quant_platform_kit.common.runtime_inputs import (
     build_semiconductor_rotation_indicators_from_history,
+    required_semiconductor_rotation_history_lookback,
 )
 
 
@@ -79,7 +80,14 @@ def calculate_rotation_indicators(
 ) -> dict[str, dict[str, float]] | None:
     from longport.openapi import AdjustType, Period
 
-    effective_lookback = lookback if lookback is not None else max(280, trend_window + 20, dynamic_rsi_quantile_window + 28)
+    effective_lookback = (
+        lookback
+        if lookback is not None
+        else required_semiconductor_rotation_history_lookback(
+            trend_ma_window=trend_window,
+            dynamic_rsi_quantile_window=dynamic_rsi_quantile_window,
+        )
+    )
     soxl_bars = q_ctx.candlesticks("SOXL.US", Period.Day, effective_lookback, AdjustType.ForwardAdjust)
     soxx_bars = q_ctx.candlesticks("SOXX.US", Period.Day, effective_lookback, AdjustType.ForwardAdjust)
     if not soxl_bars or not soxx_bars:

src/quant_platform_kit/longbridge/portfolio.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import time
 from typing import Any, Callable, Iterable
 
 from .market_data import fetch_last_prices
@@ -11,19 +12,61 @@ def fetch_strategy_account_state(
     strategy_assets: Iterable[str],
     *,
     position_log_fn: Callable[[str], None] | None = None,
+    warning_log_fn: Callable[[str], None] | None = None,
 ) -> dict[str, Any]:
+    def warn(message: str) -> None:
+        if warning_log_fn is not None:
+            warning_log_fn(message)
+
+    def load_account_balance() -> tuple[Any, ...]:
+        attempts = (
+            ("all", {}),
+            ("USD", {"currency": "USD"}),
+            ("HKD", {"currency": "HKD"}),
+            ("CNH", {"currency": "CNH"}),
+        )
+        errors: list[str] = []
+        for label, kwargs in attempts:
+            for attempt in range(1, 4):
+                try:
+                    account_balance = t_ctx.account_balance(**kwargs)
+                except TypeError as exc:
+                    errors.append(f"{label}=TypeError:{exc}")
+                    break
+                except Exception as exc:
+                    errors.append(f"{label}[attempt={attempt}]={type(exc).__name__}:{exc}")
+                    if attempt < 3:
+                        warn(
+                            "[longbridge_account_balance_retrying] "
+                            f"currency={label} attempt={attempt}/3 error_type={type(exc).__name__}"
+                        )
+                        time.sleep(0.5 * attempt)
+                        continue
+                    break
+
+                if kwargs:
+                    warn(f"[longbridge_account_balance_retry_succeeded] currency={label}")
+                return tuple(account_balance or ())
+
+        if errors:
+            warn("[longbridge_account_balance_failed] " + " | ".join(errors))
+        return ()
+
     available_cash = 0.0
     cash_by_currency: dict[str, float] = {}
-    account_balance = t_ctx.account_balance()
+    account_balance = load_account_balance()
     for account in account_balance:
+        account_buy_power = max(0.0, float(getattr(account, "buy_power", 0.0) or 0.0))
+        account_usd_cash = 0.0
         for cash_info in getattr(account, "cash_infos", []):
             currency = str(getattr(cash_info, "currency", "") or "").strip().upper()
             if not currency:
                 continue
             cash_amount = float(getattr(cash_info, "available_cash", 0.0))
             cash_by_currency[currency] = cash_by_currency.get(currency, 0.0) + cash_amount
             if currency == "USD":
-                available_cash += cash_amount
+                account_usd_cash += cash_amount
+        available_cash += max(account_buy_power, account_usd_cash)
 
     assets = [str(symbol).strip().upper() for symbol in strategy_assets if str(symbol).strip()]
     market_values = {symbol: 0.0 for symbol in assets}
@@ -32,7 +75,25 @@ def fetch_strategy_account_state(
     filter_enabled = bool(assets)
 
     position_rows: list[tuple[str, str, Any, Any]] = []
-    positions_response = t_ctx.stock_positions()
+    positions_response = None
+    position_errors: list[str] = []
+    for attempt in range(1, 4):
+        try:
+            positions_response = t_ctx.stock_positions()
+            break
+        except Exception as exc:
+            position_errors.append(f"attempt={attempt} {type(exc).__name__}:{exc}")
+            if attempt < 3:
+                warn(
+                    "[longbridge_stock_positions_retrying] "
+                    f"attempt={attempt}/3 error_type={type(exc).__name__}"
+                )
+                time.sleep(0.5 * attempt)
+                continue
+            warn(
+                "[longbridge_stock_positions_failed] "
+                f"errors={' | '.join(position_errors)}"
+            )
     if positions_response and hasattr(positions_response, "channels"):
         for channel in positions_response.channels:
             for position in getattr(channel, "positions", []):

src/quant_platform_kit/schwab/portfolio.py

@@ -25,7 +25,7 @@ def fetch_account_snapshot(
     balances = account.get("currentBalances", {})
     cash_for_equity = float(balances.get("cashAvailableForTrading", 0.0))
     raw_withdrawable = float(balances.get("cashAvailableForWithdrawal", 0.0))
-    buying_power = max(0.0, raw_withdrawable)
+    buying_power = max(0.0, cash_for_equity)
 
     allowed_symbols = set(strategy_symbols)
     positions = []
@@ -47,9 +47,11 @@ def fetch_account_snapshot(
         as_of=datetime.utcnow(),
         total_equity=total_equity,
         buying_power=buying_power,
+        cash_balance=cash_for_equity,
         positions=tuple(positions),
         metadata={
             "account_hash": account_hash,
             "cash_available_for_trading": cash_for_equity,
+            "cash_available_for_withdrawal": raw_withdrawable,
         },
     )

tests/test_cash_sweep.py

@@ -0,0 +1,44 @@
+import sys
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+from quant_platform_kit.common.cash_sweep import estimate_cash_sweep_sale_quantity_to_fund_buy  # noqa: E402
+
+
+class CashSweepHelperTests(unittest.TestCase):
+    def test_returns_zero_when_buying_power_is_sufficient(self):
+        qty = estimate_cash_sweep_sale_quantity_to_fund_buy(
+            10,
+            100.0,
+            1000.0,
+            [(500.0, 10.0)],
+        )
+        self.assertEqual(qty, 0)
+
+    def test_scales_sale_quantity_to_fund_gap(self):
+        qty = estimate_cash_sweep_sale_quantity_to_fund_buy(
+            10,
+            100.0,
+            50.0,
+            [(500.0, 10.0)],
+        )
+        self.assertEqual(qty, 5)
+
+    def test_ignores_invalid_funding_needs(self):
+        qty = estimate_cash_sweep_sale_quantity_to_fund_buy(
+            10,
+            100.0,
+            50.0,
+            [(0.0, 10.0), (500.0, 0.0), (-1.0, -1.0)],
+        )
+        self.assertEqual(qty, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_longbridge_market_data.py

@@ -71,9 +71,9 @@ def test_calculate_rotation_indicators(self) -> None:
             indicators = calculate_rotation_indicators(FakeQuoteContext(), trend_window=150)
 
         self.assertIsNotNone(indicators)
-        self.assertEqual(indicators["soxl"]["price"], 379.0)
-        self.assertEqual(indicators["soxx"]["price"], 479.0)
-        self.assertAlmostEqual(indicators["soxx"]["ma20"], sum(200.0 + i for i in range(260, 280)) / 20)
+        self.assertEqual(indicators["soxl"]["price"], 519.0)
+        self.assertEqual(indicators["soxx"]["price"], 619.0)
+        self.assertAlmostEqual(indicators["soxx"]["ma20"], sum(200.0 + i for i in range(400, 420)) / 20)
         self.assertGreater(indicators["soxx"]["ma20_slope"], 0.0)
         self.assertEqual(indicators["soxx"]["rsi14"], 100.0)
         self.assertGreaterEqual(indicators["soxx"]["rsi14_dynamic_threshold"], 70.0)