diff --git a/trading_system/execution/trade_executor.py b/trading_system/execution/trade_executor.py new file mode 100644 index 0000000..2a582f7 --- /dev/null +++ b/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) diff --git a/trading_system/execution/trade_task.py b/trading_system/execution/trade_task.py new file mode 100644 index 0000000..5ec0867 --- /dev/null +++ b/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}" diff --git a/trading_system/main.py b/trading_system/main.py index 4865dca..3d39792 100644 --- a/trading_system/main.py +++ b/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") diff --git a/trading_system/state/blob_logger.py b/trading_system/state/blob_logger.py index 64baae9..c74054b 100644 --- a/trading_system/state/blob_logger.py +++ b/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")