How to Build a Crypto Portfolio Auto-Rebalancing Bot

Why Rebalancing Beats Buy-and-Hold (With Data)

A portfolio that starts as 50% BTC / 50% ETH drifts as prices move. Without rebalancing, a strong BTC rally leaves you 80% BTC — concentrated risk you didn't intend to take.

Backtests consistently show rebalancing improves risk-adjusted returns (Sharpe ratio) even if raw returns are similar. The reason: you systematically buy low (underperformers) and sell high (overperformers).

Strategy Options

1. Threshold-Based Rebalancing

Rebalance when any asset drifts more than X% from its target:

• Target: BTC 50%, ETH 30%, SOL 20%
• If BTC reaches 65% → sell down to 50%
• Best for: reducing unnecessary trades

2. Calendar-Based Rebalancing

Rebalance on a fixed schedule (weekly, monthly) regardless of drift.

• Best for: predictability, simplicity

3. Hybrid (Recommended)

Rebalance when drift exceeds 5% AND it's been at least 7 days since last rebalance.

• Reduces unnecessary trades while preventing large drifts.

Building the Rebalancer

import ccxt
from dataclasses import dataclass
from datetime import datetime, timedelta
import json
from pathlib import Path

@dataclass
class TargetAllocation:
    symbol: str       # e.g., 'BTC'
    target_pct: float # e.g., 0.50 for 50%

class PortfolioRebalancer:
    def __init__(
        self,
        exchange: ccxt.Exchange,
        targets: list[TargetAllocation],
        rebalance_threshold: float = 0.05,  # 5% drift triggers rebalance
        min_trade_usd: float = 20.0,        # Minimum trade size (avoid dust)
        cooldown_days: int = 7,             # Minimum days between rebalances
    ):
        self.exchange = exchange
        self.targets = {t.symbol: t.target_pct for t in targets}
        self.threshold = rebalance_threshold
        self.min_trade_usd = min_trade_usd
        self.cooldown_days = cooldown_days
        self.state_file = Path("rebalance_state.json")
        self.state = self._load_state()
        
        # Validate allocations sum to 1.0
        total = sum(self.targets.values())
        if abs(total - 1.0) > 0.001:
            raise ValueError(f"Target allocations must sum to 1.0, got {total:.3f}")
    
    def _load_state(self) -> dict:
        if self.state_file.exists():
            return json.loads(self.state_file.read_text())
        return {"last_rebalance": None}
    
    def _save_state(self):
        self.state_file.write_text(json.dumps(self.state))
    
    def get_current_portfolio(self) -> dict:
        """Get current holdings as {symbol: usd_value}"""
        balance = self.exchange.fetch_balance()
        tickers = self.exchange.fetch_tickers(
            [f"{sym}/USDT" for sym in self.targets]
        )
        
        portfolio = {}
        for sym in self.targets:
            held = balance.get(sym, {}).get('total', 0)
            price = tickers.get(f"{sym}/USDT", {}).get('last', 0)
            portfolio[sym] = held * price
        
        # Include USDT in total calculation
        portfolio['USDT'] = balance.get('USDT', {}).get('total', 0)
        return portfolio
    
    def calculate_drift(self, portfolio: dict) -> dict:
        """Calculate how far each asset is from its target"""
        total_value = sum(portfolio.values())
        if total_value == 0:
            return {}
        
        drift = {}
        for sym, target_pct in self.targets.items():
            current_usd = portfolio.get(sym, 0)
            current_pct = current_usd / total_value
            drift[sym] = current_pct - target_pct  # Positive = overweight
        
        return drift
    
    def needs_rebalance(self, drift: dict) -> bool:
        """Check if rebalancing is needed"""
        # Check cooldown
        if self.state['last_rebalance']:
            last = datetime.fromisoformat(self.state['last_rebalance'])
            if datetime.now() - last < timedelta(days=self.cooldown_days):
                print(f"In cooldown ({self.cooldown_days}d). Last: {last.date()}")
                return False
        
        # Check if any asset has drifted beyond threshold
        max_drift = max(abs(d) for d in drift.values()) if drift else 0
        return max_drift > self.threshold
    
    def calculate_trades(self, portfolio: dict, drift: dict) -> list[dict]:
        """Calculate the trades needed to rebalance"""
        total_value = sum(portfolio.values())
        trades = []
        
        for sym, current_drift in drift.items():
            target_usd = total_value * self.targets[sym]
            current_usd = portfolio.get(sym, 0)
            diff_usd = target_usd - current_usd  # Positive = need to buy
            
            if abs(diff_usd) >= self.min_trade_usd:
                trades.append({
                    'symbol': sym,
                    'pair': f"{sym}/USDT",
                    'action': 'BUY' if diff_usd > 0 else 'SELL',
                    'usd_amount': abs(diff_usd),
                    'current_pct': (current_usd / total_value) * 100,
                    'target_pct': self.targets[sym] * 100,
                    'drift_pct': current_drift * 100,
                })
        
        # Execute sells before buys (ensures USDT available)
        trades.sort(key=lambda x: 0 if x['action'] == 'SELL' else 1)
        return trades
    
    def execute_rebalance(self, dry_run: bool = True) -> dict:
        """Execute rebalancing trades"""
        portfolio = self.get_current_portfolio()
        drift = self.calculate_drift(portfolio)
        total_value = sum(portfolio.values())
        
        print(f"\n📊 Portfolio Value: ${total_value:,.2f}")
        print(f"{'Symbol':<8} {'Current':>10} {'Target':>8} {'Drift':>8}")
        for sym, d in drift.items():
            curr_pct = (portfolio.get(sym, 0) / total_value) * 100
            tgt_pct = self.targets[sym] * 100
            status = "⚠️" if abs(d) > self.threshold else "✓"
            print(f"{sym:<8} {curr_pct:>9.1f}% {tgt_pct:>7.1f}% {d*100:>+7.1f}% {status}")
        
        if not self.needs_rebalance(drift):
            print("\n✅ Portfolio within bounds — no rebalancing needed")
            return {'rebalanced': False, 'trades': []}
        
        trades = self.calculate_trades(portfolio, drift)
        
        print(f"\n🔄 Rebalancing required — {len(trades)} trades:")
        for t in trades:
            print(f"  {t['action']} ${t['usd_amount']:,.2f} of {t['symbol']}")
        
        if dry_run:
            print("\n⚠️  DRY RUN — set dry_run=False to execute")
            return {'rebalanced': False, 'trades': trades, 'dry_run': True}
        
        executed = []
        for trade in trades:
            try:
                ticker = self.exchange.fetch_ticker(trade['pair'])
                price = ticker['last']
                qty = trade['usd_amount'] / price
                
                if trade['action'] == 'BUY':
                    order = self.exchange.create_market_buy_order(
                        trade['pair'], qty,
                        params={'quoteOrderQty': trade['usd_amount']}
                    )
                else:
                    order = self.exchange.create_market_sell_order(trade['pair'], qty)
                
                print(f"✅ {trade['action']} {qty:.6f} {trade['symbol']} @ ${price:,.2f}")
                executed.append(trade)
                
            except Exception as e:
                print(f"❌ Failed to {trade['action']} {trade['symbol']}: {e}")
        
        self.state['last_rebalance'] = datetime.now().isoformat()
        self._save_state()
        
        return {'rebalanced': True, 'trades': executed}

# Example usage
exchange = ccxt.binance({'apiKey': API_KEY, 'secret': SECRET})

targets = [
    TargetAllocation('BTC', 0.40),
    TargetAllocation('ETH', 0.30),
    TargetAllocation('SOL', 0.20),
    TargetAllocation('BNB', 0.10),
]

rebalancer = PortfolioRebalancer(exchange, targets, rebalance_threshold=0.05)

# Run weekly check
import schedule
schedule.every().monday.at("09:00").do(lambda: rebalancer.execute_rebalance(dry_run=False))

When to Rebalance More Aggressively

During high-volatility periods (Fear & Greed below 25), increase rebalance frequency:

def adaptive_rebalance(rebalancer: PortfolioRebalancer):
    fg = get_fear_greed_index()
    
    if fg < 25:
        # Extreme fear: rebalance if 3% drift (buy the dip more aggressively)
        rebalancer.threshold = 0.03
        rebalancer.cooldown_days = 2
    elif fg > 75:
        # Extreme greed: rebalance if 10% drift (don't over-sell the bull)
        rebalancer.threshold = 0.10
        rebalancer.cooldown_days = 14
    else:
        rebalancer.threshold = 0.05
        rebalancer.cooldown_days = 7
    
    return rebalancer.execute_rebalance(dry_run=False)

Rebalancing is one of the simplest automations to build and one of the most consistently beneficial. Start with a monthly calendar rebalance before adding sophisticated triggers.