Merge pull request #102 from valory-xyz/feat/add_retry_functionality

Retry answers functionality

packages/packages.json

@@ -3,10 +3,10 @@
         "contract/valory/fpmm_deterministic_factory/0.1.0": "bafybeihi2pozsqbgu6mkepoyqqwqxc3o7ks47tx7vuf5hxfkbsyuskzqjq",
         "contract/valory/wxdai/0.1.0": "bafybeidalocwbhmbto6ii6adldtpcughtdt6j3v4tv36utevjk2wrdyqie",
         "contract/valory/fpmm/0.1.0": "bafybeiai2ruj27nnglvn7yc5atojyojo3fkmofw6wrjgz2ybps2uwdizx4",
-        "skill/valory/market_creation_manager_abci/0.1.0": "bafybeidmzffvwvh3euspl7rub2mnhhqj4qhurwbryxvxaxsntueu2gelve",
-        "skill/valory/market_maker_abci/0.1.0": "bafybeic2zhtrusglacvb6uc66w2irybg2xm4wtwuounkmkaox3icmbwdtm",
-        "agent/valory/market_maker/0.1.0": "bafybeidsov3f5hchknxiyeouoeupqot5o44y2as6u2q4fcqeszz7bycd2a",
-        "service/valory/market_maker/0.1.0": "bafybeihvw3wntknp4sfhsd3vreth7srtbrme5wqoxj73ameabg3t6mofd4"
+        "skill/valory/market_creation_manager_abci/0.1.0": "bafybeicoymcfncfdjisnng56yu3hcyt6zbk7trpqontpjqjsk6owkwecyu",
+        "skill/valory/market_maker_abci/0.1.0": "bafybeibzabftqsyhihpv7ka6v6l4tjzhvzpijti2qwoyyl7ttrlfj3clp4",
+        "agent/valory/market_maker/0.1.0": "bafybeifdfqfrz6wdtxy2of6pazjwjxqfslyg5bosha27x3z47gu4caqove",
+        "service/valory/market_maker/0.1.0": "bafybeigviagkeq3deu6ihjvjt536lau55jz2uj5zwujx7tvwongqe3mx7a"
     },
     "third_party": {
         "protocol/valory/contract_api/1.0.0": "bafybeidgu7o5llh26xp3u3ebq3yluull5lupiyeu6iooi2xyymdrgnzq5i",

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

@@ -37,9 +37,9 @@ protocols:
 skills:
 - valory/abstract_abci:0.1.0:bafybeibvz3hi25ktgdzm25kih25o5osml62zk4kzpgmzmyrem3hzo2vtnm
 - valory/abstract_round_abci:0.1.0:bafybeic2emnylfmdtidobgdsxa4tgdelreeimtglqzrmic6cumhpsbfzhe
-- valory/market_maker_abci:0.1.0:bafybeic2zhtrusglacvb6uc66w2irybg2xm4wtwuounkmkaox3icmbwdtm
+- valory/market_maker_abci:0.1.0:bafybeibzabftqsyhihpv7ka6v6l4tjzhvzpijti2qwoyyl7ttrlfj3clp4
 - valory/registration_abci:0.1.0:bafybeibzduasg5pahk2d2glc36wlopg54ybwcdmjqbzozt2lmkp4reqsdu
-- valory/market_creation_manager_abci:0.1.0:bafybeidmzffvwvh3euspl7rub2mnhhqj4qhurwbryxvxaxsntueu2gelve
+- valory/market_creation_manager_abci:0.1.0:bafybeicoymcfncfdjisnng56yu3hcyt6zbk7trpqontpjqjsk6owkwecyu
 - valory/reset_pause_abci:0.1.0:bafybeib3geaw3o35aigmb4dhplxrjlsef7hbdwrozu4akiunlgyiegshj4
 - valory/termination_abci:0.1.0:bafybeifnfuy3svz5oz6kwglu35q6jn7ai3riqplyypskb7za5ejyvjmfza
 - valory/transaction_settlement_abci:0.1.0:bafybeifhuemoavm6hf4vypg2sl7kng3iv4ipxgpk3itbh5hp73lworpuwq
@@ -219,4 +219,5 @@ models:
       google_engine_id: ${str:google_engine_id}
       openai_api_key: ${str:openai_api_key}
       mech_contract_address: ${str:0x77af31de935740567cf4ff1986d04b2c964a786a}
+      answer_retry_intervals: ${list:[0, 86400, 259200, 604800, 1209600]}
 is_abstract: false

packages/valory/services/market_maker/service.yaml

@@ -7,7 +7,7 @@ license: Apache-2.0
 fingerprint:
   README.md: bafybeibwz3af6326msp4h3kqehijvmyhaytvyfbo3o2npc2w4b6zrg6pfq
 fingerprint_ignore_patterns: []
-agent: valory/market_maker:0.1.0:bafybeidsov3f5hchknxiyeouoeupqot5o44y2as6u2q4fcqeszz7bycd2a
+agent: valory/market_maker:0.1.0:bafybeifdfqfrz6wdtxy2of6pazjwjxqfslyg5bosha27x3z47gu4caqove
 number_of_agents: 1
 deployment:
   agent:
@@ -71,6 +71,8 @@ models:
       multisend_batch_size: ${MULTISEND_BATCH_SIZE:int:1}
       ipfs_address: ${IPFS_ADDRESS:str:https://gateway.autonolas.tech/ipfs/}
       mech_contract_address: ${MECH_CONTRACT_ADDRESS:str:0x77af31de935740567cf4ff1986d04b2c964a786a}
+      answer_retry_intervals: ${ANSWER_RETRY_INTERVALS:list:[0, 86400, 259200, 604800,
+        1209600]}
       on_chain_service_id: ${ON_CHAIN_SERVICE_ID:int:null}
       reset_tendermint_after: ${RESET_TENDERMINT_AFTER:int:1}
       reset_pause_duration: ${RESET_PAUSE_DURATION:int:1800}

packages/valory/skills/market_creation_manager_abci/behaviours.py

@@ -2291,6 +2291,52 @@ def _get_balance(self, account: str) -> Generator[None, None, Optional[int]]:
         balance = cast(int, ledger_api_response.state.body.get("get_balance_result"))
         return balance
 
+    def _eligible_questions_to_answer(
+        self, unanswered_questions: List[Dict[str, Any]]
+    ) -> List[str]:
+        """Determine the eligible questions to answer at this time"""
+        now = self.last_synced_timestamp
+        eligible_questions_id = []
+        answer_retry_intervals = self.params.answer_retry_intervals
+
+        self.context.logger.info(f"Answer retry intervals: {answer_retry_intervals}")
+
+        for question in unanswered_questions:
+            question_id = question["question"]["id"].lower()
+
+            if question_id in self.shared_state.questions_responded:
+                continue
+
+            if question_id not in self.shared_state.questions_requested_mech:
+                self.shared_state.questions_requested_mech[question_id] = {
+                    "question": question,
+                    "retries": [],
+                }
+
+            retries = self.shared_state.questions_requested_mech[question_id]["retries"]
+            n_retries = len(retries)
+            time_since_last_retry = now - retries[-1] if retries else 0
+            retry_period = answer_retry_intervals[
+                min(n_retries, len(answer_retry_intervals) - 1)
+            ]
+            if n_retries == 0 or time_since_last_retry > retry_period:
+                eligible_questions_id.append(question_id)
+
+        self.context.logger.info(
+            f"Determined {len(eligible_questions_id)} eligible questions to answer."
+        )
+
+        num_questions = min(
+            len(eligible_questions_id), self.params.multisend_batch_size
+        )
+        random.seed(self.last_synced_timestamp)
+        random_questions_id = random.sample(eligible_questions_id, num_questions)
+
+        self.context.logger.info(
+            f"Chosen {len(random_questions_id)} eligible questions to answer."
+        )
+        return random_questions_id
+
     def get_payload(self) -> Generator[None, None, str]:
         """Get the transaction payload"""
         # get the questions to that need to be answered
@@ -2300,23 +2346,15 @@ def get_payload(self) -> Generator[None, None, str]:
             self.context.logger.info("Couldn't get the questions")
             return GetPendingQuestionsRound.ERROR_PAYLOAD
 
-        filtered_questions = [
-            question
-            for question in unanswered_questions
-            if question["question"]["id"].lower()
-            not in self.shared_state.questions_requested_mech
-        ]
-        random.seed(self.last_synced_timestamp)
-        num_questions = min(len(filtered_questions), self.params.multisend_batch_size)
-        random_questions = random.sample(filtered_questions, num_questions)
-        questions = random_questions
+        eligible_questions_id = self._eligible_questions_to_answer(unanswered_questions)
 
-        if len(questions) == 0:
-            self.context.logger.info("No questions to close")
+        if len(eligible_questions_id) == 0:
+            self.context.logger.info("No eligible questions to answer")
             return GetPendingQuestionsRound.NO_TX_PAYLOAD
 
         self.context.logger.info(
-            f"Got {len(questions)} questions to close. " f"Questions: {questions}"
+            f"Got {len(eligible_questions_id)} questions to close. "
+            f"Questions ID: {eligible_questions_id}"
         )
 
         safe_address = self.synchronized_data.safe_contract_address
@@ -2327,7 +2365,7 @@ def get_payload(self) -> Generator[None, None, str]:
 
         self.context.logger.info(f"Address {safe_address!r} has balance {balance}.")
         max_num_questions = min(
-            len(questions), self.params.questions_to_close_batch_size
+            len(eligible_questions_id), self.params.questions_to_close_batch_size
         )
         bond_required = self.params.close_question_bond * max_num_questions
 
@@ -2342,14 +2380,17 @@ def get_payload(self) -> Generator[None, None, str]:
 
         # Prepare the Mech Requests for these questions
         new_mech_requests = []
-        for question in questions:
-            question_id = question["question"]["id"].lower()
-            if question_id in self.shared_state.questions_requested_mech:
-                # we already processed this question, skip it
-                self.context.logger.info(
-                    f"Question {question_id} already processed, skipping it."
-                )
-                continue
+        for question_id in eligible_questions_id:
+            question = self.shared_state.questions_requested_mech[question_id][
+                "question"
+            ]
+            retries = self.shared_state.questions_requested_mech[question_id]["retries"]
+            retries.append(self.last_synced_timestamp)
+
+            self.context.logger.info(
+                f"Requesting mech answer for question {question_id} ({question['title']})."
+            )
+            self.context.logger.info(f"Question {question_id} retries: {retries}.")
 
             new_mech_requests.append(
                 asdict(
@@ -2360,7 +2401,6 @@ def get_payload(self) -> Generator[None, None, str]:
                     )
                 )
             )
-            self.shared_state.questions_requested_mech[question_id] = question
 
         self.context.logger.info(f"new_mech_requests: {new_mech_requests}")
 
@@ -2432,9 +2472,20 @@ def _get_payload(self) -> Generator[None, None, str]:
             if question_id not in self.shared_state.questions_requested_mech:
                 continue
 
-            question = self.shared_state.questions_requested_mech[question_id]
-
+            question = self.shared_state.questions_requested_mech[question_id][
+                "question"
+            ]
+            retries = self.shared_state.questions_requested_mech[question_id]["retries"]
             answer = self._parse_mech_response(response)
+
+            if answer is None and len(retries) >= len(
+                self.params.answer_retry_intervals
+            ):
+                self.context.logger.info(
+                    f"Question {question} has been retried at timestamps {retries} without success. Assuming question is invalid."
+                )
+                answer = ANSWER_INVALID
+
             self.context.logger.info(f"Got answer {answer} for question {question}")
 
             if answer is None:
@@ -2469,6 +2520,7 @@ def _get_payload(self) -> Generator[None, None, str]:
             # try to answer the same question multiple times due to a out-of-sync issue
             # between the subgraph and the realitio contract.
             self.shared_state.questions_responded.add(question_id)
+            del self.shared_state.questions_requested_mech[question_id]
 
         if len(txs) == 0:
             # something went wrong, respond with ERROR payload for now

packages/valory/skills/market_creation_manager_abci/models.py

@@ -163,6 +163,9 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.mech_interact_round_timeout_seconds = self._ensure(
             "mech_interact_round_timeout_seconds", kwargs, type_=int
         )
+        self.answer_retry_intervals = self._ensure(
+            key="answer_retry_intervals", kwargs=kwargs, type_=List[int]
+        )
         super().__init__(*args, **kwargs)
 
 

packages/valory/skills/market_creation_manager_abci/skill.yaml

@@ -10,11 +10,11 @@ aea_version: '>=1.0.0, <2.0.0'
 fingerprint:
   .gitignore: bafybeihdfdezgtr3s2lzq5y3oaitfrdy4u4pehionju2bdez35wcjgqx6y
   __init__.py: bafybeibkyjt4covc4yhd22aw7kav45zozk3exdv344emt3lilln64soaxm
-  behaviours.py: bafybeigb3lkuhg66m7kucao3j6eknoz6pbjhqwvyaos44hqcjus7nuzpcm
+  behaviours.py: bafybeibskfldhxwww3tsybkz47vkg3pfh7tcotvde5gnwyx3c6rq7htw7a
   dialogues.py: bafybeicmaufkl7vdomnfciv7lw4536ssld7x4uemdapuhsyvfpd4ncibza
   fsm_specification.yaml: bafybeicps5t2anm7fv35fwnw7oolaxxmpmjio6mdw4sc6rzpnsolph5xlm
   handlers.py: bafybeietxjfli2i57kb7heoy772rcq2znusl36gg7jjj5g3pddw7egny3q
-  models.py: bafybeibkjv4cgvje2ozuzbmcvhscvnvd5isenlpuwmgjr4oyscmgjdobre
+  models.py: bafybeibef4czds6n7h6bgdcpfmfrhakbtegm4iffauz24pmsu7xaaw6abe
   payloads.py: bafybeibu7sptf43adazxpyzwtfpph7bgfhtwiotg5sdlcwjt6iw3idqn7a
   rounds.py: bafybeie6rgwfulbu5xkuvxgrujbvenioyfggtzlmt23pbm6auzj3w4se44
   tests/__init__.py: bafybeihfxvqnyfly72tbxnnnglshcilm2kanihqnjiasvcz3ec3csw32ti
@@ -218,6 +218,12 @@ models:
       light_slash_unit_amount: 5000000000000000
       serious_slash_unit_amount: 8000000000000000
       mech_contract_address: '0x77af31de935740567cf4ff1986d04b2c964a786a'
+      answer_retry_intervals:
+      - 0
+      - 86400
+      - 259200
+      - 604800
+      - 1209600
     class_name: MarketCreationManagerParams
   randomness_api:
     args:

packages/valory/skills/market_maker_abci/skill.yaml

@@ -24,7 +24,7 @@ skills:
 - valory/abstract_round_abci:0.1.0:bafybeic2emnylfmdtidobgdsxa4tgdelreeimtglqzrmic6cumhpsbfzhe
 - valory/registration_abci:0.1.0:bafybeibzduasg5pahk2d2glc36wlopg54ybwcdmjqbzozt2lmkp4reqsdu
 - valory/reset_pause_abci:0.1.0:bafybeib3geaw3o35aigmb4dhplxrjlsef7hbdwrozu4akiunlgyiegshj4
-- valory/market_creation_manager_abci:0.1.0:bafybeidmzffvwvh3euspl7rub2mnhhqj4qhurwbryxvxaxsntueu2gelve
+- valory/market_creation_manager_abci:0.1.0:bafybeicoymcfncfdjisnng56yu3hcyt6zbk7trpqontpjqjsk6owkwecyu
 - valory/termination_abci:0.1.0:bafybeifnfuy3svz5oz6kwglu35q6jn7ai3riqplyypskb7za5ejyvjmfza
 - valory/transaction_settlement_abci:0.1.0:bafybeifhuemoavm6hf4vypg2sl7kng3iv4ipxgpk3itbh5hp73lworpuwq
 - valory/mech_interact_abci:0.1.0:bafybeigilca6rfqu4qw7yhz5lnhzgymj5cabtusx5hjndt7w7kulkn7cga
@@ -214,6 +214,12 @@ models:
       light_slash_unit_amount: 5000000000000000
       serious_slash_unit_amount: 8000000000000000
       mech_contract_address: '0x77af31de935740567cf4ff1986d04b2c964a786a'
+      answer_retry_intervals:
+      - 0
+      - 86400
+      - 259200
+      - 604800
+      - 1209600
     class_name: Params
   randomness_api:
     args: