Moving Averages¶
ArBot incorporates moving averages as a key component of its trend filtering system, providing sophisticated market analysis to enhance arbitrage decision-making. Moving averages help identify market trends, momentum shifts, and optimal entry/exit points for arbitrage opportunities.
Overview¶
Role in Arbitrage Strategy¶
Market Context Analysis: - Determine overall market direction - Filter opportunities based on trend strength - Reduce false signals in choppy markets - Improve timing of arbitrage execution
Integration with Arbitrage: Moving averages don't replace arbitrage logic but enhance it by providing market context that improves the probability of successful trades.
# Example: Enhanced arbitrage with moving average filter
if arbitrage_opportunity and price_above_ma20 and trend_strength > 0.6:
execute_arbitrage()
Moving Average Types¶
Simple Moving Average (SMA)¶
Calculation:
Characteristics: - Equal weight to all data points - Smooth, stable trend representation - Less sensitive to recent price changes - Good for identifying long-term trends
Configuration:
Exponential Moving Average (EMA)¶
Calculation:
def calculate_ema(prices, period, alpha=None):
if alpha is None:
alpha = 2 / (period + 1)
ema = [prices[0]] # Start with first price
for price in prices[1:]:
ema_value = alpha * price + (1 - alpha) * ema[-1]
ema.append(ema_value)
return ema[-1]
Characteristics: - More weight to recent prices - Faster response to price changes - Better for short-term trend detection - Reduced lag compared to SMA
Configuration:
{
"moving_averages": {
"ema_periods": [12, 26, 50],
"ema_weight": 0.5,
"ema_alpha_custom": false
}
}
Weighted Moving Average (WMA)¶
Calculation:
def calculate_wma(prices, period):
weights = list(range(1, period + 1))
weighted_sum = sum(price * weight for price, weight in zip(prices[-period:], weights))
weight_sum = sum(weights)
return weighted_sum / weight_sum
Characteristics: - Linear weighting (most recent = highest weight) - Balance between SMA and EMA - Good for medium-term trend analysis - Customizable weighting schemes
Adaptive Moving Average (AMA)¶
Kaufman's Adaptive Moving Average:
def calculate_ama(prices, period=10, fast_period=2, slow_period=30):
direction = abs(prices[-1] - prices[-period])
volatility = sum(abs(prices[i] - prices[i-1]) for i in range(-period+1, 0))
efficiency_ratio = direction / volatility if volatility != 0 else 0
fast_sc = 2 / (fast_period + 1)
slow_sc = 2 / (slow_period + 1)
smoothing_constant = (efficiency_ratio * (fast_sc - slow_sc) + slow_sc) ** 2
return smoothing_constant * prices[-1] + (1 - smoothing_constant) * previous_ama
Characteristics: - Adapts to market volatility - Fast in trending markets - Slow in sideways markets - Reduces whipsaws
Multi-Timeframe Analysis¶
Timeframe Configuration¶
{
"moving_averages": {
"timeframes": {
"1m": {"periods": [20, 50], "weight": 0.2},
"5m": {"periods": [20, 50], "weight": 0.3},
"15m": {"periods": [20, 50], "weight": 0.3},
"1h": {"periods": [20, 50], "weight": 0.2}
}
}
}
Multi-Timeframe Signals¶
Trend Alignment:
def check_trend_alignment(symbol):
timeframes = ["1m", "5m", "15m", "1h"]
trends = {}
for tf in timeframes:
ma_short = get_moving_average(symbol, tf, 20)
ma_long = get_moving_average(symbol, tf, 50)
current_price = get_current_price(symbol)
if current_price > ma_short > ma_long:
trends[tf] = "bullish"
elif current_price < ma_short < ma_long:
trends[tf] = "bearish"
else:
trends[tf] = "neutral"
# Check alignment
bullish_count = sum(1 for trend in trends.values() if trend == "bullish")
bearish_count = sum(1 for trend in trends.values() if trend == "bearish")
if bullish_count >= 3:
return "strong_bullish"
elif bearish_count >= 3:
return "strong_bearish"
elif bullish_count > bearish_count:
return "weak_bullish"
elif bearish_count > bullish_count:
return "weak_bearish"
else:
return "neutral"
Trend Filtering Strategies¶
Basic Trend Filter¶
Price Above/Below Moving Average:
def basic_trend_filter(price, ma_period=20, ma_type="ema"):
ma_value = calculate_moving_average(price_history, ma_period, ma_type)
if price > ma_value:
return "uptrend"
elif price < ma_value:
return "downtrend"
else:
return "neutral"
Golden Cross / Death Cross¶
Moving Average Crossover Signals:
def detect_ma_crossover(short_ma, long_ma, previous_short_ma, previous_long_ma):
# Golden Cross: Short MA crosses above Long MA
if short_ma > long_ma and previous_short_ma <= previous_long_ma:
return "golden_cross"
# Death Cross: Short MA crosses below Long MA
elif short_ma < long_ma and previous_short_ma >= previous_long_ma:
return "death_cross"
return None
Trend Strength Analysis¶
Moving Average Convergence/Divergence:
def calculate_trend_strength(price, ma_short, ma_long):
# Distance from price to moving averages
price_ma_short_distance = abs(price - ma_short) / ma_short
price_ma_long_distance = abs(price - ma_long) / ma_long
# Moving average separation
ma_separation = abs(ma_short - ma_long) / ma_long
# Combine metrics
trend_strength = (price_ma_short_distance + ma_separation) / 2
return min(trend_strength, 1.0) # Cap at 1.0
Advanced Moving Average Techniques¶
Volume-Weighted Moving Average (VWMA)¶
Incorporating Volume Data:
def calculate_vwma(prices, volumes, period):
price_volume_sum = sum(price * volume for price, volume in zip(prices[-period:], volumes[-period:]))
volume_sum = sum(volumes[-period:])
return price_volume_sum / volume_sum if volume_sum > 0 else 0
Benefits: - Weights prices by trading volume - More representative of actual trading activity - Better for high-volume arbitrage analysis - Reduces impact of low-volume price spikes
Hull Moving Average (HMA)¶
Reduced Lag Moving Average:
def calculate_hma(prices, period):
half_period = period // 2
sqrt_period = int(period ** 0.5)
wma_half = calculate_wma(prices, half_period)
wma_full = calculate_wma(prices, period)
# Create intermediate series
intermediate = 2 * wma_half - wma_full
# Apply WMA to intermediate series
return calculate_wma([intermediate], sqrt_period)
Characteristics: - Significantly reduced lag - Smoother than EMA - Better trend following - Ideal for fast arbitrage decisions
Kalman Filter Moving Average¶
Adaptive Noise Reduction:
class KalmanMA:
def __init__(self, process_variance=1e-5, measurement_variance=1e-1):
self.process_variance = process_variance
self.measurement_variance = measurement_variance
self.posteri_estimate = 0
self.posteri_error_estimate = 1
def update(self, measurement):
# Prediction
priori_estimate = self.posteri_estimate
priori_error_estimate = self.posteri_error_estimate + self.process_variance
# Update
blending_factor = priori_error_estimate / (priori_error_estimate + self.measurement_variance)
self.posteri_estimate = priori_estimate + blending_factor * (measurement - priori_estimate)
self.posteri_error_estimate = (1 - blending_factor) * priori_error_estimate
return self.posteri_estimate
Integration with Arbitrage Logic¶
Trend-Filtered Arbitrage¶
Enhanced Decision Making:
def enhanced_arbitrage_decision(arbitrage_signal, symbol):
# Get moving average context
ma_context = get_ma_context(symbol)
# Base arbitrage profitability
base_profit = arbitrage_signal.profit_percent
# Trend adjustment factor
trend_factor = calculate_trend_factor(ma_context)
# Adjusted expected profit
adjusted_profit = base_profit * trend_factor
# Decision logic
if adjusted_profit > config.min_profit_threshold:
return True, adjusted_profit
else:
return False, adjusted_profit
def calculate_trend_factor(ma_context):
trend_strength = ma_context["trend_strength"]
trend_direction = ma_context["trend_direction"]
if trend_direction == "strong_bullish":
return 1.0 + (trend_strength * 0.3) # Up to 30% bonus
elif trend_direction == "strong_bearish":
return 1.0 + (trend_strength * 0.2) # Up to 20% bonus
elif trend_direction == "neutral":
return 0.8 # 20% penalty for choppy markets
else:
return 1.0
Risk Adjustment¶
Position Sizing with Moving Averages:
def adjust_position_size_by_trend(base_size, trend_context):
trend_strength = trend_context["strength"]
trend_consistency = trend_context["consistency"]
# Strong consistent trends allow larger positions
if trend_strength > 0.8 and trend_consistency > 0.7:
return base_size * 1.2
# Weak or inconsistent trends require smaller positions
elif trend_strength < 0.3 or trend_consistency < 0.4:
return base_size * 0.6
return base_size
Performance Optimization¶
Efficient Calculation¶
Incremental Updates:
class IncrementalMA:
def __init__(self, period, ma_type="sma"):
self.period = period
self.ma_type = ma_type
self.values = deque(maxlen=period)
self.current_ma = 0
def update(self, new_value):
if len(self.values) == self.period:
# Remove oldest value from sum
old_value = self.values[0]
self.current_ma = (self.current_ma * self.period - old_value + new_value) / self.period
else:
# Still building up to full period
self.values.append(new_value)
self.current_ma = sum(self.values) / len(self.values)
return self.current_ma
Parallel Calculation¶
Multi-Symbol Processing:
import asyncio
async def calculate_mas_parallel(symbols, ma_configs):
tasks = []
for symbol in symbols:
for config in ma_configs:
task = asyncio.create_task(
calculate_ma_async(symbol, config["period"], config["type"])
)
tasks.append(task)
results = await asyncio.gather(*tasks)
return organize_results(results, symbols, ma_configs)
Configuration Examples¶
Conservative Setup¶
Stable Trend Following:
{
"moving_averages": {
"enabled": true,
"primary_type": "sma",
"periods": [50, 100, 200],
"trend_filter_strength": "high",
"min_trend_consistency": 0.8,
"timeframes": ["15m", "1h", "4h"]
}
}
Aggressive Setup¶
Fast Response:
{
"moving_averages": {
"enabled": true,
"primary_type": "ema",
"periods": [12, 26, 50],
"trend_filter_strength": "medium",
"adaptive_periods": true,
"timeframes": ["1m", "5m", "15m"]
}
}
Balanced Setup¶
Multi-Type Analysis:
{
"moving_averages": {
"enabled": true,
"types": {
"sma": {"periods": [20, 50], "weight": 0.3},
"ema": {"periods": [12, 26], "weight": 0.4},
"hma": {"periods": [21], "weight": 0.3}
},
"consensus_required": 0.6,
"timeframes": ["5m", "15m", "1h"]
}
}
Monitoring and Analytics¶
Moving Average Health¶
Quality Metrics:
def assess_ma_quality(symbol, period):
ma_values = get_ma_history(symbol, period, lookback=100)
# Smoothness (lower is better)
smoothness = calculate_smoothness(ma_values)
# Responsiveness (balance needed)
responsiveness = calculate_responsiveness(ma_values)
# Trend accuracy
accuracy = backtest_ma_signals(symbol, period)
return {
"smoothness": smoothness,
"responsiveness": responsiveness,
"accuracy": accuracy,
"overall_score": (accuracy * 0.5 + (1-smoothness) * 0.3 + responsiveness * 0.2)
}
Performance Tracking¶
MA-Enhanced Trading Results:
def track_ma_performance():
trades_with_ma = get_trades_with_ma_filter()
trades_without_ma = get_trades_without_ma_filter()
comparison = {
"with_ma": {
"win_rate": calculate_win_rate(trades_with_ma),
"avg_profit": calculate_avg_profit(trades_with_ma),
"sharpe_ratio": calculate_sharpe_ratio(trades_with_ma)
},
"without_ma": {
"win_rate": calculate_win_rate(trades_without_ma),
"avg_profit": calculate_avg_profit(trades_without_ma),
"sharpe_ratio": calculate_sharpe_ratio(trades_without_ma)
}
}
return comparison
Troubleshooting¶
Common Issues¶
Lag Problems: - Use shorter periods for faster response - Consider EMA or HMA instead of SMA - Implement adaptive moving averages - Reduce number of periods calculated
False Signals: - Increase minimum trend strength requirements - Add multiple timeframe confirmation - Implement noise filters - Use volume-weighted moving averages
Performance Issues: - Implement incremental calculations - Use parallel processing for multiple symbols - Cache frequently accessed values - Optimize data structures
Optimization Tips¶
Parameter Tuning:
def optimize_ma_parameters(symbol, test_period_days=30):
best_params = None
best_performance = 0
# Test different period combinations
for short_period in range(5, 25, 5):
for long_period in range(30, 100, 10):
performance = backtest_ma_strategy(
symbol, short_period, long_period, test_period_days
)
if performance > best_performance:
best_performance = performance
best_params = (short_period, long_period)
return best_params, best_performance
Moving Average Selection
Start with EMA(20) and EMA(50) for trend filtering. These periods provide a good balance between responsiveness and stability for most arbitrage scenarios.
Overoptimization
Avoid overfitting moving average parameters to historical data. Use out-of-sample testing and walk-forward analysis to validate your configurations.
Market Adaptation
Moving average effectiveness varies with market conditions. Regularly review and adjust your parameters based on changing market volatility and trending behavior.
Created: July 12, 2025