Reduce event filtering window in case of failure

.gitleaksignore

@@ -12,3 +12,4 @@ ce811dab0e95cdd63ea00e1ed8207801f464d773:packages/valory/contracts/service_staki
 61e62bd62dcaae666f1bf25981d068e360d32322:packages/valory/skills/staking_abci/skill.yaml:generic-api-key:78
 b520b0c67a9ae55ba501e1e9a99abfc0738e8d49:packages/valory/skills/staking_abci/skill.yaml:generic-api-key:78
 61e62bd62dcaae666f1bf25981d068e360d32322:packages/valory/skills/staking_abci/skill.yaml:generic-api-key:78
+4063acab1bf2654eeaa6fca86bfbc3f6ab57e399:packages/valory/services/trader/service.yaml:generic-api-key:9

packages/packages.json

@@ -1,13 +1,13 @@
 {
     "dev": {
         "skill/valory/market_manager_abci/0.1.0": "bafybeibrjre5yka3sbvb6lsefgc5yutovb6h4ffzrpecplfkjwtl5i36te",
-        "skill/valory/decision_maker_abci/0.1.0": "bafybeiejunyhbfzk3tqn22xmyvpyn75xyjsbeg7grjrywhecak3phbwwem",
-        "skill/valory/trader_abci/0.1.0": "bafybeihjre2rppuyuxesevdmpgjvnvsvsgsevfzqi3f7l6yi6thq7xvl5a",
+        "skill/valory/decision_maker_abci/0.1.0": "bafybeighimm45dlkagx5uig2i4wa7ikmiajlcyqkpms64amm5osxrnzroi",
+        "skill/valory/trader_abci/0.1.0": "bafybeihss6lmdt3khxktp2jzuuxsargspa3wb46jpdty6nqrwhxrq5ktde",
         "contract/valory/market_maker/0.1.0": "bafybeif4mm2s3gxtvp227yypkcnna5ftec7vajcftvtbdmqddh7nprah5m",
-        "agent/valory/trader/0.1.0": "bafybeiccr6adm3ijuyp7so7js4qqnvkdrljylbjc4svc65xt75la6jykpm",
-        "service/valory/trader/0.1.0": "bafybeifeil57odcx6c22zmmfz4krqwci5gc6tttagiylwthfi34fwdahbq",
+        "agent/valory/trader/0.1.0": "bafybeigt2t6okczlbwsx7vwydq7c4cqniettxmmtidvdibvjkngadpk2wa",
+        "service/valory/trader/0.1.0": "bafybeiee5x6fx2rhzizmpcnuzwcgia2iu4qwzzfbegiwqh4okda6edlwg4",
         "contract/valory/erc20/0.1.0": "bafybeieqj7dea4tcv6z2yqkgtmhd23vbiycsr5trhwxdvlbwulpl6vhmam",
-        "skill/valory/tx_settlement_multiplexer_abci/0.1.0": "bafybeibtbginmn7tq2tluo7jyxpnlgqfwtpia5a2delxhjmsdejazvpsku",
+        "skill/valory/tx_settlement_multiplexer_abci/0.1.0": "bafybeidaamjhhyayxprfi6pgoijmjhbybnlmhjugegm2pb4kj6ax623bna",
         "contract/valory/mech/0.1.0": "bafybeihvc6btuk3nvernzcx4qpezvuhiw2wwnagqj5nkeljvszghv7mq64",
         "contract/valory/realitio/0.1.0": "bafybeiamgkwwqhray4fs2hlipwxkq7mosmi7ev7jut4vneetaaycc4ruji",
         "contract/valory/realitio_proxy/0.1.0": "bafybeidx37xzjjmapwacedgzhum6grfzhp5vhouz4zu3pvpgdy5pgb2fr4",

packages/valory/agents/trader/aea-config.yaml

@@ -43,10 +43,10 @@ skills:
 - valory/reset_pause_abci:0.1.0:bafybeicpxn2khtaesuf4cq6ypwdmdmonlqroj2q2i6cxvpizc2y4cw66pe
 - valory/termination_abci:0.1.0:bafybeieqm46zuccaagnko3qlw6p3nvoohdrfgvpmw467r5lyil2dqrzjsy
 - valory/transaction_settlement_abci:0.1.0:bafybeia6cdxdlqrcwk2maw25fo7dafzd2p3rs7syropvufophk2pitzbwy
-- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeibtbginmn7tq2tluo7jyxpnlgqfwtpia5a2delxhjmsdejazvpsku
+- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeidaamjhhyayxprfi6pgoijmjhbybnlmhjugegm2pb4kj6ax623bna
 - valory/market_manager_abci:0.1.0:bafybeibrjre5yka3sbvb6lsefgc5yutovb6h4ffzrpecplfkjwtl5i36te
-- valory/decision_maker_abci:0.1.0:bafybeiejunyhbfzk3tqn22xmyvpyn75xyjsbeg7grjrywhecak3phbwwem
-- valory/trader_abci:0.1.0:bafybeihjre2rppuyuxesevdmpgjvnvsvsgsevfzqi3f7l6yi6thq7xvl5a
+- valory/decision_maker_abci:0.1.0:bafybeighimm45dlkagx5uig2i4wa7ikmiajlcyqkpms64amm5osxrnzroi
+- valory/trader_abci:0.1.0:bafybeihss6lmdt3khxktp2jzuuxsargspa3wb46jpdty6nqrwhxrq5ktde
 - valory/staking_abci:0.1.0:bafybeifdxywwwtwgfmtufkvvjvepil6qwetbldftf5eab5v2r3b3zh574q
 default_ledger: ethereum
 required_ledgers:
@@ -191,6 +191,8 @@ models:
       realitio_proxy_address: ${str:0xAB16D643bA051C11962DA645f74632d3130c81E2}
       realitio_address: ${str:0x79e32aE03fb27B07C89c0c568F80287C01ca2E57}
       event_filtering_batch_size: ${int:5000}
+      reduce_factor: ${float:0.25}
+      max_filtering_retries: ${int:6}
       redeeming_batch_size: ${int:5}
       slippage: ${float:0.01}
       policy_epsilon: ${float:0.1}

packages/valory/services/trader/service.yaml

@@ -7,7 +7,7 @@ license: Apache-2.0
 fingerprint:
   README.md: bafybeigtuothskwyvrhfosps2bu6suauycolj67dpuxqvnicdrdu7yhtvq
 fingerprint_ignore_patterns: []
-agent: valory/trader:0.1.0:bafybeiccr6adm3ijuyp7so7js4qqnvkdrljylbjc4svc65xt75la6jykpm
+agent: valory/trader:0.1.0:bafybeigt2t6okczlbwsx7vwydq7c4cqniettxmmtidvdibvjkngadpk2wa
 number_of_agents: 4
 deployment: {}
 ---
@@ -105,6 +105,8 @@ type: skill
         realitio_proxy_address: ${REALITIO_PROXY_ADDRESS:str:0xAB16D643bA051C11962DA645f74632d3130c81E2}
         realitio_address: ${REALITIO_ADDRESS:str:0x79e32aE03fb27B07C89c0c568F80287C01ca2E57}
         event_filtering_batch_size: ${EVENT_FILTERING_BATCH_SIZE:int:5000}
+        reduce_factor: ${REDUCE_FACTOR:float:0.25}
+        max_filtering_retries: ${MAX_FILTERING_RETRIES:int:6}
         redeeming_batch_size: ${REDEEMING_BATCH_SIZE:int:1}
         slippage: ${SLIPPAGE:float:0.01}
         policy_epsilon: ${POLICY_EPSILON:float:0.1}
@@ -179,6 +181,8 @@ type: skill
         realitio_proxy_address: ${REALITIO_PROXY_ADDRESS:str:0xAB16D643bA051C11962DA645f74632d3130c81E2}
         realitio_address: ${REALITIO_ADDRESS:str:0x79e32aE03fb27B07C89c0c568F80287C01ca2E57}
         event_filtering_batch_size: ${EVENT_FILTERING_BATCH_SIZE:int:5000}
+        reduce_factor: ${REDUCE_FACTOR:float:0.25}
+        max_filtering_retries: ${MAX_FILTERING_RETRIES:int:6}
         redeeming_batch_size: ${REDEEMING_BATCH_SIZE:int:1}
         slippage: ${SLIPPAGE:float:0.01}
         policy_epsilon: ${POLICY_EPSILON:float:0.1}
@@ -251,6 +255,8 @@ type: skill
         realitio_proxy_address: ${REALITIO_PROXY_ADDRESS:str:0xAB16D643bA051C11962DA645f74632d3130c81E2}
         realitio_address: ${REALITIO_ADDRESS:str:0x79e32aE03fb27B07C89c0c568F80287C01ca2E57}
         event_filtering_batch_size: ${EVENT_FILTERING_BATCH_SIZE:int:5000}
+        reduce_factor: ${REDUCE_FACTOR:float:0.25}
+        max_filtering_retries: ${MAX_FILTERING_RETRIES:int:6}
         redeeming_batch_size: ${REDEEMING_BATCH_SIZE:int:1}
         slippage: ${SLIPPAGE:float:0.01}
         policy_epsilon: ${POLICY_EPSILON:float:0.1}
@@ -323,6 +329,8 @@ type: skill
         realitio_proxy_address: ${REALITIO_PROXY_ADDRESS:str:0xAB16D643bA051C11962DA645f74632d3130c81E2}
         realitio_address: ${REALITIO_ADDRESS:str:0x79e32aE03fb27B07C89c0c568F80287C01ca2E57}
         event_filtering_batch_size: ${EVENT_FILTERING_BATCH_SIZE:int:5000}
+        reduce_factor: ${REDUCE_FACTOR:float:0.25}
+        max_filtering_retries: ${MAX_FILTERING_RETRIES:int:6}
         redeeming_batch_size: ${REDEEMING_BATCH_SIZE:int:1}
         slippage: ${SLIPPAGE:float:0.01}
         policy_epsilon: ${POLICY_EPSILON:float:0.1}

packages/valory/skills/decision_maker_abci/behaviours/reedem.py

@@ -411,12 +411,10 @@ def _check_already_redeemed(self) -> WaitableConditionType:
             self.redeeming_progress.check_to_block = self.latest_block_number
             self.redeeming_progress.check_started = True
 
+        n_retries = 0
+        from_block = self.redeeming_progress.check_from_block
         batch_size = self.params.event_filtering_batch_size
-        for from_block in range(
-            self.redeeming_progress.check_from_block,
-            self.redeeming_progress.check_to_block,
-            batch_size,
-        ):
+        while from_block < self.redeeming_progress.check_to_block:
             max_to_block = from_block + batch_size
             to_block = min(max_to_block, self.redeeming_progress.check_to_block)
             result = yield from self._conditional_tokens_interact(
@@ -428,16 +426,33 @@ def _check_already_redeemed(self) -> WaitableConditionType:
                 to_block=to_block,
                 **kwargs,
             )
-            if not result:
+
+            if not result and n_retries == self.params.max_filtering_retries:
+                err = "Skipping the redeeming round as the RPC is misbehaving."
+                self.context.logger.error(err)
                 return False
+
+            if not result:
+                n_retries += 1
+                keep_fraction = (1 - self.params.reduce_factor) ** n_retries
+                step = int(batch_size * keep_fraction)
+                msg = f"Repeating this call with a decreased batch size of {step}."
+                self.context.logger.warning(msg)
+                from_block += step
+                continue
+
             self.redeeming_progress.payouts.update(self.payouts_batch)
             self.redeeming_progress.check_from_block = to_block
+            from_block += batch_size
 
         return True
 
-    def _clean_redeem_info(self) -> Generator:
+    def _clean_redeem_info(self) -> WaitableConditionType:
         """Clean the redeeming information based on whether any positions have already been redeemed."""
-        yield from self.wait_for_condition_with_sleep(self._check_already_redeemed)
+        success = yield from self._check_already_redeemed()
+        if not success:
+            return False
+
         payout_so_far = sum(self.redeeming_progress.payouts.values())
         if payout_so_far > 0:
             self.trades = {
@@ -449,6 +464,8 @@ def _clean_redeem_info(self) -> Generator:
             msg = f"The total payout so far has been {self.wei_to_native(payout_so_far)} wxDAI."
             self.context.logger.info(msg)
 
+        return True
+
     def _realitio_interact(
         self, contract_callable: str, data_key: str, placeholder: str, **kwargs: Any
     ) -> WaitableConditionType:
@@ -528,12 +545,10 @@ def get_claim_params(self) -> WaitableConditionType:
             )
             self.redeeming_progress.claim_started = True
 
+        n_retries = 0
+        from_block = self.redeeming_progress.claim_from_block
         batch_size = self.params.event_filtering_batch_size
-        for from_block in range(
-            self.redeeming_progress.claim_from_block,
-            self.redeeming_progress.claim_to_block,
-            batch_size,
-        ):
+        while from_block < self.redeeming_progress.claim_to_block:
             max_to_block = from_block + batch_size
             to_block = min(max_to_block, self.redeeming_progress.claim_to_block)
             result = yield from self._realitio_interact(
@@ -544,18 +559,29 @@ def get_claim_params(self) -> WaitableConditionType:
                 to_block=to_block,
                 question_id=self.current_question_id,
             )
-            if not result:
+
+            if not result and n_retries == self.params.max_filtering_retries:
+                err = "Skipping redeeming for the current position as the RPC is misbehaving."
+                self.context.logger.error(err)
                 return False
+
+            if not result:
+                n_retries += 1
+                keep_fraction = (1 - self.params.reduce_factor) ** n_retries
+                step = int(batch_size * keep_fraction)
+                msg = f"Repeating this call with a decreased batch size of {step}."
+                self.context.logger.warning(msg)
+                from_block += step
+                continue
+
             self.redeeming_progress.answered.extend(self.claim_params_batch)
             self.redeeming_progress.claim_from_block = to_block
+            from_block += batch_size
 
         return True
 
     def _build_claim_data(self) -> WaitableConditionType:
         """Prepare the safe tx to claim the winnings."""
-        if not self.redeeming_progress.claim_finished:
-            yield from self.wait_for_condition_with_sleep(self.get_claim_params)
-
         claim_params = self.redeeming_progress.claim_params
         if claim_params is None:
             self.context.logger.error(
@@ -603,7 +629,7 @@ def _build_redeem_data(self) -> WaitableConditionType:
         self.multisend_batches.append(batch)
         return True
 
-    def _prepare_single_redeem(self) -> Generator:
+    def _prepare_single_redeem(self) -> WaitableConditionType:
         """Prepare a multisend transaction for a single redeeming action."""
         yield from self.wait_for_condition_with_sleep(self._check_already_resolved)
         steps = []
@@ -614,13 +640,19 @@ def _prepare_single_redeem(self) -> Generator:
         yield from self.wait_for_condition_with_sleep(self._get_history_hash)
         if not self.is_history_hash_null:
             # 2. claim the winnings if claiming has not been done yet
+            if not self.redeeming_progress.claim_finished:
+                success = yield from self.get_claim_params()
+                if not success:
+                    return False
             steps.append(self._build_claim_data)
 
         # 3. we always redeem the position
         steps.append(self._build_redeem_data)
         for build_step in steps:
             yield from self.wait_for_condition_with_sleep(build_step)
 
+        return True
+
     def _process_candidate(
         self, redeem_candidate: Trade
     ) -> Generator[None, None, bool]:
@@ -646,7 +678,10 @@ def _process_candidate(
             self.context.logger.info("Position's redeeming amount is dust.")
             return False
 
-        yield from self._prepare_single_redeem()
+        success = yield from self._prepare_single_redeem()
+        if not success:
+            return False
+
         self._expected_winnings += self.current_claimable_amount
         return True
 
@@ -735,15 +770,17 @@ def async_act(self) -> Generator:
                 self._load_progress()
 
             if not self.redeeming_progress.check_finished:
-                yield from self._clean_redeem_info()
+                success = yield from self._clean_redeem_info()
 
             agent = self.context.agent_address
-            redeem_tx_hex = yield from self._prepare_safe_tx()
-            tx_submitter = policy = utilized_tools = None
-            if redeem_tx_hex is not None:
-                tx_submitter = self.matching_round.auto_round_id()
-                policy = self.policy.serialize()
-                utilized_tools = json.dumps(self.utilized_tools)
+            tx_submitter = policy = utilized_tools = redeem_tx_hex = None
+
+            if success:
+                redeem_tx_hex = yield from self._prepare_safe_tx()
+                if redeem_tx_hex is not None:
+                    tx_submitter = self.matching_round.auto_round_id()
+                    policy = self.policy.serialize()
+                    utilized_tools = json.dumps(self.utilized_tools)
 
             payload = RedeemPayload(
                 agent, tx_submitter, redeem_tx_hex, policy, utilized_tools

packages/valory/skills/decision_maker_abci/models.py

@@ -197,26 +197,32 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         check_prompt_template(self.prompt_template)
         multisend_address = kwargs.get("multisend_address", None)
         enforce(multisend_address is not None, "Multisend address not specified!")
-        self.multisend_address = multisend_address
-        self.dust_threshold = self._ensure("dust_threshold", kwargs, int)
-        self.conditional_tokens_address = self._ensure(
+        self.multisend_address: str = multisend_address
+        self.dust_threshold: int = self._ensure("dust_threshold", kwargs, int)
+        self.conditional_tokens_address: str = self._ensure(
             "conditional_tokens_address", kwargs, str
         )
-        self.realitio_proxy_address = self._ensure(
+        self.realitio_proxy_address: str = self._ensure(
             "realitio_proxy_address", kwargs, str
         )
-        self.realitio_address = self._ensure("realitio_address", kwargs, str)
+        self.realitio_address: str = self._ensure("realitio_address", kwargs, str)
         # this is the maximum batch size that will be used when filtering blocks for events.
         # increasing this number allows for faster filtering operations,
         # but also increases the chances of getting a timeout error from the RPC
-        self.event_filtering_batch_size = self._ensure(
+        self.event_filtering_batch_size: int = self._ensure(
             "event_filtering_batch_size", kwargs, int
         )
+        self.reduce_factor: float = self._ensure("reduce_factor", kwargs, float)
+        self.max_filtering_retries: int = self._ensure(
+            "max_filtering_retries", kwargs, int
+        )
         # this is the max number of redeeming operations that will be batched on a single multisend transaction.
         # increasing this number equals fewer fees but more chances for the transaction to fail
-        self.redeeming_batch_size = self._ensure("redeeming_batch_size", kwargs, int)
+        self.redeeming_batch_size: int = self._ensure(
+            "redeeming_batch_size", kwargs, int
+        )
         # a slippage in the range of [0, 1] to apply to the `minOutcomeTokensToBuy` when buying shares on a fpmm
-        self._slippage = 0.0
+        self._slippage: float = 0.0
         self.slippage: float = self._ensure("slippage", kwargs, float)
         self.epsilon: float = self._ensure("policy_epsilon", kwargs, float)
         self.agent_registry_address: str = self._ensure(

packages/valory/skills/decision_maker_abci/skill.yaml

@@ -18,14 +18,14 @@ fingerprint:
   behaviours/decision_receive.py: bafybeifacce2ke7oltnwnpdjdqfd74eaaw5wxnjfzk6c5tqdsxsmbzjj3m
   behaviours/decision_request.py: bafybeievr7vae43e7jr4eqqhwe3emvgiih7ysa66jcb5g2oz5lbxua232q
   behaviours/handle_failed_tx.py: bafybeidxpc6u575ymct5tdwutvzov6zqfdoio5irgldn3fw7q3lg36mmxm
-  behaviours/reedem.py: bafybeiepr4sellesn37csr7tmd4yhdsnn6y2jmy6kaculty6pffjc6tawe
+  behaviours/reedem.py: bafybeifgsgp655ml4om4ko3j24hqzwjdqcnobiege4lpy5roen5omxtnby
   behaviours/round_behaviour.py: bafybeig4tdktyu6hapoqymnxh2bgpds547st6a44heue657wkctwe4gjvm
   behaviours/sampling.py: bafybeifzhm4sspdvt227ksl5hjn26offgqpwempgbcwbr6dq7gyi2a46sm
   behaviours/tool_selection.py: bafybeigfr2frkljrxyfxs5p3j42equzehgaqtkyuxk6eiujyudr6ajqakm
   dialogues.py: bafybeigpwuzku3we7axmxeamg7vn656maww6emuztau5pg3ebsoquyfdqm
   fsm_specification.yaml: bafybeifnob3ceim2mj7lqagtnpwqjqqxs5eg3oiwc73gwm6x5i2dvvlcya
   handlers.py: bafybeihj33szgrcxnpd73s4nvluyxwwsvhjum2cuq3ilhhe6vfola3k7vy
-  models.py: bafybeiauwt4y7m3c675zs2uavz2cn45agkx4fxpgpvry4npxsstvxppsy4
+  models.py: bafybeifskab7b3nqalkenjgk745ky4h7bsqar73zzvrnygymhjl4qxl2ga
   payloads.py: bafybeifhq6stu4vp2ef4qvihxgpos3yc2zcuaja2safwt7efdidoejgvqa
   policy.py: bafybeidpmx4ek3qze63zpuwixyf6t7bdv62ewgkzt3ljrzadiwdw64cueq
   redeem_info.py: bafybeibddfxwp3577c3dl2utaowwltquu5fg6crezpumoebw563wxpbfrm
@@ -188,6 +188,8 @@ models:
       realitio_proxy_address: '0xAB16D643bA051C11962DA645f74632d3130c81E2'
       realitio_address: '0x79e32aE03fb27B07C89c0c568F80287C01ca2E57'
       event_filtering_batch_size: 5000
+      reduce_factor: 0.25
+      max_filtering_retries: 6
       redeeming_batch_size: 5
       policy_store_path: /data/
       slippage: 0.01

packages/valory/skills/trader_abci/skill.yaml

@@ -25,8 +25,8 @@ skills:
 - valory/transaction_settlement_abci:0.1.0:bafybeia6cdxdlqrcwk2maw25fo7dafzd2p3rs7syropvufophk2pitzbwy
 - valory/termination_abci:0.1.0:bafybeieqm46zuccaagnko3qlw6p3nvoohdrfgvpmw467r5lyil2dqrzjsy
 - valory/market_manager_abci:0.1.0:bafybeibrjre5yka3sbvb6lsefgc5yutovb6h4ffzrpecplfkjwtl5i36te
-- valory/decision_maker_abci:0.1.0:bafybeiejunyhbfzk3tqn22xmyvpyn75xyjsbeg7grjrywhecak3phbwwem
-- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeibtbginmn7tq2tluo7jyxpnlgqfwtpia5a2delxhjmsdejazvpsku
+- valory/decision_maker_abci:0.1.0:bafybeighimm45dlkagx5uig2i4wa7ikmiajlcyqkpms64amm5osxrnzroi
+- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeidaamjhhyayxprfi6pgoijmjhbybnlmhjugegm2pb4kj6ax623bna
 - valory/staking_abci:0.1.0:bafybeifdxywwwtwgfmtufkvvjvepil6qwetbldftf5eab5v2r3b3zh574q
 behaviours:
   main:
@@ -171,6 +171,8 @@ models:
       realitio_proxy_address: '0xAB16D643bA051C11962DA645f74632d3130c81E2'
       realitio_address: '0x79e32aE03fb27B07C89c0c568F80287C01ca2E57'
       event_filtering_batch_size: 5000
+      reduce_factor: 0.25
+      max_filtering_retries: 6
       redeeming_batch_size: 5
       slippage: 0.01
       policy_epsilon: 0.1

packages/valory/skills/tx_settlement_multiplexer_abci/skill.yaml

@@ -20,7 +20,7 @@ contracts: []
 protocols: []
 skills:
 - valory/abstract_round_abci:0.1.0:bafybeid7neqpxxe4ualp23gxgrtpf6r5u3myr2mmc4uvwameyqshfdoydq
-- valory/decision_maker_abci:0.1.0:bafybeiejunyhbfzk3tqn22xmyvpyn75xyjsbeg7grjrywhecak3phbwwem
+- valory/decision_maker_abci:0.1.0:bafybeighimm45dlkagx5uig2i4wa7ikmiajlcyqkpms64amm5osxrnzroi
 - valory/staking_abci:0.1.0:bafybeifdxywwwtwgfmtufkvvjvepil6qwetbldftf5eab5v2r3b3zh574q
 behaviours:
   main: