Skip to content

Winnower

The Winnower is the final stage of the pipeline. It prepares held-out timelines for evaluation: it splits each timeline at a configurable cut-point into a past portion (the context a model is given) and a future portion (what the model must predict), and it attaches outcome labels. Its behavior is driven by a winnowing config (winnowing.yaml).

The name is apt — the winnower also filters out subjects that cannot be fairly evaluated, for example timelines that end before the outcome horizon is reached.

What it produces

Running the winnower (via save_all) writes one file per configured split to the processed-data directory:

  • {split}_for_inference.parquet — the winnowed timelines with past/future splits and outcome flags (defaults to held_out).

How it works

prepare_winnowed_frame chains these steps:

  1. load_frame — load the tokenized timelines for the requested split and compute elapsed times.
  2. run_thresholding — compute last_valid, the cut-point between past and future. This is derived either from an elapsed-time horizon or from the first occurrence of a token of interest. Timelines that never reach the cut-point are dropped.
  3. add_outcome_flags — for each configured outcome token, add *_past and *_future boolean labels (for example DSCG//expired_past and DSCG//expired_future), so you can tell whether an outcome truly falls within the prediction window.

!!! note "Depends on tokenizer artifacts" The winnower reads the tokenizer.yaml written by the Tokenizer to resolve outcome-token patterns against the learned vocabulary, so tokenization must run first.

prepares held-out data for evaluation, adding flags to disqualify certain subjects from evaluation

Winnower

Bases: Configurable

filters held-out timelines for evaluation; assigns flags to disqualify certain subjects from evaluation, e.g. those whose timelines ends prior to the outcome horizon

Source code in src/cocoa/winnower.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class Winnower(Configurable):
    """
    filters held-out timelines for evaluation;
    assigns flags to disqualify certain subjects from evaluation,
    e.g. those whose timelines ends prior to the outcome horizon
    """

    default_file = "winnowing.yaml"

    def __init__(
        self,
        winnowing_cfg: pathlib.Path | str = None,
        processed_data_home: pathlib.Path | str = None,
        is_training: bool = True,
        **kwargs,
    ):
        super().__init__(winnowing_cfg, **kwargs)
        self.processed_data_home = (
            pathlib.Path(processed_data_home).expanduser().resolve()
        )
        self.tkzr_cfg = OmegaConf.load(self.processed_data_home / "tokenizer.yaml")
        self.grokked_outcome_tokens = [
            x
            for x in self.tkzr_cfg.lookup.keys()
            if any(fnmatch.fnmatch(x, p) for p in self.cfg.outcome_tokens)
        ]
        self.rng = np.random.default_rng(seed=42)

        self.logger.info("Winnower initialized...")
        self.logger.info(f"{self.processed_data_home=}")
        self.logger.info(
            f"Processed expressions to generate {self.grokked_outcome_tokens=}"
        )

    def load_frame(self, split="held_out") -> pl.LazyFrame:
        """
        loads held_out timelines, and performs some preliminary calculations;
        these are lazily evaluated, so only completed if used
        """
        return (
            pl.scan_parquet(self.processed_data_home / "tokens_times.parquet")
            .join(
                pl.scan_parquet(self.processed_data_home / "subject_splits.parquet"),
                on="subject_id",
                validate="1:1",
            )
            .filter(pl.col("split") == split)
            .drop("split")
            .with_columns(
                s_elapsed=pl.col("times").list.eval(
                    (pl.element() - pl.element().first()).dt.total_seconds()
                )
            )
            .with_columns(s_total_duration=pl.col("s_elapsed").list.last())
        )

    def run_thresholding(self, df: pl.LazyFrame) -> pl.LazyFrame:
        """
        evaluates configurable criteria for establishing a cut-point "last_valid";
        drops timelines that do not reach that point
        """
        if "horizon_s" in self.cfg or "duration_s" in self.cfg.get("threshold", {}):
            # run duration-based thresholding
            horizon_s = self.cfg.get("horizon_s", self.cfg.threshold.duration_s)
            return df.filter(pl.col("s_total_duration") > horizon_s).with_columns(
                last_valid=pl.col("s_elapsed")
                .list.eval(pl.element() < horizon_s)
                .list.sum()
            )
        elif "first_occurrence" in self.cfg.get("threshold", {}):
            # run first-occurrence-based thresholding
            toi = self.tkzr_cfg.lookup[self.cfg.threshold.first_occurrence]
            return df.filter(pl.col("tokens").list.contains(toi)).with_columns(
                last_valid=pl.col("tokens")
                .list.eval(pl.element() == toi)
                .list.arg_max()
                + pl.lit(1)
                # place the triggering token into the past; it is known
            )
        else:
            raise NotImplementedError("Please check the thresholding configuration.")

    def add_outcome_flags(self, df: pl.LazyFrame) -> pl.LazyFrame:
        """
        adds boolean flags for each outcome token and tense,
        e.g. DSCG//expired_past, DSCG//expired_future
        """
        df = df.with_columns(
            tokens_past=pl.col("tokens").list.head("last_valid"),
            s_elapsed_past=pl.col("s_elapsed").list.head("last_valid"),
            tokens_future=pl.col("tokens").list.tail(
                pl.col("tokens").list.len() - pl.col("last_valid")
            ),
        )  # split into past and future
        if "horizon_after_threshold_s" in self.cfg:
            df = (
                df.with_columns(
                    s_elapsed_thresh=pl.col("times")
                    .list.tail(
                        pl.col("tokens_future").list.len() + 1
                    )  # include threshold time
                    .list.eval((pl.element() - pl.element().first()).dt.total_seconds())
                )
                .with_columns(
                    valid_future_count=pl.col("s_elapsed_thresh")
                    .list.eval(pl.element() <= self.cfg.horizon_after_threshold_s)
                    .list.sum()
                    - pl.lit(1)  # threshold token was counted, drop it
                )
                .with_columns(
                    tokens_future=pl.col("tokens_future").list.head(
                        "valid_future_count"
                    )
                )
            )
        return df.with_columns(
            **{
                f"{t}_{tense}": pl.col(f"tokens_{tense}").list.contains(
                    self.tkzr_cfg.lookup[t]
                )
                for t in self.grokked_outcome_tokens
                for tense in ("past", "future")
            }
        )

    def prepare_winnowed_frame(self, split="held_out") -> pl.LazyFrame:
        """loads held-out data, splits at time threshold, and prepares labels"""
        return (
            self.load_frame(split=split)
            .pipe(self.run_thresholding)
            .pipe(self.add_outcome_flags)
        )

    def save_all(self, verbose: bool = False):
        """grabs winnowed frame, prints summary stats if requested, and saves it"""
        for split in self.cfg.get("splits", ["held_out"]):
            df = self.prepare_winnowed_frame(split=split)
            df.sink_parquet(
                self.processed_data_home / f"{split}_for_inference.parquet",
                engine="streaming",
            )
            if verbose:
                self.logger.info(f"Prepared split {split} for inference:")
                self.logger.summarize_thresholded(df, self.grokked_outcome_tokens)

add_outcome_flags(df)

adds boolean flags for each outcome token and tense, e.g. DSCG//expired_past, DSCG//expired_future

Source code in src/cocoa/winnower.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def add_outcome_flags(self, df: pl.LazyFrame) -> pl.LazyFrame:
    """
    adds boolean flags for each outcome token and tense,
    e.g. DSCG//expired_past, DSCG//expired_future
    """
    df = df.with_columns(
        tokens_past=pl.col("tokens").list.head("last_valid"),
        s_elapsed_past=pl.col("s_elapsed").list.head("last_valid"),
        tokens_future=pl.col("tokens").list.tail(
            pl.col("tokens").list.len() - pl.col("last_valid")
        ),
    )  # split into past and future
    if "horizon_after_threshold_s" in self.cfg:
        df = (
            df.with_columns(
                s_elapsed_thresh=pl.col("times")
                .list.tail(
                    pl.col("tokens_future").list.len() + 1
                )  # include threshold time
                .list.eval((pl.element() - pl.element().first()).dt.total_seconds())
            )
            .with_columns(
                valid_future_count=pl.col("s_elapsed_thresh")
                .list.eval(pl.element() <= self.cfg.horizon_after_threshold_s)
                .list.sum()
                - pl.lit(1)  # threshold token was counted, drop it
            )
            .with_columns(
                tokens_future=pl.col("tokens_future").list.head(
                    "valid_future_count"
                )
            )
        )
    return df.with_columns(
        **{
            f"{t}_{tense}": pl.col(f"tokens_{tense}").list.contains(
                self.tkzr_cfg.lookup[t]
            )
            for t in self.grokked_outcome_tokens
            for tense in ("past", "future")
        }
    )

load_frame(split='held_out')

loads held_out timelines, and performs some preliminary calculations; these are lazily evaluated, so only completed if used

Source code in src/cocoa/winnower.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def load_frame(self, split="held_out") -> pl.LazyFrame:
    """
    loads held_out timelines, and performs some preliminary calculations;
    these are lazily evaluated, so only completed if used
    """
    return (
        pl.scan_parquet(self.processed_data_home / "tokens_times.parquet")
        .join(
            pl.scan_parquet(self.processed_data_home / "subject_splits.parquet"),
            on="subject_id",
            validate="1:1",
        )
        .filter(pl.col("split") == split)
        .drop("split")
        .with_columns(
            s_elapsed=pl.col("times").list.eval(
                (pl.element() - pl.element().first()).dt.total_seconds()
            )
        )
        .with_columns(s_total_duration=pl.col("s_elapsed").list.last())
    )

prepare_winnowed_frame(split='held_out')

loads held-out data, splits at time threshold, and prepares labels

Source code in src/cocoa/winnower.py 
143
144
145
146
147
148
149
def prepare_winnowed_frame(self, split="held_out") -> pl.LazyFrame:
    """loads held-out data, splits at time threshold, and prepares labels"""
    return (
        self.load_frame(split=split)
        .pipe(self.run_thresholding)
        .pipe(self.add_outcome_flags)
    )

run_thresholding(df)

evaluates configurable criteria for establishing a cut-point "last_valid"; drops timelines that do not reach that point

Source code in src/cocoa/winnower.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def run_thresholding(self, df: pl.LazyFrame) -> pl.LazyFrame:
    """
    evaluates configurable criteria for establishing a cut-point "last_valid";
    drops timelines that do not reach that point
    """
    if "horizon_s" in self.cfg or "duration_s" in self.cfg.get("threshold", {}):
        # run duration-based thresholding
        horizon_s = self.cfg.get("horizon_s", self.cfg.threshold.duration_s)
        return df.filter(pl.col("s_total_duration") > horizon_s).with_columns(
            last_valid=pl.col("s_elapsed")
            .list.eval(pl.element() < horizon_s)
            .list.sum()
        )
    elif "first_occurrence" in self.cfg.get("threshold", {}):
        # run first-occurrence-based thresholding
        toi = self.tkzr_cfg.lookup[self.cfg.threshold.first_occurrence]
        return df.filter(pl.col("tokens").list.contains(toi)).with_columns(
            last_valid=pl.col("tokens")
            .list.eval(pl.element() == toi)
            .list.arg_max()
            + pl.lit(1)
            # place the triggering token into the past; it is known
        )
    else:
        raise NotImplementedError("Please check the thresholding configuration.")

save_all(verbose=False)

grabs winnowed frame, prints summary stats if requested, and saves it

Source code in src/cocoa/winnower.py 
151
152
153
154
155
156
157
158
159
160
161
def save_all(self, verbose: bool = False):
    """grabs winnowed frame, prints summary stats if requested, and saves it"""
    for split in self.cfg.get("splits", ["held_out"]):
        df = self.prepare_winnowed_frame(split=split)
        df.sink_parquet(
            self.processed_data_home / f"{split}_for_inference.parquet",
            engine="streaming",
        )
        if verbose:
            self.logger.info(f"Prepared split {split} for inference:")
            self.logger.summarize_thresholded(df, self.grokked_outcome_tokens)