AI Trading Bots Mastery

# Create isolated conda environment (industry standard)
conda create -n algotrading python=3.9 -y
conda activate algotrading

# Install core packages (version-locked for reproducibility)
pip install pandas==1.4.3 numpy==1.22.4 matplotlib==3.5.2 
pip install yfinance==0.1.74 backtrader==1.9.76.123 jupyter==1.0.0

# For Windows users only - fix DLL issues
conda install -c anaconda mkl-service

from sklearn.model_selection import TimeSeriesSplit
import numpy as np

def walk_forward_validation(data, n_splits=5, train_ratio=0.7):
    """
    Institutional-grade walk-forward testing
    Returns: List of (train_idx, test_idx) pairs
    """
    tscv = TimeSeriesSplit(n_splits=n_splits)
    splits = []
    
    for train_index, test_index in tscv.split(data):
        # Adjust split points for desired train/test ratio
        adjusted_point = int(len(train_index) * train_ratio)
        train_index = train_index[:adjusted_point]
        test_index = test_index  # Out-of-sample period
        
        splits.append((train_index, test_index))
    
    return splits

# Professional usage example:
data = get_historical_data('SPY')  # Your data loading function
splits = walk_forward_validation(data)

for i, (train_idx, test_idx) in enumerate(splits):
    train_data = data.iloc[train_idx]
    test_data = data.iloc[test_idx]
    
    # Optimize strategy on training window
    best_params = optimize_strategy(train_data)
    
    # Test on out-of-sample data
    results = backtest_strategy(test_data, best_params)
    
    # Store metrics
    print(f"Split {i+1}: Sharpe Ratio = {results['sharpe']:.2f}, Max DD = {results['max_dd']:.2%}")

import ccxt
import pandas as pd
import numpy as np
from typing import Dict, List

class ProfessionalDataFetcher:
    """Institutional-grade multi-source data aggregation"""
    
    def __init__(self):
        self.sources = {
            'binance': ccxt.binance({'enableRateLimit': True}),
            'alpaca': None,  # Would initialize with API keys
            'polygon': None  # Would initialize with API keys
        }
        self.cache = {}  # For memoization
        
    def fetch_ohlcv(self, symbol: str, timeframe: str = '1d', 
                   providers: List[str] = ['binance']) -> pd.DataFrame:
        """
        Fetch OHLCV data from multiple sources and merge
        Returns: Cleaned DataFrame with consensus pricing
        """
        all_data = []
        
        for provider in providers:
            try:
                if provider == 'binance':
                    raw = self.sources['binance'].fetch_ohlcv(symbol, timeframe)
                    df = pd.DataFrame(raw, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
                    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
                    df['source'] = 'binance'
                    all_data.append(df.set_index('timestamp'))
                
                # Additional providers would be implemented here
                
            except Exception as e:
                print(f"Error fetching from {provider}: {str(e)}")
                continue
        
        if not all_data:
            raise ValueError("No data fetched from any provider")
            
        # Professional data alignment and cleaning
        combined = pd.concat(all_data)
        consensus = combined.groupby(level=0).agg({
            'open': 'median',
            'high': 'max',  # Use max high across sources
            'low': 'min',   # Use min low across sources
            'close': 'median',
            'volume': 'sum'  # Sum volume across sources
        })
        
        # Handle missing data professionally
        if consensus.isnull().any().any():
            print("Warning: Missing data detected, applying interpolation")
            consensus = consensus.interpolate().ffill().bfill()
            
        return consensus

# Professional usage example:
fetcher = ProfessionalDataFetcher()
btc_data = fetcher.fetch_ohlcv('BTC/USDT', '1d')

# Institutional data validation
assert not btc_data.isnull().any().any(), "Data contains NA values"
assert (btc_data['high'] >= btc_data['low']).all(), "Invalid high/low prices"
assert (btc_data['volume'] >= 0).all(), "Negative volume"

import pandas as pd
import numpy as np
from scipy.stats import zscore

class FeatureEngineer:
    """Institutional-grade feature engineering"""
    
    @staticmethod
    def add_technical_features(df: pd.DataFrame) -> pd.DataFrame:
        """Add professional technical indicators"""
        # Log returns (more statistically stable)
        df['log_ret'] = np.log(df['close']).diff()
        
        # Volatility measures
        df['volatility_20'] = df['log_ret'].rolling(20).std() * np.sqrt(252)  # Annualized
        df['volatility_ratio'] = df['volatility_20'] / df['volatility_20'].rolling(100).mean()
        
        # Microstructure features
        df['range_pct'] = (df['high'] - df['low']) / df['close']
        df['close_to_vwap'] = df['close'] / (df['volume'] * df['close']).cumsum() * df['volume'].cumsum()
        
        # Volume features
        df['volume_z'] = zscore(df['volume'], nan_policy='omit')
        df['volume_ma_ratio'] = df['volume'] / df['volume'].rolling(20).mean()
        
        return df
    
    @staticmethod
    def add_temporal_features(df: pd.DataFrame) -> pd.DataFrame:
        """Add time-based patterns"""
        # Intraday seasonality
        df['hour'] = df.index.hour
        df['day_of_week'] = df.index.dayofweek
        df['month'] = df.index.month
        
        # Market regime indicators
        df['trend_strength'] = df['close'].rolling(50).corr(pd.Series(range(50), index=df.index[-50:]))
        
        return df
    
    @staticmethod
    def add_target_variable(df: pd.DataFrame, horizon: int = 5) -> pd.DataFrame:
        """Create professional target variable"""
        # Future return over horizon
        df[f'future_ret_{horizon}'] = df['close'].pct_change(horizon).shift(-horizon)
        
        # Binary classification target
        df['target'] = (df[f'future_ret_{horizon}'] > 0).astype(int)
        
        return df.dropna()

# Professional usage example:
engineer = FeatureEngineer()
df = engineer.add_technical_features(btc_data)
df = engineer.add_temporal_features(df)
df = engineer.add_target_variable(df)

# Institutional feature selection
professional_features = [
    'log_ret', 'volatility_20', 'volatility_ratio',
    'range_pct', 'close_to_vwap', 'volume_z',
    'hour', 'day_of_week', 'trend_strength'
]

import hashlib
import pickle
from pathlib import Path
import datetime

class DataVersionControl:
    """Institutional-grade data versioning"""
    
    def __init__(self, storage_path: str = './data_versions'):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(exist_ok=True)
    
    def save_version(self, df: pd.DataFrame, description: str = "") -> str:
        """Save dataset with version metadata"""
        # Create unique hash of data
        data_hash = hashlib.sha256(pickle.dumps(df)).hexdigest()[:16]
        
        # Version info
        version = {
            'hash': data_hash,
            'timestamp': datetime.datetime.now().isoformat(),
            'description': description,
            'columns': list(df.columns),
            'shape': df.shape,
            'sample': df.iloc[:5].to_dict()
        }
        
        # Save data and metadata
        version_path = self.storage_path / data_hash
        version_path.mkdir(exist_ok=True)
        
        df.to_parquet(version_path / 'data.parquet')
        with open(version_path / 'meta.json', 'w') as f:
            json.dump(version, f)
        
        return data_hash
    
    def load_version(self, data_hash: str) -> pd.DataFrame:
        """Load specific version"""
        version_path = self.storage_path / data_hash
        return pd.read_parquet(version_path / 'data.parquet')

# Professional usage example:
dvc = DataVersionControl()

# Save current version
version_hash = dvc.save_version(df, "BTC daily with features 2024-01")

# Later, load exact same version
same_df = dvc.load_version(version_hash)

import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import coint
from itertools import combinations
from typing import List, Tuple

class PairsFinder:
    """Institutional-grade pairs trading framework"""
    
    def __init__(self, p_threshold: float = 0.05, min_half_life: int = 5, max_half_life: int = 60):
        self.p_threshold = p_threshold
        self.min_half_life = min_half_life
        self.max_half_life = max_half_life
    
    def find_cointegrated_pairs(self, data: pd.DataFrame) -> List[Tuple[str, str, float, float]]:
        """Identify professional trading pairs with half-life filtering"""
        n = data.shape[1]
        pairs = []
        
        for i, j in combinations(range(n), 2):
            asset1 = data.iloc[:,i]
            asset2 = data.iloc[:,j]
            
            # Professional cointegration test
            score, pvalue, _ = coint(asset1, asset2)
            
            if pvalue < self.p_threshold:
                # Calculate half-life of mean reversion
                spread = asset1 - asset2
                lag_spread = spread.shift(1)
                delta_spread = spread - lag_spread
                regression = np.polyfit(lag_spread[1:], delta_spread[1:], 1)
                half_life = -np.log(2) / regression[0]
                
                if self.min_half_life <= half_life <= self.max_half_life:
                    pairs.append((
                        data.columns[i], 
                        data.columns[j], 
                        pvalue, 
                        half_life
                    ))
        
        return sorted(pairs, key=lambda x: x[2])  # Sort by p-value

    def calculate_hedge_ratio(self, x: pd.Series, y: pd.Series, method: str = 'ols') -> float:
        """Professional hedge ratio calculation"""
        if method == 'ols':
            return np.polyfit(x, y, 1)[0]  # Standard OLS
        elif method == 'tls':
            # Total least squares (more robust)
            cov = np.cov(x, y)
            eigvals, eigvecs = np.linalg.eig(cov)
            return eigvecs[1,0] / eigvecs[0,0]
        else:
            raise ValueError("Method must be 'ols' or 'tls'")

# Professional usage example:
finder = PairsFinder(p_threshold=0.01)  # 1% significance level

# Get price data for potential pairs
stocks = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'META', 'TSLA']
data = yf.download(stocks, start='2020-01-01', end='2023-12-31')['Adj Close']

# Find best pairs
pairs = finder.find_cointegrated_pairs(data)
print("Top Pairs:")
for pair in pairs[:5]:
    print(f"{pair[0]}-{pair[1]}: p={pair[2]:.4f}, half-life={pair[3]:.1f} days")

import numpy as np

class KalmanFilterPair:
    """Institutional-grade Kalman filter for pairs trading"""
    
    def __init__(self, delta=1e-4, R=0.001):
        """
        delta: Process noise variance
        R: Measurement noise variance
        """
        self.delta = delta
        self.R = R
        self.A = 1.0  # Hedge ratio
        self.C = 0.0   # Intercept
        self.P = np.eye(2)  # Covariance matrix
        self.Q = np.eye(2) * delta  # Process noise
        
    def update(self, x: float, y: float) -> Tuple[float, float]:
        """Update with new price observations"""
        # Prediction update
        self.P += self.Q
        
        # Measurement update
        H = np.array([x, 1]).reshape(1, 2)
        K = self.P @ H.T / (H @ self.P @ H.T + self.R)
        
        residual = y - (self.A * x + self.C)
        
        # Update state estimates
        state = np.array([self.A, self.C]).reshape(2, 1)
        state += K * residual
        self.A, self.C = state.flatten()
        
        # Update covariance
        self.P = (np.eye(2) - K @ H) @ self.P
        
        return self.A, self.C
    
    def get_spread(self, x: float, y: float) -> float:
        """Calculate normalized spread"""
        return y - (self.A * x + self.C)

# Professional usage example:
kf = KalmanFilterPair()

# Simulate trading
x_prices = np.random.normal(100, 5, 1000).cumsum()  # Asset X
y_prices = x_prices * 1.5 + np.random.normal(0, 3, 1000)  # Asset Y (cointegrated)

spreads = []
for x, y in zip(x_prices, y_prices):
    hedge, intercept = kf.update(x, y)
    spread = kf.get_spread(x, y)
    spreads.append(spread)

# Plot results
import matplotlib.pyplot as plt
plt.figure(figsize=(12,6))
plt.plot(spreads, label='Kalman Filter Spread')
plt.axhline(np.mean(spreads), color='r', linestyle='--', label='Mean')
plt.axhline(np.mean(spreads) + 2*np.std(spreads), color='g', linestyle=':', label='±2σ')
plt.axhline(np.mean(spreads) - 2*np.std(spreads), color='g', linestyle=':')
plt.title("Professional Kalman Filter Spread", fontsize=14)
plt.legend()
plt.show()

def kelly_position_size(returns: pd.Series, max_fraction: float = 0.2) -> float:
    """
    Calculate Kelly optimal position size
    returns: Historical strategy returns
    max_fraction: Cap on maximum position (risk management)
    Returns: Fraction of capital to allocate
    """
    win_rate = (returns > 0).mean()
    avg_win = returns[returns > 0].mean()
    avg_loss = abs(returns[returns <= 0].mean())
    
    if avg_loss == 0:  # Avoid division by zero
        return 0.0
    
    edge = win_rate * avg_win - (1 - win_rate) * avg_loss
    kelly_f = edge / (avg_win * avg_loss)
    
    return min(kelly_f, max_fraction)  # Risk cap

# Professional usage example:
strategy_returns = pd.Series([0.02, -0.01, 0.015, -0.005, 0.03, -0.01])
position_fraction = kelly_position_size(strategy_returns)
print(f"Optimal position size: {position_fraction:.1%} of capital")

import numpy as np
import pandas as pd
from sklearn.model_selection import TimeSeriesSplit
from sklearn.preprocessing import RobustScaler

class MLDataPreprocessor:
    """Institutional-grade feature engineering for trading"""
    
    def __init__(self, window: int = 30, horizon: int = 5, test_size: float = 0.2):
        self.window = window
        self.horizon = horizon
        self.test_size = test_size
        self.scaler = RobustScaler()  # Less sensitive to outliers than StandardScaler
    
    def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Create professional trading features"""
        # Price features
        df['log_ret'] = np.log(df['close']).diff()
        df['volatility'] = df['log_ret'].rolling(self.window).std()
        df['momentum'] = df['close'] / df['close'].shift(self.window) - 1
        
        # Volume features
        df['volume_ma'] = df['volume'].rolling(self.window).mean()
        df['volume_z'] = (df['volume'] - df['volume_ma']) / df['volume'].rolling(self.window).std()
        
        # Microstructure features
        df['spread_pct'] = (df['high'] - df['low']) / df['close']
        df['close_to_vwap'] = df['close'] / df['vwap']
        
        # Time features
        df['hour'] = df.index.hour
        df['day_of_week'] = df.index.dayofweek
        
        return df.dropna()
    
    def create_target(self, df: pd.DataFrame) -> pd.Series:
        """Create professional target variable"""
        # Future return over horizon
        future_ret = df['close'].pct_change(self.horizon).shift(-self.horizon)
        
        # Binary classification: will return be positive?
        target = (future_ret > 0).astype(int)
        
        return target.dropna()
    
    def prepare_ml_data(self, df: pd.DataFrame) -> tuple:
        """Create sequences for time-series prediction"""
        features = self.create_features(df)
        target = self.create_target(features)
        
        # Align features and target
        aligned_data = features.join(target, how='inner')
        aligned_data.columns = [*features.columns, 'target']
        
        # Split into train/test
        split_idx = int(len(aligned_data) * (1 - self.test_size))
        train = aligned_data.iloc[:split_idx]
        test = aligned_data.iloc[split_idx:]
        
        # Scale features (fit only on train to avoid lookahead)
        feature_cols = [col for col in train.columns if col != 'target']
        train_features = self.scaler.fit_transform(train[feature_cols])
        test_features = self.scaler.transform(test[feature_cols])
        
        # Create sequences
        X_train, y_train = self._create_sequences(train_features, train['target'])
        X_test, y_test = self._create_sequences(test_features, test['target'])
        
        return X_train, X_test, y_train, y_test
    
    def _create_sequences(self, features: np.array, target: pd.Series) -> tuple:
        """Convert tabular data to time-series sequences"""
        X, y = [], []
        
        for i in range(self.window, len(features) - self.horizon):
            X.append(features[i-self.window:i])
            y.append(target.iloc[i + self.horizon - 1])  # Predict horizon steps ahead
        
        return np.array(X), np.array(y)

# Professional usage example:
processor = MLDataPreprocessor(window=30, horizon=5)
X_train, X_test, y_train, y_test = processor.prepare_ml_data(btc_data)

print(f"Train shapes: {X_train.shape}, {y_train.shape}")
print(f"Test shapes: {X_test.shape}, {y_test.shape}")

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout, BatchNormalization
from tensorflow.keras.regularizers import l2
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau
from tensorflow.keras.optimizers import Adam

class ProfessionalLSTM:
    """Institutional-grade LSTM for trading"""
    
    def __init__(self, input_shape, learning_rate=3e-4):
        self.input_shape = input_shape
        self.learning_rate = learning_rate
    
    def build_model(self) -> Sequential:
        """Build professional trading model"""
        model = Sequential([
            # First LSTM layer with return sequences
            LSTM(64, 
                input_shape=self.input_shape,
                return_sequences=True,
                kernel_regularizer=l2(0.01),
                recurrent_dropout=0.2),
            BatchNormalization(),
            
            # Second LSTM layer
            LSTM(32,
                kernel_regularizer=l2(0.01),
                recurrent_dropout=0.1),
            BatchNormalization(),
            
            # Dense layers
            Dense(16, activation='relu', kernel_regularizer=l2(0.01)),
            Dropout(0.3),
            BatchNormalization(),
            
            # Output layer
            Dense(1, activation='sigmoid')
        ])
        
        optimizer = Adam(learning_rate=self.learning_rate, clipvalue=0.5)
        model.compile(
            optimizer=optimizer,
            loss='binary_crossentropy',
            metrics=['accuracy', 'AUC']
        )
        
        return model
    
    def get_callbacks(self) -> list:
        """Professional training callbacks"""
        return [
            EarlyStopping(patience=10, restore_best_weights=True),
            ReduceLROnPlateau(factor=0.5, patience=5)
        ]
    
    def train_model(self, X_train, y_train, X_val, y_val, 
                   epochs=100, batch_size=64) -> Sequential:
        """Professional training routine"""
        model = self.build_model()
        
        history = model.fit(
            X_train, y_train,
            validation_data=(X_val, y_val),
            epochs=epochs,
            batch_size=batch_size,
            callbacks=self.get_callbacks(),
            verbose=1
        )
        
        return model, history

# Professional usage example:
lstm = ProfessionalLSTM(input_shape=X_train.shape[1:])
model, history = lstm.train_model(X_train, y_train, X_test, y_test)

# Evaluate
test_preds = model.predict(X_test)
test_accuracy = np.mean((test_preds > 0.5) == y_test)
print(f"Test Accuracy: {test_accuracy:.2%}")

def evaluate_strategy(predictions: np.array, 
                     y_true: np.array, 
                     returns: pd.Series,
                     transaction_cost: float = 0.001) -> dict:
    """
    Professional trading strategy evaluation
    Returns: Dictionary of performance metrics
    """
    # Create signals (1=long, -1=short, 0=flat)
    signals = np.where(predictions > 0.6, 1, 
                      np.where(predictions < 0.4, -1, 0))
    
    # Calculate strategy returns
    strategy_returns = signals * returns
    
    # Account for transaction costs
    trades = np.diff(signals, prepend=0) != 0
    strategy_returns[trades] -= transaction_cost
    
    # Calculate metrics
    total_return = np.prod(1 + strategy_returns) - 1
    sharpe = np.mean(strategy_returns) / np.std(strategy_returns) * np.sqrt(252)
    max_dd = calculate_max_drawdown(strategy_returns)
    
    return {
        'total_return': total_return,
        'sharpe_ratio': sharpe,
        'max_drawdown': max_dd,
        'win_rate': np.mean(strategy_returns > 0)
    }

def calculate_max_drawdown(returns: np.array) -> float:
    """Calculate maximum drawdown"""
    cumulative = np.cumprod(1 + returns)
    peak = np.maximum.accumulate(cumulative)
    dd = (cumulative - peak) / peak
    return np.min(dd)

# Professional usage example:
test_returns = btc_data['close'].pct_change().dropna().iloc[-len(y_test):]
metrics = evaluate_strategy(test_preds, y_test, test_returns)

print("Strategy Performance:")
for k, v in metrics.items():
    print(f"{k:15}: {v:.2%}" if isinstance(v, float) else f"{k:15}: {v:.2f}")

import gym
from gym import spaces
import numpy as np
import pandas as pd

class ProfessionalTradingEnv(gym.Env):
    """Institutional-grade trading environment"""
    
    def __init__(self, df: pd.DataFrame, 
                 initial_balance: float = 100000,
                 transaction_cost: float = 0.001,
                 max_position: float = 0.1,
                 window_size: int = 30):
        """
        df: DataFrame with features and prices
        initial_balance: Starting capital
        transaction_cost: Per-trade cost (0.1%)
        max_position: Max portfolio allocation per trade (10%)
        window_size: Observation window
        """
        self.df = df
        self.features = [col for col in df.columns if col != 'close']
        self.prices = df['close'].values
        self.initial_balance = initial_balance
        self.transaction_cost = transaction_cost
        self.max_position = max_position
        self.window_size = window_size
        
        # Action space: -1 (short), 0 (hold), 1 (long)
        self.action_space = spaces.Discrete(3)
        
        # Observation space: OHLCV + features
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf,
            shape=(window_size, len(self.features) + 4),  # OHLC + features
            dtype=np.float32
        )
        
        # Reset
        self.reset()
    
    def reset(self):
        """Reset environment to initial state"""
        self.current_step = self.window_size
        self.balance = self.initial_balance
        self.position = 0
        self.shares = 0
        self.trades = []
        self.done = False
        
        # Initialize portfolio value tracking
        self.portfolio_value = [self.initial_balance]
        
        return self._get_obs()
    
    def _get_obs(self):
        """Get current observation"""
        obs = self.df.iloc[
            self.current_step - self.window_size:self.current_step
        ].copy()
        
        # Add technical indicators
        obs['returns'] = np.log(obs['close']).diff()
        obs['volatility'] = obs['returns'].rolling(5).std()
        
        return obs.values
    
    def step(self, action):
        """Execute one time step"""
        if self.done:
            return self._get_obs(), 0, True, {}
        
        # Get current price
        current_price = self.prices[self.current_step]
        prev_price = self.prices[self.current_step - 1]
        
        # Calculate position size (10% of portfolio)
        position_size = self.balance * self.max_position
        
        # Execute trade based on action
        if action == 1:  # Buy
            if self.position <= 0:  # Close short or open long
                cost = position_size * (1 + self.transaction_cost)
                if self.balance >= cost:
                    self.balance -= cost
                    self.position = position_size
                    self.trades.append(('buy', current_price))
        elif action == -1:  # Sell
            if self.position >= 0:  # Close long or open short
                self.balance += position_size * (1 - self.transaction_cost)
                self.position = -position_size
                self.trades.append(('sell', current_price))
        # Else hold
        
        # Calculate portfolio value
        portfolio_value = self.balance + self.position * (current_price / prev_price)
        self.portfolio_value.append(portfolio_value)
        
        # Calculate reward (change in portfolio value)
        reward = (portfolio_value - self.portfolio_value[-2]) / self.portfolio_value[-2]
        
        # Move to next step
        self.current_step += 1
        if self.current_step >= len(self.df) - 1:
            self.done = True
        
        return self._get_obs(), reward, self.done, {}

# Professional usage example:
env = ProfessionalTradingEnv(btc_data)
obs = env.reset()
done = False

while not done:
    action = env.action_space.sample()  # Random policy for demo
    obs, reward, done, info = env.step(action)

print(f"Final portfolio value: ${env.portfolio_value[-1]:,.2f}")

from stable_baselines3 import PPO
from stable_baselines3.common.callbacks import BaseCallback
from stable_baselines3.common.monitor import Monitor
from stable_baselines3.common.vec_env import DummyVecEnv, VecNormalize

class TensorboardCallback(BaseCallback):
    """Professional logging callback"""
    def __init__(self, verbose=0):
        super().__init__(verbose)
        self.episode_returns = []
    
    def _on_step(self) -> bool:
        # Log custom metrics
        if 'episode' in self.locals:
            for ep_info in self.locals['episode']:
                if 'r' in ep_info:
                    self.episode_returns.append(ep_info['r'])
                    self.logger.record('train/episode_return', ep_info['r'])
        
        return True

def train_rl_agent(env, total_timesteps=100000):
    """Professional RL training setup"""
    # Wrap environment
    env = Monitor(env)
    env = DummyVecEnv([lambda: env])
    env = VecNormalize(env, norm_obs=True, norm_reward=True)
    
    # Professional PPO configuration
    model = PPO(
        'MlpPolicy',
        env,
        verbose=1,
        learning_rate=3e-4,
        n_steps=2048,
        batch_size=64,
        ent_coef=0.01,  # Encourage exploration
        clip_range=0.2,
        max_grad_norm=0.5,  # Gradient clipping
        tensorboard_log="./rl_trading_logs/"
    )
    
    # Train with callbacks
    callback = TensorboardCallback()
    model.learn(
        total_timesteps=total_timesteps,
        callback=callback,
        tb_log_name="ppo_trading"
    )
    
    # Save model and normalization stats
    model.save("trading_agent")
    env.save("vec_normalize.pkl")
    
    return model

# Professional usage example:
model = train_rl_agent(env, total_timesteps=50000)

# Evaluate trained agent
obs = env.reset()
done = False
while not done:
    action, _ = model.predict(obs, deterministic=True)
    obs, _, done, _ = env.step(action)

print(f"Trained agent portfolio value: ${env.envs[0].env.portfolio_value[-1]:,.2f}")

class SafeTradingAgent:
    """Institutional-grade RL agent deployment"""
    
    def __init__(self, model_path: str, env_path: str):
        # Load trained model
        self.model = PPO.load(model_path)
        self.env = DummyVecEnv([lambda: Monitor(gym.make('TradingEnv-v0'))])
        self.env = VecNormalize.load(env_path, self.env)
        self.env.training = False  # Disable training mode
        
        # Risk management
        self.max_daily_loss = 0.05  # 5% max loss per day
        self.position_limit = 0.2   # 20% max portfolio allocation
        self.current_pnl = 0
        
    def predict(self, observation: np.array) -> int:
        """Get action with risk controls"""
        # Get raw action from policy
        action, _ = self.model.predict(observation, deterministic=True)
        
        # Apply position limits
        current_position = self.env.get_attr('position')[0]
        if abs(current_position) >= self.position_limit:
            action = 0  # Force hold if at limit
        
        # Check daily loss limit
        daily_pnl = self.env.get_attr('portfolio_value')[-1][-1] - self.env.get_attr('portfolio_value')[-1][0]
        if daily_pnl <= -self.max_daily_loss:
            action = 0  # Stop trading for the day
            
        return action
    
    def run_live(self, data_stream):
        """Run agent in live market"""
        obs = self.env.reset()
        done = False
        
        while not done:
            # Get new market data
            new_data = next(data_stream)
            self.env.env_method('update_data', new_data)
            
            # Get safe action
            action = self.predict(obs)
            
            # Execute
            obs, _, done, _ = self.env.step([action])
            
            # Professional logging
            self._log_trade(action)

# Professional usage example:
agent = SafeTradingAgent("trading_agent", "vec_normalize.pkl")

import cvxpy as cp
import numpy as np
import pandas as pd

class PortfolioOptimizer:
    """Institutional-grade portfolio optimization"""
    
    def __init__(self, returns: pd.DataFrame, 
                 risk_free_rate: float = 0.02):
        """
        returns: DataFrame of asset returns
        risk_free_rate: Annual risk-free rate
        """
        self.returns = returns
        self.risk_free_rate = risk_free_rate
        self.n_assets = returns.shape[1]
        
    def calculate_stats(self) -> tuple:
        """Calculate professional statistics"""
        annual_returns = self.returns.mean() * 252
        cov_matrix = self.returns.cov() * 252
        return annual_returns, cov_matrix
    
    def optimize_sharpe(self, target_return: float = None) -> np.array:
        """Maximize Sharpe ratio with optional target return"""
        mu, Sigma = self.calculate_stats()
        
        weights = cp.Variable(self.n_assets)
        
        # Objective: Maximize Sharpe ratio
        if target_return is None:
            objective = cp.Maximize((mu @ weights - self.risk_free_rate) / 
                                  cp.quad_form(weights, Sigma))
        else:
            # With target return constraint
            objective = cp.Minimize(cp.quad_form(weights, Sigma))
        
        constraints = [
            weights >= 0,
            cp.sum(weights) == 1,
        ]
        
        if target_return is not None:
            constraints.append(mu @ weights >= target_return)
        
        problem = cp.Problem(objective, constraints)
        problem.solve()
        
        return weights.value
    
    def black_litterman(self, P: np.array, Q: np.array, 
                       tau: float = 0.05, delta: float = 2.5) -> np.array:
        """
        Black-Litterman model for incorporating views
        P: Matrix linking assets to views
        Q: Vector of expected returns from views
        tau: Confidence in equilibrium returns
        delta: Risk aversion coefficient
        """
        mu, Sigma = self.calculate_stats()
        Pi = delta * Sigma @ np.ones(self.n_assets)  # Equilibrium returns
        
        # Posterior estimate
        omega = tau * (P @ Sigma @ P.T)  # Uncertainty in views
        mu_bl = Pi + tau * Sigma @ P.T @ np.linalg.inv(tau * P @ Sigma @ P.T + omega) @ (Q - P @ Pi)
        Sigma_bl = Sigma + np.linalg.inv(tau * Sigma) + P.T @ np.linalg.inv(omega) @ P
        
        # Optimize with new estimates
        weights = cp.Variable(self.n_assets)
        objective = cp.Maximize(mu_bl @ weights - 0.5 * delta * cp.quad_form(weights, Sigma_bl))
        constraints = [weights >= 0, cp.sum(weights) == 1]
        
        problem = cp.Problem(objective, constraints)
        problem.solve()
        
        return weights.value

# Professional usage example:
assets = ['SPY', 'TLT', 'GLD', 'VNQ']
returns = yf.download(assets, start='2010-01-01', end='2023-12-31')['Adj Close'].pct_change().dropna()

optimizer = PortfolioOptimizer(returns)
weights = optimizer.optimize_sharpe()

print("Optimal Weights:")
for asset, weight in zip(assets, weights):
    print(f"{asset}: {weight:.1%}")

# Black-Litterman example with view that SPY will outperform TLT by 5%
P = np.array([[1, -1, 0, 0]])  # SPY - TLT
Q = np.array([0.05])
bl_weights = optimizer.black_litterman(P, Q)

print("\nBlack-Litterman Weights:")
for asset, weight in zip(assets, bl_weights):
    print(f"{asset}: {weight:.1%}")

def risk_parity_allocation(cov_matrix: np.array, 
                          max_iter: int = 100, 
                          tol: float = 1e-6) -> np.array:
    """Institutional risk parity allocation"""
    n = cov_matrix.shape[0]
    weights = np.ones(n) / n  # Equal weight initialization
    
    for _ in range(max_iter):
        portfolio_vol = np.sqrt(weights.T @ cov_matrix @ weights)
        marginal_risk = cov_matrix @ weights / portfolio_vol
        risk_contributions = weights * marginal_risk
        
        # Check convergence
        if np.max(np.abs(risk_contributions - portfolio_vol/n)) < tol:
            break
            
        # Update weights
        weights = risk_contributions / marginal_risk
        weights = weights / np.sum(weights)  # Normalize
    
    return weights

# Professional enhancement with constraints
def constrained_risk_parity(cov_matrix: np.array,
                          min_weight: float = 0.05,
                          max_weight: float = 0.4) -> np.array:
    """Risk parity with weight constraints"""
    n = cov_matrix.shape[0]
    weights = cp.Variable(n)
    
    # Risk contributions
    portfolio_vol = cp.quad_form(weights, cov_matrix)
    marginal_risk = cov_matrix @ weights / portfolio_vol
    risk_contributions = cp.multiply(weights, marginal_risk)
    
    # Objective: Equal risk contributions
    objective = cp.Minimize(
        cp.sum_squares(risk_contributions - portfolio_vol/n)
    )
    
    constraints = [
        weights >= min_weight,
        weights <= max_weight,
        cp.sum(weights) == 1
    ]
    
    problem = cp.Problem(objective, constraints)
    problem.solve()
    
    return weights.value

# Professional usage example:
_, cov_matrix = optimizer.calculate_stats()
rp_weights = risk_parity_allocation(cov_matrix.values)

print("\nRisk Parity Weights:")
for asset, weight in zip(assets, rp_weights):
    print(f"{asset}: {weight:.1%}")

# Constrained version
crp_weights = constrained_risk_parity(cov_matrix.values)

print("\nConstrained Risk Parity Weights:")
for asset, weight in zip(assets, crp_weights):
    print(f"{asset}: {weight:.1%}")

class DrawdownManager:
    """Institutional drawdown control"""
    
    def __init__(self, max_drawdown: float = 0.2, 
                 lookback: int = 252):
        """
        max_drawdown: Maximum allowed drawdown (20%)
        lookback: Window for volatility calculation (1 year)
        """
        self.max_dd = max_drawdown
        self.lookback = lookback
        self.portfolio_values = []
    
    def update(self, current_value: float) -> float:
        """Update and return required position adjustment"""
        self.portfolio_values.append(current_value)
        
        if len(self.portfolio_values) < 2:
            return 1.0  # Full allocation
        
        # Calculate current drawdown
        peak = max(self.portfolio_values)
        current_dd = (peak - current_value) / peak
        
        if current_dd >= self.max_dd:
            # Calculate required reduction
            reduction = 1 - (current_dd / self.max_dd)
            return max(reduction, 0.1)  # Never go below 10%
        
        # Calculate volatility-scaling
        returns = np.diff(np.log(self.portfolio_values[-self.lookback:]))
        if len(returns) >= 5:  # Enough data
            vol = np.std(returns)
            target_vol = 0.15 / np.sqrt(252)  # 15% annualized
            scaling = target_vol / (vol + 1e-6)  # Avoid division by zero
            return min(scaling, 1.5)  # Cap at 1.5x
        else:
            return 1.0

# Professional usage example:
manager = DrawdownManager(max_drawdown=0.15)  # 15% max drawdown

portfolio_values = [100000]
allocations = [1.0]

for _ in range(252):  # Simulate 1 year
    # Simulate random returns
    ret = np.random.normal(0.0005, 0.01)
    current_value = portfolio_values[-1] * (1 + ret * allocations[-1])
    portfolio_values.append(current_value)
    
    # Get new allocation
    alloc = manager.update(current_value)
    allocations.append(alloc)

# Plot results
import matplotlib.pyplot as plt
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
ax1.plot(portfolio_values)
ax1.set_title('Portfolio Value')
ax2.plot(allocations)
ax2.set_title('Dynamic Allocation')
plt.tight_layout()
plt.show()

import time
from typing import Optional
from datetime import datetime, timedelta

class TWAPExecutor:
    """Institutional-grade TWAP execution"""
    
    def __init__(self, symbol: str, 
                 quantity: float,
                 duration: timedelta,
                 exchange: str = 'binance',
                 slippage: float = 0.001,
                 min_slice: float = 0.01):
        """
        symbol: Trading pair (e.g., 'BTC/USDT')
        quantity: Total quantity to execute
        duration: Total execution duration
        exchange: Target exchange
        slippage: Estimated slippage per trade
        min_slice: Minimum order size (1%)
        """
        self.symbol = symbol
        self.total_qty = quantity
        self.duration = duration.total_seconds()
        self.slippage = slippage
        self.min_slice = min(0.01, min_slice)  # Never less than 1%
        
        # Initialize exchange connection
        self.exchange = self._init_exchange(exchange)
        
        # Tracking
        self.executed_qty = 0
        self.avg_fill_price = 0
        self.start_time = None
    
    def _init_exchange(self, exchange: str):
        """Initialize professional exchange connection"""
        # In production, would include:
        # - API rate limit handling
        # - Error recovery
        # - Connection pooling
        if exchange == 'binance':
            return ccxt.binance({
                'enableRateLimit': True,
                'options': {
                    'adjustForTimeDifference': True
                }
            })
        else:
            raise ValueError(f"Unsupported exchange: {exchange}")
    
    def _get_market_data(self) -> dict:
        """Get professional market snapshot"""
        try:
            orderbook = self.exchange.fetch_order_book(self.symbol)
            return {
                'bid': orderbook['bids'][0][0],
                'ask': orderbook['asks'][0][0],
                'spread': orderbook['asks'][0][0] - orderbook['bids'][0][0]
            }
        except Exception as e:
            print(f"Market data error: {str(e)}")
            return None
    
    def _submit_order(self, qty: float, is_buy: bool) -> Optional[float]:
        """Professional order submission"""
        market = self._get_market_data()
        if not market:
            return None
        
        try:
            price = market['ask'] if is_buy else market['bid']
            adjusted_price = price * (1 + self.slippage) if is_buy else price * (1 - self.slippage)
            
            # In production, would use limit orders with sophisticated logic
            order = self.exchange.create_order(
                self.symbol,
                'limit',
                'buy' if is_buy else 'sell',
                qty,
                adjusted_price
            )
            
            # Professional order tracking
            self.executed_qty += qty
            self.avg_fill_price = (
                (self.avg_fill_price * (self.executed_qty - qty) + 
                 adjusted_price * qty) / self.executed_qty
            )
            
            return adjusted_price
        except Exception as e:
            print(f"Order failed: {str(e)}")
            return None
    
    def execute(self, is_buy: bool = True) -> bool:
        """Run TWAP execution"""
        self.start_time = datetime.utcnow()
        end_time = self.start_time + self.duration
        
        remaining_qty = self.total_qty
        slice_count = max(1, int(self.duration / 60))  # 1-minute slices
        
        while datetime.utcnow() < end_time and remaining_qty > 0:
            # Calculate next slice
            elapsed = (datetime.utcnow() - self.start_time).total_seconds()
            progress = min(1.0, elapsed / self.duration)
            
            target_qty = self.total_qty * progress
            slice_qty = max(
                self.total_qty * self.min_slice,  # Minimum size
                min(target_qty - self.executed_qty, remaining_qty)  # Don't over-execute
            )
            
            if slice_qty > 0:
                fill_price = self._submit_order(slice_qty, is_buy)
                if fill_price:
                    remaining_qty -= slice_qty
            
            # Professional pacing
            time_to_wait = max(
                0,
                (end_time - datetime.utcnow()).total_seconds() / 
                (self.total_qty / slice_qty)
            )
            time.sleep(time_to_wait)
        
        return remaining_qty == 0

# Professional usage example:
twap = TWAPExecutor(
    symbol='BTC/USDT',
    quantity=1.0,  # 1 BTC
    duration=timedelta(hours=4),  # Execute over 4 hours
    slippage=0.0005  # 0.05% slippage
)

success = twap.execute(is_buy=True)
print(f"Execution {'succeeded' if success else 'failed'}")
print(f"Avg fill price: {twap.avg_fill_price:.2f}")
print(f"Executed qty: {twap.executed_qty:.4f}")

import time
import logging
from enum import Enum, auto
from threading import Thread

class TradingState(Enum):
    INIT = auto()
    RUNNING = auto()
    PAUSED = auto()
    STOPPED = auto()
    ERROR = auto()

class ProfessionalTradingBot:
    """Institutional-grade trading system"""
    
    def __init__(self, strategy, max_retries=3, heartbeat_interval=60):
        self.strategy = strategy
        self.max_retries = max_retries
        self.heartbeat_interval = heartbeat_interval
        self.state = TradingState.INIT
        self.thread = None
        self.retry_count = 0
        
        # Professional logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('trading_bot.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def start(self):
        """Start trading thread"""
        if self.state != TradingState.INIT:
            raise RuntimeError("Bot already started")
        
        self.thread = Thread(target=self._run_loop, daemon=True)
        self.state = TradingState.RUNNING
        self.thread.start()
        self.logger.info("Trading bot started")
    
    def _run_loop(self):
        """Main trading loop"""
        while self.state != TradingState.STOPPED:
            try:
                if self.state == TradingState.RUNNING:
                    self._execute_strategy()
                elif self.state == TradingState.PAUSED:
                    time.sleep(1)
                    continue
                
                time.sleep(self.heartbeat_interval)
                self.retry_count = 0  # Reset on successful iteration
                
            except Exception as e:
                self.logger.error(f"Strategy error: {str(e)}", exc_info=True)
                self._handle_error(e)
    
    def _execute_strategy(self):
        """Execute strategy with professional safeguards"""
        try:
            # Check market conditions
            if not self._market_open():
                self.logger.warning("Market closed, pausing")
                self.state = TradingState.PAUSED
                return
            
            # Check position limits
            if self._exceeds_limits():
                self.logger.error("Position limits exceeded")
                self.state = TradingState.ERROR
                return
            
            # Execute strategy
            self.strategy.execute()
            self.logger.info("Strategy executed successfully")
            
        except Exception as e:
            raise RuntimeError(f"Execution failed: {str(e)}")
    
    def _handle_error(self, error):
        """Professional error handling"""
        self.retry_count += 1
        
        if self.retry_count >= self.max_retries:
            self.logger.critical("Max retries exceeded, stopping")
            self.state = TradingState.ERROR
            self._shutdown()
        else:
            self.logger.warning(f"Retrying ({self.retry_count}/{self.max_retries})")
            time.sleep(2 ** self.retry_count)  # Exponential backoff
    
    def _shutdown(self):
        """Graceful shutdown procedure"""
        self.logger.info("Initiating shutdown")
        self.state = TradingState.STOPPED
        
        try:
            # Close all positions
            self.strategy.close_positions()
            
            # Cancel all orders
            self.strategy.cancel_orders()
            
            # Save state
            self.strategy.save_state()
            
            self.logger.info("Shutdown complete")
        except Exception as e:
            self.logger.error(f"Shutdown error: {str(e)}", exc_info=True)
    
    def _market_open(self) -> bool:
        """Check if market is open"""
        # In production, would check exchange calendar
        return True
    
    def _exceeds_limits(self) -> bool:
        """Check position limits"""
        # In production, would verify against risk limits
        return False

# Professional usage example:
class SampleStrategy:
    def execute(self):
        # Simulate occasional failure
        if time.time() % 10 < 2:  # 20% chance of failure
            raise ValueError("Simulated error")
        print("Strategy executed")

    def close_positions(self):
        print("Positions closed")

    def cancel_orders(self):
        print("Orders cancelled")

    def save_state(self):
        print("State saved")

bot = ProfessionalTradingBot(SampleStrategy(), max_retries=3)
bot.start()

# Let it run for a while
time.sleep(30)
bot.state = TradingState.STOPPED

import prometheus_client
from prometheus_client import Gauge, Counter
import time
from threading import Thread

class TradingMonitor:
    """Institutional-grade trading monitoring"""
    
    def __init__(self):
        # Metrics
        self.pnl = Gauge('trading_pnl', 'Current Profit and Loss')
        self.drawdown = Gauge('trading_drawdown', 'Current Drawdown')
        self.trade_count = Counter('trades_total', 'Total trades executed')
        self.order_latency = Gauge('order_latency_ms', 'Order execution latency')
        self.slippage = Gauge('execution_slippage', 'Average execution slippage')
        
        # Alert thresholds
        self.max_drawdown = 0.1  # 10%
        self.max_slippage = 0.002  # 0.2%
        
        # Start web server
        prometheus_client.start_http_server(8000)
        
        # Background updater
        self._running = True
        self.thread = Thread(target=self._update_metrics)
        self.thread.start()
    
    def _update_metrics(self):
        """Simulate metric updates"""
        while self._running:
            # In production, would get real values from trading system
            self.pnl.set(np.random.normal(1000, 500))
            self.drawdown.set(abs(np.random.normal(0.05, 0.02)))
            self.trade_count.inc(np.random.poisson(2))
            self.order_latency.set(np.random.gamma(1, 50))
            self.slippage.set(abs(np.random.normal(0.001, 0.0005)))
            
            # Check alerts
            if self.drawdown._value.get() > self.max_drawdown:
                print(f"ALERT: Drawdown {self.drawdown._value.get():.1%} exceeds threshold")
            
            if self.slippage._value.get() > self.max_slippage:
                print(f"ALERT: Slippage {self.slippage._value.get():.1%} exceeds threshold")
            
            time.sleep(5)
    
    def stop(self):
        """Stop monitoring"""
        self._running = False
        self.thread.join()

# Professional usage example:
monitor = TradingMonitor()

# Simulate running for a while
time.sleep(60)
monitor.stop()