Algorithmic Trading

Tools and patterns for algorithmic trading systems.

Strategy Base Class

import pandas as pd
import numpy as np
from abc import ABC, abstractmethod

class TradingStrategy(ABC):
    def __init__(self, initial_capital=100000):
        self.capital = initial_capital
        self.positions = {}
        self.trades = []

    @abstractmethod
    def generate_signals(self, data):
        """Generate buy/sell signals from market data."""
        pass

    def execute_trade(self, symbol, quantity, price, side):
        trade = {
            'symbol': symbol,
            'quantity': quantity,
            'price': price,
            'side': side,
            'value': quantity * price
        }
        self.trades.append(trade)

        if side == 'buy':
            self.capital -= trade['value']
            self.positions[symbol] = self.positions.get(symbol, 0) + quantity
        else:
            self.capital += trade['value']
            self.positions[symbol] = self.positions.get(symbol, 0) - quantity

Moving Average Crossover

class MACrossover(TradingStrategy):
    def __init__(self, short_window=20, long_window=50, **kwargs):
        super().__init__(**kwargs)
        self.short_window = short_window
        self.long_window = long_window

    def generate_signals(self, data):
        signals = pd.DataFrame(index=data.index)
        signals['price'] = data['close']
        signals['short_ma'] = data['close'].rolling(window=self.short_window).mean()
        signals['long_ma'] = data['close'].rolling(window=self.long_window).mean()

        signals['signal'] = 0
        signals.loc[signals['short_ma'] > signals['long_ma'], 'signal'] = 1
        signals.loc[signals['short_ma'] < signals['long_ma'], 'signal'] = -1

        signals['positions'] = signals['signal'].diff()

        return signals

Momentum Strategy

class MomentumStrategy(TradingStrategy):
    def __init__(self, lookback=20, threshold=0.02, **kwargs):
        super().__init__(**kwargs)
        self.lookback = lookback
        self.threshold = threshold

    def generate_signals(self, data):
        signals = pd.DataFrame(index=data.index)
        signals['price'] = data['close']

        # Calculate momentum
        signals['returns'] = data['close'].pct_change(self.lookback)

        signals['signal'] = 0
        signals.loc[signals['returns'] > self.threshold, 'signal'] = 1
        signals.loc[signals['returns'] < -self.threshold, 'signal'] = -1

        return signals

Backtesting Engine

class Backtester:
    def __init__(self, strategy, data, commission=0.001):
        self.strategy = strategy
        self.data = data
        self.commission = commission

    def run(self):
        signals = self.strategy.generate_signals(self.data)

        portfolio = pd.DataFrame(index=signals.index)
        portfolio['holdings'] = signals['signal'] * self.data['close']
        portfolio['cash'] = self.strategy.capital - (signals['signal'].diff().abs() *
                                                      self.data['close'] *
                                                      (1 + self.commission)).cumsum()
        portfolio['total'] = portfolio['holdings'] + portfolio['cash']
        portfolio['returns'] = portfolio['total'].pct_change()

        return portfolio

    def calculate_metrics(self, portfolio):
        returns = portfolio['returns'].dropna()

        metrics = {
            'total_return': (portfolio['total'].iloc[-1] / self.strategy.capital - 1) * 100,
            'sharpe_ratio': returns.mean() / returns.std() * np.sqrt(252),
            'max_drawdown': ((portfolio['total'].cummax() - portfolio['total']) /
                            portfolio['total'].cummax()).max() * 100,
            'win_rate': (returns > 0).sum() / len(returns) * 100
        }
        return metrics

Risk Management

class RiskManager:
    def __init__(self, max_position_size=0.1, max_drawdown=0.2, stop_loss=0.05):
        self.max_position_size = max_position_size
        self.max_drawdown = max_drawdown
        self.stop_loss = stop_loss

    def check_position_size(self, portfolio_value, trade_value):
        return trade_value / portfolio_value <= self.max_position_size

    def check_stop_loss(self, entry_price, current_price):
        loss = (entry_price - current_price) / entry_price
        return loss < self.stop_loss

    def calculate_position_size(self, portfolio_value, risk_per_trade, stop_distance):
        """Kelly criterion position sizing."""
        return (portfolio_value * risk_per_trade) / stop_distance