Update oracle

.dockerignore

@@ -1,7 +1,8 @@
 /.git
 /.idea
 /.github
-venv
+/venv
+/dist
 .mypy_cache
 Dockerfile
 .dockerignore

.gitignore

@@ -1,5 +1,6 @@
 __pycache__/
 venv
+dist
 .mypy_cache
 .idea
 local.env

main.py

@@ -5,13 +5,13 @@
 
 import aiohttp
 
-from src.distributor.controller import DistributorController
-from src.eth1 import check_oracle_account, get_finalized_block, get_voting_parameters
-from src.ipfs import check_or_create_ipns_keys
-from src.rewards.controller import RewardsController
-from src.rewards.eth2 import get_finality_checkpoints, get_genesis
-from src.settings import LOG_LEVEL, PROCESS_INTERVAL
-from src.validators.controller import ValidatorsController
+from oracle.distributor.controller import DistributorController
+from oracle.eth1 import check_oracle_account, get_finalized_block, get_voting_parameters
+from oracle.ipfs import check_or_create_ipns_keys
+from oracle.rewards.controller import RewardsController
+from oracle.rewards.eth2 import get_finality_checkpoints, get_genesis
+from oracle.settings import LOG_LEVEL, PROCESS_INTERVAL
+from oracle.validators.controller import ValidatorsController
 
 logging.basicConfig(
     format="%(asctime)s %(name)-12s %(levelname)-8s %(message)s",
@@ -42,12 +42,12 @@ def exit_gracefully(self, signum: int, frame: Any) -> None:
 
 
 async def main() -> None:
-    # aiohttp session
-    session = aiohttp.ClientSession()
-
     # check stakewise graphql connection
     await get_finalized_block()
 
+    # aiohttp session
+    session = aiohttp.ClientSession()
+
     # check ETH2 API connection
     await get_finality_checkpoints(session)
 

oracle/clients.py

@@ -0,0 +1,27 @@
+import logging
+from typing import Dict
+
+import backoff
+from gql import Client
+from gql.transport.aiohttp import AIOHTTPTransport
+from graphql import DocumentNode
+
+from oracle.settings import STAKEWISE_SUBGRAPH_URL, UNISWAP_V3_SUBGRAPH_URL
+
+logger = logging.getLogger(__name__)
+
+
+@backoff.on_exception(backoff.expo, Exception, max_time=300)
+async def execute_sw_gql_query(query: DocumentNode, variables: Dict) -> Dict:
+    """Executes GraphQL query."""
+    transport = AIOHTTPTransport(url=STAKEWISE_SUBGRAPH_URL)
+    async with Client(transport=transport, fetch_schema_from_transport=True) as session:
+        return await session.execute(query, variable_values=variables)
+
+
+@backoff.on_exception(backoff.expo, Exception, max_time=300)
+async def execute_uniswap_v3_gql_query(query: DocumentNode, variables: Dict) -> Dict:
+    """Executes GraphQL query."""
+    transport = AIOHTTPTransport(url=UNISWAP_V3_SUBGRAPH_URL)
+    async with Client(transport=transport, fetch_schema_from_transport=True) as session:
+        return await session.execute(query, variable_values=variables)

src/distributor/controller.py → oracle/distributor/controller.py

@@ -3,8 +3,11 @@
 
 from web3 import Web3
 
-from src.ipfs import submit_ipns_vote
-from src.settings import REWARD_ETH_TOKEN_CONTRACT_ADDRESS, SWISE_TOKEN_CONTRACT_ADDRESS
+from oracle.ipfs import submit_ipns_vote
+from oracle.settings import (
+    REWARD_ETH_TOKEN_CONTRACT_ADDRESS,
+    SWISE_TOKEN_CONTRACT_ADDRESS,
+)
 
 from .eth1 import (
     get_active_tokens_allocations,

src/distributor/eth1.py → oracle/distributor/eth1.py

@@ -6,14 +6,14 @@
 from web3 import Web3
 from web3.types import BlockNumber, Wei
 
-from src.clients import execute_graphql_query, sw_gql_client
-from src.graphql_queries import (
+from oracle.clients import execute_sw_gql_query
+from oracle.graphql_queries import (
     ACTIVE_TOKEN_DISTRIBUTIONS_QUERY,
     DISABLED_STAKER_ACCOUNTS_QUERY,
     DISTRIBUTOR_CLAIMED_ACCOUNTS_QUERY,
     SWISE_HOLDERS_QUERY,
 )
-from src.settings import (
+from oracle.settings import (
     REWARD_ETH_TOKEN_CONTRACT_ADDRESS,
     STAKED_ETH_TOKEN_CONTRACT_ADDRESS,
     SWISE_TOKEN_CONTRACT_ADDRESS,
@@ -36,8 +36,7 @@ async def get_active_tokens_allocations(
 ) -> TokenAllocations:
     """Fetches active token allocations."""
     last_id = ""
-    result: Dict = await execute_graphql_query(
-        client=sw_gql_client,
+    result: Dict = await execute_sw_gql_query(
         query=ACTIVE_TOKEN_DISTRIBUTIONS_QUERY,
         variables=dict(from_block=from_block, to_block=to_block, last_id=last_id),
     )
@@ -50,8 +49,7 @@ async def get_active_tokens_allocations(
         if not last_id:
             break
 
-        result: Dict = await execute_graphql_query(
-            client=sw_gql_client,
+        result: Dict = await execute_sw_gql_query(
             query=ACTIVE_TOKEN_DISTRIBUTIONS_QUERY,
             variables=dict(from_block=from_block, to_block=to_block, last_id=last_id),
         )
@@ -89,8 +87,7 @@ async def get_disabled_stakers_reward_eth_distributions(
         return []
 
     last_id = ""
-    result: Dict = await execute_graphql_query(
-        client=sw_gql_client,
+    result: Dict = await execute_sw_gql_query(
         query=DISABLED_STAKER_ACCOUNTS_QUERY,
         variables=dict(block_number=to_block, last_id=last_id),
     )
@@ -103,8 +100,7 @@ async def get_disabled_stakers_reward_eth_distributions(
         if not last_id:
             break
 
-        result: Dict = await execute_graphql_query(
-            client=sw_gql_client,
+        result: Dict = await execute_sw_gql_query(
             query=DISABLED_STAKER_ACCOUNTS_QUERY,
             variables=dict(block_number=to_block, last_id=last_id),
         )
@@ -128,7 +124,7 @@ async def get_disabled_stakers_reward_eth_distributions(
             continue
 
         principals[staker_address] = staker_principal
-        distributor_principal += staker_principal
+        distributor_principal += Wei(staker_principal)
 
     if distributor_principal <= 0:
         return []
@@ -154,7 +150,7 @@ async def get_disabled_stakers_reward_eth_distributions(
             reward=reward,
         )
         distributions.append(distribution)
-        distributed += reward
+        distributed += Wei(reward)
 
     return distributions
 
@@ -163,8 +159,7 @@ async def get_disabled_stakers_reward_eth_distributions(
 async def get_distributor_claimed_accounts(merkle_root: HexStr) -> ClaimedAccounts:
     """Fetches addresses that have claimed their tokens from the `MerkleDistributor` contract."""
     last_id = ""
-    result: Dict = await execute_graphql_query(
-        client=sw_gql_client,
+    result: Dict = await execute_sw_gql_query(
         query=DISTRIBUTOR_CLAIMED_ACCOUNTS_QUERY,
         variables=dict(merkle_root=merkle_root, last_id=last_id),
     )
@@ -177,8 +172,7 @@ async def get_distributor_claimed_accounts(merkle_root: HexStr) -> ClaimedAccoun
         if not last_id:
             break
 
-        result: Dict = await execute_graphql_query(
-            client=sw_gql_client,
+        result: Dict = await execute_sw_gql_query(
             query=DISTRIBUTOR_CLAIMED_ACCOUNTS_QUERY,
             variables=dict(merkle_root=merkle_root, last_id=last_id),
         )
@@ -194,8 +188,7 @@ async def get_swise_holders(
 ) -> Balances:
     """Fetches SWISE holding points."""
     last_id = ""
-    result: Dict = await execute_graphql_query(
-        client=sw_gql_client,
+    result: Dict = await execute_sw_gql_query(
         query=SWISE_HOLDERS_QUERY,
         variables=dict(block_number=to_block, last_id=last_id),
     )
@@ -208,8 +201,7 @@ async def get_swise_holders(
         if not last_id:
             break
 
-        result: Dict = await execute_graphql_query(
-            client=sw_gql_client,
+        result: Dict = await execute_sw_gql_query(
             query=SWISE_HOLDERS_QUERY,
             variables=dict(block_number=to_block, last_id=last_id),
         )
@@ -241,18 +233,25 @@ async def get_swise_holders(
         total_points += account_holding_points
 
     # process unclaimed SWISE
-    for account, rewards in unclaimed_rewards.items():
-        balance = int(rewards.get(SWISE_TOKEN_CONTRACT_ADDRESS, "0"))
-        if balance <= 0:
+    for account in unclaimed_rewards:
+        origins = unclaimed_rewards.get(account, {}).get(
+            SWISE_TOKEN_CONTRACT_ADDRESS, {}
+        )
+        if not origins:
             continue
 
-        account_holding_points = balance * (to_block - from_block)
-        if account_holding_points <= 0:
-            continue
+        for origin, balance in origins.items():
+            balance = int(balance)
+            if balance <= 0:
+                continue
 
-        holding_points[account] = (
-            holding_points.setdefault(account, 0) + account_holding_points
-        )
-        total_points += account_holding_points
+            account_holding_points = balance * (to_block - from_block)
+            if account_holding_points <= 0:
+                continue
+
+            holding_points[account] = (
+                holding_points.setdefault(account, 0) + account_holding_points
+            )
+            total_points += account_holding_points
 
     return Balances(total_supply=total_points, balances=holding_points)

src/distributor/ipfs.py → oracle/distributor/ipfs.py

@@ -1,10 +1,9 @@
 import logging
-from typing import Dict
 
 import backoff
 import ipfshttpclient
 
-from src.settings import IPFS_ENDPOINT
+from oracle.settings import IPFS_ENDPOINT
 
 from .types import ClaimedAccounts, Claims, Rewards
 
@@ -19,27 +18,31 @@ def get_unclaimed_balances(
     merkle_proofs = merkle_proofs.replace("ipfs://", "").replace("/ipfs/", "")
 
     with ipfshttpclient.connect(IPFS_ENDPOINT) as client:
-        prev_claims: Dict = client.get_json(merkle_proofs)
+        prev_claims: Claims = client.get_json(merkle_proofs)
 
-    unclaimed_rewards: Rewards = Rewards({})
+    unclaimed_rewards: Rewards = {}
     for account, claim in prev_claims.items():
         if account in claimed_accounts:
             continue
 
-        # TODO: remove after first v2 merkle root update
-        key = "reward_tokens" if "reward_tokens" in claim else "tokens"
-        for token, reward in zip(claim[key], claim["values"]):
-            prev_unclaimed = unclaimed_rewards.setdefault(account, {}).setdefault(
-                token, "0"
-            )
-            unclaimed_rewards[account][token] = str(int(prev_unclaimed) + int(reward))
+        for i, reward_token in enumerate(claim["reward_tokens"]):
+            for origin, value in zip(claim["origins"][i], claim["values"][i]):
+                prev_unclaimed = (
+                    unclaimed_rewards.setdefault(account, {})
+                    .setdefault(reward_token, {})
+                    .setdefault(origin, "0")
+                )
+                unclaimed_rewards[account][reward_token][origin] = str(
+                    int(prev_unclaimed) + int(value)
+                )
 
     return unclaimed_rewards
 
 
 @backoff.on_exception(backoff.expo, Exception, max_time=900)
 def upload_claims(claims: Claims) -> str:
     """Submits claims to the IPFS and pins the file."""
+    # TODO: split claims into files up to 1000 entries
     with ipfshttpclient.connect(IPFS_ENDPOINT) as client:
         ipfs_id = client.add_json(claims)
         client.pin.add(ipfs_id)

src/distributor/merkle_tree.py → oracle/distributor/merkle_tree.py

@@ -101,7 +101,7 @@ def get_merkle_node(
     index: int,
     tokens: List[ChecksumAddress],
     account: ChecksumAddress,
-    values: List[str],
+    values: List[int],
 ) -> bytes:
     """Generates node for merkle tree."""
     encoded_data: bytes = w3.codec.encode_abi(
@@ -117,16 +117,31 @@ def calculate_merkle_root(rewards: Rewards) -> Tuple[HexStr, Claims]:
     accounts: List[ChecksumAddress] = sorted(rewards.keys())
     claims: Claims = OrderedDict()
     for i, account in enumerate(accounts):
-        tokens: List[ChecksumAddress] = sorted(rewards[account].keys())
-        values: List[str] = [rewards[account][token] for token in tokens]
-        claim: Claim = OrderedDict(index=i, tokens=tokens, values=values)
+        reward_tokens: List[ChecksumAddress] = sorted(rewards[account].keys())
+        claim: Claim = OrderedDict(index=i, reward_tokens=reward_tokens)
+
+        reward_token_amounts: Dict[ChecksumAddress, int] = {}
+        for reward_token in reward_tokens:
+            origins: List[ChecksumAddress] = sorted(
+                rewards[account][reward_token].keys()
+            )
+            values: List[str] = []
+            for origin in origins:
+                value: str = rewards[account][reward_token][origin]
+                values.append(value)
+                prev_value = reward_token_amounts.setdefault(reward_token, 0)
+                reward_token_amounts[reward_token] = prev_value + int(value)
+
+            claim.setdefault("origins", []).append(origins)
+            claim.setdefault("values", []).append(values)
+
         claims[account] = claim
 
-        merkle_element = get_merkle_node(
+        merkle_element: bytes = get_merkle_node(
             index=i,
             account=account,
-            tokens=tokens,
-            values=values,
+            tokens=reward_tokens,
+            values=[reward_token_amounts[token] for token in reward_tokens],
         )
         merkle_elements.append(merkle_element)