Add TaskScheduler, PDT gate, and task classes

trading_system/execution/trade_executor.py

@@ -0,0 +1,95 @@
+"""
+Executor — submission gateway for TradeTask objects.
+
+Responsibilities:
+  - Run PDT gate before every order
+  - PDT-blocked tasks are deferred to the next trading day
+  - drain_deferred() is called at the top of each engine cycle
+"""
+from __future__ import annotations
+
+from datetime import date, timedelta
+
+from trading_system.execution.trade_task import TradeTask, ScheduledTask, DeferredTask
+from trading_system.execution.pdt_gate import PDTGate
+
+
+def _next_trading_day(today: date) -> date:
+    """Return the next weekday after today."""
+    delta = 1
+    while True:
+        candidate = today + timedelta(days=delta)
+        if candidate.weekday() < 5:  # Mon–Fri
+            return candidate
+        delta += 1
+
+
+class Executor:
+    """Submits TradeTask objects through the PDT gate; defers on block."""
+
+    def __init__(self, trading_bot):
+        self._bot = trading_bot
+        self._pdt_gate = PDTGate(trading_bot=trading_bot)
+        self._deferred: list[DeferredTask] = []
+
+    def submit(self, task: TradeTask) -> dict | None:
+        """Gate the task through PDT, then dispatch or defer."""
+        ready, reason = task.should_execute()
+        if not ready:
+            print(f"  [executor] task not ready: {reason}")
+            return None
+
+        allowed, pdt_reason = self._pdt_gate.can_place_order(task.symbol, task.side)
+        if not allowed:
+            print(f"  [executor] PDT blocked ({pdt_reason}) — deferring {task.symbol} {task.side}")
+            self._enqueue_deferred(task, pdt_reason)
+            return None
+
+        return self._dispatch(task)
+
+    def drain_deferred(self) -> None:
+        """Re-attempt deferred tasks whose execute_date has arrived."""
+        if not self._deferred:
+            return
+
+        remaining = []
+        for task in self._deferred:
+            ready, _ = task.should_execute()
+            if ready:
+                print(f"  [executor] retrying deferred {task.symbol} {task.side} "
+                      f"(attempt {task.retry_count + 1})")
+                result = self.submit(task)
+                if result is None:
+                    task.retry_count += 1
+                    remaining.append(task)
+            else:
+                remaining.append(task)
+        self._deferred = remaining
+
+    def _dispatch(self, task: TradeTask) -> dict | None:
+        """Call the appropriate broker method based on task type."""
+        try:
+            if task.order_type == "stop_limit":
+                return self._bot.place_stop_limit_sell(
+                    task.symbol, task.quantity, task.stop_price, task.price
+                )
+            elif task.side == "buy":
+                return self._bot.place_limit_buy(task.symbol, task.quantity, task.price)
+            else:
+                return self._bot.place_limit_sell(task.symbol, task.quantity, task.price)
+        except Exception as e:
+            print(f"  [executor] dispatch error for {task.symbol}: {e}")
+            return None
+
+    def _enqueue_deferred(self, task: TradeTask, reason: str) -> None:
+        deferred = DeferredTask(
+            symbol=task.symbol,
+            side=task.side,
+            quantity=task.quantity,
+            price=task.price,
+            order_type=task.order_type,
+            execute_date=_next_trading_day(date.today()),
+            deferred_reason=reason,
+            stop_price=getattr(task, "stop_price", None),
+        )
+        self._deferred.append(deferred)

trading_system/execution/trade_task.py

@@ -0,0 +1,51 @@
+"""
+Trade task classes — Command + Strategy pattern.
+
+TradeTask   : base ABC, defines should_execute()
+ScheduledTask: executes on/after execute_date
+DeferredTask : blocked by a gate (e.g. PDT), retried next trading day
+"""
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from datetime import date
+
+
+class TradeTask(ABC):
+    @abstractmethod
+    def should_execute(self) -> tuple[bool, str]: ...
+
+
+@dataclass
+class ScheduledTask(TradeTask):
+    symbol: str
+    side: str          # "buy" | "sell"
+    quantity: float
+    price: float
+    order_type: str    # "limit" | "stop_limit"
+    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):
+    symbol: str
+    side: str
+    quantity: float
+    price: float
+    order_type: str
+    execute_date: date
+    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}"

trading_system/main.py

@@ -5,7 +5,7 @@
 
 import os
 import sys
-from datetime import datetime
+from datetime import datetime, date
 from pathlib import Path
 from typing import List, Dict
 import time
@@ -25,6 +25,8 @@
 from trading_system.utils.slack import send_slack_alert  # noqa: E402
 from trading_system.entities.OrderType import OrderSide  # noqa: E402
 from utils.safe_cash_bot import SafeCashBot  # noqa: E402
+from trading_system.execution.trade_executor import Executor  # noqa: E402
+from trading_system.execution.trade_task import ScheduledTask  # noqa: E402
 
 
 class TradingSystem:
@@ -61,6 +63,7 @@ def __init__(self, twelve_data_api_key: str, symbols: List[str],
         self.metrics_calculator = MetricsCalculator()
         self.state_manager = StateManager()
         self.trading_bot = SafeCashBot()
+        self.executor = Executor(trading_bot=self.trading_bot)
 
         # Initialize execution quality layer
         self.fill_logger = None
@@ -598,6 +601,7 @@ def print_portfolio_allocation(self):
 
     def run_once(self):
         """Run trading system once for all symbols"""
+        self.executor.drain_deferred()
         if self.verbose:
             print(f"\n{'='*70}")
             print("RUNNING TRADING SYSTEM")

trading_system/state/blob_logger.py

@@ -73,6 +73,9 @@ def _serialize_state(state_manager, order_book=None, portfolio=None,
 
 def _serialize_value(obj):
     """JSON serializer for objects not serializable by default."""
+    import datetime
+    if isinstance(obj, (datetime.datetime, datetime.date)):
+        return obj.isoformat()
     if hasattr(obj, "value"):
         return obj.value
     raise TypeError(f"Object of type {type(obj)} is not JSON serializable")