create_dca_order
Automate recurring cryptocurrency purchases through dollar-cost averaging to build positions gradually and reduce market timing risks.
Instructions
Create a DCA order.
Expects a DCAOrderRequestContainer, returns a list of DCAOrderResponse.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| dca_order_requests | Yes |
Implementation Reference
- armor_crypto_mcp/armor_mcp.py:418-432 (handler)MCP tool handler for 'create_dca_order'. This is the primary execution function for the MCP tool, decorated with @mcp.tool() which handles registration. It validates input via type hints, checks authentication, calls the underlying client API, and returns results or errors.@mcp.tool() async def create_dca_order(dca_order_requests: DCAOrderRequestContainer) -> List[DCAOrderResponse]: """ Create a DCA order. Expects a DCAOrderRequestContainer, returns a list of DCAOrderResponse. """ if not armor_client: return [{"error": "Not logged in"}] try: result: List[DCAOrderResponse] = await armor_client.create_dca_order(dca_order_requests) return result except Exception as e: return [{"error": str(e)}]
- Underlying client method in ArmorWalletAPIClient that serializes the DCAOrderRequestContainer to JSON payload and makes a POST API call to the backend endpoint 'transactions/dca-order/create/' to execute the DCA order creation.async def create_dca_order(self, data: DCAOrderRequestContainer) -> List[DCAOrderResponse]: """Create a DCA order.""" # payload = [v.model_dump() for v in data.dca_order_requests] payload = data.model_dump(exclude_none=True)['dca_order_requests'] return await self._api_call("POST", "transactions/dca-order/create/", payload)
- Pydantic container model for input: wraps a list of DCAOrderRequest objects, used for batch operations.class DCAOrderRequestContainer(BaseModel): dca_order_requests: List[DCAOrderRequest]
- Pydantic model defining the input schema for a single DCA order request, including wallet, tokens, amount, schedule, duration, execution type, and optional conditional watchers.class DCAOrderRequest(BaseModel): wallet: str = Field(description="name of the wallet") input_token: str = Field(description="public address of the input token. To get the address from a token symbol use `get_token_details`") output_token: str = Field(description="public address of the output token. To get the address from a token symbol use `get_token_details`") amount: float = Field(description="total amount of input token to invest") cron_expression: str = Field(description="cron expression for the DCA worker execution frequency") strategy_duration_unit: Literal["MINUTE", "HOUR", "DAY", "WEEK", "MONTH", "YEAR"] = Field(description="unit of the duration of the DCA order") strategy_duration: int = Field(description="Total running time of the DCA order given in strategy duration units, should be more than 0") execution_type: Literal["MULTIPLE", "SINGLE"] = Field(description="set to SINGLE only if the user is asking for a single scheduled order, MULTIPLE if it is a true DCA") token_address_watcher: Optional[str] = Field(description="If the DCA is conditional, public address of the token to watch.") watch_field: Optional[Literal["liquidity", "marketCap", "price"]] = Field(description="If the DCA is conditional, field to watch for the condition") delta_type: Optional[Literal["INCREASE", "DECREASE", "MOVE", "MOVE_DAILY", "AVERAGE_MOVE"]] = Field(description="If the DCA is conditional, the operator of the watch field in the conditional statement") delta_percentage: Optional[float] = Field(description="If the DCA is conditional, percentage of the change to watch for given the delta_type") time_zone: Optional[str] = Field(description="user's time zone. Defaults to UTC")
- Pydantic model defining the output schema for a DCA order response, including order details, status, token info, watchers, and transactions.class DCAOrderResponse(BaseModel): id: str = Field(description="id of the DCA order") amount: float = Field(description="amount of tokens to invest") investment_per_cycle: float = Field(description="amount of tokens to invest per cycle") cycles_completed: int = Field(description="number of cycles completed") total_cycles: int = Field(description="total number of cycles") human_readable_expiry: str = Field(description="human readable expiry date of the DCA order") status: str = Field(description="status of the DCA order") input_token_data: TokenData = Field(description="details of the input token") output_token_data: TokenData = Field(description="details of the output token") wallet_name: str = Field(description="name of the wallet") watchers: List[DCAWatcher] = Field(description="list of watchers for the DCA order") dca_transactions: List[dict] = Field(description="list of DCA transactions") # Can be further typed if structure is known created: str = Field(description="Linux timestamp of the creation of the order")