From bc40a671806d482f389ac06dc3f1940a17a7745a Mon Sep 17 00:00:00 2001 From: Florian Egger Date: Tue, 26 May 2026 14:49:57 +0200 Subject: [PATCH] Refactor intraday trading system for correctness and maintainability Extract timestamp generation to shared helper, fix attention gating from broken softmax to sigmoid per-feature gates, and detach LSTM hidden state to prevent gradient accumulation. Backtester now uses real prices, proper position sizing, and sell logic instead of placeholder buys. Alpha calculation uses consistent geometric annualization. Add optional mark-to-market pricing to paper broker and close-position safety in real- time trader. Add sqlite3 import and fix pandas view warning in live data. --- live_trading.py | 40 +------------ src/data/live_data.py | 6 +- src/evaluation/intraday_backtester.py | 85 ++++++++++++++++++++++----- src/evaluation/metrics.py | 9 +-- src/models/intraday_gnn.py | 37 +++++------- src/trading/paper_broker.py | 12 +++- src/trading/real_time_trader.py | 20 +++++-- 7 files changed, 122 insertions(+), 87 deletions(-) diff --git a/live_trading.py b/live_trading.py index a917e65..f6f834f 100644 --- a/live_trading.py +++ b/live_trading.py @@ -1,7 +1,7 @@ import asyncio import logging import time -from datetime import datetime, timedelta +from datetime import datetime from typing import Dict, List import numpy as np @@ -15,6 +15,7 @@ from src.data.live_data import LiveDataService from src.data.pipeline import StockDataPipeline from src.models.intraday_gnn import IntradayGNN from src.models.trainer import GNNTrainer +from src.utils.helpers import generate_intraday_timestamps from src.utils.memory_manager import MemoryManager # Configure logging @@ -166,7 +167,7 @@ class LiveTradingSystem: continue # Get sequence of features - timestamps = self._generate_timestamps_for_date(current_date) + timestamps = generate_intraday_timestamps(current_date) current_idx = ( timestamps.index(timestamp) if timestamp in timestamps @@ -278,41 +279,6 @@ class LiveTradingSystem: return data - def _generate_timestamps_for_date(self, date: str) -> List[str]: - """Generate timestamps for a given date""" - # Get market open and close times - market_open = datetime.strptime(config.TRADING_HOURS["start"], "%H:%M").time() - market_close = datetime.strptime(config.TRADING_HOURS["end"], "%H:%M").time() - - # Create datetime objects for open and close - open_datetime = datetime.strptime(f"{date} {market_open}", "%Y-%m-%d %H:%M:%S") - close_datetime = datetime.strptime( - f"{date} {market_close}", "%Y-%m-%d %H:%M:%S" - ) - - # Generate timestamps based on trading frequency - if config.TRADING_FREQUENCY == "1min": - delta = timedelta(minutes=1) - elif config.TRADING_FREQUENCY == "5min": - delta = timedelta(minutes=5) - elif config.TRADING_FREQUENCY == "15min": - delta = timedelta(minutes=15) - elif config.TRADING_FREQUENCY == "30min": - delta = timedelta(minutes=30) - elif config.TRADING_FREQUENCY == "1h": - delta = timedelta(hours=1) - else: # Default to 1 minute - delta = timedelta(minutes=1) - - timestamps = [] - current = open_datetime - - while current <= close_datetime: - timestamps.append(current.strftime("%Y-%m-%d %H:%M:%S")) - current += delta - - return timestamps - def _get_feature_sequence(self, ticker: str, start: str, end: str): """Get feature sequence for a ticker between start and end timestamps""" if ( diff --git a/src/data/live_data.py b/src/data/live_data.py index b799e4b..0af500b 100644 --- a/src/data/live_data.py +++ b/src/data/live_data.py @@ -1,6 +1,7 @@ import asyncio import json import logging +import sqlite3 import time from datetime import datetime, timedelta from typing import Callable, Dict, List, Optional @@ -390,10 +391,11 @@ class LiveDataService: if ticker not in self.price_bars or self.price_bars[ticker].empty: return - # Get the most recent bars + # Get the most recent bars (.copy() prevents SettingWithCopyWarning + # and ensures assignments below write to this local frame, not a view) recent_bars = self.price_bars[ticker].iloc[ -config.REALTIME_FEATURE_WINDOW : - ] + ].copy() if len(recent_bars) < 5: # Need at least 5 bars for meaningful features return diff --git a/src/evaluation/intraday_backtester.py b/src/evaluation/intraday_backtester.py index 4c52cf6..2f6d313 100644 --- a/src/evaluation/intraday_backtester.py +++ b/src/evaluation/intraday_backtester.py @@ -3,9 +3,10 @@ Intraday backtesting framework. """ import logging -from typing import Dict, List, Tuple +from typing import Dict, List, Optional, Tuple import pandas as pd +import torch from config import config from src.trading.paper_broker import PaperTradingBroker @@ -21,6 +22,30 @@ class IntradayBacktester: self.pipeline = pipeline self.broker = PaperTradingBroker(initial_cash=config.INITIAL_CAPITAL) + def _get_current_price(self, ticker: str, date) -> Optional[float]: + """Look up the adjusted close price for a ticker on or before the given date.""" + price_df = self.pipeline.price_data.get(ticker) + if price_df is None or price_df.empty: + return None + try: + if date in price_df.index: + return float(price_df.loc[date]["Close"]) + idx = price_df.index.get_indexer([date], method="ffill")[0] + if idx >= 0: + return float(price_df.iloc[idx]["Close"]) + except Exception: + pass + return None + + def _portfolio_value(self, date) -> float: + """Compute total portfolio value using current market prices.""" + total = self.broker.cash + for ticker, qty in self.broker.positions.items(): + price = self._get_current_price(ticker, date) + if price: + total += qty * price + return total + def run_backtest(self, dataset: List) -> Tuple[pd.Series, List]: """Run an intraday backtest.""" logger.info(f"Starting intraday backtest with {len(dataset)} samples") @@ -30,26 +55,54 @@ class IntradayBacktester: trade_log = [] for data in dataset: - predictions = self.model(data) + date = getattr(data, "date", None) + + with torch.no_grad(): + predictions = self.model(data.to(config.DEVICE)) + + portfolio_val = self._portfolio_value(date) for i, ticker in enumerate(getattr(data, "tickers", [])): pred = predictions[i].item() + price = self._get_current_price(ticker, date) - if pred > 0.002: - order = { - "ticker": ticker, - "action": "buy", - "quantity": 100, - "price": 100, - "timestamp": getattr(data, "timestamp", ""), - "type": "market", - } - order_id = self.broker.submit_order(order) - if order_id: - trade_log.append({**order, "order_id": order_id}) + if price is None or price <= 0: + continue - account = self.broker.get_account_summary() - portfolio_values.append(account["total_value"]) + held_qty = self.broker.positions.get(ticker, 0) + + if held_qty > 0: + if pred < -0.002: + order = { + "ticker": ticker, + "action": "sell", + "quantity": held_qty, + "price": price, + "timestamp": getattr(data, "timestamp", ""), + "type": "market", + } + order_id = self.broker.submit_order(order) + if order_id: + trade_log.append({**order, "order_id": order_id}) + else: + if pred > 0.002: + max_spend = portfolio_val * config.MAX_POSITION_SIZE + quantity = int(max_spend / price) + if quantity <= 0: + continue + order = { + "ticker": ticker, + "action": "buy", + "quantity": quantity, + "price": price, + "timestamp": getattr(data, "timestamp", ""), + "type": "market", + } + order_id = self.broker.submit_order(order) + if order_id: + trade_log.append({**order, "order_id": order_id}) + + portfolio_values.append(self._portfolio_value(date)) timestamps.append(getattr(data, "timestamp", None)) portfolio_series = pd.Series(portfolio_values, index=timestamps) diff --git a/src/evaluation/metrics.py b/src/evaluation/metrics.py index 0dcce1b..703dc00 100644 --- a/src/evaluation/metrics.py +++ b/src/evaluation/metrics.py @@ -70,10 +70,11 @@ def calculate_performance_metrics( benchmark_variance = benchmark_returns.var() metrics["beta"] = covariance / benchmark_variance if benchmark_variance > 0 else 0 - # Alpha - metrics["alpha"] = ( - metrics["annualized_return"] - metrics["beta"] * benchmark_returns.mean() * 252 - ) + # Alpha — use same geometric annualization as portfolio so both are comparable + benchmark_annualized = (1 + benchmark_returns).prod() ** ( + 252 / len(benchmark_returns) + ) - 1 + metrics["alpha"] = metrics["annualized_return"] - metrics["beta"] * benchmark_annualized return metrics diff --git a/src/models/intraday_gnn.py b/src/models/intraday_gnn.py index 30c70d3..904482c 100644 --- a/src/models/intraday_gnn.py +++ b/src/models/intraday_gnn.py @@ -86,12 +86,13 @@ class IntradayGNN(nn.Module): # Initialize LSTM weights self._init_lstm_weights() - # Attention mechanism for final prediction + # Per-feature gating after GNN+LSTM (Softmax over a single element is always 1, + # so the original design collapsed to a scalar no-op; Sigmoid gives proper gates) self.attention = nn.Sequential( nn.Linear(config.HIDDEN_CHANNELS, config.HIDDEN_CHANNELS), nn.SiLU(), - nn.Linear(config.HIDDEN_CHANNELS, 1), - nn.Softmax(dim=1), + nn.Linear(config.HIDDEN_CHANNELS, config.HIDDEN_CHANNELS), + nn.Sigmoid(), ) # Final prediction layer @@ -151,14 +152,18 @@ class IntradayGNN(nn.Module): else: lstm_out, self.hidden_state = self.lstm(lstm_input) + # Detach hidden state so gradients don't accumulate across inference calls + if self.hidden_state is not None: + h, c = self.hidden_state + self.hidden_state = (h.detach(), c.detach()) + # Remove sequence dimension - lstm_out = lstm_out.squeeze(1) + lstm_out = lstm_out.squeeze(1) # (N, HIDDEN_CHANNELS) - # Apply attention to LSTM outputs - attention_weights = self.attention(lstm_out) - attended = (lstm_out * attention_weights).sum(dim=1, keepdim=True) + # Per-feature gating: sigmoid weights scale each channel independently + attention_weights = self.attention(lstm_out) # (N, HIDDEN_CHANNELS) + attended = lstm_out * attention_weights # (N, HIDDEN_CHANNELS) - # Final prediction return self.linear(attended) def _gnn_forward(self, x, edge_index, edge_attr): @@ -184,16 +189,8 @@ class IntradayGNN(nn.Module): """Get attention weights for interpretability""" x, edge_index, edge_attr = data.x, data.edge_index, data.edge_attr - # Process through the model up to attention - batch_size, seq_len, num_features = x.size() - temporal_features = [] - - for i in range(batch_size): - stock_sequence = x[i].unsqueeze(0) - temporal_feature = self.temporal_attention(stock_sequence) - temporal_features.append(temporal_feature) - - temporal_features = torch.cat(temporal_features, dim=0) + # TemporalAttention already handles batched input — no need to loop + temporal_features = self.temporal_attention(x) processed_features = self.feature_processor(temporal_features) x = self.conv1(processed_features, edge_index, edge_attr) @@ -204,6 +201,4 @@ class IntradayGNN(nn.Module): lstm_out, _ = self.lstm(lstm_input) lstm_out = lstm_out.squeeze(1) - attention_weights = self.attention(lstm_out) - - return attention_weights + return self.attention(lstm_out) diff --git a/src/trading/paper_broker.py b/src/trading/paper_broker.py index d437fd7..b53d97c 100644 --- a/src/trading/paper_broker.py +++ b/src/trading/paper_broker.py @@ -72,10 +72,16 @@ class PaperTradingBroker(Broker): """Get current simulated positions.""" return self.positions.copy() - def get_account_summary(self) -> Dict: - """Get simulated account summary using fill prices for position valuation.""" + def get_account_summary(self, prices: Optional[Dict[str, float]] = None) -> Dict: + """Get simulated account summary. + + Args: + prices: Current market prices for mark-to-market valuation. + Falls back to fill price for any ticker not in prices. + """ + effective_prices = prices or {} position_value = sum( - qty * self.position_prices.get(ticker, 0) + qty * effective_prices.get(ticker, self.position_prices.get(ticker, 0)) for ticker, qty in self.positions.items() ) total_value = self.cash + position_value diff --git a/src/trading/real_time_trader.py b/src/trading/real_time_trader.py index 7b98a1f..5277b1e 100644 --- a/src/trading/real_time_trader.py +++ b/src/trading/real_time_trader.py @@ -4,7 +4,7 @@ Real-time trader for live trading execution. import logging from datetime import datetime -from typing import Dict +from typing import Dict, Optional from config import config @@ -24,9 +24,17 @@ class RealTimeTrader: self.max_drawdown = 0.0 self.entry_times = {} + def _get_last_known_price(self, ticker: str) -> Optional[float]: + """Look up the most recent daily close for a ticker from the pipeline.""" + price_df = self.pipeline.price_data.get(ticker) + if price_df is not None and not price_df.empty: + return float(price_df.iloc[-1]["Close"]) + return None + def _calculate_position_size(self, ticker: str, price: float) -> int: """Calculate position size based on risk management rules.""" - account_summary = self.broker.get_account_summary() + # Pass current price so the broker marks this position to market + account_summary = self.broker.get_account_summary(prices={ticker: price}) total_value = account_summary.get("total_value", config.INITIAL_CAPITAL) max_position_value = total_value * config.MAX_POSITION_SIZE position_size = int(max_position_value / price) @@ -75,13 +83,17 @@ class RealTimeTrader: return False def _close_all_positions(self): - """Close all open positions.""" + """Close all open positions using last known price from the pipeline.""" for ticker in list(self.current_positions.keys()): + price = self._get_last_known_price(ticker) + if price is None or price <= 0: + logger.warning(f"Cannot close {ticker}: no price available, skipping") + continue order = { "ticker": ticker, "action": "sell", "quantity": self.current_positions[ticker], - "price": 0, # market order + "price": price, "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "type": "market", }