the0
Open source algorithmic
trading platform

1from typing import Dict, Any
2import logging
3from datetime import datetime
4from alpaca.trading.client import TradingClient
5from alpaca.trading.requests import MarketOrderRequest
6from alpaca.trading.enums import OrderSide, TimeInForce
7
8# Configure logging
9logging.basicConfig(level=logging.INFO)
10logger = logging.getLogger(__name__)
11
12def main(id: str, config: Dict[str, Any]) -> Dict[str, Any]:
13    """
14    Main DCA bot execution function using Alpaca.
15    
16    Args:
17        id: Unique bot instance ID
18        config: Bot configuration from user
19        
20    Returns:
21        Dict with execution status and results
22    """
23    logger.info(f"Starting DCA bot {id}")
24    
25    try:
26        # Extract configuration
27        api_key = config["api_key"]
28        secret_key = config["secret_key"]
29        paper = config.get("paper", True)
30        symbol = config["symbol"]
31        amount = config["amount"]
32        
33        # Initialize Alpaca trading client
34        client = TradingClient(
35            api_key=api_key,
36            secret_key=secret_key,
37            paper=paper
38        )
39        
40        # Calculate and execute purchase
41        order = MarketOrderRequest(
42            symbol=symbol,
43            notional=amount,
44            side=OrderSide.BUY,
45            time_in_force=TimeInForce.DAY
46        )
47        
48        result = client.submit_order(order_data=order)
49        
50        return {
51            "status": "success",
52            "message": f"Purchased ${amount} of {symbol}",
53            "order_id": str(result.id)
54        }
55        
56    except Exception as e:
57        logger.error(f"Bot execution failed: {e}")
58        return {
59            "status": "error",
60            "message": str(e),
61            "timestamp": datetime.now().isoformat()
62        }