SKILL.md
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_ddOrder Execution
from enum import Enum
from decimal import Decimal
from datetime import datetime
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
class OrderType(Enum):
MARKET = "MARKET"
LIMIT = "LIMIT"
STOP = "STOP"
STOP_LIMIT = "STOP_LIMIT"
class Order:
def __init__(self, symbol: str, side: OrderSide, order_type: OrderType,
quantity: int, price: Optional[Decimal] = None):
self.id = self.generate_order_id()
self.symbol = symbol
self.side = side
self.type = order_type
self.quantity = quantity
self.price = price
self.filled_quantity = 0
self.status = "NEW"
self.created_at = datetime.now()
def generate_order_id(self) -> str:
import uuid
return str(uuid.uuid4())
class OrderManager:
def __init__(self):
self.orders = {}
self.positions = {}
def place_order(self, order: Order) -> str:
"""Place new order"""
self.orders[order.id] = order
# Route to exchange/broker
self.route_order(order)
return order.id
def cancel_order(self, order_id: str) -> bool:
"""Cancel existing order"""
if order_id in self.orders:
order = self.orders[order_id]
if order.status in ["NEW", "PARTIALLY_FILLED"]:
order.status = "CANCELLED"
return True
return False
def route_order(self, order: Order):
"""Smart order routing"""
# Check for best execution venue
venues = self.get_venue_quotes(order.symbol)
best_venue = self.select_best_venue(venues, order)
# Send order to venue
self.send_to_venue(order, best_venue)Risk Management
class RiskManager:
def __init__(self, max_position_size: float = 0.1,
max_portfolio_risk: float = 0.02,
stop_loss_pct: float = 0.05):
self.max_position_size = max_position_size
self.max_portfolio_risk = max_portfolio_risk
self.stop_loss_pct = stop_loss_pct
def calculate_position_size(self, capital: float, price: float,
volatility: float) -> int:
"""Calculate optimal position size using Kelly Criterion"""
max_position_value = capital * self.max_position_size
shares = int(max_position_value / price)
# Adjust for volatility
risk_adjusted_shares = int(shares * (1 - volatility))
return max(0, risk_adjusted_shares)
def check_risk_limits(self, portfolio: dict) -> bool:
"""Check if portfolio is within risk limits"""
total_value = portfolio['cash'] + sum(p['value'] for p in portfolio['positions'])
total_risk = sum(p['risk'] for p in portfolio['positions'])
if total_risk / total_value > self.max_portfolio_risk:
return False
return True
def calculate_var(self, returns: pd.Series, confidence: float = 0.95) -> float:
"""Calculate Value at Risk"""
return returns.quantile(1 - confidence)Market Data Processing
class MarketDataProcessor:
def __init__(self):
self.order_book = {'bids': [], 'asks': []}
def process_tick(self, tick: dict):
"""Process real-time tick data"""
if tick['type'] == 'trade':
self.process_trade(tick)
elif tick['type'] == 'quote':
self.update_order_book(tick)
def update_order_book(self, quote: dict):
"""Update order book with new quote"""
if quote['side'] == 'bid':
self.order_book['bids'] = sorted(
self.order_book['bids'] + [(quote['price'], quote['size'])],
key=lambda x: x[0],
reverse=True
)[:100] # Keep top 100
else:
self.order_book['asks'] = sorted(
self.order_book['asks'] + [(quote['price'], quote['size'])],
key=lambda x: x[0]
)[:100]
def calculate_vwap(self, trades: list) -> float:
"""Calculate Volume Weighted Average Price"""
total_volume = sum(t['volume'] for t in trades)
vwap = sum(t['price'] * t['volume'] for t in trades) / total_volume
return vwap
def calculate_spread(self) -> float:
"""Calculate bid-ask spread"""
if self.order_book['bids'] and self.order_book['asks']:
best_bid = self.order_book['bids'][0][0]
best_ask = self.order_book['asks'][0][0]
return best_ask - best_bid
return 0Best Practices
- Always backtest strategies on historical data
- Implement proper risk management
- Monitor execution quality (slippage, fill rates)
- Use limit orders to control execution price
- Implement circuit breakers for risk control
- Log all trades and orders for audit
- Test in paper trading before live deployment
- Monitor latency in real-time systems
- Implement failover mechanisms
- Regular strategy performance review
Anti-Patterns
❌ No backtesting before live trading
❌ Ignoring transaction costs
❌ Over-optimization (curve fitting)
❌ No risk management
❌ Trading without stop losses
❌ Ignoring market microstructure
❌ No position sizing strategy
Resources
- QuantConnect: https://www.quantconnect.com/
- Zipline: https://www.zipline.io/
- Backtrader: https://www.backtrader.com/
- Interactive Brokers API: https://interactivebrokers.github.io/