Разработка бота для DCA (Dollar Cost Averaging)

Разработка бота для DCA

DCA (Dollar-Cost Averaging) — это стратегия покупки фиксированной суммы актива через равные промежутки времени независимо от цены. Купили по $50,000, купили по $45,000, купили по $55,000 — средняя цена входа сглаживается. Бот автоматизирует эту дисциплину.

Как работает DCA-бот

Логика проста: каждый N период (час, день, неделя) бот исполняет market или limit ордер на фиксированную сумму в USD. Никакого анализа, никаких индикаторов — только расписание.

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:
            # Проверяем наличие баланса
            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

            # Исполняем покупку
            order = await self.exchange.place_market_order(
                symbol=self.config.symbol,
                side='buy',
                quote_order_qty=float(self.config.amount_per_order)  # в 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: куплено {order.filled_quantity:.6f} {self.config.base_currency} "
                f"по {order.fill_price:.2f} USDT\n"
                f"Средняя цена входа: {avg_price:.2f} USDT"
            )

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

Конфигурация

@dataclass
class DCAConfig:
    symbol: str = 'BTCUSDT'
    base_currency: str = 'BTC'
    quote_currency: str = 'USDT'
    amount_per_order: Decimal = Decimal('100')  # $100 за раз

    # Расписание
    interval: str = 'daily'  # 'hourly', 'daily', 'weekly'
    time_utc: str = '12:00'  # время исполнения

    # Опциональные условия
    dip_buying: bool = False     # покупать больше при падении
    dip_threshold: float = 5.0  # % падение = дополнительная покупка
    dip_multiplier: float = 2.0 # удвоить сумму при dip

    # Лимиты
    max_total_investment: Decimal = Decimal('10000')  # максимум за всё время
    stop_above_price: float = None  # остановить при цене выше N

Расширенный DCA: покупка на просадках

Vanilla DCA покупает всегда одинаково. Улучшение: удваиваем покупку при падении цены:

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)

Планировщик задач

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)

Статистика и аналитика

Бот должен показывать:

• Среднюю цену входа (avg cost basis)
• Текущий unrealized PnL
• Количество и суммы всех DCA покупок
• График equity curve

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-бот — это один из самых простых, но при этом один из наиболее эффективных инструментов для долгосрочных инвесторов. Исследования показывают, что DCA стратегия в BTC за любой 4-летний период исторически давала положительный результат. Разработка занимает 1-2 дня, но ценность для пользователя — долгосрочная.