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", }