Portfolio Rebalancing Bot Development
A rebalancing bot maintains specified asset allocation in your portfolio. If BTC grew from 50% to 65% due to price rise, bot sells some BTC and buys other assets to target weights. This is disciplined portfolio management: automatically locks profits in grown assets and buys underperformers.
Rebalancing Strategies
Calendar Rebalancing (Time-based)
Rebalance on schedule regardless of deviations:
# Weekly rebalancing every Monday at 12:00 UTC
schedule = {
'type': 'calendar',
'interval': 'weekly',
'day': 'monday',
'time': '12:00'
}
Simple and predictable, but may coincide with unfavorable market moment.
Threshold Rebalancing (Deviation-based)
Rebalance only when asset weight deviates from target by N%:
def needs_rebalancing(
current_weights: dict,
target_weights: dict,
threshold_percent: float = 5.0
) -> bool:
for asset, target_weight in target_weights.items():
current = current_weights.get(asset, 0)
deviation = abs(current - target_weight)
if deviation >= threshold_percent:
return True
return False
More efficient: rebalance only when truly needed.
Hybrid Approach
Check daily but rebalance only when deviation > 5%:
class HybridRebalancer:
def should_rebalance(self, portfolio: Portfolio) -> bool:
current_weights = portfolio.get_weights()
max_deviation = max(
abs(current_weights[a] - self.target_weights[a])
for a in self.target_weights
)
return max_deviation >= self.threshold
Bot Implementation
Calculate Rebalancing Orders
from decimal import Decimal
from typing import dict
class RebalancingCalculator:
def calculate_trades(
self,
current_balances: dict[str, Decimal],
current_prices: dict[str, Decimal],
target_weights: dict[str, float] # sum = 100
) -> list[RebalanceTrade]:
# Calculate total portfolio value
total_value = sum(
current_balances[asset] * current_prices[asset]
for asset in current_balances
)
trades = []
for asset, target_weight in target_weights.items():
target_value = total_value * Decimal(str(target_weight / 100))
current_value = current_balances.get(asset, Decimal(0)) * current_prices[asset]
diff_value = target_value - current_value
if abs(diff_value) < Decimal('10'): # Minimum $10 to trade
continue
diff_quantity = diff_value / current_prices[asset]
trades.append(RebalanceTrade(
asset=asset,
side='buy' if diff_value > 0 else 'sell',
quantity=abs(diff_quantity),
value_usd=abs(diff_value),
current_weight=float(current_value / total_value * 100),
target_weight=target_weight
))
return trades
def optimize_trade_order(self, trades: list[RebalanceTrade]) -> list[RebalanceTrade]:
"""Sell first, then buy — no need for additional USDT"""
sells = [t for t in trades if t.side == 'sell']
buys = [t for t in trades if t.side == 'buy']
return sells + buys
Execute Rebalancing
class RebalancingBot:
async def execute_rebalance(self) -> RebalanceReport:
# Get current state
balances = await self.exchange.get_balances()
prices = await self.exchange.get_prices(list(self.target_weights.keys()))
current_weights = self.calculate_weights(balances, prices)
if not self.should_rebalance_now(current_weights):
logger.info("Rebalancing not needed")
return RebalanceReport(skipped=True)
trades = self.calculator.calculate_trades(balances, prices, self.target_weights)
optimized_trades = self.calculator.optimize_trade_order(trades)
executed = []
for trade in optimized_trades:
try:
order = await self.exchange.place_market_order(
symbol=f"{trade.asset}/USDT",
side=trade.side,
quantity=float(trade.quantity)
)
executed.append({
'trade': trade,
'fill_price': order.fill_price,
'actual_quantity': order.filled_quantity,
'fee': order.fee
})
logger.info(f"Rebalanced {trade.asset}: {trade.side} {trade.quantity:.4f}")
except Exception as e:
logger.error(f"Failed to execute {trade.asset} trade: {e}")
return RebalanceReport(
executed_trades=executed,
portfolio_before=current_weights,
total_cost_usd=sum(t['fee'] for t in executed),
timestamp=datetime.utcnow()
)
Target Allocations
Example Configuration
portfolio:
target_weights:
BTC: 40
ETH: 30
SOL: 15
BNB: 10
USDT: 5 # cash cushion
rebalancing:
strategy: hybrid
threshold_percent: 5.0
check_interval_hours: 24
min_trade_usd: 20
# Time restrictions
rebalance_window:
days: ['monday', 'tuesday', 'wednesday', 'thursday', 'friday']
hours_utc: [8, 20] # only 8 to 20 UTC
risk:
max_single_trade_percent: 20 # no more than 20% of portfolio per trade
slippage_tolerance: 0.5
Dynamic Weights
Advanced: weights change based on market conditions:
def calculate_dynamic_weights(market_data: dict) -> dict:
"""
Risk-parity: distribute weight inversely proportional to volatility.
Lower volatility assets get higher weight.
"""
assets = ['BTC', 'ETH', 'SOL', 'BNB']
volatilities = {a: market_data[a]['vol_30d'] for a in assets}
# Inverse volatility
inv_vol = {a: 1 / v for a, v in volatilities.items()}
total_inv_vol = sum(inv_vol.values())
weights = {a: inv_vol[a] / total_inv_vol * 100 for a in assets}
return weights
Tax Considerations
Each rebalancing is a taxable event in most jurisdictions (capital gain/loss realization). With high rebalancing frequency, tax obligations may exceed rebalancing benefits.
Optimization: rebalance through loss positions (tax-loss harvesting) — sell losing assets, allocate to gainers. Automating tax-loss harvesting is separate and much-needed feature for tax jurisdictions with detailed crypto reporting (USA, Germany).
Rebalancing bot is discipline automated into code. Research shows: systematic quarterly rebalancing outperforms buy-and-hold by 0.5-2% per year on volatile assets due to rebalancing premium.