feat: Checkpoint setter (#145)

* feat(checkpoint_setter): Checkpoint setter added

* feat(checkpoint_setter): Price pusher quick update

* feat(checkpoint_setter): Lock updates

* feat(checkpoint_setter): Checkpoints

* feat(checkpoint_setter): Checkpoint almost gud

* feat(checkpoint_setter): Added TODO

* feat(checkpoint_setter): Added tests for checkpoints

* feat(checkpoint_setter): Documentation

* feat(checkpoint_setter): Removed unused logs

* feat(checkpoint_setter): README update

* Update README.md

* feat(checkpoint_setter): Updated checkpoint logs

* feat(checkpoint_setter): Comments

* feat(checkpoint_setter): Fixes from review

* feat(checkpoint_setter): Fixes from review

* feat(checkpoint_setter): Fixed CI branch main

* feat(checkpoint_setter): Fixed our tests

* feat(checkpoint_setter): Triggering CI

* feat(checkpoint_setter): Trigger CI

* feat(checkpoint_setter): Trigger CI

* feat(checkpoint_setter): Trigger CI

* feat(checkpoint_setter): Updated coverage errors

* feat(checkpoint_setter):

.github/workflows/docker-build.yml

@@ -13,7 +13,7 @@
     build_containers:
       strategy:
         matrix:
-          package: [price-pusher, vrf-listener]
+          package: [price-pusher, vrf-listener, checkpointer]
       runs-on: ubuntu-latest
       steps:
         - uses: actions/checkout@v4
@@ -44,7 +44,7 @@
   
     bump_version:
       needs: [build_containers]
-      if: github.event_name == 'push' && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/master')
+      if: github.event_name == 'push' && (github.ref == 'refs/heads/master' || github.ref == 'refs/heads/master')
       runs-on: ubuntu-latest
       steps:
         - uses: actions/checkout@v3

.github/workflows/linters.yml

@@ -14,7 +14,7 @@
       runs-on: ubuntu-latest
       strategy:
         matrix:
-          package: [pragma-utils, pragma-sdk, price-pusher, vrf-listener]
+          package: [pragma-utils, pragma-sdk, price-pusher, vrf-listener, checkpointer]
         fail-fast: false
       steps:
         - uses: actions/checkout@v3

.github/workflows/pull-request.yml

@@ -4,7 +4,7 @@
   on:
     workflow_dispatch:
     pull_request:
-      branches: [main]
+      branches: [master]
 
   concurrency:
     group:

.github/workflows/tests.yml

@@ -14,7 +14,7 @@
       runs-on: ubuntu-latest
       strategy:
         matrix:
-          package: [pragma-utils, pragma-sdk, price-pusher, vrf-listener]
+          package: [pragma-utils, pragma-sdk, price-pusher, vrf-listener, checkpointer]
         fail-fast: false
       steps:
         - uses: actions/checkout@v3
@@ -79,9 +79,9 @@
               poetry run poe test_client
               poetry run poe test_vrf
               poetry run poe test_fetchers
-              poetry run poe test_all_unit
+              poetry run poe test_unit
             else
-              poetry run poe test_all
+              poetry run poe test
             fi
         - name: Generate coverage in XML
           run: |

README.md

@@ -24,7 +24,6 @@ For more information, see the [project's repository](https://github.com/Astraly-
 - <a href="./vrf_listener">VRF Listener</a>
 - <a href="./pragma-utils">Pragma utils</a>
 
-
 ## Contributing
 
 See the [CONTRIBUTING](./CONTRIBUTING.md) guide.

checkpointer/README.md

@@ -0,0 +1,72 @@
+# Checkpointer
+
+Service used to automatically create checkpoints periodically for a set of pairs.
+
+### Usage
+
+The service is ran through the CLI.
+
+To specify for which assets you want to set checkpoints, you will need to provide a yaml configuration file formatted as follow:
+
+```yaml
+# config/config.example.yaml
+spot:
+  - pair: BTC/USD
+  - pair: ETH/USD
+
+future:
+  - pair: BTC/USD
+    expiry: 102425525524
+    # You can have the same pair multiple time for different expiries
+  - pair: BTC/USD
+    expiry: 0
+  - pair: ETH/USD
+    expiry: 234204249042
+  - pair: SOL/USD
+    expiry: 0
+```
+
+For spot pairs, we simply list them, but for future ones, you need to add for which expiry timestamp you wish to create new checkpoints.
+
+To have more information on how to run the CLI, you can use the `--help` command:
+
+```bash
+.venv ❯ python checkpointer/main.py --help
+
+Usage: main.py [OPTIONS]
+
+  Checkpoints setter entry point.
+
+Options:
+  -c, --config-file PATH          Path to YAML configuration file.  [required]
+  
+  --log-level [DEBUG|INFO|WARNING|ERROR|CRITICAL]
+                                  Logging level.
+  
+  -n, --network [sepolia|mainnet]
+                                  On which networks the checkpoints will be
+                                  set. Defaults to SEPOLIA.  [required]
+  
+  --rpc-url TEXT                  RPC url used by the onchain client.
+  
+  --oracle-address TEXT           Address of the Pragma Oracle  [required]
+  
+  --admin-address TEXT            Address of the Admin contract for the
+                                  Oracle.  [required]
+  
+  -p, --private-key TEXT          Secret key of the signer. Format:
+                                  aws:secret_name, plain:secret_key, or
+                                  env:ENV_VAR_NAME  [required]
+  
+  -t, --set-checkpoint-interval
+                                  Delay in minutes between each new
+                                  checkpoints. Defaults to 60 minutes.  [x>=0]
+  
+  -help                          Show this message and exit.
+```
+
+For example:
+
+```sh
+poetry run checkpointer -c config/config.example.yaml --oracle-address $PRAGMA_ORACLE_ADDRESS --admin-address $PRAGMA_ADMIN_ACCOUNT -p plain:$MY_PRIVATE_KEY
+```

checkpointer/VERSION

@@ -0,0 +1 @@
+0.0.1

checkpointer/checkpointer/configs/pairs_config.py

@@ -0,0 +1,72 @@
+import yaml
+from typing import List, Optional, Tuple, Union
+from datetime import datetime
+from zoneinfo import ZoneInfo
+from dataclasses import dataclass, field
+from pydantic import BaseModel, ConfigDict, field_validator
+
+from pragma_sdk.common.types.pair import Pair
+from pragma_sdk.common.utils import str_to_felt
+
+
+@dataclass(frozen=True)
+class SpotPairConfig:
+    pair: Pair
+
+
+@dataclass(frozen=True)
+class FuturePairConfig:
+    pair: Pair
+    expiry: datetime = field(default_factory=lambda: datetime.fromtimestamp(0, ZoneInfo("UTC")))
+
+
+class PairsConfig(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    spot: List[SpotPairConfig] = field(default_factory=list)
+    future: List[FuturePairConfig] = field(default_factory=list)
+
+    @field_validator("spot", "future", mode="before")
+    def validate_pairs(
+        cls, value: Optional[List[dict]], info
+    ) -> List[SpotPairConfig | FuturePairConfig]:
+        if not value:
+            return []
+
+        pairs: List[Union[SpotPairConfig, FuturePairConfig]] = []
+        for raw_pair in value:
+            pair_str = raw_pair["pair"].replace(" ", "").upper()
+            base, quote = pair_str.split("/")
+            if len(base) == 0 or len(quote) == 0:
+                raise ValueError("Pair should be formatted as 'BASE/QUOTE'")
+
+            pair = Pair.from_tickers(base, quote)
+            if pair is None:
+                raise ValueError(f"⛔ Could not create Pair object for {base}/{quote}")
+
+            if info.field_name == "future":
+                expiry = datetime.fromtimestamp(raw_pair.get("expiry", 0), tz=ZoneInfo("UTC"))
+                pairs.append(FuturePairConfig(pair=pair, expiry=expiry))
+            else:
+                pairs.append(SpotPairConfig(pair=pair))
+
+        return list(set(pairs))
+
+    @classmethod
+    def from_yaml(cls, path: str) -> "PairsConfig":
+        with open(path, "r") as file:
+            pairs_config = yaml.safe_load(file)
+        config = cls(**pairs_config)
+
+        if not config.spot and not config.future:
+            raise ValueError(
+                "⛔ No pair found: you need to specify at least one spot/future pair in the configuration file."
+            )
+        return config
+
+    def get_spot_ids(self) -> List[int]:
+        return [str_to_felt(str(pair_config.pair)) for pair_config in self.spot]
+
+    def get_future_ids_and_expiries(self) -> Tuple[List[int], List[int]]:
+        pair_ids = [str_to_felt(str(pair_config.pair)) for pair_config in self.future]
+        expiries = [int(pair_config.expiry.timestamp()) for pair_config in self.future]
+        return pair_ids, expiries

checkpointer/checkpointer/main.py

@@ -0,0 +1,178 @@
+import asyncio
+import click
+import logging
+
+from pydantic import HttpUrl
+from typing import Optional, Literal
+
+from pragma_utils.logger import setup_logging
+from pragma_utils.cli import load_private_key_from_cli_arg
+
+from pragma_sdk.common.types.types import AggregationMode, DataTypes
+from pragma_sdk.onchain.client import PragmaOnChainClient
+from pragma_sdk.onchain.types import ContractAddresses
+
+from checkpointer.configs.pairs_config import PairsConfig
+
+logger = logging.getLogger(__name__)
+
+
+async def main(
+    pairs_config: PairsConfig,
+    network: Literal["mainnet", "sepolia"],
+    oracle_address: str,
+    admin_address: str,
+    private_key: str,
+    set_checkpoint_interval: int,
+    rpc_url: Optional[HttpUrl] = None,
+) -> None:
+    pragma_client = PragmaOnChainClient(
+        chain_name=network,
+        network=network if rpc_url is None else rpc_url,
+        account_contract_address=int(admin_address, 16),
+        account_private_key=int(private_key, 16),
+        contract_addresses_config=ContractAddresses(
+            publisher_registry_address=0x0,
+            oracle_proxy_addresss=int(oracle_address, 16),
+            summary_stats_address=0x0,
+        ),
+    )
+    _log_handled_pairs(pairs_config, set_checkpoint_interval)
+    logger.info("🧩 Starting Checkpointer...")
+    try:
+        while True:
+            tasks = []
+            if pairs_config.spot:
+                tasks.append(_set_checkpoints(pragma_client, pairs_config, DataTypes.SPOT))
+            if pairs_config.future:
+                tasks.append(_set_checkpoints(pragma_client, pairs_config, DataTypes.FUTURE))
+            await asyncio.gather(*tasks)
+            await asyncio.sleep(set_checkpoint_interval)
+    except asyncio.CancelledError:
+        logger.info("... Checkpointer service stopped! 👋")
+        return
+
+
+async def _set_checkpoints(
+    client: PragmaOnChainClient,
+    pairs_config: PairsConfig,
+    pairs_type: DataTypes,
+) -> None:
+    try:
+        match pairs_type:
+            case DataTypes.SPOT:
+                pair_ids = pairs_config.get_spot_ids()
+                tx = await client.set_checkpoints(
+                    pair_ids=pair_ids,
+                    aggregation_mode=AggregationMode.MEDIAN,
+                )
+            case DataTypes.FUTURE:
+                pair_ids, expiries = pairs_config.get_future_ids_and_expiries()
+                tx = await client.set_future_checkpoints(
+                    pair_ids=pair_ids,
+                    expiry_timestamps=expiries,
+                    aggregation_mode=AggregationMode.MEDIAN,
+                )
+        await tx.wait_for_acceptance()
+        logger.info(f"✅ Successfully set {pairs_type} checkpoints")
+    except Exception as e:
+        logger.error(f"⛔ Error while setting {pairs_type} checkpoint: {e}")
+
+
+def _log_handled_pairs(pairs_config: PairsConfig, set_checkpoint_interval: int) -> None:
+    log_message = (
+        "👇 New checkpoints will automatically be set every "
+        f"{set_checkpoint_interval} seconds for those pairs:"
+    )
+    for pair_type, pairs in pairs_config:
+        if pairs:
+            log_message += f"\n{pair_type.upper()}: {pairs}"
+    logger.info(log_message)
+
+
+@click.command()
+@click.option(
+    "-c",
+    "--config-file",
+    type=click.Path(exists=True),
+    required=True,
+    help="Path to YAML configuration file.",
+)
+@click.option(
+    "--log-level",
+    default="INFO",
+    type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False),
+    help="Logging level.",
+)
+@click.option(
+    "-n",
+    "--network",
+    required=True,
+    default="sepolia",
+    type=click.Choice(["sepolia", "mainnet"], case_sensitive=False),
+    help="On which networks the checkpoints will be set. Defaults to SEPOLIA.",
+)
+@click.option(
+    "--rpc-url",
+    type=click.STRING,
+    required=False,
+    help="RPC url used by the onchain client.",
+)
+@click.option(
+    "--oracle-address",
+    type=click.STRING,
+    required=True,
+    help="Address of the Pragma Oracle",
+)
+@click.option(
+    "--admin-address",
+    type=click.STRING,
+    required=True,
+    help="Address of the Admin contract for the Oracle.",
+)
+@click.option(
+    "-p",
+    "--private-key",
+    type=click.STRING,
+    required=True,
+    help="Secret key of the signer. Format: aws:secret_name, plain:secret_key, or env:ENV_VAR_NAME",
+)
+@click.option(
+    "-t",
+    "--set-checkpoint-interval",
+    type=click.IntRange(min=0),
+    required=False,
+    default=3600,
+    help="Delay in seconds between each new checkpoints. Defaults to 1 hour (3600s).",
+)
+def cli_entrypoint(
+    config_file: str,
+    log_level: str,
+    network: Literal["mainnet", "sepolia"],
+    rpc_url: Optional[HttpUrl],
+    oracle_address: str,
+    admin_address: str,
+    private_key: str,
+    set_checkpoint_interval: int,
+) -> None:
+    """
+    Checkpointer entry point.
+    """
+    setup_logging(logger, log_level)
+    private_key = load_private_key_from_cli_arg(private_key)
+    pairs_config = PairsConfig.from_yaml(config_file)
+    asyncio.run(
+        main(
+            pairs_config=pairs_config,
+            network=network,
+            rpc_url=rpc_url,
+            oracle_address=oracle_address,
+            admin_address=admin_address,
+            private_key=private_key,
+            set_checkpoint_interval=set_checkpoint_interval,
+        )
+    )
+
+
+if __name__ == "__main__":
+    cli_entrypoint()