Executor pattern, RobinhoodClient, deferred task scheduler

tests/state/test_state_manager.py

@@ -96,7 +96,8 @@ def test_side_and_order_id_preserved(self):
         ticker = mgr.get_ticker('BTC')
         assert ticker.orders[0].side == OrderSide.SELL
         assert ticker.orders[0].order_id == 'ORD-001'
-        assert ticker.orders[0].created_at == '2025-06-01 10:00:00'
+        from datetime import datetime
+        assert ticker.orders[0].created_at == datetime(2025, 6, 1, 10, 0)
         assert ticker.orders[1].side == OrderSide.BUY
         assert ticker.orders[1].order_id == 'ORD-002'
 

tests/test_momentum_dca.py

@@ -537,7 +537,7 @@ def test_created_at_passed_through(self):
              'quantity': 20, 'limit_price': 443.0, 'stop_price': 445.0,
              'created_at': '2026-02-07 10:30:00', 'order_id': 'abc-123'},
         ]
-        mgr.load_broker_sell_orders('SPY', broker_orders)
+        mgr.load_broker_orders('SPY', broker_orders)
         order = mgr.get_ticker('SPY').orders[0]
         assert order.created_at is not None
         assert order.created_at.year == 2026
@@ -549,7 +549,7 @@ def test_missing_created_at_is_none(self):
             {'symbol': 'SPY', 'side': 'SELL', 'order_type': 'Limit',
              'quantity': 10, 'limit_price': 460.0, 'stop_price': None},
         ]
-        mgr.load_broker_sell_orders('SPY', broker_orders)
+        mgr.load_broker_orders('SPY', broker_orders)
         order = mgr.get_ticker('SPY').orders[0]
         assert order.created_at is None
         assert order.order_id is None

tests/test_order_replacement.py

@@ -10,14 +10,14 @@
 def _make_system():
     """Build a TradingSystem with mocked dependencies so no real I/O occurs."""
     with patch('trading_system.main.TwelveDataProvider'), \
-         patch('trading_system.main.SafeCashBot') as MockBot:
-        bot_instance = MockBot.return_value
+         patch('trading_system.main.RobinhoodClient') as MockBot:
+        bot_instance = MockBot.create.return_value
         bot_instance.get_pdt_status.return_value = {
             'day_trade_count': 0,
             'flagged': False,
             'trades': [],
         }
-        bot_instance.cancel_order_by_id.return_value = True
+        bot_instance.cancel_order.return_value = True
 
         system = TradingSystem(
             twelve_data_api_key='fake',
@@ -94,7 +94,7 @@ def test_both_sells_cancelled_and_replaced(self):
         system._handle_order_replacement('SPY', signal, symbol_orders)
 
         # Both sells should be cancelled
-        calls = system.trading_bot.cancel_order_by_id.call_args_list
+        calls = system.trading_bot.cancel_order.call_args_list
         cancelled_ids = {c[0][0] for c in calls}
         assert cancelled_ids == {'S1', 'S2'}
 
@@ -109,7 +109,7 @@ def test_replaces_all_lot_orders(self):
 
         system._handle_order_replacement('SPY', signal, symbol_orders)
 
-        calls = system.trading_bot.cancel_order_by_id.call_args_list
+        calls = system.trading_bot.cancel_order.call_args_list
         cancelled_ids = {c[0][0] for c in calls}
         assert cancelled_ids == {'SELL-OLD', 'BUY-OLD'}
 
@@ -124,7 +124,7 @@ def test_qty_mismatch_still_cancels_all(self):
         system._handle_order_replacement('SPY', signal, symbol_orders)
 
         # Both should be cancelled regardless of quantity
-        calls = system.trading_bot.cancel_order_by_id.call_args_list
+        calls = system.trading_bot.cancel_order.call_args_list
         cancelled_ids = {c[0][0] for c in calls}
         assert cancelled_ids == {'SELL-001', 'BUY-001'}
 
@@ -146,7 +146,7 @@ def test_pdt_count_2_alerts_and_skips(self, mock_slack):
 
         mock_slack.assert_called_once()
         assert 'PDT day trade count at 2/3' in mock_slack.call_args[0][0]
-        system.trading_bot.cancel_order_by_id.assert_not_called()
+        system.trading_bot.cancel_order.assert_not_called()
 
 
 class TestPdtFlaggedAlertsAndSkips:
@@ -166,7 +166,7 @@ def test_pdt_flagged_alerts_and_skips(self, mock_slack):
 
         mock_slack.assert_called_once()
         assert 'PDT FLAGGED' in mock_slack.call_args[0][0]
-        system.trading_bot.cancel_order_by_id.assert_not_called()
+        system.trading_bot.cancel_order.assert_not_called()
 
 
 class TestPdtSafeProceeds:
@@ -186,7 +186,7 @@ def test_pdt_safe_proceeds(self, mock_slack):
 
         mock_slack.assert_not_called()
         # Both sell and buy should be cancelled
-        assert system.trading_bot.cancel_order_by_id.call_count == 2
+        assert system.trading_bot.cancel_order.call_count == 2
 
 
 class TestPdtNoneProceeds:
@@ -200,15 +200,15 @@ def test_pdt_none_proceeds(self):
         system._handle_order_replacement('SPY', signal, symbol_orders)
 
         # Both sell and buy should be cancelled
-        assert system.trading_bot.cancel_order_by_id.call_count == 2
+        assert system.trading_bot.cancel_order.call_count == 2
 
 
 class TestCancelFailsStillPlaces:
     def test_cancel_fails_still_places(self):
-        """cancel_order_by_id returns False → placement still proceeds
+        """cancel_order returns False → placement still proceeds
         (order likely already filled/cancelled)."""
         system = _make_system()
-        system.trading_bot.cancel_order_by_id.return_value = False
+        system.trading_bot.cancel_order.return_value = False
         signal = _make_signal()
         symbol_orders = [_sell_order(), _buy_order()]
 
@@ -235,9 +235,9 @@ def test_momentum_pricing_used(self):
 
         system._handle_order_replacement('SPY', signal, symbol_orders)
 
-        # strategy defaults: stop_offset_pct=0.0125, buy_offset=0.50
-        expected_stop = round(current_price * (1 - 0.0125), 2)  # 493.75
-        expected_buy = round(expected_stop - 0.50, 2)  # 493.25
+        # strategy defaults: stop_offset_pct=0.015, buy_offset=0.20
+        expected_stop = round(current_price * (1 - 0.015), 2)  # 492.50
+        expected_buy = round(expected_stop - 0.20, 2)  # 492.30
 
         sell_call = system._execute_stop_limit_sell_order.call_args
         assert sell_call[0][1]['stop_price'] == expected_stop
@@ -264,7 +264,7 @@ def test_stop_limit_sell_cancels_existing_sell(self):
         system.process_signal('SPY', signal, open_orders)
 
         # Existing sell should be cancelled before new pair is placed
-        system.trading_bot.cancel_order_by_id.assert_called_once_with('EXISTING-SELL')
+        system.trading_bot.cancel_order.assert_called_once_with('EXISTING-SELL')
         system._execute_stop_limit_sell_order.assert_called_once()
         system._execute_paired_limit_buy.assert_called_once()
 
@@ -284,7 +284,7 @@ def test_uses_target_qty_for_new_order(self):
         system.process_signal('SPY', signal, open_orders)
 
         # Old sell cancelled
-        system.trading_bot.cancel_order_by_id.assert_called_once_with('OLD-SELL')
+        system.trading_bot.cancel_order.assert_called_once_with('OLD-SELL')
 
         # New sell uses target_qty (250), not gap_qty (100)
         sell_order = system._execute_stop_limit_sell_order.call_args[0][1]
@@ -323,6 +323,6 @@ def test_non_lot_sized_sell_is_cancelled(self):
         system.process_signal('SPY', signal, open_orders)
 
         # Non-lot-sized sell should still be cancelled
-        system.trading_bot.cancel_order_by_id.assert_called_once_with('PARTIAL-SELL')
+        system.trading_bot.cancel_order.assert_called_once_with('PARTIAL-SELL')
         system._execute_stop_limit_sell_order.assert_called_once()
         system._execute_paired_limit_buy.assert_called_once()

trading_system/execution/trade_executor.py

@@ -0,0 +1,153 @@
+"""
+Executor — execution quality layer + deferred order queue.
+
+Sits between the strategy (which creates TradeTasks) and the Brokerage
+(which talks to the broker API). Owns:
+  - PDT gate, spread check, price optimizer, fill logger
+  - In-memory deferred queue for PDT-blocked orders
+
+The Brokerage is injected (Dependency Inversion), so the Executor never
+imports robin_stocks or any platform-specific code.
+"""
+
+from datetime import date, timedelta
+from typing import Optional
+
+from trading_system.execution.trade_task import TradeTask, DeferredTask
+
+
+def _next_trading_day(from_date: date) -> date:
+    d = from_date + timedelta(days=1)
+    while d.weekday() >= 5:  # skip Saturday=5, Sunday=6
+        d += timedelta(days=1)
+    return d
+
+
+class Executor:
+    """
+    Processes TradeTasks. Applies pre-flight checks, dispatches to the
+    Brokerage, and defers PDT-blocked orders to the next trading day.
+    """
+
+    def __init__(self, brokerage, pdt_gate=None, spread_checker=None,
+                 price_optimizer=None, fill_logger=None, fill_auditor=None):
+        self._brokerage = brokerage
+        self._pdt_gate = pdt_gate
+        self._spread_checker = spread_checker
+        self._price_optimizer = price_optimizer
+        self._fill_logger = fill_logger
+        self._fill_auditor = fill_auditor
+        self._deferred_queue: list[DeferredTask] = []
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def submit(self, task: TradeTask) -> Optional[dict]:
+        """
+        Try to execute a task now.
+        Returns broker response on success, None if blocked or deferred.
+        """
+        can_execute, reason = task.should_execute()
+        if not can_execute:
+            self._enqueue_deferred(task, reason)
+            return None
+
+        if self._pdt_gate:
+            allowed, pdt_reason = self._pdt_gate.can_place_order(task.symbol, task.side)
+            print(f"   PDT Gate: {pdt_reason}")
+            if not allowed:
+                self._enqueue_deferred(task, pdt_reason)
+                return None
+
+        if self._spread_checker:
+            spread_info = self._spread_checker.check_spread(task.symbol)
+            if spread_info and not spread_info.get('is_acceptable', True):
+                print(f"   Spread Check: BLOCKED — {spread_info.get('reason', '')}")
+                return None
+            if spread_info and spread_info.get('should_wait'):
+                print(f"   Spread Check: WARNING — {spread_info.get('reason', '')}")
+
+        return self._dispatch(task)
+
+    def drain_deferred(self):
+        """
+        Re-attempt deferred tasks whose execute_date has been reached.
+        Call at the top of each engine cycle.
+        """
+        ready = [t for t in self._deferred_queue if t.should_execute()[0]]
+        submitted = []
+        for task in ready:
+            print(f"   [executor] Retrying deferred {task.symbol} {task.side} "
+                  f"(was: {task.deferred_reason})")
+            result = self.submit(task)
+            if result:
+                submitted.append(task)
+        for task in submitted:
+            self._deferred_queue.remove(task)
+
+    def get_deferred(self) -> list[DeferredTask]:
+        """Read-only view of the current deferred queue."""
+        return list(self._deferred_queue)
+
+    # ------------------------------------------------------------------
+    # Internal
+    # ------------------------------------------------------------------
+
+    def _dispatch(self, task: TradeTask) -> Optional[dict]:
+        """Send the task to the brokerage after pre-flight passes."""
+        submission_id = None
+        bid, ask = None, None
+
+        if self._fill_auditor:
+            nbbo = self._fill_auditor.get_nbbo_now(task.symbol)
+            if nbbo:
+                bid = nbbo.get('bid')
+                ask = nbbo.get('ask')
+                mid = nbbo.get('mid')
+                if mid and mid > 0:
+                    print(f"   NBBO: bid=${bid:.2f} ask=${ask:.2f} mid=${mid:.2f}")
+
+        if self._fill_logger:
+            submission_id = self._fill_logger.log_submission(
+                task.symbol, task.side, task.price, bid, ask)
+
+        try:
+            if task.side == 'buy':
+                result = self._brokerage.place_limit_buy(
+                    task.symbol, task.quantity, task.price)
+            elif task.order_type == 'stop_limit':
+                result = self._brokerage.place_stop_limit_sell(
+                    task.symbol, task.quantity, task.stop_price, task.price)
+            else:
+                result = self._brokerage.place_limit_sell(
+                    task.symbol, task.quantity, task.price)
+        except Exception as e:
+            if self._fill_logger and submission_id:
+                self._fill_logger.log_cancel(submission_id, reason=str(e))
+            raise
+
+        if result and self._fill_logger and submission_id:
+            order_id = result.get('id') if isinstance(result, dict) else None
+            if order_id:
+                self._fill_logger.log_fill(submission_id, task.price)
+
+        return result
+
+    def _enqueue_deferred(self, task: TradeTask, reason: str):
+        next_day = _next_trading_day(date.today())
+        retry_count = getattr(task, 'retry_count', 0) + 1
+        deferred = DeferredTask(
+            symbol=task.symbol,
+            side=task.side,
+            quantity=task.quantity,
+            price=task.price,
+            order_type=task.order_type,
+            stop_price=getattr(task, 'stop_price', None),
+            execute_date=next_day,
+            checks=["pdt_gate"],
+            deferred_reason=reason,
+            retry_count=retry_count,
+        )
+        print(f"   [executor] Deferred {task.symbol} {task.side} → {next_day} | {reason}")
+        self._deferred_queue.append(deferred)

trading_system/execution/trade_task.py

@@ -0,0 +1,59 @@
+"""
+TradeTask — Command Pattern base for deferred and scheduled orders.
+
+Each subclass implements should_execute() (Strategy Pattern) so the
+Executor can ask "is this task ready?" without knowing which type
+it holds.
+"""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from datetime import date
+
+
+class TradeTask(ABC):
+    """Base command. Holds trade data; does not execute itself."""
+
+    @abstractmethod
+    def should_execute(self) -> tuple[bool, str]:
+        """Returns (can_execute, reason)."""
+        ...
+
+
+@dataclass
+class ScheduledTask(TradeTask):
+    """Executes on or after execute_date."""
+
+    symbol: str
+    side: str           # "buy" | "sell"
+    quantity: float
+    price: float
+    order_type: str     # "limit" | "stop_limit" | "market"
+    execute_date: date
+    stop_price: float | None = None
+
+    def should_execute(self) -> tuple[bool, str]:
+        if date.today() >= self.execute_date:
+            return True, "scheduled date reached"
+        return False, f"scheduled for {self.execute_date}"
+
+
+@dataclass
+class DeferredTask(TradeTask):
+    """Blocked by PDT or another gate. Retried each cycle once execute_date is reached."""
+
+    symbol: str
+    side: str
+    quantity: float
+    price: float
+    order_type: str
+    execute_date: date
+    checks: list[str]       # gates to re-run: ["pdt_gate", "spread_check"]
+    deferred_reason: str = ""
+    stop_price: float | None = None
+    retry_count: int = 0
+
+    def should_execute(self) -> tuple[bool, str]:
+        if date.today() >= self.execute_date:
+            return True, "deferred date reached"
+        return False, f"deferred until {self.execute_date}"