trading-expert

Trading Expert

Expert guidance for algorithmic trading systems, quantitative analysis, market data processing, and trading platform development.

Core Concepts

Trading Systems

• Algorithmic trading strategies
• High-frequency trading (HFT)
• Market making
• Arbitrage strategies
• Portfolio optimization
• Risk management

Market Data

• Order book processing
• Tick data analysis
• Market microstructure
• Real-time data feeds
• Historical data analysis

Execution

• Order routing
• Smart order routing (SOR)
• Execution algorithms (TWAP, VWAP)
• Slippage minimization
• Transaction cost analysis

Trading Strategy Implementation

import pandas as pd
import numpy as np
from typing import Optional

class TradingStrategy:
    def __init__(self, symbol: str, capital: float = 100000):
        self.symbol = symbol
        self.capital = capital
        self.position = 0
        self.cash = capital
        self.trades = []

    def moving_average_crossover(self, data: pd.DataFrame,
                                  short_window: int = 50,
                                  long_window: int = 200) -> pd.Series:
        """Simple Moving Average Crossover Strategy"""
        data['SMA_short'] = data['close'].rolling(window=short_window).mean()
        data['SMA_long'] = data['close'].rolling(window=long_window).mean()

        # Generate signals
        data['signal'] = 0
        data.loc[data['SMA_short'] > data['SMA_long'], 'signal'] = 1
        data.loc[data['SMA_short'] < data['SMA_long'], 'signal'] = -1

        return data['signal']

    def mean_reversion(self, data: pd.DataFrame,
                       window: int = 20,
                       num_std: float = 2.0) -> pd.Series:
        """Mean Reversion Strategy using Bollinger Bands"""
        data['MA'] = data['close'].rolling(window=window).mean()
        data['STD'] = data['close'].rolling(window=window).std()
        data['upper_band'] = data['MA'] + (data['STD'] * num_std)
        data['lower_band'] = data['MA'] - (data['STD'] * num_std)

        # Generate signals
        data['signal'] = 0
        data.loc[data['close'] < data['lower_band'], 'signal'] = 1  # Buy
        data.loc[data['close'] > data['upper_band'], 'signal'] = -1  # Sell

        return data['signal']

    def momentum_strategy(self, data: pd.DataFrame, period: int = 14) -> pd.Series:
        """Momentum Strategy using RSI"""
        delta = data['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()

        rs = gain / loss
        data['RSI'] = 100 - (100 / (1 + rs))

        # Generate signals
        data['signal'] = 0
        data.loc[data['RSI'] < 30, 'signal'] = 1  # Oversold - Buy
        data.loc[data['RSI'] > 70, 'signal'] = -1  # Overbought - Sell

        return data['signal']

class Backtester:
    def __init__(self, initial_capital: float = 100000):
        self.initial_capital = initial_capital
        self.capital = initial_capital
        self.position = 0
        self.trades = []

    def run(self, data: pd.DataFrame, signals: pd.Series) -> dict:
        """Run backtest on historical data"""
        portfolio_value = []

        for i in range(len(data)):
            if signals.iloc[i] == 1 and self.position == 0:  # Buy signal
                shares = self.capital // data['close'].iloc[i]
                cost = shares * data['close'].iloc[i]
                self.capital -= cost
                self.position = shares
                self.trades.append({
                    'type': 'BUY',
                    'price': data['close'].iloc[i],
                    'shares': shares,
                    'date': data.index[i]
                })

            elif signals.iloc[i] == -1 and self.position > 0:  # Sell signal
                proceeds = self.position * data['close'].iloc[i]
                self.capital += proceeds
                self.trades.append({
                    'type': 'SELL',
                    'price': data['close'].iloc[i],
                    'shares': self.position,
                    'date': data.index[i]
                })
                self.position = 0

            # Calculate portfolio value
            current_value = self.capital + (self.position * data['close'].iloc[i])
            portfolio_value.append(current_value)

        return self.calculate_metrics(portfolio_value, data)

    def calculate_metrics(self, portfolio_value: list, data: pd.DataFrame) -> dict:
        """Calculate performance metrics"""
        returns = pd.Series(portfolio_value).pct_change()

        total_return = (portfolio_value[-1] - self.initial_capital) / self.initial_capital
        sharpe_ratio = returns.mean() / returns.std() * np.sqrt(252)
        max_drawdown = self.calculate_max_drawdown(portfolio_value)

        return {
            'total_return': total_return,
            'sharpe_ratio': sharpe_ratio,
            'max_drawdown': max_drawdown,
            'total_trades': len(self.trades),
            'final_value': portfolio_value[-1]
        }

    def calculate_max_drawdown(self, portfolio_value: list) -> float:
        """Calculate maximum drawdown"""
        peak = portfolio_value[0]
        max_dd = 0

        for value in portfolio_value:
            if value > peak:
                peak = value
            dd = (peak - value) / peak
            if dd > max_dd:
                max_dd = dd

        return max_dd