Fix Firstrade buying-power sizing and dry-run isolation

application/execution_service.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from math import floor
 from typing import Any
 
 from quant_platform_kit.common.order_status import compute_confirmed_sell_release_value
@@ -139,6 +140,7 @@ class ExecutionCycleResult:
 DEFAULT_SAFE_HAVEN_CASH_SUBSTITUTE_THRESHOLD_USD = 1000.0
 SMALL_ACCOUNT_SAFE_HAVEN_CASH_SUBSTITUTE_LIMIT_USD = 2000.0
 MIN_NOTIONAL_BUY_USD = 1.0
+NOTIONAL_BUY_CASH_UTILIZATION_RATIO = 0.98
 _ACCEPTED_ORDER_STATUSES = frozenset(
     {"accepted", "filled", "partiallyfilled", "previewed", "submitted"}
 )
@@ -567,6 +569,19 @@ def _submit_notional_buy_order(
     }
 
 
+def _apply_notional_cash_buffer(*, buy_budget: float, investable_cash: float) -> float:
+    budget = max(0.0, float(buy_budget or 0.0))
+    available = max(0.0, float(investable_cash or 0.0))
+    if budget + 0.005 < available:
+        return budget
+    buffered_available = floor(
+        available * NOTIONAL_BUY_CASH_UTILIZATION_RATIO * 100
+    ) / 100
+    if budget >= MIN_NOTIONAL_BUY_USD and available >= MIN_NOTIONAL_BUY_USD:
+        buffered_available = max(buffered_available, MIN_NOTIONAL_BUY_USD)
+    return min(budget, buffered_available)
+
+
 def _order_submission_accepted(order: dict[str, Any]) -> bool:
     status = "".join(
         ch for ch in str(order.get("status") or "").strip().lower() if ch.isalnum()
@@ -770,6 +785,10 @@ def execute_value_target_plan(
             if order_notional_cap is not None:
                 buy_budget = min(buy_budget, order_notional_cap)
             if notional_buy_execution:
+                buy_budget = _apply_notional_cash_buffer(
+                    buy_budget=buy_budget,
+                    investable_cash=investable_cash,
+                )
                 if buy_budget >= MIN_NOTIONAL_BUY_USD:
                     estimated_buy_cost += buy_budget
                 elif float(delta_value) >= MIN_NOTIONAL_BUY_USD:
@@ -820,6 +839,10 @@ def execute_value_target_plan(
         if order_notional_cap is not None:
             buy_budget = min(buy_budget, order_notional_cap)
         if notional_buy_execution:
+            buy_budget = _apply_notional_cash_buffer(
+                buy_budget=buy_budget,
+                investable_cash=investable_cash,
+            )
             if buy_budget < MIN_NOTIONAL_BUY_USD:
                 skipped.append(
                     {

application/runtime_broker_adapters.py

@@ -110,6 +110,21 @@ def _positive_or_none(value: float | None) -> float | None:
     return resolved if resolved > 0.0 else None
 
 
+def _resolve_buying_power(
+    *,
+    cash_balance: float | None,
+    reported_buying_power: float | None,
+    cash_only_execution: bool,
+) -> float | None:
+    if not cash_only_execution:
+        return reported_buying_power if reported_buying_power is not None else cash_balance
+    if cash_balance is None:
+        return None
+    if reported_buying_power is None:
+        return cash_balance
+    return max(0.0, min(float(cash_balance), float(reported_buying_power)))
+
+
 def _resolve_total_equity(
     *,
     balances,
@@ -281,8 +296,10 @@ def build_portfolio_snapshot(self) -> PortfolioSnapshot:
             )
         cash_balance = _first_numeric_by_keyword_groups(balances, _CASH_BALANCE_KEYWORD_GROUPS)
         reported_buying_power = _first_numeric_by_keyword_groups(balances, _BUYING_POWER_KEYWORD_GROUPS)
-        buying_power = cash_balance if self.cash_only_execution else (
-            reported_buying_power if reported_buying_power is not None else cash_balance
+        buying_power = _resolve_buying_power(
+            cash_balance=cash_balance,
+            reported_buying_power=reported_buying_power,
+            cash_only_execution=self.cash_only_execution,
         )
         position_market_value = sum(position.market_value for position in positions)
         total_equity, total_equity_source = _resolve_total_equity(

application/strategy_run_persistence.py

@@ -85,9 +85,16 @@ def resolve_strategy_run_period(
     return f"{now.year:04d}-{now.month:02d}"
 
 
-def strategy_run_state_key(*, account: str, strategy_profile: str, run_period: str) -> str:
+def strategy_run_state_key(
+    *,
+    account: str,
+    strategy_profile: str,
+    run_period: str,
+    dry_run_only: bool = False,
+) -> str:
+    prefix = "strategy-runs/dry-run" if dry_run_only else "strategy-runs"
     return (
-        f"strategy-runs/{safe_key(account)}/{safe_key(strategy_profile)}/"
+        f"{prefix}/{safe_key(account)}/{safe_key(strategy_profile)}/"
         f"{safe_key(run_period)}/latest.json"
     )
 
@@ -99,10 +106,12 @@ def strategy_run_history_key(
     run_period: str,
     stage: str,
     now: datetime,
+    dry_run_only: bool = False,
 ) -> str:
     stamp = now.strftime("%Y%m%dT%H%M%SZ")
+    prefix = "strategy-runs/dry-run" if dry_run_only else "strategy-runs"
     return (
-        f"strategy-runs/{safe_key(account)}/{safe_key(strategy_profile)}/"
+        f"{prefix}/{safe_key(account)}/{safe_key(strategy_profile)}/"
         f"{safe_key(run_period)}/history/{now:%Y/%m/%d}/{stamp}-{safe_key(stage)}.json"
     )
 
@@ -189,11 +198,13 @@ def persist_strategy_run_state(
     strategy_profile = str(state.get("strategy_profile") or "unknown")
     run_period = str(state.get("run_period") or f"{as_of.year:04d}-{as_of.month:02d}")
     stage = str(state.get("stage") or "UNKNOWN")
+    dry_run_only = bool(state.get("dry_run_only"))
     store.write_json(
         strategy_run_state_key(
             account=account,
             strategy_profile=strategy_profile,
             run_period=run_period,
+            dry_run_only=dry_run_only,
         ),
         dict(state),
     )
@@ -204,6 +215,7 @@ def persist_strategy_run_state(
             run_period=run_period,
             stage=stage,
             now=as_of,
+            dry_run_only=dry_run_only,
         ),
         dict(state),
     )

entrypoints/cloud_run.py

@@ -17,6 +17,11 @@ def is_market_open_now(*, calendar_name="NYSE", timezone_name="America/New_York"
         schedule = calendar.schedule(start_date=now_market.date(), end_date=now_market.date())
         if schedule.empty:
             return False, None
-        return calendar.open_at_time(schedule, now_market), None
+        try:
+            return calendar.open_at_time(schedule, now_market), None
+        except ValueError as exc:
+            if "not covered by the schedule" not in str(exc):
+                raise
+            return False, None
     except Exception as exc:
         return False, exc

scripts/reconcile_cloud_runtime.py

@@ -3,9 +3,7 @@
 
 This script keeps the runtime logic minimal and explicit:
 - reconcile Cloud Run traffic to the latest ready revision and verify commit-sha
-- delete only explicit legacy session-check Cloud Scheduler jobs
-
-It deliberately does not touch probe/precheck bridge jobs or unknown schedulers.
+- delete only explicit legacy Cloud Scheduler jobs
 """
 from __future__ import annotations
 
@@ -52,9 +50,28 @@ def _first_target(plan: Mapping[str, Any]) -> Mapping[str, Any]:
 def _resolve_context(env: Mapping[str, str] = os.environ) -> tuple[RuntimeContext, dict[str, Any]]:
     plan = _parse_sync_plan(str(env.get("SYNC_PLAN_JSON", "") or ""))
     target = _first_target(plan)
+    configured_service = str(env.get("CLOUD_RUN_SERVICE", "") or "").strip()
+    targets = plan.get("targets")
+    if configured_service and isinstance(targets, list) and targets:
+        matching_targets = [
+            candidate
+            for candidate in targets
+            if isinstance(candidate, Mapping)
+            and configured_service
+            in {
+                str(candidate.get("service_name") or "").strip(),
+                str(candidate.get("service") or "").strip(),
+                str(candidate.get("cloud_run_service") or "").strip(),
+            }
+        ]
+        if len(matching_targets) != 1:
+            raise ValueError(
+                f"CLOUD_RUN_SERVICE {configured_service} does not match any sync-plan target"
+            )
+        target = matching_targets[0]
 
     service_name = (
-        str(env.get("CLOUD_RUN_SERVICE", "") or "").strip()
+        configured_service
         or str(target.get("service_name") or "").strip()
         or str(target.get("service") or "").strip()
         or str(target.get("cloud_run_service") or "").strip()
@@ -256,11 +273,18 @@ def reconcile_traffic(
         time.sleep(5)
 
 
-def _legacy_session_check_jobs(service_name: str) -> list[str]:
+def _legacy_scheduler_jobs(service_name: str) -> list[str]:
     candidates = [f"{service_name}-session-check-scheduler"]
     alias = service_name.removesuffix("-service")
     if alias and alias != service_name:
-        candidates.append(f"{alias}-session-check-scheduler")
+        candidates.extend(
+            [
+                f"{alias}-session-check-scheduler",
+                f"{alias}-probe-scheduler",
+                f"{alias}-precheck-scheduler",
+            ]
+        )
+    candidates.append("firstrade-monitor-dispatcher-scheduler")
     seen: list[str] = []
     for candidate in candidates:
         if candidate not in seen:
@@ -273,9 +297,53 @@ def cleanup_legacy_scheduler_jobs(
     *,
     run_gcloud: RunGcloud = _run_gcloud,
 ) -> None:
-    ctx, _plan = _resolve_context(env)
+    ctx, plan = _resolve_context(env)
     deleted: list[str] = []
-    for job_name in _legacy_session_check_jobs(ctx.service_name):
+    legacy_jobs = _legacy_scheduler_jobs(ctx.service_name)
+    dispatcher_job = "firstrade-monitor-dispatcher-scheduler"
+    direct_jobs = (
+        f"{ctx.service_name}-probe-scheduler",
+        f"{ctx.service_name}-precheck-scheduler",
+    )
+    targets = plan.get("targets")
+    has_single_sync_target = not str(env.get("SYNC_PLAN_JSON", "") or "").strip() or (
+        isinstance(targets, list) and len(targets) == 1
+    )
+    migration_confirmed = (
+        str(env.get("DIRECT_MONITOR_MIGRATION_COMPLETE") or "").strip() == "true"
+    )
+    current_sync_confirmed = (
+        str(env.get("DIRECT_MONITOR_SCHEDULERS_RECONCILED") or "").strip().lower()
+        == "true"
+    )
+    direct_jobs_exist = (
+        migration_confirmed
+        and current_sync_confirmed
+        and has_single_sync_target
+        and all(
+            run_gcloud(
+                [
+                    "scheduler",
+                    "jobs",
+                    "describe",
+                    job_name,
+                    "--project",
+                    ctx.project_id,
+                    "--location",
+                    ctx.scheduler_location,
+                ]
+            ).returncode
+            == 0
+            for job_name in direct_jobs
+        )
+    )
+    if dispatcher_job in legacy_jobs and not direct_jobs_exist:
+        legacy_jobs.remove(dispatcher_job)
+        print(
+            f"Keeping legacy Cloud Scheduler job {dispatcher_job} until direct monitor jobs exist."
+        )
+
+    for job_name in legacy_jobs:
         result = run_gcloud(
             [
                 "scheduler",
@@ -317,7 +385,7 @@ def build_parser() -> argparse.ArgumentParser:
     subparsers = parser.add_subparsers(dest="command", required=True)
     subparsers.add_parser("traffic", help="reconcile Cloud Run traffic")
     subparsers.add_parser(
-        "scheduler-cleanup", help="delete explicit legacy session-check scheduler jobs"
+        "scheduler-cleanup", help="delete explicit legacy scheduler jobs"
     )
     return parser
 

tests/test_cloud_run_entrypoint.py

@@ -0,0 +1,17 @@
+from entrypoints import cloud_run
+
+
+def test_market_hours_after_close_is_closed_without_calendar_error(monkeypatch):
+    class FakeSchedule:
+        empty = False
+
+    class FakeCalendar:
+        def schedule(self, **_kwargs):
+            return FakeSchedule()
+
+        def open_at_time(self, _schedule, _now):
+            raise ValueError("The provided timestamp is not covered by the schedule")
+
+    monkeypatch.setattr(cloud_run.mcal, "get_calendar", lambda _name: FakeCalendar())
+
+    assert cloud_run.is_market_open_now() == (False, None)

tests/test_execution_service.py

@@ -4,6 +4,7 @@
 from datetime import datetime, timezone
 
 from application.execution_service import (
+    _apply_notional_cash_buffer,
     execute_value_target_plan,
     substitute_small_safe_haven_targets_with_cash,
 )
@@ -582,6 +583,33 @@ def test_execute_value_target_plan_uses_notional_buy_when_enabled():
     assert result.execution_notes == ()
 
 
+def test_notional_buy_keeps_cash_buffer_when_order_would_use_all_available_cash():
+    execution_port = FakeExecutionPort()
+    result = execute_value_target_plan(
+        plan={
+            "allocation": {"targets": {"IBIT": 150.0}},
+            "portfolio": {
+                "market_values": {"IBIT": 70.0},
+                "quantities": {"IBIT": 2.0},
+                "liquid_cash": 80.0,
+                "total_equity": 150.0,
+            },
+            "execution": {"current_min_trade": 1.0, "investable_cash": 80.0},
+        },
+        market_data_port=FakeMarketDataPort({"IBIT": 35.0}),
+        execution_port=execution_port,
+        dry_run_only=False,
+        notional_buy_execution=True,
+    )
+
+    assert result.action_done is True
+    assert execution_port.orders[0].metadata["notional_usd"] == 78.4
+
+
+def test_notional_cash_buffer_preserves_minimum_eligible_order():
+    assert _apply_notional_cash_buffer(buy_budget=1.02, investable_cash=1.02) == 1.0
+
+
 def test_execute_value_target_plan_routes_rejected_notional_buy_to_skipped_orders():
     class RejectedExecutionPort(FakeExecutionPort):
         def submit_order(self, order_intent) -> ExecutionReport:
@@ -622,7 +650,7 @@ def submit_order(self, order_intent) -> ExecutionReport:
     assert result.action_done is False
     assert result.submitted_orders == ()
     assert result.skipped_orders[0]["reason"] == "fractional_trading_disclosure_required"
-    assert result.skipped_orders[0]["notional_usd"] == 80.0
+    assert result.skipped_orders[0]["notional_usd"] == 78.4
     assert result.execution_notes == ()