""" Optimus Research Center - Laboratory 1: Quantitative & Financial Research Strictly experimental and simulated quantitative analysis module. STRICT BAN ON REAL TRADING: Enforced by design. Implements historical data import, technical indicators, non-leaking walk-forward validation, and commission/slippage-aware backtesting. """ import os import math import numpy as np import pandas as pd from typing import Dict, Any, List, Tuple, Optional from pathlib import Path from app.config import settings class RealTradingProhibitedError(RuntimeError): """Raised if any trading or order execution attempt is made.""" pass class QuantitativeResearchEngine: """Quantitative simulation and validation engine.""" DISCLAIMER = ( "STRICT RESEARCH & SIMULATION ONLY: This module performs historical backtesting " "and time-series analysis. Real order execution, broker connections, and live financial " "transactions are prohibited and physically impossible in this architecture." ) LIVE_TRADING_ENABLED = False BROKER_SAFETY_LOCK = True @classmethod def assert_no_live_trading( cls, live_order: bool = False, broker_endpoint: Optional[str] = None ) -> Dict[str, Any]: """Security guard guaranteeing no external trading broker or live order can be called.""" env_override = os.environ.get("ALLOW_LIVE_TRADING", "false").strip().lower() in {"1", "true", "yes"} if ( cls.LIVE_TRADING_ENABLED or not cls.BROKER_SAFETY_LOCK or live_order or bool(broker_endpoint) or env_override ): raise RealTradingProhibitedError( "SAFETY VIOLATION: Live financial trading and broker connections are strictly prohibited. " "QuantitativeResearchEngine operates exclusively in offline historical/synthetic backtesting mode." ) return { "live_trading_enabled": False, "broker_safety_lock": True, "mode": "OFFLINE_BACKTEST_ONLY" } @classmethod def generate_or_load_benchmark_series( cls, ticker: str = "BENCHMARK_SPY", n_days: int = 500, random_seed: int = 42, force_regenerate: bool = False ) -> pd.DataFrame: """ Loads or generates realistic benchmark historical price series with date, open, high, low, close, volume. """ cls.assert_no_live_trading() dataset_file = settings.datasets_path / f"{ticker.lower()}_historical.csv" if dataset_file.exists() and not force_regenerate: df = pd.read_csv(dataset_file) if len(df) >= n_days: df["date"] = pd.to_datetime(df["date"]) return df.iloc[:n_days].copy() # Generate realistic geometric Brownian motion benchmark with drift and volatility rng = np.random.RandomState(random_seed) dates = pd.date_range(end=pd.Timestamp.now(), periods=n_days, freq="B") initial_price = 100.0 daily_returns = rng.normal(loc=0.0004, scale=0.012, size=n_days) price_series = initial_price * np.exp(np.cumsum(daily_returns)) highs = price_series * (1 + rng.uniform(0.002, 0.015, size=n_days)) lows = price_series * (1 - rng.uniform(0.002, 0.015, size=n_days)) opens = lows + (highs - lows) * rng.uniform(0.2, 0.8, size=n_days) volumes = rng.randint(1000000, 50000000, size=n_days) df = pd.DataFrame({ "date": dates, "open": np.round(opens, 2), "high": np.round(highs, 2), "low": np.round(lows, 2), "close": np.round(price_series, 2), "volume": volumes }) df.to_csv(dataset_file, index=False) return df @classmethod def calculate_indicators(cls, df: pd.DataFrame) -> pd.DataFrame: """Calculates standard technical indicators without lookahead bias.""" data = df.copy() close = data["close"] # Simple Moving Averages data["sma_20"] = close.rolling(window=20).mean() data["sma_50"] = close.rolling(window=50).mean() # Exponential Moving Averages data["ema_12"] = close.ewm(span=12, adjust=False).mean() data["ema_26"] = close.ewm(span=26, adjust=False).mean() # MACD & Signal line data["macd"] = data["ema_12"] - data["ema_26"] data["macd_signal"] = data["macd"].ewm(span=9, adjust=False).mean() # Relative Strength Index (RSI - 14) delta = close.diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / (loss + 1e-9) data["rsi"] = 100 - (100 / (1 + rs)) # Bollinger Bands (20, 2 std) rolling_std = close.rolling(window=20).std() data["bb_upper"] = data["sma_20"] + (rolling_std * 2) data["bb_lower"] = data["sma_20"] - (rolling_std * 2) # Average True Range (ATR - 14) high_low = data["high"] - data["low"] high_close = (data["high"] - close.shift()).abs() low_close = (data["low"] - close.shift()).abs() true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) data["atr"] = true_range.rolling(window=14).mean() # Log returns data["log_return"] = np.log(close / close.shift(1)) return data.dropna().reset_index(drop=True) @classmethod def walk_forward_validation( cls, df: pd.DataFrame, n_splits: int = 5, train_window: int = 150, test_window: int = 30 ) -> Dict[str, Any]: """ Strict non-leaking Walk-Forward Analysis (WFA) / Expanding Window validation. Guarantees train data strictly precedes test data in time. """ cls.assert_no_live_trading() total_len = len(df) if total_len < train_window + test_window: raise ValueError(f"Dataset length {total_len} insufficient for WFA.") split_results = [] step = max(10, (total_len - (train_window + test_window)) // max(1, n_splits - 1)) for fold in range(n_splits): start_idx = fold * step train_end = start_idx + train_window test_end = min(total_len, train_end + test_window) if test_end <= train_end: break train_df = df.iloc[start_idx:train_end] test_df = df.iloc[train_end:test_end] # Fit baseline autoregressive trend on train y_train = train_df["close"].values X_train = np.arange(len(y_train)).reshape(-1, 1) # Fit OLS slope x_mean = np.mean(X_train) y_mean = np.mean(y_train) slope = np.sum((X_train.ravel() - x_mean) * (y_train - y_mean)) / np.sum((X_train.ravel() - x_mean) ** 2) intercept = y_mean - slope * x_mean # Forecast on out-of-sample test window (strictly future relative to train) X_test = (len(y_train) + np.arange(len(test_df))).reshape(-1, 1) y_pred = intercept + slope * X_test.ravel() y_actual = test_df["close"].values mse = float(np.mean((y_actual - y_pred) ** 2)) mae = float(np.mean(np.abs(y_actual - y_pred))) pct_err = float(np.mean(np.abs((y_actual - y_pred) / y_actual)) * 100) split_results.append({ "fold": fold + 1, "train_dates": f"{train_df['date'].iloc[0].strftime('%Y-%m-%d')} to {train_df['date'].iloc[-1].strftime('%Y-%m-%d')}", "test_dates": f"{test_df['date'].iloc[0].strftime('%Y-%m-%d')} to {test_df['date'].iloc[-1].strftime('%Y-%m-%d')}", "train_samples": len(train_df), "test_samples": len(test_df), "mse": round(mse, 4), "mae": round(mae, 4), "mape_percent": round(pct_err, 2) }) avg_mse = float(np.mean([r["mse"] for r in split_results])) avg_mae = float(np.mean([r["mae"] for r in split_results])) return { "validation_type": "WALK_FORWARD_EXPANDING_WINDOW", "leakage_prevented": True, "folds_evaluated": len(split_results), "splits": split_results, "average_mse": round(avg_mse, 4), "average_mae": round(avg_mae, 4), "uncertainty_margin_pct": round(float(np.std([r['mape_percent'] for r in split_results])), 2) } @classmethod def backtest_strategy( cls, df: pd.DataFrame, initial_capital: float = 100000.0, commission_bps: float = 10.0, # 0.10% per trade slippage_bps: float = 5.0 # 0.05% slippage ) -> Dict[str, Any]: """ Simulates technical trend strategy (SMA Golden Cross + RSI confirmation). Includes realistic transaction costs (commission & slippage). Compares strictly against Buy & Hold benchmark. """ cls.assert_no_live_trading() data = cls.calculate_indicators(df) if len(data) < 60: raise ValueError("Data insufficient for backtesting.") cash = initial_capital position_shares = 0 portfolio_values = [] trades = [] comm_pct = commission_bps / 10000.0 slip_pct = slippage_bps / 10000.0 for i in range(1, len(data)): row = data.iloc[i] prev = data.iloc[i - 1] price = row["close"] # Strategy signal: SMA20 crosses above SMA50 and RSI < 70 (Buy) # SMA20 crosses below SMA50 (Sell) buy_signal = (prev["sma_20"] <= prev["sma_50"]) and (row["sma_20"] > row["sma_50"]) and (row["rsi"] < 70) sell_signal = (prev["sma_20"] >= prev["sma_50"]) and (row["sma_20"] < row["sma_50"]) # Execute simulated buy if buy_signal and position_shares == 0: exec_price = price * (1 + slip_pct) shares_to_buy = int(cash / exec_price) if shares_to_buy > 0: cost = shares_to_buy * exec_price fee = cost * comm_pct if cash >= cost + fee: cash -= (cost + fee) position_shares = shares_to_buy trades.append({ "type": "BUY", "date": str(row["date"]), "shares": shares_to_buy, "price": round(exec_price, 2), "fee": round(fee, 2) }) # Execute simulated sell elif sell_signal and position_shares > 0: exec_price = price * (1 - slip_pct) revenue = position_shares * exec_price fee = revenue * comm_pct cash += (revenue - fee) trades.append({ "type": "SELL", "date": str(row["date"]), "shares": position_shares, "price": round(exec_price, 2), "fee": round(fee, 2) }) position_shares = 0 # Current total equity equity = cash + (position_shares * price) portfolio_values.append(equity) portfolio_values = np.array(portfolio_values) initial_val = initial_capital final_val = float(portfolio_values[-1]) total_return_pct = round(((final_val - initial_val) / initial_val) * 100, 2) # Benchmark: Buy and Hold bnh_shares = initial_capital / data["close"].iloc[0] bnh_final = float(bnh_shares * data["close"].iloc[-1]) bnh_return_pct = round(((bnh_final - initial_val) / initial_val) * 100, 2) # Drawdown calculation peaks = np.maximum.accumulate(portfolio_values) drawdowns = (portfolio_values - peaks) / peaks max_drawdown_pct = round(float(np.min(drawdowns)) * 100, 2) # Sharpe ratio daily_returns = np.diff(portfolio_values) / portfolio_values[:-1] std_returns = np.std(daily_returns) sharpe = round(float((np.mean(daily_returns) / (std_returns + 1e-9)) * np.sqrt(252)), 2) return { "disclaimer": cls.DISCLAIMER, "initial_capital": initial_capital, "final_portfolio_value": round(final_val, 2), "strategy_return_pct": total_return_pct, "benchmark_buy_and_hold_return_pct": bnh_return_pct, "excess_return_pct": round(total_return_pct - bnh_return_pct, 2), "max_drawdown_pct": max_drawdown_pct, "sharpe_ratio": sharpe, "total_trades_simulated": len(trades), "trades": trades[:10], # sample first 10 for audit "transaction_costs_modeled": { "commission_bps": commission_bps, "slippage_bps": slippage_bps } } @classmethod def walk_forward_backtest( cls, strategy_name: str = "momentum_crossover", train_ratio: float = 0.7, ticker: str = "BENCHMARK_SPY", n_days: int = 500, random_seed: int = 42, initial_capital: float = 100000.0, commission_bps: float = 10.0, slippage_bps: float = 5.0 ) -> Dict[str, Any]: """ Unified non-leaking Walk-Forward Validation + commission/slippage-aware Backtest invoked by ScientificCoordinator and AlgorithmLabAdapter for QUANTITATIVE_RESEARCH tasks. """ safety_state = cls.assert_no_live_trading() train_ratio = max(0.3, min(0.9, float(train_ratio))) df = cls.generate_or_load_benchmark_series( ticker=ticker, n_days=n_days, random_seed=random_seed ) train_window = max(60, int(len(df) * train_ratio * 0.5)) test_window = max(20, int(len(df) * (1.0 - train_ratio) * 0.25)) wfa = cls.walk_forward_validation( df, n_splits=5, train_window=train_window, test_window=test_window ) bt = cls.backtest_strategy( df, initial_capital=initial_capital, commission_bps=commission_bps, slippage_bps=slippage_bps ) return { "status": "COMPLETED", "strategy_name": strategy_name, "ticker": ticker, "train_ratio": round(train_ratio, 4), "random_seed": random_seed, "walk_forward": wfa, "backtest": bt, "strategy_return_pct": bt["strategy_return_pct"], "benchmark_buy_and_hold_return_pct": bt["benchmark_buy_and_hold_return_pct"], "excess_return_pct": bt["excess_return_pct"], "sharpe_ratio": bt["sharpe_ratio"], "max_drawdown_pct": bt["max_drawdown_pct"], "walk_forward_avg_mse": wfa["average_mse"], "leakage_prevented": True, "live_trading_blocked": True, "safety_check": safety_state }