Data and features

The point-in-time store, the panel builder, the leakage guards, and the feature transforms.

FeatureSnapshotStore

FeatureSnapshotStore(root: Path | str)

Append-only, point-in-time store of per-market feature snapshots.

root instance-attribute

root = Path(root)

append

append(family: str, snapshots: list[dict[str, Any]]) -> int

Append point-in-time snapshots for a market family.

Each snapshot dict must contain market_id and captured_at (an ISO-8601 timestamp) plus any feature_* values. Existing (market_id, captured_at) rows are preserved and duplicates dropped, so the store is append-only and idempotent.

Returns:

Type	Description
int	The number of rows in the family store after the append.

Raises:

Type	Description
ValueError	if a snapshot is missing market_id or captured_at.

load

load(family: str) -> pd.DataFrame

Load all snapshots for a family (empty DataFrame if none).

as_of

as_of(family: str, cutoff: str) -> pd.DataFrame

Return the latest snapshot per market captured strictly before cutoff.

This is the no-lookahead read: for each market, the most recent snapshot whose captured_at < cutoff. Markets whose only snapshots are at or after cutoff are excluded (their features were not yet observable).

Parameters:

Name	Type	Description	Default
family	str	Market family.	required
cutoff	str	ISO-8601 timestamp; snapshots at or after this are excluded.	required

Returns:

Type	Description
DataFrame	One row per market, or an empty DataFrame if nothing qualifies.

build_panel

build_panel(
    rows: Iterable[
        dict[str, Any] | list[dict[str, Any]] | None
    ],
    check_leakage: bool = False,
    max_auc: float = 0.97,
) -> pd.DataFrame

Build a validated training panel from plugin rows.

Parameters:

Name	Type	Description	Default
rows	Iterable[dict[str, Any] | list[dict[str, Any]] | None]	Iterable of row dicts, or of lists of row dicts, so the output of either MarketPlugin.build_training_row (one row or None) or MarketPlugin.build_training_rows (a market's rows) can be passed per market unchanged. None entries and empty lists are dropped.	required
check_leakage	bool	If True, run check_no_leakage on the assembled panel and raise LeakageError if any feature leaks the label. On a one-row-per-bin panel the label is the per-row outcome outcome_label == winning_label, the one the backtest trains on (see docs/leakage.md).	False
max_auc	float	Threshold forwarded to the leakage check.	0.97

A panel whose rows carry neither price_ts nor decision_ts is built, but logs a single WARNING (logger pmlab.data.panel_builder, once per call): its market_price cannot be checked against the decision it informs. That is the case for every plugin written before the check existed, for the sports_f1 plugin, and for build_panel_from_snapshots. Silence it by stamping the rows, or filter the logger.

Returns:

Type	Description
DataFrame	A schema-validated panel DataFrame with float feature columns.

Raises:

Type	Description
PanelSchemaError	on empty input, missing required columns, no feature_* column, malformed decision_date, or a feature that cannot be coerced to numeric.
LeakageError	if check_leakage is True and leakage is detected, or, always, if the rows carry price_ts and decision_ts and a price was observed at or after its decision, if they carry market_close_ts and a decision was taken at or after it, or if they carry only one of price_ts and decision_ts (see check_price_timing).

build_panel_from_snapshots

build_panel_from_snapshots(
    store: FeatureSnapshotStore,
    family: str,
    resolutions: DataFrame,
    outcome_label: str = "Yes",
    price_col: str | None = None,
) -> pd.DataFrame

Assemble a no-lookahead training panel from snapshots + resolutions.

Parameters:

Name	Type	Description	Default
store	FeatureSnapshotStore	The snapshot store to read point-in-time features from.	required
family	str	Market family to read.	required
resolutions	DataFrame	DataFrame with market_id, decision_date (YYYY-MM-DD) and winning_label — one row per resolved market.	required
outcome_label	str	Value for the panel's outcome_label column (the bin the row's features/price describe). Defaults to "Yes".	'Yes'
price_col	str | None	Feature column to copy into market_price. If None (the default), the first feature_* column is used; market_price must be numeric so the panel is backtestable.	None

Returns:

Type	Description
DataFrame	A training panel (one row per resolvable market) with market_id,
DataFrame	decision_date, outcome_label, winning_label, market_price
DataFrame	and the snapshot's feature_* columns. Markets with no snapshot before
DataFrame	their decision date, or absent from the store, are dropped.
DataFrame	The panel is untimed: market_price is a snapshot feature, taken strictly
DataFrame	before decision_date by as_of, but the panel carries no price_ts or
DataFrame	decision_ts, so check_price_timing has nothing to verify and build_panel
DataFrame	warns about it. One row per market, labelled with a single outcome_label.

Raises:

Type	Description
ValueError	if resolutions is missing a required column.

PanelSchemaError

Bases: Exception

Raised when plugin rows do not form a valid training panel.

check_no_leakage

check_no_leakage(
    df: DataFrame,
    label_col: str | None = None,
    max_auc: float = 0.97,
    raise_on_leak: bool = True,
) -> LeakageReport

Check that no feature_* column leaks the label.

Each feature is scored by how well it alone separates a binary target, measured as max(auc, 1 - auc) so leakage toward either class is caught. Features scoring above max_auc are considered leaking.

The target is what the model learns from. For a per-bin panel (a market is several rows, one per outcome bin, at one decision date: what build_training_rows returns) it is the per-row outcome outcome_label == winning_label, exactly the label rolling_origin_eval trains on. winning_label itself is the same for every bin of a market, so scoring against it asks which market resolved to the rarest label, and an honest forecast answers that well. For any other panel (one row per market) the target is winning_label, as it always was. LeakageReport.target says which.

A feature that is the same for every bin of a market scores 0.5 against the per-row outcome, by construction: it cannot tell a market's winning bin from its losing ones. So that a market-level copy of the realised value does not hide there, such a feature is also scored by its distance to each bin (0 inside the bin), read from the numbers or ranges in outcome_label ("18°C", "28-30°C", "<=14°C", "38°C or above"). A copy is inside the winning bin and outside every other, and scores about 1.0; an honest forecast is only sometimes nearest to it. The feature's score is the larger of the two, and it is listed in LeakageReport.bin_relative. Labels that are not numbers leave the feature at 0.5, as before. The feature must be in the labels' unit to be measured against them.

Parameters:

Name	Type	Description	Default
df	DataFrame	Panel/trade frame containing feature_* columns and the label.	required
label_col	str | None	Score against this column, as given, whatever the panel's shape. Default None: choose the target as described above (winning_label, or the per-row outcome for a per-bin panel).	None
max_auc	float	Separation score above which a feature is flagged (default 0.97).	0.97
raise_on_leak	bool	If True (default), raise LeakageError when any feature leaks; otherwise return the report for inspection.	True

Returns:

Type	Description
LeakageReport	A LeakageReport. When raise_on_leak is True and leakage is
LeakageReport	found, raises instead of returning.

Raises:

Type	Description
ValueError	if the label column (label_col, or winning_label by default) is absent.
LeakageError	if leakage is found and raise_on_leak is True.

LeakageReport dataclass

LeakageReport(
    scores: dict[str, float] = dict(),
    leaked: list[str] = list(),
    max_auc: float = 0.97,
    target: str = _LABEL_COL,
    bin_relative: list[str] = list(),
)

Per-feature leakage diagnostics.

Attributes:

Name	Type	Description
scores	dict[str, float]	feature column → separation score in [0.5, 1.0] (|AUC| folded so that a feature which perfectly predicts either class scores 1.0).
leaked	list[str]	feature columns whose score exceeds max_auc.
max_auc	float	the threshold used.
target	str	what the features were scored against: a column name, or "outcome_label == winning_label" for the per-row outcome of a per-bin panel.
bin_relative	list[str]	features that are the same for every bin of a market, which were scored by their distance to each bin (their scores entry is that score). Empty unless the panel is per-bin and its outcome_label values read as numbers or ranges.

scores class-attribute instance-attribute

scores: dict[str, float] = field(default_factory=dict)

leaked class-attribute instance-attribute

leaked: list[str] = field(default_factory=list)

max_auc class-attribute instance-attribute

max_auc: float = 0.97

target class-attribute instance-attribute

target: str = _LABEL_COL

bin_relative class-attribute instance-attribute

bin_relative: list[str] = field(default_factory=list)

LeakageError

Bases: Exception

Raised when a feature column separates the label too well to be legitimate.

check_price_timing

check_price_timing(
    df: DataFrame,
    price_col: str = "price_ts",
    cutoff_col: str = "decision_ts",
    close_col: str = "market_close_ts",
    raise_on_leak: bool = True,
) -> PriceTimingReport

Check that every price was observed strictly before the decision that uses it.

"Strictly before" is the rule FeatureSnapshotStore.as_of applies to features (captured_at < cutoff), so a price and the features beside it are held to one standard. Timestamps are compared in UTC, so offsets do not matter.

A row with a decision time but no readable price time cannot be shown to be in time, and is reported (and raised on) like a late one.

A price observed before the decision is still no decision-time price if the decision was taken after the market had closed: the outcome was known, and so was the price that reflects it. When the frame has a close_col (default market_close_ts), a row whose decision is at or after its close is reported, and raised on, too. A row whose close is missing or unreadable cannot be judged against it; it is counted in close_unknown and not refused (a plugin that wants it refused excludes the market itself, as weather_tmax does). Without the column no close is checked.

The guard applies to a frame that has both price_col and cutoff_col, and a frame that has neither is the documented unchecked case: a plugin that never stamps its prices (every plugin written before this guard) is left alone. A frame with exactly one of them is neither, and is refused: a price time with no decision to compare it to, or a decision with no price time, would otherwise read as an untimed panel and pass.

Parameters:

Name	Type	Description	Default
df	DataFrame	Panel containing price_col and cutoff_col (ISO-8601 or datetimes).	required
price_col	str	When the row's market_price was observed.	'price_ts'
cutoff_col	str	The decision time the row informs.	'decision_ts'
close_col	str	When the market closed (or its outcome became known), if the frame says.	'market_close_ts'
raise_on_leak	bool	If True (default), raise LeakageError on a late or untimed row; otherwise return the report.	True

Returns:

Type	Description
PriceTimingReport	A PriceTimingReport.

Raises:

Type	Description
LeakageError	a price was observed at or after its decision, has no time, or the decision was at or after the market's close, and raise_on_leak is True; or, whatever raise_on_leak says, the frame has exactly one of price_col and cutoff_col (a half-timed panel has nothing to report on).

PriceTimingReport dataclass

PriceTimingReport(
    n_rows: int = 0,
    n_checked: int = 0,
    late: int = 0,
    untimed: int = 0,
    after_close: int = 0,
    close_unknown: int = 0,
    late_index: list[Any] = list(),
    untimed_index: list[Any] = list(),
    after_close_index: list[Any] = list(),
)

What check_price_timing found.

Attributes:

Name	Type	Description
n_rows	int	rows in the frame.
n_checked	int	rows the guard applied to: all of them when the frame carries both timing columns, none when it carries neither.
late	int	rows whose price was observed at or after the decision.
untimed	int	rows with a decision time but no readable price time.
after_close	int	rows whose decision was taken at or after the market's close (only when the frame has a close column).
close_unknown	int	rows with a decision time but no readable close, which the close check cannot judge (counted, not refused).
late_index	list[Any]	index labels of the late rows.
untimed_index	list[Any]	index labels of the untimed rows.
after_close_index	list[Any]	index labels of the rows decided after the close.

n_rows class-attribute instance-attribute

n_rows: int = 0

n_checked class-attribute instance-attribute

n_checked: int = 0

late class-attribute instance-attribute

late: int = 0

untimed class-attribute instance-attribute

untimed: int = 0

after_close class-attribute instance-attribute

after_close: int = 0

close_unknown class-attribute instance-attribute

close_unknown: int = 0

late_index class-attribute instance-attribute

late_index: list[Any] = field(default_factory=list)

untimed_index class-attribute instance-attribute

untimed_index: list[Any] = field(default_factory=list)

after_close_index class-attribute instance-attribute

after_close_index: list[Any] = field(default_factory=list)

add_lags

add_lags(
    df: DataFrame,
    cols: list[str],
    lags: list[int],
    group_by: str | None = None,
    fill_value: float = float("nan"),
) -> pd.DataFrame

Add lagged copies of columns, named {col}_lag{n}.

Parameters:

Name	Type	Description	Default
df	DataFrame	The frame; not modified (a copy is returned).	required
cols	list[str]	Columns to lag.	required
lags	list[int]	How many rows to shift each column by.	required
group_by	str | None	Shift within each group of this column, so the first row of one entity does not see the last row of the previous one. Pass it whenever the frame holds several entities (cities, markets).	None
fill_value	float	What the leading rows, which have no earlier row, get.	float('nan')

Returns:

Type	Description
DataFrame	A copy of df with len(cols) * len(lags) extra columns.

Raises:

Type	Description
KeyError	A column in cols is not in df.

add_rolling_stats

add_rolling_stats(
    df: DataFrame,
    cols: list[str],
    windows: list[int],
    stats: list[
        Literal["mean", "std", "min", "max", "median"]
    ]
    | None = None,
    group_by: str | None = None,
    min_periods: int = 1,
) -> pd.DataFrame

Add rolling-window statistics of columns, named {col}_roll{window}_{stat}.

Windows follow the frame's row order (sort by time within each group first) and include the current row. With group_by each row's window is computed over the rows of its own group and lands on that same row, whether or not the frame is sorted by the group key.

Parameters:

Name	Type	Description	Default
df	DataFrame	The frame; not modified (a copy is returned).	required
cols	list[str]	Columns to summarise.	required
windows	list[int]	Window sizes, in rows.	required
stats	list[Literal['mean', 'std', 'min', 'max', 'median']] | None	Which statistics: "mean", "std", "min", "max", "median". Default ["mean", "std"].	None
group_by	str | None	Compute the windows within each group of this column.	None
min_periods	int	Minimum rows in a window for it to produce a value (earlier rows are NaN).	1

Returns:

Type	Description
DataFrame	A copy of df with len(cols) * len(windows) * len(stats) extra columns.

Raises:

Type	Description
KeyError	A column in cols is not in df.

encode_cyclical

encode_cyclical(
    series: Series, period: float
) -> tuple[pd.Series, pd.Series]

Encode a cyclical value (hour, month, day of year) as a point on the unit circle.

The sine and cosine put the end of the cycle next to its start, which a plain integer cannot.

Parameters:

Name	Type	Description	Default
series	Series	The values; its name prefixes the new series (x if it has none).	required
period	float	The length of the cycle, e.g. 24 for hours or 12 for months.	required

Returns:

Type	Description
tuple[Series, Series]	(sin, cos), two series named {name}_sin and {name}_cos on series' index.

encode_onehot

encode_onehot(
    df: DataFrame,
    cols: list[str],
    drop_first: bool = True,
    dtype: type = float,
) -> pd.DataFrame

One-hot encode categorical columns; the originals are replaced by the indicator columns.

Fit nothing: the categories are read from df itself, so encode the training and live frames together, or use TargetEncoder when the live frame may hold categories the training one did not.

Parameters:

Name	Type	Description	Default
df	DataFrame	The frame; not modified (a copy is returned).	required
cols	list[str]	Columns to encode.	required
drop_first	bool	Drop each column's first category (avoids a redundant indicator).	True
dtype	type	The type of the indicator columns.	float

Returns:

Type	Description
DataFrame	df without cols and with one indicator column per remaining category.

Raises:

Type	Description
KeyError	A column in cols is not in df.

clip_outliers

clip_outliers(
    df: DataFrame,
    cols: list[str],
    method: Literal["iqr", "zscore"] = "iqr",
    iqr_factor: float = 3.0,
    zscore_threshold: float = 4.0,
) -> pd.DataFrame

Clip the outliers of columns to a bound computed from the column itself.

Parameters:

Name	Type	Description	Default
df	DataFrame	The frame; not modified (a copy is returned).	required
cols	list[str]	Columns to clip.	required
method	Literal['iqr', 'zscore']	"iqr" clips to [Q1 - iqr_factor * IQR, Q3 + iqr_factor * IQR]; "zscore" clips to mean +/- zscore_threshold * std.	'iqr'
iqr_factor	float	The IQR multiple, for method="iqr".	3.0
zscore_threshold	float	The number of standard deviations, for method="zscore".	4.0

Returns:

Type	Description
DataFrame	A copy of df with the columns clipped in place (same names). The bounds are computed on
DataFrame	df as given, so clip the training frame, not the live one, if the bounds must not see
DataFrame	live data.

Raises:

Type	Description
KeyError	A column in cols is not in df.

add_missing_indicators

add_missing_indicators(
    df: DataFrame, cols: list[str], suffix: str = "_isna"
) -> pd.DataFrame

Add one int8 0/1 column per entry of cols flagging missing values.

{col}{suffix} is 1 where col is NaN/None/NaT and 0 elsewhere. Row-local (it looks at no other row), so it cannot leak. Apply it before imputing the column, otherwise the missingness is already gone. The input frame is never mutated.

add_interactions

add_interactions(
    df: DataFrame,
    pairs: list[tuple[str, str]],
    op: InteractionOp = "mul",
) -> pd.DataFrame

Add one pairwise interaction column per (a, b) in pairs.

op is one of "mul" (a * b, column a_x_b), "div" (a / b, a_div_b), "add" (a + b, a_plus_b) or "sub" (a - b, a_minus_b). Names follow the order of the pair, so (a, b) and (b, a) are different columns. Division by zero gives NaN (never inf), so downstream models and scalers see a missing value rather than an infinity. Row-local, so it cannot leak. The input frame is never mutated.

TargetEncoder dataclass

TargetEncoder(
    col: str,
    target: str,
    smoothing: float = 10.0,
    cv: CVScheme = "time",
    n_folds: int = 5,
    time_col: str | None = None,
    group_col: str | None = None,
    seed: int = 0,
    suffix: str = "_te",
    output_col: str | None = None,
)

Configuration of a leakage-safe target encoding of one categorical column.

See the module docstring for the leakage contract. col is encoded into output_col (default f"{col}{suffix}") using target. Give output_col a feature_ prefix to have rolling_origin_eval pick the column up as a feature; the same name is used on the training path and on the fitted (live) path.

n_folds is capped by the number of distinct timestamps / groups / rows, so a tiny frame degrades to fewer folds rather than failing.

col instance-attribute

col: str

target instance-attribute

target: str

smoothing class-attribute instance-attribute

smoothing: float = 10.0

cv class-attribute instance-attribute

cv: CVScheme = 'time'

n_folds class-attribute instance-attribute

n_folds: int = 5

time_col class-attribute instance-attribute

time_col: str | None = None

group_col class-attribute instance-attribute

group_col: str | None = None

seed class-attribute instance-attribute

seed: int = 0

suffix class-attribute instance-attribute

suffix: str = '_te'

output_col class-attribute instance-attribute

output_col: str | None = None

encoded_col property

encoded_col: str

Name of the column that holds the encoding: output_col, else col + suffix.

assign_folds

assign_folds(df: DataFrame) -> npt.NDArray[np.int64]

Fold id (0..K-1) of every row of df, positionally. Deterministic given seed.

fit_transform

fit_transform(df: DataFrame) -> pd.DataFrame

Encode the training rows out-of-fold; returns a copy with output_col added.

The row's own target (and every target of its own fold) is never used, and with cv="time" neither is any target at the same or a later timestamp. df is not mutated. Do not use this on live rows: use fit(train).transform(live).

fit

fit(df: DataFrame) -> FittedTargetEncoder

Compute the full-train statistics, for encoding new/live rows at inference time.

FittedTargetEncoder dataclass

FittedTargetEncoder(
    col: str,
    output_col: str,
    smoothing: float,
    prior: float,
    nan_value: float,
    mapping: dict[Any, float],
)

Full-train smoothed statistics, applied to new rows. Build it with TargetEncoder.fit.

mapping is read-only by convention (it is a plain dict so the object pickles).

col instance-attribute

col: str

output_col instance-attribute

output_col: str

smoothing instance-attribute

smoothing: float

prior instance-attribute

prior: float

nan_value instance-attribute

nan_value: float

mapping instance-attribute

mapping: dict[Any, float]

transform

transform(df: DataFrame) -> pd.DataFrame

Return a copy of df with output_col added; df is not mutated.

Unseen categories get prior; missing categories get nan_value. The target column is not needed (nor read) here, so this works on live rows.