Oracles 🔮

Introduction

Oracles are entities that can update state variables in smart contracts and whose goal is usually to accurately estimate or predict some real world quantity or quantities. These quantities can then be used in the logic of other smart contracts. This guide shows how to write a CosmPy script that deploys and updates an oracle contract with a coin price, and another script that deploys a contract that queries this coin price.

Walk-through

Here we provide an overview guide for setting up your own aerial oracle in few steps.

We initially need to download the binaries for both contracts, which can be done as follows:

wget https://raw.githubusercontent.com/fetchai/agents-aea/develop/packages/fetchai/contracts/oracle/build/oracle.wasm
wget https://raw.githubusercontent.com/fetchai/agents-aea/develop/packages/fetchai/contracts/oracle_client/build/oracle_client.wasm

Aerial oracle

First of all, create a Python script and name it: touch aerial_oracle.py

We would then also require the following imports:

from time import sleep
import requests
from cosmpy.aerial.client import LedgerClient, NetworkConfig
from cosmpy.aerial.contract import LedgerContract
from cosmpy.aerial.faucet import FaucetApi
from cosmpy.aerial.wallet import LocalWallet
from cosmpy.crypto.address import Address

We then need to choose a data source for the coin price, the update interval, the decimal precision, and the decimal timeout for the oracle value:

COIN_PRICE_URL = (
    "https://api.coingecko.com/api/v3/simple/price?ids=fetch-ai&vs_currencies=usd"
)
UPDATE_INTERVAL_SECONDS = 10
ORACLE_VALUE_DECIMALS = 5
DEFAULT_TIMEOUT = 60.0

We then proceed and define a _parse_commandline() function by first importing the argparse module, which is a standard Python module for parsing command-line arguments:
```
def _parse_commandline():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "contract_path", help="The path to the oracle contract to upload"
    )
    parser.add_argument(
        "contract_address",
        nargs="?",
        type=Address,
        help="The address of the oracle contract if already deployed",
    )
    return parser.parse_args()
```
This first creates an argument parser object. The ArgumentParser class provides a way to specify the arguments your script should accept and automatically generates help messages and error messages. We then use add_argument() to add a positional argument named contract_path. This argument is required and should be a path to the oracle contract that you want to upload. The help argument provides a description of what this argument does. We further add another positional argument named contract_address. This argument is optional (nargs="?" allows it to be omitted), and it should be of type Address. The type argument specifies the type of the argument. In this case, Address is a custom type or class used to represent addresses. The help argument provides a description of what this argument does. At the end, we parse the command-line arguments provided when the script is executed. It returns an object that contains the values of the parsed arguments.
We then need to proceed and define our main() function:
```
def main():
    """Run main."""
    args = _parse_commandline()
 
    wallet = LocalWallet.generate()
 
    ledger = LedgerClient(NetworkConfig.fetchai_stable_testnet())
    faucet_api = FaucetApi(NetworkConfig.fetchai_stable_testnet())
 
    wallet_balance = ledger.query_bank_balance(wallet.address())
 
    while wallet_balance < (10**18):
        print("Providing wealth to wallet...")
        faucet_api.get_wealth(wallet.address())
        wallet_balance = ledger.query_bank_balance(wallet.address())
 
    contract = LedgerContract(args.contract_path, ledger, address=args.contract_address)
 
    if not args.contract_address:
        instantiation_message = {"fee": "100"}
        contract.deploy(instantiation_message, wallet, funds="1atestfet")
 
    print(f"Oracle contract deployed at: {contract.address}")
 
    grant_role_message = {"grant_oracle_role": {"address": wallet}}
    contract.execute(grant_role_message, wallet).wait_to_complete()
 
    print(f"Oracle role granted to address: {wallet}")
 
    while True:
        resp = requests.get(COIN_PRICE_URL, timeout=DEFAULT_TIMEOUT).json()
        price = resp["fetch-ai"]["usd"]
        value = int(price * 10**ORACLE_VALUE_DECIMALS)
 
        update_message = {
            "update_oracle_value": {
                "value": str(value),
                "decimals": str(ORACLE_VALUE_DECIMALS),
            }
        }
        contract.execute(update_message, wallet).wait_to_complete()
 
        print(f"Oracle value updated to: {price} USD")
        print(f"Next update in {UPDATE_INTERVAL_SECONDS} seconds...")
        sleep(UPDATE_INTERVAL_SECONDS)
 
 
if __name__ == "__main__":
    main()
```
This defines our main() function. When we run the script, the code inside main() will be executed. args = _parse_commandline() calls the _parse_commandline() function that we defined earlier. It parses the command-line arguments and returns an object (args) containing the values of the parsed arguments. We then generate a new local wallet, and then create a client for interacting with a blockchain ledger, using LedgerClient() class. We configured it to use the Fetch.ai stable testnet. We then create a client for interacting with a faucet API and query the balance of the wallet's address using the query_bank_balance() method. We also define an initial while loop which continues as long as the wallet_balance is less than 10**18. Inside this first loop: it prints a message indicating that wealth is being provided to the wallet, then it calls the faucet API to get wealth for the wallet, and it updates the wallet_balance by querying the bank balance again.

After this, we create a contract object using LedgerContract(): this takes the path to the oracle contract file, the ledger client, and optionally, the contract address. if not args.contract_address: condition checks if args.contract_address is not provided. If it has not been provided, it means the contract has not been deployed yet. We then set up an instantiation message with a fee of 100. We can then deploy the contract using the provided instantiation message, the wallet, and a specified fund source ("1atestfet" in this case).

The print() function prints the address of the deployed oracle contract. After this, we define a grant_role_message object which sets up a message to grant the oracle role to the address associated with the wallet, and execute the message to grant the oracle role and wait for the transaction to complete. The following print() function prints a message indicating that the oracle role has been granted to the address associated with the wallet.

We can finally define a second while loop which runs indefinitely: it sends a GET request to a URL (COIN_PRICE_URL) to retrieve coin prices, then extracts the price in USD. It then calculates a value based on the price and the specified decimal precision (ORACLE_VALUE_DECIMALS), and sets up an update message with the new oracle value. Lastly, it executes the update message, waits for the transaction to complete, prints the updated oracle value and indicates when the next update will occur.

This script let us interact with a blockchain ledger, deploy a contract, and perform oracle-related tasks such as updating values based on external data.

Save the script.

The overall script should be as follows:

aerial_oracle.py

import argparse
from time import sleep
 
import requests
 
from cosmpy.aerial.client import LedgerClient, NetworkConfig
from cosmpy.aerial.contract import LedgerContract
from cosmpy.aerial.faucet import FaucetApi
from cosmpy.aerial.wallet import LocalWallet
from cosmpy.crypto.address import Address
 
 
COIN_PRICE_URL = (
    "https://api.coingecko.com/api/v3/simple/price?ids=fetch-ai&vs_currencies=usd"
)
UPDATE_INTERVAL_SECONDS = 10
ORACLE_VALUE_DECIMALS = 5
DEFAULT_TIMEOUT = 60.0
 
 
def _parse_commandline():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "contract_path", help="The path to the oracle contract to upload"
    )
    parser.add_argument(
        "contract_address",
        nargs="?",
        type=Address,
        help="The address of the oracle contract if already deployed",
    )
    return parser.parse_args()
 
def main():
    """Run main."""
    args = _parse_commandline()
 
    wallet = LocalWallet.generate()
 
    ledger = LedgerClient(NetworkConfig.fetchai_stable_testnet())
    faucet_api = FaucetApi(NetworkConfig.fetchai_stable_testnet())
 
    wallet_balance = ledger.query_bank_balance(wallet.address())
 
    while wallet_balance < (10**18):
        print("Providing wealth to wallet...")
        faucet_api.get_wealth(wallet.address())
        wallet_balance = ledger.query_bank_balance(wallet.address())
 
    contract = LedgerContract(args.contract_path, ledger, address=args.contract_address)
 
    if not args.contract_address:
        instantiation_message = {"fee": "100"}
        contract.deploy(instantiation_message, wallet, funds="1atestfet")
 
    print(f"Oracle contract deployed at: {contract.address}")
 
    grant_role_message = {"grant_oracle_role": {"address": wallet}}
    contract.execute(grant_role_message, wallet).wait_to_complete()
 
    print(f"Oracle role granted to address: {wallet}")
 
    while True:
        resp = requests.get(COIN_PRICE_URL, timeout=DEFAULT_TIMEOUT).json()
        price = resp["fetch-ai"]["usd"]
        value = int(price * 10**ORACLE_VALUE_DECIMALS)
 
        update_message = {
            "update_oracle_value": {
                "value": str(value),
                "decimals": str(ORACLE_VALUE_DECIMALS),
            }
        }
        contract.execute(update_message, wallet).wait_to_complete()
 
        print(f"Oracle value updated to: {price} USD")
        print(f"Next update in {UPDATE_INTERVAL_SECONDS} seconds...")
        sleep(UPDATE_INTERVAL_SECONDS)
 
if __name__ == "__main__":
    main()

Oracle client

Now, we will write a script that deploys a contract that can request the oracle value in exchange for the required fee.

Let's first create a Python script and name it: touch aerial_oracle_client.py

We start by importing the needed classes and define a REQUEST_INTERVAL_SECONDS variable:

import argparse
from time import sleep
 
from cosmpy.aerial.client import LedgerClient, NetworkConfig
from cosmpy.aerial.contract import LedgerContract
from cosmpy.aerial.faucet import FaucetApi
from cosmpy.aerial.wallet import LocalWallet
from cosmpy.crypto.address import Address
 
REQUEST_INTERVAL_SECONDS = 10

Like before, we proceed and define a _parse_commandline() function:
```
def _parse_commandline():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "contract_path", help="The path to the oracle client contract to upload"
    )
    parser.add_argument(
        "oracle_contract_address",
        type=Address,
        help="The address of the oracle contract",
    )
    parser.add_argument(
        "contract_address",
        nargs="?",
        type=Address,
        help="The address of the oracle client contract if already deployed",
    )
    return parser.parse_args()
```
This _parse_commandline() function is designed to parse command-line arguments. We first create a parser object. This object is used to specify what command-line arguments the program should expect. We then use the add_argument() method to define the arguments that the program expects. In this function, there are three arguments being defined:
- contract_path: this is a required argument. It expects a string representing the path to the oracle client contract to upload.
- oracle_contract_address: this is also a required argument. It expects an Address object representing the address of the oracle contract.
- contract_address: this is an optional argument. It expects an Address object and is used to specify the address of the oracle client contract if it has already been deployed. The nargs="?" indicates that this argument is optional.
The function returns an object containing the parsed values.
We can now define our main() function.
```
def main():
    """Run main."""
    args = _parse_commandline()
 
    wallet = LocalWallet.generate()
 
    ledger = LedgerClient(NetworkConfig.fetchai_stable_testnet())
    faucet_api = FaucetApi(NetworkConfig.fetchai_stable_testnet())
 
    wallet_balance = ledger.query_bank_balance(wallet.address())
 
    while wallet_balance < (10**18):
        print("Providing wealth to wallet...")
        faucet_api.get_wealth(wallet.address())
        wallet_balance = ledger.query_bank_balance(wallet.address())
 
    contract = LedgerContract(args.contract_path, ledger, address=args.contract_address)
 
    if not args.contract_address:
        instantiation_message = {
            "oracle_contract_address": str(args.oracle_contract_address)
        }
        contract.deploy(instantiation_message, wallet)
 
    print(f"Oracle client contract deployed at: {contract.address}")
 
    while True:
        request_message = {"query_oracle_value": {}}
        contract.execute(
            request_message, wallet, funds="100atestfet"
        ).wait_to_complete()
 
        result = contract.query({"oracle_value": {}})
        print(f"Oracle value successfully retrieved: {result}")
 
        sleep(REQUEST_INTERVAL_SECONDS)
 
if __name__ == "__main__":
    main()
```
The first line calls the _parse_commandline() function that we defined earlier. It will parse the command-line arguments and return an object (args) containing the parsed values. We proceed and generate a new local wallet, wallet, and then create a new ledger object for interacting with the blockchain or ledger system, using LedgerClient(). Afterwards, we create a FaucetApi object, faucet_api, which is used for interacting with the faucet service. We use the query_bank_balance() method to query the balance associated with the wallet's address. We then define a while loop which will continue as long as the wallet_balance is less than 10**18. This is to ensure the wallet has a sufficient balance. Afterwards, we use the get_wealth() method to add wealth to the wallet, and then create a new LedgerContract() object which takes the contract_path, the ledger object, and an optional contract_address. if not args.contract_address: checks if args.contract_address is not provided. If it has not been provided, it means that the contract has not been deployed yet. We then create an instantiation_message, which contains the data needed for deploying the contract. contract.deploy() deploys the contract with the provided instantiation_message and the wallet. The code then prints out the address of the deployed contract. Finally, we define a second loop starting with while True: which repeatedly executes the following steps:
- It creates a request message, which is used to query the oracle value.
- It executes the contract function call with the request message, using the wallet for authorization. The funds argument is set to "100atestfet".
- It queries the contract for the oracle value.
- It prints out the retrieved oracle value.
- It finally waits for a specified number of seconds (defined by REQUEST_INTERVAL_SECONDS) before the next iteration. This is likely to prevent overloading the system with requests.
Save the script.

The overall script should be as follows:

aerial_oracle_client.py

import argparse
from time import sleep
 
from cosmpy.aerial.client import LedgerClient, NetworkConfig
from cosmpy.aerial.contract import LedgerContract
from cosmpy.aerial.faucet import FaucetApi
from cosmpy.aerial.wallet import LocalWallet
from cosmpy.crypto.address import Address
 
REQUEST_INTERVAL_SECONDS = 10
 
def _parse_commandline():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "contract_path", help="The path to the oracle client contract to upload"
    )
    parser.add_argument(
        "oracle_contract_address",
        type=Address,
        help="The address of the oracle contract",
    )
    parser.add_argument(
        "contract_address",
        nargs="?",
        type=Address,
        help="The address of the oracle client contract if already deployed",
    )
    return parser.parse_args()
 
def main():
    """Run main."""
    args = _parse_commandline()
 
    wallet = LocalWallet.generate()
 
    ledger = LedgerClient(NetworkConfig.fetchai_stable_testnet())
    faucet_api = FaucetApi(NetworkConfig.fetchai_stable_testnet())
 
    wallet_balance = ledger.query_bank_balance(wallet.address())
 
    while wallet_balance < (10**18):
        print("Providing wealth to wallet...")
        faucet_api.get_wealth(wallet.address())
        wallet_balance = ledger.query_bank_balance(wallet.address())
 
    contract = LedgerContract(args.contract_path, ledger, address=args.contract_address)
 
    if not args.contract_address:
        instantiation_message = {
            "oracle_contract_address": str(args.oracle_contract_address)
        }
        contract.deploy(instantiation_message, wallet)
 
    print(f"Oracle client contract deployed at: {contract.address}")
 
    while True:
        request_message = {"query_oracle_value": {}}
        contract.execute(
            request_message, wallet, funds="100atestfet"
        ).wait_to_complete()
 
        result = contract.query({"oracle_value": {}})
        print(f"Oracle value successfully retrieved: {result}")
 
        sleep(REQUEST_INTERVAL_SECONDS)
 
if __name__ == "__main__":
    main()

Bear in mind that specific data related to the oracle's address and contract need to be provided by hand based on your personalized information!

Was this page helpful?

You can also leave detailed feedbackon Github

Stake optimizer Wallet top-up