Skip to content

Propensity Score

OnlinePropensityScore

onlinecml.propensity.propensity_score.OnlinePropensityScore

Bases: Base

Online propensity score estimator wrapping any River classifier.

Estimates P(W=1 | X) incrementally, one observation at a time. The wrapped classifier is updated after each call to learn_one. Predicted probabilities are clipped to [clip_min, clip_max] to prevent extreme importance weights.

This class does NOT inherit from BaseOnlineEstimator because it is a helper component, not a CATE estimator. It is used internally by OnlineIPW, OnlineAIPW, and the meta-learners.

Parameters:

Name Type Description Default
classifier Classifier

Any River binary classifier with learn_one and predict_proba_one methods.

 required
clip_min float

Lower clip bound for predicted probabilities. Default 0.01.

 0.01
clip_max float

Upper clip bound for predicted probabilities. Default 0.99.

 0.99
Notes

Before any observations are seen, predict_one returns 0.5 (uniform prior). This means early IPW weights equal 2.0 regardless of treatment — the variance is high during warm-up (~50–100 obs).

River's predict_proba_one returns {True: p, False: 1-p} with boolean keys. This class accesses the probability via proba[True].

Examples:

>>> from river.linear_model import LogisticRegression
>>> ps = OnlinePropensityScore(LogisticRegression())
>>> ps.predict_one({"age": 30, "income": 50000})
0.5
>>> ps.learn_one({"age": 30, "income": 50000}, treatment=1)
>>> ps.n_seen
1
Source code in onlinecml/propensity/propensity_score.py 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 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
class OnlinePropensityScore(Base):
    """Online propensity score estimator wrapping any River classifier.

    Estimates P(W=1 | X) incrementally, one observation at a time. The
    wrapped classifier is updated after each call to ``learn_one``.
    Predicted probabilities are clipped to ``[clip_min, clip_max]`` to
    prevent extreme importance weights.

    This class does NOT inherit from ``BaseOnlineEstimator`` because it
    is a helper component, not a CATE estimator. It is used internally
    by ``OnlineIPW``, ``OnlineAIPW``, and the meta-learners.

    Parameters
    ----------
    classifier : river.base.Classifier
        Any River binary classifier with ``learn_one`` and
        ``predict_proba_one`` methods.
    clip_min : float
        Lower clip bound for predicted probabilities. Default 0.01.
    clip_max : float
        Upper clip bound for predicted probabilities. Default 0.99.

    Notes
    -----
    Before any observations are seen, ``predict_one`` returns 0.5
    (uniform prior). This means early IPW weights equal 2.0 regardless
    of treatment — the variance is high during warm-up (~50–100 obs).

    River's ``predict_proba_one`` returns ``{True: p, False: 1-p}``
    with boolean keys. This class accesses the probability via
    ``proba[True]``.

    Examples
    --------
    >>> from river.linear_model import LogisticRegression
    >>> ps = OnlinePropensityScore(LogisticRegression())
    >>> ps.predict_one({"age": 30, "income": 50000})
    0.5
    >>> ps.learn_one({"age": 30, "income": 50000}, treatment=1)
    >>> ps.n_seen
    1
    """

    def __init__(
        self,
        classifier: Classifier,
        clip_min: float = 0.01,
        clip_max: float = 0.99,
    ) -> None:
        self.classifier = classifier
        self.clip_min = clip_min
        self.clip_max = clip_max
        self._n_seen: int = 0

    def learn_one(self, x: dict, treatment: int) -> None:
        """Update the propensity model with one observation.

        Parameters
        ----------
        x : dict
            Feature dictionary for this observation.
        treatment : int
            Treatment indicator (0 = control, 1 = treated).
        """
        self.classifier.learn_one(x, bool(treatment))
        self._n_seen += 1

    def predict_one(self, x: dict) -> float:
        """Predict P(W=1 | X) for a single unit.

        Parameters
        ----------
        x : dict
            Feature dictionary for the unit.

        Returns
        -------
        float
            Estimated propensity score, clipped to ``[clip_min, clip_max]``.
            Returns 0.5 before any training observations are seen.
        """
        if self._n_seen == 0:
            return 0.5
        proba = self.classifier.predict_proba_one(x)
        p = proba.get(True, 0.5)
        return max(self.clip_min, min(self.clip_max, p))

    def ipw_weight(self, x: dict, treatment: int) -> float:
        """Compute the inverse probability weight for this observation.

        Parameters
        ----------
        x : dict
            Feature dictionary for the unit.
        treatment : int
            Treatment indicator (0 or 1).

        Returns
        -------
        float
            IPW weight: ``1/p`` for treated units, ``1/(1-p)`` for control.
        """
        p = self.predict_one(x)
        return 1.0 / p if treatment == 1 else 1.0 / (1.0 - p)

    def overlap_weight(self, x: dict, treatment: int) -> float:
        """Compute the overlap (trimming) weight for this observation.

        Overlap weights are bounded and proportional to the probability
        of being in the opposite treatment group, providing more stable
        estimates under near-positivity violations.

        Parameters
        ----------
        x : dict
            Feature dictionary for the unit.
        treatment : int
            Treatment indicator (0 or 1).

        Returns
        -------
        float
            Overlap weight: ``(1-p)`` for treated, ``p`` for control.
        """
        p = self.predict_one(x)
        return (1.0 - p) if treatment == 1 else p

    @property
    def n_seen(self) -> int:
        """Number of observations processed."""
        return self._n_seen

n_seen property

Number of observations processed.

ipw_weight(x, treatment)

Compute the inverse probability weight for this observation.

Parameters:

Name Type Description Default
x dict

Feature dictionary for the unit.

 required
treatment int

Treatment indicator (0 or 1).

 required

Returns:

Type Description
float

IPW weight: 1/p for treated units, 1/(1-p) for control.
Source code in onlinecml/propensity/propensity_score.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def ipw_weight(self, x: dict, treatment: int) -> float:
    """Compute the inverse probability weight for this observation.

    Parameters
    ----------
    x : dict
        Feature dictionary for the unit.
    treatment : int
        Treatment indicator (0 or 1).

    Returns
    -------
    float
        IPW weight: ``1/p`` for treated units, ``1/(1-p)`` for control.
    """
    p = self.predict_one(x)
    return 1.0 / p if treatment == 1 else 1.0 / (1.0 - p)

learn_one(x, treatment)

Update the propensity model with one observation.

Parameters:

Name Type Description Default
x dict

Feature dictionary for this observation.

 required
treatment int

Treatment indicator (0 = control, 1 = treated).

 required
Source code in onlinecml/propensity/propensity_score.py 
60
61
62
63
64
65
66
67
68
69
70
71
def learn_one(self, x: dict, treatment: int) -> None:
    """Update the propensity model with one observation.

    Parameters
    ----------
    x : dict
        Feature dictionary for this observation.
    treatment : int
        Treatment indicator (0 = control, 1 = treated).
    """
    self.classifier.learn_one(x, bool(treatment))
    self._n_seen += 1

overlap_weight(x, treatment)

Compute the overlap (trimming) weight for this observation.

Overlap weights are bounded and proportional to the probability of being in the opposite treatment group, providing more stable estimates under near-positivity violations.

Parameters:

Name Type Description Default
x dict

Feature dictionary for the unit.

 required
treatment int

Treatment indicator (0 or 1).

 required

Returns:

Type Description
float

Overlap weight: (1-p) for treated, p for control.
Source code in onlinecml/propensity/propensity_score.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def overlap_weight(self, x: dict, treatment: int) -> float:
    """Compute the overlap (trimming) weight for this observation.

    Overlap weights are bounded and proportional to the probability
    of being in the opposite treatment group, providing more stable
    estimates under near-positivity violations.

    Parameters
    ----------
    x : dict
        Feature dictionary for the unit.
    treatment : int
        Treatment indicator (0 or 1).

    Returns
    -------
    float
        Overlap weight: ``(1-p)`` for treated, ``p`` for control.
    """
    p = self.predict_one(x)
    return (1.0 - p) if treatment == 1 else p

predict_one(x)

Predict P(W=1 | X) for a single unit.

Parameters:

Name Type Description Default
x dict

Feature dictionary for the unit.

 required

Returns:

Type Description
float

Estimated propensity score, clipped to [clip_min, clip_max]. Returns 0.5 before any training observations are seen.
Source code in onlinecml/propensity/propensity_score.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def predict_one(self, x: dict) -> float:
    """Predict P(W=1 | X) for a single unit.

    Parameters
    ----------
    x : dict
        Feature dictionary for the unit.

    Returns
    -------
    float
        Estimated propensity score, clipped to ``[clip_min, clip_max]``.
        Returns 0.5 before any training observations are seen.
    """
    if self._n_seen == 0:
        return 0.5
    proba = self.classifier.predict_proba_one(x)
    p = proba.get(True, 0.5)
    return max(self.clip_min, min(self.clip_max, p))