Merge pull request #258 from valory-xyz/fix/benchmarking-mode

Update the data reading logic for the mocked version

packages/packages.json

@@ -13,13 +13,13 @@
         "contract/valory/transfer_nft_condition/0.1.0": "bafybeicgpoag2lymofz3vnen76q7gtig5hzimn32o57php4uerr6t25em4",
         "contract/valory/relayer/0.1.0": "bafybeiaabvxim4blp5fxb6qjlzjivtvkme3fk24h5jte7w6vr6rsx72j6u",
         "skill/valory/market_manager_abci/0.1.0": "bafybeidygkw7mwhbk3ry3au5c5265vms5eti375v5jthd4be5dfnnoache",
-        "skill/valory/decision_maker_abci/0.1.0": "bafybeibjy72eh5fdqknvi7eabiojhksz6ls7do6ltedpspawtw7ns4v3ka",
-        "skill/valory/trader_abci/0.1.0": "bafybeif5vg5onbu5j6327pq76uoawmgri2katotwpgw7w37zmdox5rfh64",
-        "skill/valory/tx_settlement_multiplexer_abci/0.1.0": "bafybeifjwsosgd5hfx26uok23s2z6wzkeh6khjl2fmqfvc3onhk6naya34",
+        "skill/valory/decision_maker_abci/0.1.0": "bafybeiglbuwdmfhz5qsqretj63abdl6cuy6jbjqcn5qnqcklvnybgvsp4u",
+        "skill/valory/trader_abci/0.1.0": "bafybeiecqxvx366glo7oka72lppzgxf5wd6huvuegmqgzzlnhk3byuri2u",
+        "skill/valory/tx_settlement_multiplexer_abci/0.1.0": "bafybeicqiobktulmsnivmooll4hgpsj4y4uh36xqalyfbqnpyuynmk4mcu",
         "skill/valory/staking_abci/0.1.0": "bafybeicsydq6fdansf7qrmrygzchl3h6rtkdw5rmx2jyrwecj4laj5nehy",
         "skill/valory/check_stop_trading_abci/0.1.0": "bafybeidyc5fvw5wosbc3anxxxog5b67cfmvrsrltjh3cfllye3bb43r3z4",
-        "agent/valory/trader/0.1.0": "bafybeiebv47c2772agpgeswjum4cqws7t3uc7elmpdau6wd2vysqi5sexe",
-        "service/valory/trader/0.1.0": "bafybeic2zqf2i6auwl326xrvzkust3ubt2ropb6ubxn55dlbo6kqz6iucq"
+        "agent/valory/trader/0.1.0": "bafybeihdzxtn2xosfatylv4zjrjbfxq7aapfvp5zp45p7b2fdiveahfcku",
+        "service/valory/trader/0.1.0": "bafybeigvzdb7itfxrcemcig4waqwtj7zxmathgwgealgmmbw52c4rg2thm"
     },
     "third_party": {
         "protocol/open_aea/signing/1.0.0": "bafybeihv62fim3wl2bayavfcg3u5e5cxu3b7brtu4cn5xoxd6lqwachasi",

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

@@ -44,10 +44,10 @@ skills:
 - valory/reset_pause_abci:0.1.0:bafybeidw4mbx3os3hmv7ley7b3g3gja7ydpitr7mxbjpwzxin2mzyt5yam
 - valory/termination_abci:0.1.0:bafybeihq6qtbwt6i53ayqym63vhjexkcppy26gguzhhjqywfmiuqghvv44
 - valory/transaction_settlement_abci:0.1.0:bafybeigtzlk4uakmd54rxnznorcrstsr52kta474lgrnvx5ovr546vj7sq
-- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeifjwsosgd5hfx26uok23s2z6wzkeh6khjl2fmqfvc3onhk6naya34
+- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeicqiobktulmsnivmooll4hgpsj4y4uh36xqalyfbqnpyuynmk4mcu
 - valory/market_manager_abci:0.1.0:bafybeidygkw7mwhbk3ry3au5c5265vms5eti375v5jthd4be5dfnnoache
-- valory/decision_maker_abci:0.1.0:bafybeibjy72eh5fdqknvi7eabiojhksz6ls7do6ltedpspawtw7ns4v3ka
-- valory/trader_abci:0.1.0:bafybeif5vg5onbu5j6327pq76uoawmgri2katotwpgw7w37zmdox5rfh64
+- valory/decision_maker_abci:0.1.0:bafybeiglbuwdmfhz5qsqretj63abdl6cuy6jbjqcn5qnqcklvnybgvsp4u
+- valory/trader_abci:0.1.0:bafybeiecqxvx366glo7oka72lppzgxf5wd6huvuegmqgzzlnhk3byuri2u
 - valory/staking_abci:0.1.0:bafybeicsydq6fdansf7qrmrygzchl3h6rtkdw5rmx2jyrwecj4laj5nehy
 - valory/check_stop_trading_abci:0.1.0:bafybeidyc5fvw5wosbc3anxxxog5b67cfmvrsrltjh3cfllye3bb43r3z4
 - valory/mech_interact_abci:0.1.0:bafybeigkvcluq2kejpxdcb54iwqtvwhov5elg3cv4v2yomwjxyu5u7g7hi

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:bafybeiebv47c2772agpgeswjum4cqws7t3uc7elmpdau6wd2vysqi5sexe
+agent: valory/trader:0.1.0:bafybeihdzxtn2xosfatylv4zjrjbfxq7aapfvp5zp45p7b2fdiveahfcku
 number_of_agents: 4
 deployment: {}
 ---

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

@@ -21,17 +21,11 @@
 
 import csv
 import json
-import os
-import shutil
 from math import prod
-from tempfile import NamedTemporaryFile
 from typing import Any, Dict, Generator, Optional, Tuple, Union
 
 from packages.valory.skills.decision_maker_abci.behaviours.base import (
     DecisionMakerBaseBehaviour,
-    NEW_LINE,
-    QUOTE,
-    TWO_QUOTES,
     remove_fraction_wei,
 )
 from packages.valory.skills.decision_maker_abci.io_.loader import ComponentPackageLoader
@@ -91,62 +85,31 @@ def mech_response(self) -> MechInteractionResponse:
             return MechInteractionResponse(error=error)
         return self._mech_response
 
-    def _cut_dataset_row(self) -> Optional[Dict[str, str]]:
-        """Read and delete a row from the input dataset which is used during the benchmarking mode.
+    def _next_dataset_row(self) -> Optional[Dict[str, str]]:
+        """Read the next row from the input dataset which is used during the benchmarking mode.
 
         :return: a dictionary with the header fields mapped to the values of the first row.
-            If no rows are in the file, returns `None`
+            If no rows are left to process in the file, returns `None`.
         """
         sep = self.benchmarking_mode.sep
         dataset_filepath = (
             self.params.store_path / self.benchmarking_mode.dataset_filename
         )
-        write_dataset = NamedTemporaryFile(WRITE_TEXT_MODE, delete=False)
+        next_mock_data_row = self.synchronized_data.next_mock_data_row
+
+        row_with_headers: Optional[Dict[str, str]] = None
         with open(dataset_filepath) as read_dataset:
             reader = csv.DictReader(read_dataset, delimiter=sep)
-            row_with_headers: Dict[str, str] = next(reader, {})
+
+            for _ in range(next_mock_data_row):
+                row_with_headers = next(reader, {})
+
             if not row_with_headers:
                 # if no rows are in the file, then we finished the benchmarking
                 return None
 
-            # write the header into the temporary file
-            headers = reader.fieldnames
-            if headers is None:
-                self.context.logger.error(
-                    "The dataset has no headers! Cannot perform the benchmarking with a corrupted dataset file."
-                )
-                return None
-            serialized_header = sep.join(headers) + NEW_LINE
-            write_dataset.write(serialized_header)
-
-            # write the next rows into the temporary file
-            while True:
-                next_row: Dict[str, str] = next(reader, {})
-                if not next_row:
-                    break
-
-                quotes_preserved = []
-                for field in next_row.values():
-                    if QUOTE in field:
-                        # the reader removes the non-duplicated quotes, and replaces duplicated quotes with single ones;
-                        # therefore, we need to reintroduce duplicated quotes to match the original csv representation
-                        field = field.replace(QUOTE, TWO_QUOTES)
-                    if any(trigger in field for trigger in (COMMA, TWO_QUOTES)):
-                        # next, we need to reintroduce the quotes around fields that contain duplicated quotes or commas
-                        field = QUOTE + field + QUOTE
-                    quotes_preserved.append(field)
-
-                serialized_row = sep.join(quotes_preserved) + NEW_LINE
-                write_dataset.write(serialized_row)
-
-        # replace the current file with the temporary one, effectively removing the first row, excluding the header
-        shutil.move(write_dataset.name, dataset_filepath)
-        write_dataset.close()
-        try:
-            os.unlink(write_dataset.name)
-        except FileNotFoundError:
-            pass
-
+        msg = f"Processing question in row with index {next_mock_data_row}: {row_with_headers}"
+        self.context.logger.info(msg)
         return row_with_headers
 
     def _parse_dataset_row(self, row: Dict[str, str]) -> str:
@@ -176,7 +139,7 @@ def _parse_dataset_row(self, row: Dict[str, str]) -> str:
 
     def _mock_response(self) -> None:
         """Mock the response data."""
-        dataset_row = self._cut_dataset_row()
+        dataset_row = self._next_dataset_row()
         if dataset_row is None:
             return
         mech_response = self._parse_dataset_row(dataset_row)
@@ -393,6 +356,7 @@ def async_act(self) -> Generator:
             vote, p_yes, p_no, win_probability, confidence = self._get_decision()
             is_profitable = None
             bet_amount = None
+            next_mock_data_row = None
             if (
                 vote is not None
                 and p_yes is not None
@@ -403,12 +367,17 @@ def async_act(self) -> Generator:
                 is_profitable, bet_amount = yield from self._is_profitable(
                     vote, p_yes, p_no, win_probability, confidence
                 )
+
+                if self.benchmarking_mode.enabled:
+                    next_mock_data_row = self.synchronized_data.next_mock_data_row + 1
+
             payload = DecisionReceivePayload(
                 self.context.agent_address,
                 is_profitable,
                 vote,
                 confidence,
                 bet_amount,
+                next_mock_data_row,
             )
 
         yield from self.finish_behaviour(payload)

packages/valory/skills/decision_maker_abci/payloads.py

@@ -34,6 +34,7 @@ class DecisionReceivePayload(BaseTxPayload):
     vote: Optional[int]
     confidence: Optional[float]
     bet_amount: Optional[int]
+    next_mock_data_row: Optional[int]
 
 
 @dataclass(frozen=True)

packages/valory/skills/decision_maker_abci/rounds.py

@@ -303,6 +303,8 @@ class DecisionMakerAbciApp(AbciApp[Event]):
             get_name(SynchronizedData.redeemed_condition_ids),
             get_name(SynchronizedData.payout_so_far),
             get_name(SynchronizedData.mech_price),
+            get_name(SynchronizedData.mocking_mode),
+            get_name(SynchronizedData.next_mock_data_row),
         }
     )
     final_states: Set[AppState] = {
@@ -341,7 +343,8 @@ class DecisionMakerAbciApp(AbciApp[Event]):
         },
         BenchmarkingModeDisabledRound: set(),
         FinishedBenchmarkingRound: {
-            get_name(SynchronizedData.sampled_bet_index),
+            get_name(SynchronizedData.mocking_mode),
+            get_name(SynchronizedData.next_mock_data_row),
         },
         FinishedDecisionRequestRound: set(),
         FinishedSubscriptionRound: {
@@ -352,5 +355,8 @@ class DecisionMakerAbciApp(AbciApp[Event]):
         FinishedWithoutRedeemingRound: set(),
         RefillRequiredRound: set(),
         ImpossibleRound: set(),
-        BenchmarkingDoneRound: set(),
+        BenchmarkingDoneRound: {
+            get_name(SynchronizedData.mocking_mode),
+            get_name(SynchronizedData.next_mock_data_row),
+        },
     }

packages/valory/skills/decision_maker_abci/skill.yaml

@@ -17,7 +17,7 @@ fingerprint:
   behaviours/blacklisting.py: bafybeigamu43kwg4wdooviw3fb73casqeigokzy5zqshgybe4tmn6utcrm
   behaviours/check_benchmarking.py: bafybeiao2lyj7apezkqrpgsyzb3dwvrdgsrgtprf6iuhsmlsufvxfl5bci
   behaviours/claim_subscription.py: bafybeihv5dg74deifzk46ppdwcvz6lgamgl6m7qr6sgqv2zie35j2576ca
-  behaviours/decision_receive.py: bafybeieuqggknaggmxkhcxa47jh7yiyv4xjzi4zyfjuamwvowqun7lx3b4
+  behaviours/decision_receive.py: bafybeig2h6aokyrvzdebhwremhkvxvyg4z6wo3n7uw4yfiazysoxy2zaru
   behaviours/decision_request.py: bafybeiabjzzcwcfbfmtoftjhewmkgbhxfnigbc5cwmmxl6cob5gv64jwwa
   behaviours/handle_failed_tx.py: bafybeidxpc6u575ymct5tdwutvzov6zqfdoio5irgldn3fw7q3lg36mmxm
   behaviours/order_subscription.py: bafybeibl5ktq5lcwzr3komg77o67yebiokqyii7r3fsshatpww5tr5r3ke
@@ -32,17 +32,17 @@ fingerprint:
   io_/__init__.py: bafybeifxgmmwjqzezzn3e6keh2bfo4cyo7y5dq2ept3stfmgglbrzfl5rq
   io_/loader.py: bafybeih3sdsx5dhe4kzhtoafexjgkutsujwqy3zcdrlrkhtdks45bc7exa
   models.py: bafybeiax4skemmboginzgiozay5svzwgdvd62lvn4yksmfgxqoeflatht4
-  payloads.py: bafybeibqsrudf2b3g7qyhv3grbvzirh7tz3jcyt3untjm3az2ltcrpcaee
+  payloads.py: bafybeiheazhwvvg4e5p6g32vlawc7imswjznku2i47u5nsruk4agaxtf7m
   policy.py: bafybeidbsu5zn456jpku65jajwlxyg5kn2ltlkctkculkj2i2tfmmwm4jq
   redeem_info.py: bafybeifiiix4gihfo4avraxt34sfw35v6dqq45do2drrssei2shbps63mm
-  rounds.py: bafybeicnzjh7shniwojx52tgeoeea5ijqu3tzs4nol5zvhtluswfj3m5ti
+  rounds.py: bafybeib2p6omkn3r4m2tcf6avx2omxzjhd76cpkokmg45ukw6vrlot6v6e
   states/__init__.py: bafybeid23llnyp6j257dluxmrnztugo5llsrog7kua53hllyktz4dqhqoy
-  states/base.py: bafybeicdxqa2qdz4eklmywqsxdmdr4ia2bnymyn4fyerozc4bfirqsur5a
+  states/base.py: bafybeier54hauln3hb6aqzsrt56q3txjwdlgzehugmx3ip2ee7zjzbn4i4
   states/bet_placement.py: bafybeibalhxhp2c4oljmiwqi6ds3g36fgtabmf42mb5sgq6z22znrcbhda
   states/blacklisting.py: bafybeiapelgjhbjjn4uq4z5gspyirqzwzgccg5anktrp5kxdwamfnfw5mi
   states/check_benchmarking.py: bafybeiabv6pq7q45jd3nkor5afmlycqgec5ctuwcfbdukkjjm4imesv4ni
   states/claim_subscription.py: bafybeiampifhdoztggwj6gthl2hfzecmjcwnm6nic2o47q4je7j4x3ujne
-  states/decision_receive.py: bafybeicghjnwk5fhlp5psmxxz3b4fhbru4nm2i2z6kdqvehyly4w45g3lm
+  states/decision_receive.py: bafybeibc5o5stxfm47jdwrjkldirgbvvdaprdtkevwahsnedimcyso6wtm
   states/decision_request.py: bafybeiarv3r5j7cfvxmudki2llbdl2pvf24p5mvsva6bdgrylnwdyag5xy
   states/final_states.py: bafybeidtmnr3ybzmkvk4twt4dz4iqgr6ddfkgs2gqfmohxhdf2i5vkj2s4
   states/handle_failed_tx.py: bafybeihewm2vernvhktuorljdupjqcg2p5vs6wvsira2d62wkoyo5xlzjm

packages/valory/skills/decision_maker_abci/states/base.py

@@ -200,6 +200,14 @@ def mocking_mode(self) -> Optional[bool]:
             return None
         return bool(mode)
 
+    @property
+    def next_mock_data_row(self) -> int:
+        """Get the next_mock_data_row."""
+        next_mock_data_row = self.db.get("next_mock_data_row", 1)
+        if next_mock_data_row is None:
+            return 1
+        return int(next_mock_data_row)
+
     @property
     def mech_responses(self) -> List[MechInteractionResponse]:
         """Get the mech responses."""

packages/valory/skills/decision_maker_abci/states/decision_receive.py

@@ -46,6 +46,7 @@ class DecisionReceiveRound(CollectSameUntilThresholdRound):
         get_name(SynchronizedData.vote),
         get_name(SynchronizedData.confidence),
         get_name(SynchronizedData.bet_amount),
+        get_name(SynchronizedData.next_mock_data_row),
     )
     collection_key = get_name(SynchronizedData.participant_to_decision)
 

packages/valory/skills/trader_abci/skill.yaml

@@ -25,8 +25,8 @@ skills:
 - valory/transaction_settlement_abci:0.1.0:bafybeigtzlk4uakmd54rxnznorcrstsr52kta474lgrnvx5ovr546vj7sq
 - valory/termination_abci:0.1.0:bafybeihq6qtbwt6i53ayqym63vhjexkcppy26gguzhhjqywfmiuqghvv44
 - valory/market_manager_abci:0.1.0:bafybeidygkw7mwhbk3ry3au5c5265vms5eti375v5jthd4be5dfnnoache
-- valory/decision_maker_abci:0.1.0:bafybeibjy72eh5fdqknvi7eabiojhksz6ls7do6ltedpspawtw7ns4v3ka
-- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeifjwsosgd5hfx26uok23s2z6wzkeh6khjl2fmqfvc3onhk6naya34
+- valory/decision_maker_abci:0.1.0:bafybeiglbuwdmfhz5qsqretj63abdl6cuy6jbjqcn5qnqcklvnybgvsp4u
+- valory/tx_settlement_multiplexer_abci:0.1.0:bafybeicqiobktulmsnivmooll4hgpsj4y4uh36xqalyfbqnpyuynmk4mcu
 - valory/staking_abci:0.1.0:bafybeicsydq6fdansf7qrmrygzchl3h6rtkdw5rmx2jyrwecj4laj5nehy
 - valory/check_stop_trading_abci:0.1.0:bafybeidyc5fvw5wosbc3anxxxog5b67cfmvrsrltjh3cfllye3bb43r3z4
 - valory/mech_interact_abci:0.1.0:bafybeigkvcluq2kejpxdcb54iwqtvwhov5elg3cv4v2yomwjxyu5u7g7hi

packages/valory/skills/tx_settlement_multiplexer_abci/skill.yaml

@@ -21,7 +21,7 @@ protocols:
 - valory/ledger_api:1.0.0:bafybeihdk6psr4guxmbcrc26jr2cbgzpd5aljkqvpwo64bvaz7tdti2oni
 skills:
 - valory/abstract_round_abci:0.1.0:bafybeih3enhagoql7kzpeyzzu2scpkif6y3ubakpralfnwxcvxexdyvy5i
-- valory/decision_maker_abci:0.1.0:bafybeibjy72eh5fdqknvi7eabiojhksz6ls7do6ltedpspawtw7ns4v3ka
+- valory/decision_maker_abci:0.1.0:bafybeiglbuwdmfhz5qsqretj63abdl6cuy6jbjqcn5qnqcklvnybgvsp4u
 - valory/staking_abci:0.1.0:bafybeicsydq6fdansf7qrmrygzchl3h6rtkdw5rmx2jyrwecj4laj5nehy
 - valory/mech_interact_abci:0.1.0:bafybeigkvcluq2kejpxdcb54iwqtvwhov5elg3cv4v2yomwjxyu5u7g7hi
 behaviours: