Repository: RasaHQ/rasa
Pull request: marker stats
**Proposed changes**:
- closes #9961 

**Status (please check what you already did)**:
- [x] added some tests for the functionality
- [ ] updated the documentation
- [ ] updated the changelog (please check [changelog](https://github.com/RasaHQ/rasa/tree/main/changelog) for instructions)
- [x] reformat files using `black` (please check [Readme](https://github.com/RasaHQ/rasa#code-style) for instructions)


Problem statement:
Success Markers: cleanup/fix stats
**Definition of Done**
- [ ] Ensure it can deal with multiple evaluated dialogues for the same sender_id
- [ ] Ensure that stat computation works with new extractor
- [ ] Document what stats are computed
- [ ] Replace timestamp by event index
- [ ] In the current implementation the "number of times that the marker applied" is the statistic called "n". Rename it.
- [ ] output flat file format

Hints from issue discussion:

Commit message: small fixes (types, forgotten assert)

Patch to review:
diff --git a/rasa/core/evaluation/marker_stats.py b/rasa/core/evaluation/marker_stats.py
index 45d4aabd1998..2ced0bee715e 100644
--- a/rasa/core/evaluation/marker_stats.py
+++ b/rasa/core/evaluation/marker_stats.py
@@ -1,143 +1,292 @@
-import os
-import sys
-
-if sys.version_info >= (3, 8):
-    from typing import TypedDict
-else:
-    from typing_extensions import TypedDict
-from typing import Dict, Text, Union, Any, List, Tuple
+from __future__ import annotations
+from typing import Dict, Text, Union, List, Tuple
 from pathlib import Path
-import json
+import io
+import csv
+
 import numpy as np
 
 
-class MarkerStats(TypedDict):
-    """A TypedDict for statistics computed over extracted markers."""
+from rasa.core.evaluation.marker_base import EventMetaData
 
-    n: int
-    mean: float
-    median: float
-    min: float
-    max: float
 
+def compute_statistics(
+    values: List[Union[float, int]]
+) -> Dict[Text, Union[int, np.float]]:
+    """Computes some statistics over the given numbers."""
+    return {
+        "count": len(values) if values else 0,
+        "mean": np.mean(values) if values else np.nan,
+        "median": np.median(values) if values else np.nan,
+        "min": min(values) if values else np.nan,
+        "max": max(values) if values else np.nan,
+    }
 
-def load_extracted_markers_json_file(path: Union[Text, Path]) -> List:
-    """Reads a json marker file.
 
-    Args:
-        path: path to a json file.
-    """
-    path = os.path.abspath(path)
-    with open(path) as json_file:
-        extracted_markers = json.load(json_file)
-        return extracted_markers
+class _CSVWriter:
+    """A csv writer."""
 
+    def __init__(self, stream: io.TextIOBase) -> None:
+        self.table_writer = csv.writer(stream)
 
-def compute_summary_stats(data_points: Union[List[float], np.ndarray]) -> MarkerStats:
-    """Computes summary statistics for a given array.
+    def writerow(self, row: List[Text]) -> None:
+        """Write the given row.
 
-    Args:
-        data_points: can be a numpy array or a list of numbers.
+        Args:
+            row: the entries of a row as a list of strings
+        """
+        self.table_writer.writerow(row)
 
-    Returns:
-        A MarkerStats object containing size, mean, median, min, and max.
-        If the given array of data points is empty, it returns 0 for size, and
-        `np.nan` for every statistic.
-    """
-    if np.size(data_points) > 0:
-        stats: MarkerStats = {
-            "n": int(np.size(data_points)),
-            "mean": float(np.mean(data_points)),
-            "median": float(np.median(data_points)),
-            "min": int(np.min(data_points)),
-            "max": int(np.max(data_points)),
-        }
-        return stats
-    else:
-        empty_stats: MarkerStats = {
-            "n": 0,
-            "mean": np.nan,
-            "median": np.nan,
-            "min": np.nan,
-            "max": np.nan,
-        }
-        return empty_stats
-
-
-def compute_single_tracker_stats(
-    single_tracker_markers: Dict[str, Any]
-) -> Dict[str, MarkerStats]:
-    """Computes summary statistics for a single tracker.
-
-    Args:
-        single_tracker_markers: a dictionary containing the extracted
-        markers for one tracker.
-
-    Returns:
-        A dictionary containing statistics computed for each marker.
-    """
-    tracker_stats = {}
-    for marker in single_tracker_markers["markers"]:
-        tracker_stats[marker["marker"]] = compute_summary_stats(
-            marker["num_preceding_user_turns"]
-        )
-    return tracker_stats
 
+class MarkerStatistics:
+    """Computes some statistics on marker extraction results.
+
+    (1) Number of sessions where markers apply:
+
+    For each marker, we compute the total number (as well as the percentage) of the
+    sessions in which a marker applies at least once.
+    Moreover, we output the total number of sessions that were parsed.
 
-def compute_multi_tracker_stats(
-    multi_tracker_markers: List[Dict[str, Any]],
-) -> Tuple[Dict[str, Union[int, MarkerStats]], Dict[Any, Dict[str, MarkerStats]]]:
-    """Computes summary statistics for multiple trackers.
+    (2) Number of user turns preceding relevant events - per sessions:
 
-    Args:
-        multi_tracker_markers: a list of dictionaries each containing the
-        extracted markers for one tracker.
+    For each marker, we consider all relevant events where that marker applies.
+    Everytime a marker applies, we check how many user turns precede that event.
+    We collect all these numbers and compute basic statistics (e.g. count and mean)
+    on them.
 
-    Returns:
-         A dictionary containing summary statistics computed per
-         marker over all trackers.
-         A dictionary containing summary statistics computed
-         per tracker.
+    This means, per session, we compute how often a marker applies and how many
+    user turns preceding a relevant marker application on average, in that session.
+
+    (3) Number of user turns preceding relevant events - over all sessions:
+
+    Here, for each marker, we consider all relevant events where a marker applies
+    *in any of the sessions*. Then, we again calculate basic statistics over the
+    respective number of user turns that precedes each of these events.
+
+    This means, we compute to how many events the marker applies in total and we
+    compute an estimate of the expected number of user turns preceding that
+    precede an (relevant) event where a marker applies.
     """
-    per_marker_stats: Dict[str, Union[int, MarkerStats]] = {
-        "num_trackers": len(multi_tracker_markers)
-    }
-    per_tracker_stats = {}
-    per_marker_values = {}
 
-    for tracker in multi_tracker_markers:
-        # compute statistics per tracker
-        per_tracker_stats[tracker["tracker_ID"]] = compute_single_tracker_stats(tracker)
+    NO_MARKER = "-"
+    STAT_NUM_SESSIONS = "total_number_of_sessions"
+    STAT_NUM_SESSIONS_WHERE_APPLIES = (
+        "number_of_sessions_where_marker_applies_at_least_once"
+    )
+    STAT_PERCENTAGE_SESSIONS_WHERE_APPLIES = (
+        "percentage_of_sessions_where_marker_applies_at_least_once"
+    )
+
+    @staticmethod
+    def _add_num_user_turns_str_to(stat_name: Text) -> Text:
+        return f"{stat_name}(number of preceding user turns)"
+
+    ALL_SESSIONS = np.nan
+    ALL_SENDERS = "all"
+
+    def __init__(self) -> None:
+        """Creates a new marker statistic."""
+        # to ensure consistency of processed rows
+        self._marker_names = []
+
+        # (1) For collecting the per-session analysis:
+        # NOTE: we could stream these instead of collecting them...
+        self.session_results: Dict[Text, Dict[Text, List[Union[int, float]]]] = {}
+        self.session_identifier: List[Tuple[Text, int]] = []
 
-        for marker in tracker["markers"]:
-            # append raw values
-            per_marker_values.setdefault(marker["marker"], []).extend(
-                marker["num_preceding_user_turns"]
+        # (2) For the overall statistics:
+        self.num_preceding_user_turns_collected: Dict[Text, List[int]] = {}
+        self.count_if_applied_at_least_once: Dict[Text, int] = {}
+        self.num_sessions = 0
+
+    def process(
+        self,
+        extracted_markers: Dict[Text, List[EventMetaData]],
+        session_idx: int,
+        sender_id: Text,
+    ) -> None:
+        """Adds the restriction result to the statistics evaluation.
+
+        Args:
+            extracted_markers: marker extraction results, i.e. a dictionary mapping
+                from a marker name to a list of meta data describing relevant events
+                for that marker
+            session_idx: an index that, together with the `sender_id` identifies
+                the session from which the markers where extracted
+            sender_id: an id that, together with the `session_idx` identifies
+                the session from which the markers where extracted
+        """
+        if len(self._marker_names) == 0:
+            # sort and initialise here once so our result tables are sorted
+            self._marker_names = sorted(extracted_markers.keys())
+            self.count_if_applied_at_least_once = {
+                marker_name: 0 for marker_name in self._marker_names
+            }
+            self.num_preceding_user_turns_collected = {
+                marker_name: [] for marker_name in self._marker_names
+            }
+            # NOTE: we could stream these instead of collecting them...
+            stat_names = sorted(compute_statistics([]).keys())
+            self.session_results = {
+                marker_name: {stat_name: [] for stat_name in stat_names}
+                for marker_name in self._marker_names
+            }
+        else:
+            if set(extracted_markers.keys()) != set(self.session_results):
+                raise RuntimeError(
+                    f"Expected all processed extraction results to contain information"
+                    f"for the same set of markers. But found "
+                    f"{set(extracted_markers.keys())} which differs from "
+                    f"the marker extracted so far (i.e. {self.session_results})."
+                )
+
+        # update session identifiers / count
+        self.num_sessions += 1
+        self.session_identifier.append((sender_id, session_idx))
+
+        for marker_name, meta_data in extracted_markers.items():
+
+            num_preceding_user_turns = [
+                event_meta_data.preceding_user_turns for event_meta_data in meta_data
+            ]
+
+            # update per session statistics
+            statistics = compute_statistics(num_preceding_user_turns)
+            for stat_name, stat_value in statistics.items():
+                self.session_results[marker_name][stat_name].append(stat_value)
+
+            # update overall statistics
+            self.num_preceding_user_turns_collected[marker_name].extend(
+                num_preceding_user_turns
             )
+            if len(num_preceding_user_turns):
+                self.count_if_applied_at_least_once[marker_name] += 1
 
-    for marker_name in per_marker_values.keys():
-        # compute statistics per marker over all trackers
-        per_marker_stats[marker_name] = compute_summary_stats(
-            per_marker_values[marker_name]
-        )
+    def to_csv(self, path: Path, overwrite: bool = False) -> None:
+        """Exports the resulting statistics to a csv file.
 
-    return per_marker_stats, per_tracker_stats
+        Args:
+            path: path to where the csv file should be written.
+            overwrite: set to `True` to enable overwriting an existing file
+        """
+        if path.is_file() and not overwrite:
+            raise FileExistsError(f"Expected that there was no file at {path}.")
+        with path.open(mode="w") as f:
+            table_writer = _CSVWriter(f)
+            table_writer.writerow(self._header())
+            self._write_overview(table_writer)
+            self._write_overall_statistics(table_writer)
+            # NOTE: we could stream these instead of collecting them...
+            self._write_per_session_statistics(table_writer)
 
+    @staticmethod
+    def _header() -> List[Text]:
+        return [
+            "sender_id",
+            "session_idx",
+            "marker",
+            "statistic",
+            "value",
+        ]
 
-def write_stats(path: Union[Text, Path], stats: dict, per_tracker_stats: dict) -> None:
-    """Outputs statistics to JSON file."""
-    path = os.path.abspath(path)
-    data = {"marker_stats": stats, "tracker_stats": per_tracker_stats}
-    with open(path, "w") as outfile:
-        json_str = json.dumps(data, default=np_encoder, indent=2)
-        outfile.write(json_str)
+    def _write_overview(self, table_writer: _CSVWriter) -> None:
+        special_sender_idx = self.ALL_SENDERS
+        special_session_idx = self.ALL_SESSIONS
+        self._write_row(
+            table_writer=table_writer,
+            sender_id=special_sender_idx,
+            session_idx=special_session_idx,
+            marker_name=self.NO_MARKER,
+            statistic_name=self.STAT_NUM_SESSIONS,
+            statistic_value=self.num_sessions,
+        )
+        for marker_name, count in self.count_if_applied_at_least_once.items():
+            self._write_row(
+                table_writer=table_writer,
+                sender_id=special_sender_idx,
+                session_idx=special_session_idx,
+                marker_name=marker_name,
+                statistic_name=self.STAT_NUM_SESSIONS_WHERE_APPLIES,
+                statistic_value=count,
+            )
+            self._write_row(
+                table_writer=table_writer,
+                sender_id=special_sender_idx,
+                session_idx=special_session_idx,
+                marker_name=marker_name,
+                statistic_name=self.STAT_PERCENTAGE_SESSIONS_WHERE_APPLIES,
+                statistic_value=(count / self.num_sessions * 100)
+                if self.num_sessions
+                else 100.0,
+            )
 
+    def _write_overall_statistics(self, table_writer: _CSVWriter) -> None:
+        for marker_name, num_list in self.num_preceding_user_turns_collected.items():
+            for statistic_name, value in compute_statistics(num_list).items():
+                MarkerStatistics._write_row(
+                    table_writer=table_writer,
+                    sender_id=self.ALL_SENDERS,
+                    session_idx=self.ALL_SESSIONS,
+                    marker_name=marker_name,
+                    statistic_name=self._add_num_user_turns_str_to(statistic_name),
+                    statistic_value=value,
+                )
 
-def np_encoder(obj: Any) -> Any:
-    """Encodes numpy array values to make them JSON serializable.
+    def _write_per_session_statistics(self, table_writer: _CSVWriter) -> None:
+        for marker_name, collected_statistics in self.session_results.items():
+            for statistic_name, values in collected_statistics.items():
+                self._write_per_session_statistic(
+                    table_writer=table_writer,
+                    marker_name=marker_name,
+                    statistic_name=statistic_name,
+                    session_identifiers=self.session_identifier,
+                    values=values,
+                )
 
-    adapted from: https://bit.ly/3ajjTwp
-    """
-    if isinstance(obj, np.generic):
-        return obj.item()
-    return obj
+    @staticmethod
+    def _write_per_session_statistic(
+        table_writer: _CSVWriter,
+        marker_name: Text,
+        statistic_name: Text,
+        session_identifiers: List[Tuple[Text, int]],
+        values: List[Union[np.float, int]],
+    ) -> None:
+        for record_idx, (sender_id, session_idx) in enumerate(session_identifiers):
+            MarkerStatistics._write_row(
+                table_writer=table_writer,
+                sender_id=sender_id,
+                session_idx=session_idx,
+                marker_name=marker_name,
+                statistic_name=MarkerStatistics._add_num_user_turns_str_to(
+                    statistic_name
+                ),
+                statistic_value=values[record_idx],
+            )
+
+    @staticmethod
+    def _write_row(
+        table_writer: _CSVWriter,
+        sender_id: Text,
+        session_idx: Union[int, np.float],
+        marker_name: Text,
+        statistic_name: Text,
+        statistic_value: Union[int, np.float],
+    ) -> None:
+        if isinstance(statistic_value, int):
+            value_str = str(statistic_value)
+        elif np.isnan(statistic_value):
+            value_str = str(np.nan)
+        else:
+            value_str = np.round(statistic_value, 3)
+        table_writer.writerow(
+            [
+                str(item)
+                for item in [
+                    sender_id,
+                    session_idx,
+                    marker_name,
+                    statistic_name,
+                    value_str,
+                ]
+            ]
+        )
diff --git a/tests/core/evaluation/test_marker_stats.py b/tests/core/evaluation/test_marker_stats.py
new file mode 100644
index 000000000000..a2f2b33330b9
--- /dev/null
+++ b/tests/core/evaluation/test_marker_stats.py
@@ -0,0 +1,266 @@
+import csv
+from pathlib import Path
+from typing import Dict, List, Text, Tuple
+import itertools
+
+import pytest
+import numpy as np
+
+from rasa.core.evaluation.marker_stats import (
+    EventMetaData,
+    MarkerStatistics,
+    compute_statistics,
+)
+
+
+def test_compute_statistics_on_empty_set():
+    result = compute_statistics([])
+    assert result["count"] == 0
+    assert all(np.isnan(value) for key, value in result.items() if key != "count")
+    # we see the same keys as in those cases where we can evaluate something
+    dummy_result_non_empty = compute_statistics([1, 2])
+    assert set(dummy_result_non_empty.keys()) == set(result.keys())
+
+
+def test_compute_statistics_simple_check():
+    stats = compute_statistics([1, 2, 9, 0])
+    assert stats["count"] == 4
+    assert stats["min"] == 0
+    assert stats["max"] == 9
+    assert stats["mean"] == 3
+    assert stats["median"] == 1.5  # this is no bug, it is a convention numpy follows
+
+
+def _generate_random_example_for_one_session_and_one_marker(
+    rng: np.random.Generator,
+) -> Tuple[List[EventMetaData], List[int]]:
+    """Generates a random marker extraction result for a single session and marker.
+
+    Args:
+        rng: a random number generator
+    Returns:
+        the event list representing the marker extraction result as well as
+        the plain list of numbers used as "preceding user turns" in that extraction
+        result
+    """
+    applies = int(rng.choice(10))
+    all_preceding_user_turn_numbers = [int(rng.choice(20)) for _ in range(applies)]
+    event_list = [
+        EventMetaData(
+            idx=int(rng.choice(100)), preceding_user_turns=preceding_user_turns
+        )
+        for preceding_user_turns in all_preceding_user_turn_numbers
+    ]
+    return event_list, all_preceding_user_turn_numbers
+
+
+def _generate_random_examples(
+    rng: np.random.Generator,
+    num_markers: int = 3,
+    num_sessions_min: int = 2,
+    num_sessions_max: int = 10,
+) -> Tuple[List[Dict[Text, List[EventMetaData]]], Dict[Text, List[List[int]]]]:
+    """Generates a random number of random marker extraction results for some markers.
+
+    Args:
+        rng: a random number generator
+        num_markers: the number of markers to be imitated
+    Returns:
+        a list containing a dictionary of the marker extraction results per marker,
+        as well as a collection of the plain list of numbers used as "preceding user
+        turns" in that extraction results
+    """
+    num_sessions = int(rng.integers(low=num_sessions_min, high=num_sessions_max + 1))
+    markers = [f"marker{idx}" for idx in range(num_markers)]
+    per_session_results: List[Dict[Text, List[EventMetaData]]] = []
+    preceding_user_turn_numbers_used_per_marker: Dict[Text, List[List[int]]] = {
+        marker: [] for marker in markers
+    }
+    for _ in range(num_sessions - 1):  # we append one later
+        result_dict = {}
+        for marker in markers:
+            (
+                event_list,
+                num_list,
+            ) = _generate_random_example_for_one_session_and_one_marker(rng=rng)
+            result_dict[marker] = event_list
+            preceding_user_turn_numbers_used_per_marker[marker].append(num_list)
+        per_session_results.append(result_dict)
+    # append a session where we didn't find any marker
+    per_session_results.append({marker: [] for marker in markers})
+    for marker in preceding_user_turn_numbers_used_per_marker:
+        preceding_user_turn_numbers_used_per_marker[marker].append([])
+    return per_session_results, preceding_user_turn_numbers_used_per_marker
+
+
+@pytest.mark.parametrize("seed", [2345, 5654, 2345234,])
+def test_process_results_per_session(seed: int):
+
+    rng = np.random.default_rng(seed=seed)
+
+    (
+        per_session_results,
+        preceding_user_turn_numbers_used_per_marker,
+    ) = _generate_random_examples(num_markers=3, rng=rng)
+    markers = sorted(preceding_user_turn_numbers_used_per_marker.keys())
+    num_sessions = len(per_session_results)
+
+    stats = MarkerStatistics()
+    sender_ids = []
+    session_indices = []
+    for session_idx, results in enumerate(per_session_results):
+        sender_id = str(rng.choice(100))
+        session_idx = int(rng.choice(100))
+        stats.process(
+            session_idx=session_idx, sender_id=sender_id, extracted_markers=results,
+        )
+        sender_ids.append(sender_id)
+        session_indices.append(session_idx)
+
+    assert stats.num_sessions == len(per_session_results)
+    for marker in markers:
+        for idx in range(num_sessions):
+            expected_stats = compute_statistics(
+                preceding_user_turn_numbers_used_per_marker[marker][idx]
+            )
+            for stat_name, stat_value in expected_stats.items():
+                assert pytest.approx(
+                    stats.session_results[marker][stat_name][idx], stat_value
+                )
+    for idx in range(num_sessions):
+        assert stats.session_identifier[idx] == (sender_ids[idx], session_indices[idx],)
+
+
+@pytest.mark.parametrize("seed", [2345, 5654, 2345234,])
+def test_process_results_overall(seed: int):
+
+    rng = np.random.default_rng(seed=seed)
+    (
+        per_session_results,
+        preceding_user_turn_numbers_used_per_marker,
+    ) = _generate_random_examples(num_markers=3, rng=rng)
+    markers = sorted(preceding_user_turn_numbers_used_per_marker.keys())
+    num_sessions = len(per_session_results)
+
+    stats = MarkerStatistics()
+    for session_idx, results in enumerate(per_session_results):
+        stats.process(
+            session_idx=session_idx,
+            sender_id=str(rng.choice(100)),
+            extracted_markers=results,
+        )
+
+    assert stats.num_sessions == num_sessions
+    for marker in markers:
+        # count how often we generated some results for a session:
+        number_lists = preceding_user_turn_numbers_used_per_marker[marker]
+        applies_at_least_once = sum(len(sub_list) > 0 for sub_list in number_lists)
+        # and compare that to the expected count:
+        assert stats.count_if_applied_at_least_once[marker] == applies_at_least_once
+        # check if we collected the all the "preceding user turn numbers"
+        concatenated_numbers = list(
+            itertools.chain.from_iterable(
+                preceding_user_turn_numbers_used_per_marker[marker]
+            )
+        )
+        assert stats.num_preceding_user_turns_collected[marker] == concatenated_numbers
+
+
+@pytest.mark.parametrize("seed", [2345, 5654, 2345234,])
+def test_to_csv(tmp_path: Path, seed: int):
+
+    rng = np.random.default_rng(seed=seed)
+    (
+        per_session_results,
+        preceding_user_turn_numbers_used_per_marker,
+    ) = _generate_random_examples(
+        num_markers=3, rng=rng, num_sessions_min=10, num_sessions_max=20
+    )
+    markers = sorted(preceding_user_turn_numbers_used_per_marker.keys())
+    num_sessions = len(per_session_results)
+
+    stats = MarkerStatistics()
+    for session_idx, results in enumerate(per_session_results):
+        stats.process(
+            session_idx=session_idx,
+            sender_id=str(rng.choice(100)),
+            extracted_markers=results,
+        )
+
+    tmp_file = tmp_path / "test.csv"
+    stats.to_csv(path=tmp_file)
+
+    with tmp_file.open(mode="r") as f:
+        reader = csv.DictReader(f)
+        rows = [row for row in reader]
+
+    assert rows[0] == {
+        "sender_id": "all",
+        "session_idx": "nan",
+        "marker": "-",
+        "statistic": "total_number_of_sessions",
+        "value": str(num_sessions),
+    }
+
+    num_digits = 3
+    row_idx = 1
+    for marker_name in markers:
+        assert rows[row_idx] == {
+            "sender_id": "all",
+            "session_idx": "nan",
+            "marker": marker_name,
+            "statistic": "number_of_sessions_where_marker_applies_at_least_once",
+            "value": str(stats.count_if_applied_at_least_once[marker_name]),
+        }
+        row_idx += 1
+        assert rows[row_idx] == {
+            "sender_id": "all",
+            "session_idx": "nan",
+            "marker": marker_name,
+            "statistic": "percentage_of_sessions_where_marker_applies_at_least_once",
+            "value": str(
+                round(
+                    stats.count_if_applied_at_least_once[marker_name]
+                    / num_sessions
+                    * 100,
+                    num_digits,
+                )
+            ),
+        }
+        row_idx += 1
+
+    for marker_name in markers:
+        statistics = compute_statistics(
+            stats.num_preceding_user_turns_collected[marker_name]
+        )
+        for stat_name, stat_value in statistics.items():
+            assert rows[row_idx] == {
+                "sender_id": "all",
+                "session_idx": "nan",
+                "marker": marker_name,
+                "statistic": MarkerStatistics._add_num_user_turns_str_to(stat_name),
+                "value": str(round(stat_value, num_digits)),
+            }
+            row_idx += 1
+
+    actual_information = {
+        (row["sender_id"], row["session_idx"], row["marker"], row["statistic"]): row[
+            "value"
+        ]
+        for row in rows[row_idx:]
+    }
+    expected_information = {
+        (
+            sender_id,
+            str(session_idx),
+            marker_name,
+            MarkerStatistics._add_num_user_turns_str_to(stat_name),
+        ): str(value)
+        if np.isnan(value)
+        else str(round(value, num_digits))
+        for marker_name in markers
+        for stat_name, values in stats.session_results[marker_name].items()
+        for (sender_id, session_idx), value in zip(stats.session_identifier, values)
+    }
+
+    assert actual_information == expected_information
diff --git a/tests/core/evaluation/test_stats.py b/tests/core/evaluation/test_stats.py
deleted file mode 100644
index 887fd5f5f6cc..000000000000
--- a/tests/core/evaluation/test_stats.py
+++ /dev/null
@@ -1,217 +0,0 @@
-import pytest
-import os
-import math
-import numpy as np
-from typing import Text
-from rasa.core.evaluation.marker_stats import (
-    load_extracted_markers_json_file,
-    compute_multi_tracker_stats,
-    compute_summary_stats,
-    compute_single_tracker_stats,
-    write_stats,
-)
-
-
-@pytest.fixture
-def overall_stats() -> dict:
-    return {
-        "num_trackers": 3,
-        "no_restart": {"n": 2, "mean": 16.5, "median": 16.5, "min": 13, "max": 20},
-        "all_required_data_gathered": {
-            "n": 4,
-            "mean": 11.0,
-            "median": 8.0,
-            "min": 8,
-            "max": 20,
-        },
-        "carbon_offset_calculated": {
-            "n": 6,
-            "mean": 12.0,
-            "median": 11.0,
-            "min": 8,
-            "max": 22,
-        },
-        "chitchat_and_FAQ": {"n": 4, "mean": 6.5, "median": 5.5, "min": 2, "max": 13},
-        "dialog_expected_order": {
-            "n": 4,
-            "mean": 10.0,
-            "median": 8.0,
-            "min": 5,
-            "max": 19,
-        },
-    }
-
-
-@pytest.fixture
-def multi_tracker_stats() -> dict:
-    return {
-        "dialogue_1": {
-            "no_restart": {"n": 1, "mean": 20.0, "median": 20.0, "min": 20, "max": 20},
-            "all_required_data_gathered": {
-                "n": 1,
-                "mean": 8.0,
-                "median": 8.0,
-                "min": 8,
-                "max": 8,
-            },
-            "carbon_offset_calculated": {
-                "n": 2,
-                "mean": 10.0,
-                "median": 10.0,
-                "min": 8,
-                "max": 12,
-            },
-            "chitchat_and_FAQ": {
-                "n": 3,
-                "mean": 8.0,
-                "median": 9.0,
-                "min": 2,
-                "max": 13,
-            },
-            "dialog_expected_order": {
-                "n": 1,
-                "mean": 5.0,
-                "median": 5.0,
-                "min": 5,
-                "max": 5,
-            },
-        },
-        "dialogue_2": {
-            "no_restart": {"n": 1, "mean": 13.0, "median": 13.0, "min": 13, "max": 13},
-            "all_required_data_gathered": {
-                "n": 1,
-                "mean": 8.0,
-                "median": 8.0,
-                "min": 8,
-                "max": 8,
-            },
-            "carbon_offset_calculated": {
-                "n": 2,
-                "mean": 10.0,
-                "median": 10.0,
-                "min": 8,
-                "max": 12,
-            },
-            "chitchat_and_FAQ": {
-                "n": 0,
-                "mean": np.nan,
-                "median": np.nan,
-                "min": np.nan,
-                "max": np.nan,
-            },
-            "dialog_expected_order": {
-                "n": 1,
-                "mean": 7.0,
-                "median": 7.0,
-                "min": 7,
-                "max": 7,
-            },
-        },
-        "dialogue_3": {
-            "no_restart": {
-                "n": 0,
-                "mean": np.nan,
-                "median": np.nan,
-                "min": np.nan,
-                "max": np.nan,
-            },
-            "all_required_data_gathered": {
-                "n": 2,
-                "mean": 14.0,
-                "median": 14.0,
-                "min": 8,
-                "max": 20,
-            },
-            "carbon_offset_calculated": {
-                "n": 2,
-                "mean": 16.0,
-                "median": 16.0,
-                "min": 10,
-                "max": 22,
-            },
-            "chitchat_and_FAQ": {
-                "n": 1,
-                "mean": 2.0,
-                "median": 2.0,
-                "min": 2,
-                "max": 2,
-            },
-            "dialog_expected_order": {
-                "n": 2,
-                "mean": 14.0,
-                "median": 14.0,
-                "min": 9,
-                "max": 19,
-            },
-        },
-    }
-
-
-def test_compute_summary_stats():
-    """Tests compute summary stats."""
-    a = [20, 30, 32]
-    stats_a = compute_summary_stats(a)
-    assert stats_a["n"] == 3
-    assert math.isclose(stats_a["mean"], 27.33, abs_tol=0.01)
-    assert stats_a["median"] == 30
-    assert stats_a["min"] == 20
-    assert stats_a["max"] == 32
-
-    b = [15]
-    stats_b = compute_summary_stats(b)
-    assert stats_b["n"] == 1
-    assert stats_b["mean"] == 15
-    assert stats_b["median"] == 15
-    assert stats_b["min"] == 15
-    assert stats_b["max"] == 15
-
-
-def test_compute_summary_stats_empty():
-    """Tests compute summary stats on an empty array."""
-    stats = compute_summary_stats([])
-    assert stats == {
-        "n": 0,
-        "mean": np.nan,
-        "median": np.nan,
-        "min": np.nan,
-        "max": np.nan,
-    }
-
-
-def test_compute_single_tracker_stats(
-    extracted_markers_json: Text, multi_tracker_stats: dict
-):
-    """Tests compute summary stats for a single tracker."""
-    expected_output = multi_tracker_stats["dialogue_1"]
-    extracted_markers = load_extracted_markers_json_file(extracted_markers_json)
-    stats = compute_single_tracker_stats(extracted_markers[0])
-    assert stats == expected_output
-
-
-def test_compute_single_tracker_stats_some_empty(
-    extracted_markers_json: Text, multi_tracker_stats: dict
-):
-    """Tests compute summary stats for a single tracker with empty values."""
-    expected_output = multi_tracker_stats["dialogue_2"]
-    extracted_markers = load_extracted_markers_json_file(extracted_markers_json)
-    stats = compute_single_tracker_stats(extracted_markers[1])
-    assert stats == expected_output
-
-
-def test_compute_multi_tracker_stats(
-    extracted_markers_json: Text, overall_stats: dict, multi_tracker_stats: dict
-):
-    """Tests compute summary stats on multiple trackers."""
-    extracted_markers = load_extracted_markers_json_file(extracted_markers_json)
-    stats, per_tracker_stats = compute_multi_tracker_stats(extracted_markers)
-    assert stats == overall_stats
-    assert per_tracker_stats == multi_tracker_stats
-
-
-def test_write_stats(extracted_markers_json: Text, marker_stats_output_json: Text):
-    """Tests write summary stats."""
-    extracted_markers = load_extracted_markers_json_file(extracted_markers_json)
-    stats, per_tracker_stats = compute_multi_tracker_stats(extracted_markers)
-    write_stats(marker_stats_output_json, stats, per_tracker_stats)
-    assert os.path.exists(marker_stats_output_json)
-    os.remove(marker_stats_output_json)


Code review comments:
diff_hunk: @@ -1,143 +1,292 @@
-import os
-import sys
-
-if sys.version_info >= (3, 8):
-    from typing import TypedDict
-else:
-    from typing_extensions import TypedDict
-from typing import Dict, Text, Union, Any, List, Tuple
+from __future__ import annotations
+from typing import Dict, Text, Union, List, Tuple
 from pathlib import Path
-import json
+import io
+import csv
+
 import numpy as np
 
 
-class MarkerStats(TypedDict):
-    """A TypedDict for statistics computed over extracted markers."""
+from rasa.core.evaluation.marker_base import EventMetaData
 
-    n: int
-    mean: float
-    median: float
-    min: float
-    max: float
 
+def compute_statistics(
+    values: List[Union[float, int]]
+) -> Dict[Text, Union[int, np.float]]:
+    """Computes some statistics over the given numbers."""
+    return {
+        "count": len(values) if values else 0,
+        "mean": np.mean(values) if values else np.nan,
+        "median": np.median(values) if values else np.nan,
+        "min": min(values) if values else np.nan,
+        "max": max(values) if values else np.nan,
+    }
 
-def load_extracted_markers_json_file(path: Union[Text, Path]) -> List:
-    """Reads a json marker file.
 
-    Args:
-        path: path to a json file.
-    """
-    path = os.path.abspath(path)
-    with open(path) as json_file:
-        extracted_markers = json.load(json_file)
-        return extracted_markers
+class _CSVWriter:
+    """A csv writer."""
 
+    def __init__(self, stream: io.TextIOBase) -> None:
+        self.table_writer = csv.writer(stream)
 
-def compute_summary_stats(data_points: Union[List[float], np.ndarray]) -> MarkerStats:
-    """Computes summary statistics for a given array.
+    def writerow(self, row: List[Text]) -> None:
+        """Write the given row.
 
-    Args:
-        data_points: can be a numpy array or a list of numbers.
+        Args:
+            row: the entries of a row as a list of strings
+        """
+        self.table_writer.writerow(row)
 
-    Returns:
-        A MarkerStats object containing size, mean, median, min, and max.
-        If the given array of data points is empty, it returns 0 for size, and
-        `np.nan` for every statistic.
-    """
-    if np.size(data_points) > 0:
-        stats: MarkerStats = {
-            "n": int(np.size(data_points)),
-            "mean": float(np.mean(data_points)),
-            "median": float(np.median(data_points)),
-            "min": int(np.min(data_points)),
-            "max": int(np.max(data_points)),
-        }
-        return stats
-    else:
-        empty_stats: MarkerStats = {
-            "n": 0,
-            "mean": np.nan,
-            "median": np.nan,
-            "min": np.nan,
-            "max": np.nan,
-        }
-        return empty_stats
-
-
-def compute_single_tracker_stats(
-    single_tracker_markers: Dict[str, Any]
-) -> Dict[str, MarkerStats]:
-    """Computes summary statistics for a single tracker.
-
-    Args:
-        single_tracker_markers: a dictionary containing the extracted
-        markers for one tracker.
-
-    Returns:
-        A dictionary containing statistics computed for each marker.
-    """
-    tracker_stats = {}
-    for marker in single_tracker_markers["markers"]:
-        tracker_stats[marker["marker"]] = compute_summary_stats(
-            marker["num_preceding_user_turns"]
-        )
-    return tracker_stats
 
+class MarkerStatistics:
+    """Computes some statistics on marker extraction results.
+
+    (1) Number of sessions where markers apply:
+
+    For each marker, we compute the total number (as well as the percentage) of the
+    sessions in which a marker applies at least once.
+    Moreover, we output the total number of sessions that were parsed.
 
-def compute_multi_tracker_stats(
-    multi_tracker_markers: List[Dict[str, Any]],
-) -> Tuple[Dict[str, Union[int, MarkerStats]], Dict[Any, Dict[str, MarkerStats]]]:
-    """Computes summary statistics for multiple trackers.
+    (2) Number of user turns preceding relevant events - per sessions:
 
-    Args:
-        multi_tracker_markers: a list of dictionaries each containing the
-        extracted markers for one tracker.
+    For each marker, we consider all relevant events where that marker applies.
+    Everytime a marker applies, we check how many user turns precede that event.
+    We collect all these numbers and compute basic statistics (e.g. count and mean)
+    on them.
 
-    Returns:
-         A dictionary containing summary statistics computed per
-         marker over all trackers.
-         A dictionary containing summary statistics computed
-         per tracker.
+    This means, per session, we compute how often a marker applies and how many
+    user turns preceding a relevant marker application on average, in that session.
+
+    (3) Number of user turns preceding relevant events - over all sessions:
+
+    Here, for each marker, we consider all relevant events where a marker applies
+    *in any of the sessions*. Then, we again calculate basic statistics over the
+    respective number of user turns that precedes each of these events.
+
+    This means, we compute to how many events the marker applies in total and we
+    compute an estimate of the expected number of user turns preceding that
+    precede an (relevant) event where a marker applies.
     """
-    per_marker_stats: Dict[str, Union[int, MarkerStats]] = {
-        "num_trackers": len(multi_tracker_markers)
-    }
-    per_tracker_stats = {}
-    per_marker_values = {}
 
-    for tracker in multi_tracker_markers:
-        # compute statistics per tracker
-        per_tracker_stats[tracker["tracker_ID"]] = compute_single_tracker_stats(tracker)
+    NO_MARKER = "-"
+    STAT_NUM_SESSIONS = "total_number_of_sessions"
+    STAT_NUM_SESSIONS_WHERE_APPLIES = (
+        "number_of_sessions_where_marker_applies_at_least_once"
+    )
+    STAT_PERCENTAGE_SESSIONS_WHERE_APPLIES = (
+        "percentage_of_sessions_where_marker_applies_at_least_once"
+    )
+
+    @staticmethod
+    def _add_num_user_turns_str_to(stat_name: Text) -> Text:
+        return f"{stat_name}(number of preceding user turns)"
+
+    ALL_SESSIONS = np.nan
+    ALL_SENDERS = "all"
+
+    def __init__(self) -> None:
+        """Creates a new marker statistic."""
+        # to ensure consistency of processed rows
+        self._marker_names = []
+
+        # (1) For collecting the per-session analysis:
+        # NOTE: we could stream these instead of collecting them...
+        self.session_results: Dict[Text, Dict[Text, List[Union[int, float]]]] = {}
+        self.session_identifier: List[Tuple[Text, int]] = []
 
-        for marker in tracker["markers"]:
-            # append raw values
-            per_marker_values.setdefault(marker["marker"], []).extend(
-                marker["num_preceding_user_turns"]
+        # (2) For the overall statistics:
+        self.num_preceding_user_turns_collected: Dict[Text, List[int]] = {}
+        self.count_if_applied_at_least_once: Dict[Text, int] = {}
+        self.num_sessions = 0
+
+    def process(
+        self,
+        extracted_markers: Dict[Text, List[EventMetaData]],
+        session_idx: int,
+        sender_id: Text,
+    ) -> None:
+        """Adds the restriction result to the statistics evaluation.
+
+        Args:
+            extracted_markers: marker extraction results, i.e. a dictionary mapping
+                from a marker name to a list of meta data describing relevant events
+                for that marker
+            session_idx: an index that, together with the `sender_id` identifies
+                the session from which the markers where extracted
+            sender_id: an id that, together with the `session_idx` identifies
+                the session from which the markers where extracted
+        """
+        if len(self._marker_names) == 0:
+            # sort and initialise here once so our result tables are sorted
+            self._marker_names = sorted(extracted_markers.keys())
+            self.count_if_applied_at_least_once = {
+                marker_name: 0 for marker_name in self._marker_names
+            }
+            self.num_preceding_user_turns_collected = {
+                marker_name: [] for marker_name in self._marker_names
+            }
+            # NOTE: we could stream these instead of collecting them...
+            stat_names = sorted(compute_statistics([]).keys())
+            self.session_results = {
+                marker_name: {stat_name: [] for stat_name in stat_names}
+                for marker_name in self._marker_names
+            }
+        else:
+            if set(extracted_markers.keys()) != set(self.session_results):
+                raise RuntimeError(
+                    f"Expected all processed extraction results to contain information"
+                    f"for the same set of markers. But found "
+                    f"{set(extracted_markers.keys())} which differs from "
+                    f"the marker extracted so far (i.e. {self.session_results})."
+                )
+
+        # update session identifiers / count
+        self.num_sessions += 1
+        self.session_identifier.append((sender_id, session_idx))
+
+        for marker_name, meta_data in extracted_markers.items():
+
+            num_preceding_user_turns = [
+                event_meta_data.preceding_user_turns for event_meta_data in meta_data
+            ]
+
+            # update per session statistics
+            statistics = compute_statistics(num_preceding_user_turns)
+            for stat_name, stat_value in statistics.items():
+                self.session_results[marker_name][stat_name].append(stat_value)
+
+            # update overall statistics
+            self.num_preceding_user_turns_collected[marker_name].extend(
+                num_preceding_user_turns
             )
+            if len(num_preceding_user_turns):
+                self.count_if_applied_at_least_once[marker_name] += 1
 
-    for marker_name in per_marker_values.keys():
-        # compute statistics per marker over all trackers
-        per_marker_stats[marker_name] = compute_summary_stats(
-            per_marker_values[marker_name]
-        )
+    def to_csv(self, path: Path, overwrite: bool = False) -> None:
+        """Exports the resulting statistics to a csv file.
 
-    return per_marker_stats, per_tracker_stats
+        Args:
+            path: path to where the csv file should be written.
+            overwrite: set to `True` to enable overwriting an existing file
+        """
+        if path.is_file() and not overwrite:
+            raise FileExistsError(f"Expected that there was no file at {path}.")
+        with path.open(mode="w") as f:
+            table_writer = _CSVWriter(f)
+            table_writer.writerow(self._header())
+            self._write_overview(table_writer)
+            self._write_overall_statistics(table_writer)
+            # NOTE: we could stream these instead of collecting them...
+            self._write_per_session_statistics(table_writer)
 
+    @staticmethod
+    def _header() -> List[Text]:
+        return [
+            "sender_id",
+            "session_idx",
path: rasa/core/evaluation/marker_stats.py
text: @user1:
Since we use `dialogue_id` in the output file, we probably want to be consistent here too - which do you prefer? I can change it in the CLI PR as well to match if that's better

@author:
oh, I wasn't thinking of that -- I just remembered Gregs comment and thought this was more precise and that we should change it in marker.py later (can do that in the validate PR) -- yep, I think we should rename 👍 

@user2:
Yeah I agree that we should keep session as it's more precise.

@author:
kept as is :)

diff_hunk: @@ -1,143 +1,292 @@
-import os
-import sys
-
-if sys.version_info >= (3, 8):
-    from typing import TypedDict
-else:
-    from typing_extensions import TypedDict
-from typing import Dict, Text, Union, Any, List, Tuple
+from __future__ import annotations
+from typing import Dict, Text, Union, List, Tuple
 from pathlib import Path
-import json
+import io
+import csv
+
 import numpy as np
 
 
-class MarkerStats(TypedDict):
-    """A TypedDict for statistics computed over extracted markers."""
+from rasa.core.evaluation.marker_base import EventMetaData
 
-    n: int
-    mean: float
-    median: float
-    min: float
-    max: float
 
+def compute_statistics(
+    values: List[Union[float, int]]
+) -> Dict[Text, Union[int, np.float]]:
+    """Computes some statistics over the given numbers."""
+    return {
+        "count": len(values) if values else 0,
+        "mean": np.mean(values) if values else np.nan,
+        "median": np.median(values) if values else np.nan,
+        "min": min(values) if values else np.nan,
+        "max": max(values) if values else np.nan,
+    }
 
-def load_extracted_markers_json_file(path: Union[Text, Path]) -> List:
-    """Reads a json marker file.
 
-    Args:
-        path: path to a json file.
-    """
-    path = os.path.abspath(path)
-    with open(path) as json_file:
-        extracted_markers = json.load(json_file)
-        return extracted_markers
+class _CSVWriter:
+    """A csv writer."""
 
+    def __init__(self, stream: io.TextIOBase) -> None:
+        self.table_writer = csv.writer(stream)
 
-def compute_summary_stats(data_points: Union[List[float], np.ndarray]) -> MarkerStats:
-    """Computes summary statistics for a given array.
+    def writerow(self, row: List[Text]) -> None:
+        """Write the given row.
 
-    Args:
-        data_points: can be a numpy array or a list of numbers.
+        Args:
+            row: the entries of a row as a list of strings
+        """
+        self.table_writer.writerow(row)
 
-    Returns:
-        A MarkerStats object containing size, mean, median, min, and max.
-        If the given array of data points is empty, it returns 0 for size, and
-        `np.nan` for every statistic.
-    """
-    if np.size(data_points) > 0:
-        stats: MarkerStats = {
-            "n": int(np.size(data_points)),
-            "mean": float(np.mean(data_points)),
-            "median": float(np.median(data_points)),
-            "min": int(np.min(data_points)),
-            "max": int(np.max(data_points)),
-        }
-        return stats
-    else:
-        empty_stats: MarkerStats = {
-            "n": 0,
-            "mean": np.nan,
-            "median": np.nan,
-            "min": np.nan,
-            "max": np.nan,
-        }
-        return empty_stats
-
-
-def compute_single_tracker_stats(
-    single_tracker_markers: Dict[str, Any]
-) -> Dict[str, MarkerStats]:
-    """Computes summary statistics for a single tracker.
-
-    Args:
-        single_tracker_markers: a dictionary containing the extracted
-        markers for one tracker.
-
-    Returns:
-        A dictionary containing statistics computed for each marker.
-    """
-    tracker_stats = {}
-    for marker in single_tracker_markers["markers"]:
-        tracker_stats[marker["marker"]] = compute_summary_stats(
-            marker["num_preceding_user_turns"]
-        )
-    return tracker_stats
 
+class MarkerStatistics:
+    """Computes some statistics on marker extraction results.
+
+    (1) Number of sessions where markers apply:
+
+    For each marker, we compute the total number (as well as the percentage) of the
+    sessions in which a marker applies at least once.
+    Moreover, we output the total number of sessions that were parsed.
 
-def compute_multi_tracker_stats(
-    multi_tracker_markers: List[Dict[str, Any]],
-) -> Tuple[Dict[str, Union[int, MarkerStats]], Dict[Any, Dict[str, MarkerStats]]]:
-    """Computes summary statistics for multiple trackers.
+    (2) Number of user turns preceding relevant events - per sessions:
 
-    Args:
-        multi_tracker_markers: a list of dictionaries each containing the
-        extracted markers for one tracker.
+    For each marker, we consider all relevant events where that marker applies.
+    Everytime a marker applies, we check how many user turns precede that event.
+    We collect all these numbers and compute basic statistics (e.g. count and mean)
+    on them.
 
-    Returns:
-         A dictionary containing summary statistics computed per
-         marker over all trackers.
-         A dictionary containing summary statistics computed
-         per tracker.
+    This means, per session, we compute how often a marker applies and how many
+    user turns preceding a relevant marker application on average, in that session.
+
+    (3) Number of user turns preceding relevant events - over all sessions:
+
+    Here, for each marker, we consider all relevant events where a marker applies
+    *in any of the sessions*. Then, we again calculate basic statistics over the
+    respective number of user turns that precedes each of these events.
path: rasa/core/evaluation/marker_stats.py
text: @user1:
```suggestion
    respective number of user turns that precede each of these events.
```

diff_hunk: @@ -1,143 +1,292 @@
-import os
-import sys
-
-if sys.version_info >= (3, 8):
-    from typing import TypedDict
-else:
-    from typing_extensions import TypedDict
-from typing import Dict, Text, Union, Any, List, Tuple
+from __future__ import annotations
+from typing import Dict, Text, Union, List, Tuple
 from pathlib import Path
-import json
+import io
+import csv
+
 import numpy as np
 
 
-class MarkerStats(TypedDict):
-    """A TypedDict for statistics computed over extracted markers."""
+from rasa.core.evaluation.marker_base import EventMetaData
 
-    n: int
-    mean: float
-    median: float
-    min: float
-    max: float
 
+def compute_statistics(
+    values: List[Union[float, int]]
+) -> Dict[Text, Union[int, np.float]]:
+    """Computes some statistics over the given numbers."""
+    return {
+        "count": len(values) if values else 0,
+        "mean": np.mean(values) if values else np.nan,
+        "median": np.median(values) if values else np.nan,
+        "min": min(values) if values else np.nan,
+        "max": max(values) if values else np.nan,
+    }
 
-def load_extracted_markers_json_file(path: Union[Text, Path]) -> List:
-    """Reads a json marker file.
 
-    Args:
-        path: path to a json file.
-    """
-    path = os.path.abspath(path)
-    with open(path) as json_file:
-        extracted_markers = json.load(json_file)
-        return extracted_markers
+class _CSVWriter:
+    """A csv writer."""
 
+    def __init__(self, stream: io.TextIOBase) -> None:
+        self.table_writer = csv.writer(stream)
 
-def compute_summary_stats(data_points: Union[List[float], np.ndarray]) -> MarkerStats:
-    """Computes summary statistics for a given array.
+    def writerow(self, row: List[Text]) -> None:
+        """Write the given row.
 
-    Args:
-        data_points: can be a numpy array or a list of numbers.
+        Args:
+            row: the entries of a row as a list of strings
+        """
+        self.table_writer.writerow(row)
 
-    Returns:
-        A MarkerStats object containing size, mean, median, min, and max.
-        If the given array of data points is empty, it returns 0 for size, and
-        `np.nan` for every statistic.
-    """
-    if np.size(data_points) > 0:
-        stats: MarkerStats = {
-            "n": int(np.size(data_points)),
-            "mean": float(np.mean(data_points)),
-            "median": float(np.median(data_points)),
-            "min": int(np.min(data_points)),
-            "max": int(np.max(data_points)),
-        }
-        return stats
-    else:
-        empty_stats: MarkerStats = {
-            "n": 0,
-            "mean": np.nan,
-            "median": np.nan,
-            "min": np.nan,
-            "max": np.nan,
-        }
-        return empty_stats
-
-
-def compute_single_tracker_stats(
-    single_tracker_markers: Dict[str, Any]
-) -> Dict[str, MarkerStats]:
-    """Computes summary statistics for a single tracker.
-
-    Args:
-        single_tracker_markers: a dictionary containing the extracted
-        markers for one tracker.
-
-    Returns:
-        A dictionary containing statistics computed for each marker.
-    """
-    tracker_stats = {}
-    for marker in single_tracker_markers["markers"]:
-        tracker_stats[marker["marker"]] = compute_summary_stats(
-            marker["num_preceding_user_turns"]
-        )
-    return tracker_stats
 
+class MarkerStatistics:
+    """Computes some statistics on marker extraction results.
+
+    (1) Number of sessions where markers apply:
+
+    For each marker, we compute the total number (as well as the percentage) of the
+    sessions in which a marker applies at least once.
+    Moreover, we output the total number of sessions that were parsed.
 
-def compute_multi_tracker_stats(
-    multi_tracker_markers: List[Dict[str, Any]],
-) -> Tuple[Dict[str, Union[int, MarkerStats]], Dict[Any, Dict[str, MarkerStats]]]:
-    """Computes summary statistics for multiple trackers.
+    (2) Number of user turns preceding relevant events - per sessions:
 
-    Args:
-        multi_tracker_markers: a list of dictionaries each containing the
-        extracted markers for one tracker.
+    For each marker, we consider all relevant events where that marker applies.
+    Everytime a marker applies, we check how many user turns precede that event.
+    We collect all these numbers and compute basic statistics (e.g. count and mean)
+    on them.
 
-    Returns:
-         A dictionary containing summary statistics computed per
-         marker over all trackers.
-         A dictionary containing summary statistics computed
-         per tracker.
+    This means, per session, we compute how often a marker applies and how many
+    user turns preceding a relevant marker application on average, in that session.
+
+    (3) Number of user turns preceding relevant events - over all sessions:
+
+    Here, for each marker, we consider all relevant events where a marker applies
+    *in any of the sessions*. Then, we again calculate basic statistics over the
+    respective number of user turns that precedes each of these events.
+
+    This means, we compute to how many events the marker applies in total and we
+    compute an estimate of the expected number of user turns preceding that
+    precede an (relevant) event where a marker applies.
     """
-    per_marker_stats: Dict[str, Union[int, MarkerStats]] = {
-        "num_trackers": len(multi_tracker_markers)
-    }
-    per_tracker_stats = {}
-    per_marker_values = {}
 
-    for tracker in multi_tracker_markers:
-        # compute statistics per tracker
-        per_tracker_stats[tracker["tracker_ID"]] = compute_single_tracker_stats(tracker)
+    NO_MARKER = "-"
+    STAT_NUM_SESSIONS = "total_number_of_sessions"
+    STAT_NUM_SESSIONS_WHERE_APPLIES = (
+        "number_of_sessions_where_marker_applies_at_least_once"
+    )
+    STAT_PERCENTAGE_SESSIONS_WHERE_APPLIES = (
+        "percentage_of_sessions_where_marker_applies_at_least_once"
+    )
+
+    @staticmethod
+    def _add_num_user_turns_str_to(stat_name: Text) -> Text:
+        return f"{stat_name}(number of preceding user turns)"
+
+    ALL_SESSIONS = np.nan
+    ALL_SENDERS = "all"
+
+    def __init__(self) -> None:
+        """Creates a new marker statistic."""
+        # to ensure consistency of processed rows
+        self._marker_names = []
+
+        # (1) For collecting the per-session analysis:
+        # NOTE: we could stream these instead of collecting them...
+        self.session_results: Dict[Text, Dict[Text, List[Union[int, float]]]] = {}
+        self.session_identifier: List[Tuple[Text, int]] = []
 
-        for marker in tracker["markers"]:
-            # append raw values
-            per_marker_values.setdefault(marker["marker"], []).extend(
-                marker["num_preceding_user_turns"]
+        # (2) For the overall statistics:
+        self.num_preceding_user_turns_collected: Dict[Text, List[int]] = {}
+        self.count_if_applied_at_least_once: Dict[Text, int] = {}
+        self.num_sessions = 0
+
+    def process(
+        self,
+        extracted_markers: Dict[Text, List[EventMetaData]],
path: rasa/core/evaluation/marker_stats.py
text: @user1:
This is for a single dialogue/session right? Might be good to mention in the docs to disambiguate between functions over dialogues, trackers, and collections of trackers

@user1:
From what I understand the workflow is we feed dialogues into this one at a time, and then the state of this class tracks all of them for later output?

@author:
Oh good point - the `extracted_marker` and the comment is not so informative -- and you understand it exactly as I meant it :) 

@user2:
I understood it as per-session markers when I saw `sender_id` and `session_idx`, but I agree it's better to say explicitly what it does in the docstrings. 
Does it make sense to order the arugments as: 
```
  sender_id: Text,
  session_idx: int,
  extracted_markers: Dict[Text, List[EventMetaData]],
```
I think that would make it clearer (and consistent with the final output).

@author:
done

diff_hunk: @@ -1,143 +1,292 @@
-import os
-import sys
-
-if sys.version_info >= (3, 8):
-    from typing import TypedDict
-else:
-    from typing_extensions import TypedDict
-from typing import Dict, Text, Union, Any, List, Tuple
+from __future__ import annotations
+from typing import Dict, Text, Union, List, Tuple
 from pathlib import Path
-import json
+import io
+import csv
+
 import numpy as np
 
 
-class MarkerStats(TypedDict):
-    """A TypedDict for statistics computed over extracted markers."""
+from rasa.core.evaluation.marker_base import EventMetaData
 
-    n: int
-    mean: float
-    median: float
-    min: float
-    max: float
 
+def compute_statistics(
+    values: List[Union[float, int]]
+) -> Dict[Text, Union[int, np.float]]:
+    """Computes some statistics over the given numbers."""
+    return {
+        "count": len(values) if values else 0,
+        "mean": np.mean(values) if values else np.nan,
+        "median": np.median(values) if values else np.nan,
+        "min": min(values) if values else np.nan,
+        "max": max(values) if values else np.nan,
+    }
 
-def load_extracted_markers_json_file(path: Union[Text, Path]) -> List:
-    """Reads a json marker file.
 
-    Args:
-        path: path to a json file.
-    """
-    path = os.path.abspath(path)
-    with open(path) as json_file:
-        extracted_markers = json.load(json_file)
-        return extracted_markers
+class _CSVWriter:
+    """A csv writer."""
path: rasa/core/evaluation/marker_stats.py
text: @user1:
Do we need a CSVWriter wrapper class as it is not doing much at the moment? Why not use csv module directly?
```
csv_writer = CSVWriter(stream)
csv_writer.writerow(some_text)
# vs
csv_writer = csv.writer(stream)
csv_writer.writerow(some_text_
```

@author:
It was meant as a workaround because of the type of that csv_writer,  because I pass the csv_writer to some functions later (and hence need typing there). But in the solution above the BaseIO thing is also not right....

Hm... I guess I could replace this attempt by a Protocol. Then I can get rid of the need to type the stream 🤔 

@author:
done :)

diff_hunk: @@ -1,143 +1,292 @@
-import os
-import sys
-
-if sys.version_info >= (3, 8):
-    from typing import TypedDict
-else:
-    from typing_extensions import TypedDict
-from typing import Dict, Text, Union, Any, List, Tuple
+from __future__ import annotations
+from typing import Dict, Text, Union, List, Tuple
 from pathlib import Path
-import json
+import io
+import csv
+
 import numpy as np
 
 
-class MarkerStats(TypedDict):
-    """A TypedDict for statistics computed over extracted markers."""
+from rasa.core.evaluation.marker_base import EventMetaData
 
-    n: int
-    mean: float
-    median: float
-    min: float
-    max: float
 
+def compute_statistics(
+    values: List[Union[float, int]]
+) -> Dict[Text, Union[int, np.float]]:
+    """Computes some statistics over the given numbers."""
+    return {
+        "count": len(values) if values else 0,
+        "mean": np.mean(values) if values else np.nan,
+        "median": np.median(values) if values else np.nan,
+        "min": min(values) if values else np.nan,
+        "max": max(values) if values else np.nan,
+    }
 
-def load_extracted_markers_json_file(path: Union[Text, Path]) -> List:
-    """Reads a json marker file.
 
-    Args:
-        path: path to a json file.
-    """
-    path = os.path.abspath(path)
-    with open(path) as json_file:
-        extracted_markers = json.load(json_file)
-        return extracted_markers
+class _CSVWriter:
+    """A csv writer."""
 
+    def __init__(self, stream: io.TextIOBase) -> None:
+        self.table_writer = csv.writer(stream)
 
-def compute_summary_stats(data_points: Union[List[float], np.ndarray]) -> MarkerStats:
-    """Computes summary statistics for a given array.
+    def writerow(self, row: List[Text]) -> None:
+        """Write the given row.
 
-    Args:
-        data_points: can be a numpy array or a list of numbers.
+        Args:
+            row: the entries of a row as a list of strings
+        """
+        self.table_writer.writerow(row)
 
-    Returns:
-        A MarkerStats object containing size, mean, median, min, and max.
-        If the given array of data points is empty, it returns 0 for size, and
-        `np.nan` for every statistic.
-    """
-    if np.size(data_points) > 0:
-        stats: MarkerStats = {
-            "n": int(np.size(data_points)),
-            "mean": float(np.mean(data_points)),
-            "median": float(np.median(data_points)),
-            "min": int(np.min(data_points)),
-            "max": int(np.max(data_points)),
-        }
-        return stats
-    else:
-        empty_stats: MarkerStats = {
-            "n": 0,
-            "mean": np.nan,
-            "median": np.nan,
-            "min": np.nan,
-            "max": np.nan,
-        }
-        return empty_stats
-
-
-def compute_single_tracker_stats(
-    single_tracker_markers: Dict[str, Any]
-) -> Dict[str, MarkerStats]:
-    """Computes summary statistics for a single tracker.
-
-    Args:
-        single_tracker_markers: a dictionary containing the extracted
-        markers for one tracker.
-
-    Returns:
-        A dictionary containing statistics computed for each marker.
-    """
-    tracker_stats = {}
-    for marker in single_tracker_markers["markers"]:
-        tracker_stats[marker["marker"]] = compute_summary_stats(
-            marker["num_preceding_user_turns"]
-        )
-    return tracker_stats
 
+class MarkerStatistics:
+    """Computes some statistics on marker extraction results.
+
+    (1) Number of sessions where markers apply:
+
+    For each marker, we compute the total number (as well as the percentage) of the
+    sessions in which a marker applies at least once.
+    Moreover, we output the total number of sessions that were parsed.
 
-def compute_multi_tracker_stats(
-    multi_tracker_markers: List[Dict[str, Any]],
-) -> Tuple[Dict[str, Union[int, MarkerStats]], Dict[Any, Dict[str, MarkerStats]]]:
-    """Computes summary statistics for multiple trackers.
+    (2) Number of user turns preceding relevant events - per sessions:
 
-    Args:
-        multi_tracker_markers: a list of dictionaries each containing the
-        extracted markers for one tracker.
+    For each marker, we consider all relevant events where that marker applies.
+    Everytime a marker applies, we check how many user turns precede that event.
+    We collect all these numbers and compute basic statistics (e.g. count and mean)
+    on them.
 
-    Returns:
-         A dictionary containing summary statistics computed per
-         marker over all trackers.
-         A dictionary containing summary statistics computed
-         per tracker.
+    This means, per session, we compute how often a marker applies and how many
+    user turns preceding a relevant marker application on average, in that session.
path: rasa/core/evaluation/marker_stats.py
text: @user1:
```suggestion
    user turns precede a relevant marker application on average, in that session.
```