DCA Bot Development (Dollar Cost Averaging)

DCA Bot Development

DCA (Dollar-Cost Averaging) is a strategy of buying a fixed amount of an asset at regular intervals regardless of price. Bought at $50,000, bought at $45,000, bought at $55,000 — the average entry price smooths out. The bot automates this discipline.

How DCA Bot Works

The logic is simple: every N periods (hour, day, week), the bot executes a market or limit order for a fixed USD amount. No analysis, no indicators — just schedule.

import asyncio
from decimal import Decimal
from datetime import datetime

class DCABot:
    def __init__(self, config: DCAConfig, exchange_client):
        self.config = config
        self.exchange = exchange_client
        self.total_invested = Decimal(0)
        self.total_purchased = Decimal(0)

    async def execute_dca_order(self):
        try:
            # Check balance
            balance = await self.exchange.get_balance(self.config.quote_currency)
            if balance < self.config.amount_per_order:
                await self.alert(f"Insufficient balance: {balance} < {self.config.amount_per_order}")
                return

            # Execute purchase
            order = await self.exchange.place_market_order(
                symbol=self.config.symbol,
                side='buy',
                quote_order_qty=float(self.config.amount_per_order)  # in USDT
            )

            self.total_invested += self.config.amount_per_order
            self.total_purchased += Decimal(str(order.filled_quantity))

            avg_price = self.total_invested / self.total_purchased

            await self.log_purchase(order, avg_price)
            await self.telegram_notify(
                f"DCA: bought {order.filled_quantity:.6f} {self.config.base_currency} "
                f"at {order.fill_price:.2f} USDT\n"
                f"Average entry price: {avg_price:.2f} USDT"
            )

        except Exception as e:
            await self.alert(f"DCA order failed: {e}")

Configuration

@dataclass
class DCAConfig:
    symbol: str = 'BTCUSDT'
    base_currency: str = 'BTC'
    quote_currency: str = 'USDT'
    amount_per_order: Decimal = Decimal('100')  # $100 at a time

    # Schedule
    interval: str = 'daily'  # 'hourly', 'daily', 'weekly'
    time_utc: str = '12:00'  # execution time

    # Optional conditions
    dip_buying: bool = False     # buy more on dips
    dip_threshold: float = 5.0  # % drop = additional purchase
    dip_multiplier: float = 2.0 # double amount on dip

    # Limits
    max_total_investment: Decimal = Decimal('10000')  # maximum ever
    stop_above_price: float = None  # stop when price above N

Enhanced DCA: Buying on Dips

Vanilla DCA buys the same amount always. Improvement: double the purchase on price drops:

async def enhanced_dca_order(self):
    current_price = await self.exchange.get_price(self.config.symbol)
    last_purchase_price = await self.db.get_last_purchase_price(self.config.symbol)

    amount = self.config.amount_per_order

    if last_purchase_price and self.config.dip_buying:
        price_drop = (last_purchase_price - current_price) / last_purchase_price * 100
        if price_drop >= self.config.dip_threshold:
            amount *= Decimal(str(self.config.dip_multiplier))
            logger.info(f"Dip detected ({price_drop:.1f}%), buying {self.config.dip_multiplier}x")

    await self.execute_order(amount)

Task Scheduler

import schedule
import time

def start_scheduler(bot: DCABot, config: DCAConfig):
    if config.interval == 'hourly':
        schedule.every().hour.do(lambda: asyncio.run(bot.execute_dca_order()))
    elif config.interval == 'daily':
        schedule.every().day.at(config.time_utc).do(lambda: asyncio.run(bot.execute_dca_order()))
    elif config.interval == 'weekly':
        schedule.every().monday.at(config.time_utc).do(lambda: asyncio.run(bot.execute_dca_order()))

    while True:
        schedule.run_pending()
        time.sleep(60)

Statistics and Analytics

Bot should display:

• Average entry price (avg cost basis)
• Current unrealized PnL
• Count and amounts of all DCA purchases
• Equity curve chart

def get_statistics(self) -> dict:
    current_price = self.get_current_price()
    current_value = self.total_purchased * Decimal(str(current_price))
    unrealized_pnl = current_value - self.total_invested
    unrealized_pnl_percent = unrealized_pnl / self.total_invested * 100

    return {
        'total_invested': str(self.total_invested),
        'total_purchased': str(self.total_purchased),
        'avg_purchase_price': str(self.total_invested / self.total_purchased),
        'current_value': str(current_value),
        'unrealized_pnl': str(unrealized_pnl),
        'unrealized_pnl_percent': float(unrealized_pnl_percent),
        'num_orders': self.order_count,
    }

DCA bot is one of the simplest, yet one of the most effective tools for long-term investors. Research shows that DCA strategy in BTC over any 4-year historical period has been profitable. Development takes 1-2 days, but value to the user is long-term.