the0
Self-hosted execution engine
for algorithmic trading bots

1"""
2DCA Bot - Dollar Cost Averaging with Alpaca
3A scheduled bot that executes recurring purchases.
4"""
5
6import structlog
7from typing import Dict, Any
8from alpaca.trading.client import TradingClient
9from alpaca.trading.requests import MarketOrderRequest
10from alpaca.trading.enums import OrderSide, TimeInForce
11
12structlog.configure(
13    processors=[
14        structlog.processors.TimeStamper(fmt="iso"),
15        structlog.processors.JSONRenderer()
16    ]
17)
18logger = structlog.get_logger()
19
20
21def main(id: str, config: Dict[str, Any]) -> Dict[str, Any]:
22    """Bot entry point - called on each scheduled run."""
23    api_key = config["api_key"]
24    secret_key = config["secret_key"]
25    symbol = config.get("symbol", "BTC/USD")
26    amount = config.get("amount", 100)
27    paper = config.get("paper", True)
28
29    logger.info("bot_started", bot_id=id, symbol=symbol, amount=amount)
30
31    # Initialize Alpaca client
32    client = TradingClient(api_key=api_key, secret_key=secret_key, paper=paper)
33
34    # Execute DCA purchase
35    order = MarketOrderRequest(
36        symbol=symbol,
37        notional=amount,
38        side=OrderSide.BUY,
39        time_in_force=TimeInForce.DAY
40    )
41    result = client.submit_order(order_data=order)
42
43    # Emit trade metric
44    logger.info(
45        "order_executed",
46        _metric="trade",
47        symbol=symbol,
48        side="buy",
49        amount=amount,
50        order_id=str(result.id)
51    )
52
53    return {
54        "status": "success",
55        "message": f"Purchased ${amount} of {symbol}",
56        "data": {"order_id": str(result.id)}
57    }