Skip to content

Fix PDT gate to detect same-day filled orders (critical PDT protection) #45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
IamJasonBian wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix PDT gate to detect same-day filled orders (critical PDT protection) #45

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
28c2235
Use state-logs blob store for state uploads
IamJasonBian Mar 2, 2026
cfdf627
Fix PDT gate to detect same-day filled orders (not just open orders)
IamJasonBian Mar 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 118 additions & 0 deletions tests/test_pdt_guard.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
"""

@IamJasonBian IamJasonBian Jul 13, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a manual script, not a pytest test. It requires live Robinhood credentials, uses sys.path.insert hacks, and prints results instead of asserting. It will fail in any CI environment and doesn't validate behavior deterministically.

Suggestion: replace with proper unit tests that mock SafeCashBot methods (e.g., get_recent_orders, get_recent_option_orders, get_pdt_status) and assert on PDTGate.can_place_order return values. You can delete this file and still have better test coverage.

Generated by Claude Code

Test PDT Gate with current live positions and fills
"""

import sys
import os
from datetime import date

# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from utils.safe_cash_bot import SafeCashBot
from trading_system.execution.pdt_gate import PDTGate


def test_pdt_guard():
"""Test PDT guard against current live positions and fills."""
print("\n" + "="*70)
print("PDT GATE TEST - LIVE DATA")
print("="*70 + "\n")

# Initialize bot and PDT gate
bot = SafeCashBot()
pdt_gate = PDTGate(trading_bot=bot)

# Get recent fills (stock and options)
print("Fetching recent fills (last 1 day)...\n")
recent_stock_orders = bot.get_recent_orders(days=1)
recent_option_orders = bot.get_recent_option_orders(days=1)

stock_fills = [o for o in recent_stock_orders if o['state'] in ('filled', 'confirmed')]
option_fills = [o for o in recent_option_orders if o['state'] == 'filled']

print(f"Stock fills today: {len(stock_fills)}")
print(f"Option fills today: {len(option_fills)}\n")

# Display stock fills
if stock_fills:
print("STOCK FILLS TODAY:")
for fill in stock_fills:
print(f" {fill['side']} {fill['symbol']} x{fill['quantity']:.0f} @ ${fill.get('average_price', 0):.2f}")
print(f" Filled: {fill.get('last_transaction_at', 'N/A')}")
print()

# Display option fills
if option_fills:
print("OPTION FILLS TODAY:")
for fill in option_fills:
for leg in fill.get('legs', []):
print(f" {leg['side']} {leg['position_effect']} {leg['chain_symbol']} "
f"${leg['strike']:.2f} {leg['option_type'].upper()} exp {leg['expiration']}")
print(f" Premium: ${fill['processed_premium']:.2f}")
print(f" Updated: {fill['updated_at']}")
print()

# Test PDT checks for various scenarios
print("="*70)
print("PDT CHECK SCENARIOS")
print("="*70 + "\n")

# Collect unique symbols from fills
test_symbols = set()
for fill in stock_fills:
test_symbols.add(fill['symbol'])
for fill in option_fills:
for leg in fill.get('legs', []):
test_symbols.add(leg['chain_symbol'])

if not test_symbols:
print("No fills today - testing with common symbols: BTC, SPY, QQQ\n")
test_symbols = {'BTC', 'SPY', 'QQQ'}

# Test each symbol for both buy and sell
for symbol in sorted(test_symbols):
print(f"Testing {symbol}:")

# Test BUY
can_buy, buy_reason = pdt_gate.can_place_order(symbol, 'buy')
status = "✅ ALLOWED" if can_buy else "🚫 BLOCKED"
print(f" BUY: {status}")
print(f" {buy_reason}")

# Test SELL
can_sell, sell_reason = pdt_gate.can_place_order(symbol, 'sell')
status = "✅ ALLOWED" if can_sell else "🚫 BLOCKED"
print(f" SELL: {status}")
print(f" {sell_reason}")
print()

# Summary
print("="*70)
print("PDT GUARD STATUS")
print("="*70 + "\n")

pdt_info = bot.get_pdt_status()
if pdt_info:
count = pdt_info.get('day_trade_count', 0)
flagged = pdt_info.get('flagged', False)

print(f"Day Trades Used: {count}/3")
print(f"PDT Flagged: {'YES ⚠️' if flagged else 'NO ✅'}")
print(f"Remaining Day Trades: {3 - count}")

if flagged:
print("\n⚠️ ACCOUNT IS PDT FLAGGED")
print(" Only position-closing sells are allowed")
elif count >= 2:
print("\n⚠️ WARNING: Only 1 day trade remaining")
else:
print(f"\n✅ Safe to trade ({3 - count} day trades remaining)")
else:
print("❌ Could not fetch PDT status")

print("\n" + "="*70 + "\n")


if __name__ == "__main__":
test_pdt_guard()
117 changes: 107 additions & 10 deletions trading_system/execution/pdt_gate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,9 @@ class PDTGate:

def __init__(self, trading_bot=None):
self._trading_bot = trading_bot
# Cache of same-day transactions to prevent round trips
# Format: {symbol: {'buy': date_str, 'sell': date_str}}
self._same_day_fills = {}

def can_place_order(self, symbol: str, side: str) -> tuple:
"""Check if an order is safe to place from a PDT perspective.
Expand Down Expand Up @@ -59,7 +62,13 @@ def can_place_order(self, symbol: str, side: str) -> tuple:
f"PDT warning: {day_trade_count}/3 day trades used, "
f"this {side} on {symbol} may create another (2/3)")

# Step 4: Safe to proceed
# Step 4: Would create day trade but we have capacity
if would_dt and day_trade_count == 0:
return (True,
f"⚠️ WILL CREATE DAY TRADE on {symbol} (opposite side filled today) — "
f"allowed ({day_trade_count}/3 used)")

# Step 5: Safe to proceed
return (True, f"PDT OK: {day_trade_count}/3 day trades")

except Exception as e:
Expand All @@ -69,29 +78,117 @@ def _would_create_day_trade(self, symbol: str, side: str) -> bool:
"""Check if placing this order would create a day trade (round trip).

Returns True if there's a same-day opposite fill for this symbol.
Checks both:
1. Recent filled orders from Robinhood API
2. Local cache of same-day fills (updated on successful order placement)

Defensive: returns True on API failure (assume worst case).
"""
try:
open_orders = self._trading_bot.get_open_orders()
if not open_orders:
return False
# Clear old cached fills from previous days
self._clear_old_fills()

today = date.today().isoformat()
opposite_side = 'SELL' if side.lower() == 'buy' else 'BUY'
opposite_side = 'sell' if side.lower() == 'buy' else 'buy'

for order in open_orders:
# Check 1: Local cache (updated when orders fill)
if symbol in self._same_day_fills:
if opposite_side in self._same_day_fills[symbol]:
cached_date = self._same_day_fills[symbol][opposite_side]
if cached_date == today:
print(f" [pdt_gate] CACHE HIT: {symbol} {opposite_side.upper()} filled today")
return True

# Check 2: Recent filled stock orders from API (last 1 day)
recent_orders = self._trading_bot.get_recent_orders(days=1)

for order in recent_orders:
if order.get('symbol') != symbol:
continue
order_side = order.get('side', '').upper()

# Only check filled/confirmed orders
order_state = order.get('state', '').lower()
if order_state not in ('filled', 'confirmed'):
continue

# Check if it's the opposite side
order_side = order.get('side', '').lower()
if order_side != opposite_side:
continue
# Check if created today
created = order.get('created_at', '')
if isinstance(created, str) and created.startswith(today):

# Check if it FILLED today (not just created)
# Use last_transaction_at (when filled) instead of created_at
filled_at = order.get('last_transaction_at') or order.get('updated_at')
if isinstance(filled_at, str) and filled_at.startswith(today):
print(f" [pdt_gate] API HIT: {symbol} {order_side.upper()} filled today at {filled_at}")
# Update cache for future checks
self._record_fill(symbol, order_side)
return True

# Check 3: Recent filled option orders (options also count for PDT)
try:
recent_option_orders = self._trading_bot.get_recent_option_orders(days=1)

for order in recent_option_orders:
if order.get('state', '').lower() != 'filled':
continue

# Check each leg for same-day open+close on the same underlying
for leg in order.get('legs', []):
if leg.get('chain_symbol') != symbol:
continue

position_effect = leg.get('position_effect', '').lower()
leg_side = leg.get('side', '').lower()

# Map option position_effect to stock buy/sell equivalent
# BUY to open = debit (like buying stock)
# SELL to close = credit (like selling stock)
option_equiv_side = None
if position_effect == 'open':
option_equiv_side = 'buy' if leg_side == 'buy' else 'sell'
elif position_effect == 'close':
option_equiv_side = 'sell' if leg_side == 'sell' else 'buy'

if option_equiv_side == opposite_side:
filled_at = order.get('updated_at', '')
if isinstance(filled_at, str) and filled_at.startswith(today):

@IamJasonBian IamJasonBian Jul 13, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_clear_old_fills can be simplified. The double-loop-with-deferred-deletes pattern is verbose. A dict comprehension does the same thing in fewer lines:

def _clear_old_fills(self):
    today = date.today().isoformat()
    self._same_day_fills = {
        sym: {side: d for side, d in sides.items() if d == today}
        for sym, sides in self._same_day_fills.items()
    }
    self._same_day_fills = {s: v for s, v in self._same_day_fills.items() if v}

Generated by Claude Code

print(f" [pdt_gate] OPTION HIT: {symbol} option {position_effect} filled today")
self._record_fill(symbol, option_equiv_side)
return True
except Exception as e:
print(f" [pdt_gate] Option order check failed (continuing): {e}")

return False

except Exception as e:
print(f" [pdt_gate] _would_create_day_trade error: {e}")
return True # Assume worst case on failure

def _record_fill(self, symbol: str, side: str):
"""Record that we filled a buy/sell for this symbol today."""
today = date.today().isoformat()
if symbol not in self._same_day_fills:
self._same_day_fills[symbol] = {}
self._same_day_fills[symbol][side.lower()] = today
print(f" [pdt_gate] Recorded {symbol} {side.upper()} fill for today")

def _clear_old_fills(self):
"""Remove fills from previous days to keep cache fresh."""
today = date.today().isoformat()
symbols_to_remove = []

for symbol in self._same_day_fills:
sides_to_remove = []
for side, fill_date in self._same_day_fills[symbol].items():
if fill_date != today:
sides_to_remove.append(side)

for side in sides_to_remove:
del self._same_day_fills[symbol][side]

if not self._same_day_fills[symbol]:
symbols_to_remove.append(symbol)

for symbol in symbols_to_remove:
del self._same_day_fills[symbol]
25 changes: 21 additions & 4 deletions trading_system/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,6 +236,24 @@ def process_signal(self, symbol: str, signal: Dict, open_orders: list = None):
if target_qty:
paired_buy['quantity'] = target_qty

# PDT safety check for paired orders
# Don't place opposite-side paired order if we already filled that side today
pdt_gate = getattr(self.trading_bot, '_pdt_gate', None)
if pdt_gate:
# Check if placing the paired buy would violate PDT (sell filled today)
can_place_buy, buy_reason = pdt_gate.can_place_order(symbol, 'buy')
# Check if placing the stop-limit sell would violate PDT (buy filled today)
can_place_sell, sell_reason = pdt_gate.can_place_order(symbol, 'sell')

if not can_place_buy or not can_place_sell:
print(f" 🚫 PDT BLOCKED paired orders for {symbol}:")
if not can_place_buy:
print(f" BUY: {buy_reason}")
if not can_place_sell:
print(f" SELL: {sell_reason}")
print(f" Skipping paired order placement to prevent PDT violation")
return

# Default: place buy first, then sell.
# If first existing order is a buy, place sell first instead.
sell_first = False
Expand Down Expand Up @@ -297,13 +315,12 @@ def _handle_order_replacement(self, symbol: str, signal: Dict, symbol_orders: li

# Build replacement orders using momentum pricing from the signal
current_price = signal['order'].get('current_price', 0)
stop_offset_pct = getattr(self.strategy, 'stop_offset_pct', 0.01)
buy_offset = getattr(self.strategy, 'buy_offset', 0.20)
buy_offset_pct = getattr(self.strategy, 'buy_offset_pct', 0.0125)
hedge_symbol_map = getattr(self.strategy, 'hedge_symbol_map', {})

stop_price = round(current_price * (1 - stop_offset_pct), 2)
stop_price = round(current_price + 1.00, 2)
limit_price = stop_price
buy_price = round(stop_price - buy_offset, 2)
buy_price = round(current_price * (1 - buy_offset_pct), 2)
order_symbol = hedge_symbol_map.get(symbol, symbol)
qty = lot_size

Expand Down
2 changes: 1 addition & 1 deletion trading_system/state/blob_logger.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@


NETLIFY_BLOBS_URL = "https://api.netlify.com/api/v1/blobs"
STORE_NAME = "order-book"
STORE_NAME = "state-logs" # Using state-logs while UI manages cut-overs
LOCAL_LOG_DIR = Path(__file__).resolve().parent.parent.parent / "state_logs"


Expand Down
Loading