Modeling¶
Forecasters, calibration, diagnostics, and the champion: the published model, its manifest, and the registry that archives each promotion.
Bases: ABC
Abstract base class for all market forecasters.
abstractmethod
¶
predict_proba(X: DataFrame) -> np.ndarray
Return probability estimates.
LGBMForecaster(
objective: str = "binary", **lgbm_params: Any
)
Bases: MarketForecaster
LightGBM wrapper supporting binary and multiclass classification.
resolved_params() -> dict[str, Any]
The parameters LightGBM is actually given: the defaults, then the caller's overrides.
This is what a reproducibility record stores as the model config, so a later change of
a default here does not change what an old record means. The seed is part of it only
when the caller passed random_state (LightGBM otherwise uses its own fixed internal
seeds: deterministic, but not a value pmlab can record).
predict_proba(X: DataFrame) -> np.ndarray
Return probability estimates, shape (n_samples, n_classes).
classmethod
¶
load(path: Path) -> LGBMForecaster
Load a previously saved LGBMForecaster from path.
SklearnForecaster(
estimator: str
| ClassifierMixin = "logistic_regression",
**kwargs: Any,
)
Bases: MarketForecaster
sklearn-based market forecaster supporting named presets and custom estimators.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
estimator
|
str | ClassifierMixin
|
One of 'logistic_regression' (default), 'random_forest', 'gradient_boosting', or a pre-instantiated sklearn classifier. |
'logistic_regression'
|
**kwargs
|
Any
|
Extra kwargs forwarded to the preset constructor (ignored when a pre-instantiated estimator is passed). |
{}
|
Examples::
# Named preset
f = SklearnForecaster(estimator="random_forest", n_estimators=50)
# Custom estimator
from sklearn.tree import DecisionTreeClassifier
f = SklearnForecaster(estimator=DecisionTreeClassifier(max_depth=4))
EnsembleForecaster(
forecasters: Sequence[MarketForecaster],
weights: Sequence[float] | None = None,
)
Bases: MarketForecaster
Weighted-average ensemble of member forecasters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
forecasters
|
Sequence[MarketForecaster]
|
One or more fitted-or-unfitted |
required |
weights
|
Sequence[float] | None
|
Optional non-negative blend weights, one per member. Defaults to
equal weight. Normalized internally, so |
None
|
Example::
ens = EnsembleForecaster(
forecasters=[LGBMForecaster(), SklearnForecaster("random_forest")],
weights=[2.0, 1.0],
)
ens.fit(X_train, y_train)
proba = ens.predict_proba(X_test)
ConformalForecaster(
base: MarketForecaster,
alpha: float = 0.1,
calib_fraction: float = 0.3,
random_state: int = 0,
)
Bases: MarketForecaster
Wrap a MarketForecaster to emit calibrated prediction sets.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
base
|
MarketForecaster
|
The forecaster to wrap. |
required |
alpha
|
float
|
Miscoverage level in (0, 1). Target coverage is |
0.1
|
calib_fraction
|
float
|
Fraction of |
0.3
|
random_state
|
int
|
Seed for the calibration split. |
0
|
predict_proba passes through the base model's probabilities (so an
instance is still a drop-in forecaster); predict_set returns the
coverage-guaranteed sets.
fit(X: DataFrame, y: Series) -> None
Fit base on a proper-train split, calibrate qhat on the held-out split.
predict_proba(X: DataFrame) -> np.ndarray
Pass through the base model's probability matrix, shape (n, n_classes).
predict_set(X: DataFrame) -> list[set[object]]
Return one prediction set per row: classes with 1 - p_k <= qhat.
Every set is guaranteed non-empty (the argmax class always qualifies), so a downstream consumer never has to special-case the empty set. Label types are preserved (int, str, etc.) — categorical markets (e.g. F1 driver names) get their real labels back, not coerced integers.
classmethod
¶
load(path: Path) -> ConformalForecaster
Load a previously saved ConformalForecaster from path.
TunedForecaster(
build_fn: Callable[[dict[str, Any]], MarketForecaster],
param_space: Callable[[Trial], dict[str, Any]],
n_trials: int = 20,
metric: TuneMetric = "total_pnl",
random_state: int = 0,
stride: int = 10,
min_train_rows: int = 20,
flat_stake: float = 1.0,
taker_bps: float | None = None,
policy: TradePolicy | None = None,
costs: CostModel | None = None,
)
Bases: MarketForecaster
Optuna hyperparameter search scored by walk-forward backtest.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
build_fn
|
Callable[[dict[str, Any]], MarketForecaster]
|
|
required |
param_space
|
Callable[[Trial], dict[str, Any]]
|
|
required |
n_trials
|
int
|
Number of Optuna trials (> 0). |
20
|
metric
|
TuneMetric
|
What to maximize — |
'total_pnl'
|
random_state
|
int
|
Seed for Optuna's sampler (reproducible search). |
0
|
stride
|
int
|
Forwarded to |
10
|
min_train_rows
|
int
|
Forwarded to |
20
|
flat_stake
|
float
|
Forwarded to |
1.0
|
taker_bps
|
float | None
|
Legacy fee-only knob (default 30). Mutually exclusive with |
None
|
policy
|
TradePolicy | None
|
The decision rule every trial is scored under, a
|
None
|
costs
|
CostModel | None
|
The |
None
|
After tune(panel), the instance is a fitted forecaster (best params,
refit on the full panel) and exposes best_params_, best_score_,
and study_. tuning_config returns the objective it maximised, resolved,
for the reproducibility record (BacktestConfig.resolve(tuning=...)).
A trial that makes no trade scores -inf; if every trial does (a threshold nothing
clears), tune warns, because the search learned nothing and best_params_ is
just the first trial.
instance-attribute
¶
costs: CostModel = (
costs
if costs is not None
else CostModel()
if taker_bps is None
else CostModel(taker_bps=taker_bps)
)
tune(panel: DataFrame) -> TunedForecaster
Run the Optuna search over panel, then refit best on the full panel.
tuning_config() -> dict[str, Any]
The objective the search maximised, resolved and strict-JSON, for the record.
Pass it to BacktestConfig.resolve(tuning=...) so the reproducibility record of the
backtest says how its parameters were chosen: which metric, under which trade policy
and costs, walking forward how. Every default is spelled out. The sampler seed is not
in here: a run's seeds live in one place, resolve_seeds(seed, optuna=tuner.random_state).
best_score is None before tune and for a search that never traded.
fit(X: DataFrame, y: Series) -> None
Fit the current best model directly (bypasses search).
Available after tune; use tune for the walk-forward search. This
exists so a TunedForecaster still satisfies the MarketForecaster contract.
classmethod
¶
load(path: Path) -> TunedForecaster
Load a tuned forecaster. The Optuna study itself is not restored.
CalibratedForecaster(
base: MarketForecaster,
method: CalibrationMethod = "isotonic",
calib_fraction: float = 0.3,
random_state: int = 0,
)
Bases: MarketForecaster
Wrap a MarketForecaster and calibrate its probability output.
On fit, the data is split into a training part (fits the base model) and a
held-out calibration part (fits the calibrator on the base model's out-of-fold
scores). predict_proba then returns the calibrated matrix. Binary targets
calibrate the positive class; multiclass targets use one-vs-rest calibration
(MulticlassCalibrator), so F1 / political multi-outcome markets are
supported too.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
base
|
MarketForecaster
|
The forecaster to wrap. |
required |
method
|
CalibrationMethod
|
|
'isotonic'
|
calib_fraction
|
float
|
Fraction of |
0.3
|
random_state
|
int
|
Seed for the calibration split. |
0
|
fit(X: DataFrame, y: Series) -> None
Fit base model on a train split, calibrator on a held-out split.
predict_proba(X: DataFrame) -> np.ndarray
Return calibrated (n, n_classes) probabilities.
classmethod
¶
load(path: Path) -> CalibratedForecaster
Load a previously saved CalibratedForecaster from path.
IsotonicCalibrator()
SigmoidCalibrator()
Platt scaling — a 1-D logistic regression mapping raw scores to probabilities.
More sample-efficient than isotonic when calibration data is scarce, at the cost of assuming a sigmoidal distortion. Output is monotonic in the input.
MulticlassCalibrator(
method: CalibrationMethod = "isotonic",
)
One-vs-rest probability calibration for multiclass score matrices.
Fits an independent 1-D calibrator (isotonic or sigmoid) per class on the one-vs-rest problem "is this row class k?", then renormalizes the calibrated columns to sum to 1. This is the standard reduction sklearn uses for multiclass calibration, and it keeps each class's mapping monotonic.
Input to fit/transform is a (n, n_classes) probability matrix whose
columns are ordered by np.unique of the training labels.
fit(proba: ndarray, labels: ndarray) -> None
Fit one calibrator per class on the one-vs-rest target.
transform(proba: ndarray) -> np.ndarray
Calibrate each column, then renormalize rows to sum to 1.
classmethod
¶
load(path: Path) -> MulticlassCalibrator
Load a previously saved MulticlassCalibrator from path.
dataclass
¶
BrierDecomposition(
uncertainty: float,
resolution: float,
reliability: float,
brier_score: float,
skill_score: float,
n_samples: int,
)
brier_decomposition(
y_true: ArrayLike, y_prob: ArrayLike, n_bins: int = 10
) -> BrierDecomposition
Murphy (1973) Brier score decomposition.
reliability_data(
y_true: ArrayLike, y_prob: ArrayLike, n_bins: int = 10
) -> tuple[
NDArray[np.float64],
NDArray[np.float64],
NDArray[np.float64],
]
Reliability diagram data: (bin_centers, mean_predicted_prob, fraction_positive).
dataclass
¶
MulticlassBrier(
brier_score: float,
skill_score: float,
n_samples: int,
n_classes: int,
)
multiclass_brier(
y_true: ArrayLike, y_prob: ArrayLike
) -> MulticlassBrier
Multiclass Brier score with a climatology skill score.
y_true is a length-n vector of integer class indices (0..k-1); y_prob
is the (n, k) probability matrix. Skill is measured against the base-rate
(climatology) forecast that predicts the training class frequencies for every
row: skill = 1 - BS_model / BS_climatology.
reliability_data_multiclass(
y_true: ArrayLike, y_prob: ArrayLike, n_bins: int = 10
) -> list[
tuple[
NDArray[np.float64],
NDArray[np.float64],
NDArray[np.float64],
]
]
Per-class reliability data via one-vs-rest.
Returns one (bin_centers, mean_predicted_prob, fraction_positive) tuple
per class, in the column order of the probability matrix.
dataclass
¶
ChampionManifest(
model_name: str,
model_path: Path,
calibrator_path: Path | None,
gate: HoldoutGateResult,
published_at: str,
plugin_family: str,
costs: CostModel | None = None,
expectation: BacktestExpectation | None = None,
reproducibility: ReproducibilityRecord | None = None,
)
class-attribute
instance-attribute
¶
reproducibility: ReproducibilityRecord | None = None
classmethod
¶
publish(
model: MarketForecaster,
gate: HoldoutGateResult,
output_dir: Path,
plugin_family: str,
model_name: str = "champion",
calibrator: IsotonicCalibrator | None = None,
costs: CostModel | None = None,
expectation: BacktestExpectation | None = None,
reproducibility: ReproducibilityRecord | None = None,
) -> ChampionManifest
Publish a champion model.
HARD GATE: raises ValueError if gate.decision != "GO".
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
costs
|
CostModel | None
|
The cost model the gate was computed under. Recording it lets a live run assert it is trading with the same friction the promotion assumed; without it, live and backtest PnL can diverge for reasons that have nothing to do with the market. |
None
|
expectation
|
BacktestExpectation | None
|
The backtest's per-trade PnL and Brier statistics
( |
None
|
reproducibility
|
ReproducibilityRecord | None
|
The backtest's |
None
|
to_dict() -> dict[str, Any]
The champion.json payload: every field of the manifest, JSON-ready.
The one place a manifest is serialized. publish, ModelRegistry.record and
ModelRegistry.rollback all write this, so a field added to the manifest reaches every
copy of it, instead of being dropped by whichever writer was not updated (#33).
ChampionManifest.load is its inverse (the JSON key for expectation is
backtest_expectation, and the gate is stored as publish_gate).
classmethod
¶
load(json_path: Path) -> ChampionManifest
Load a ChampionManifest from champion.json.
get_allowed_segments() -> set[str]
Return segments where the gate result passes.
A champion promoted on an aggregate gate returns {"all"}: it was graded on
the whole trade log and allows every segment (see allows_segment).
allows_segment(segment: str) -> bool
May this champion trade segment? "all" among the allowed means yes.
ModelRegistry(registry_dir: Path | str)
Append-only, versioned store of promoted champions.
record(manifest: ChampionManifest) -> str
Archive a published champion as an immutable version.
Re-asserts the hard gate: refuses to record anything whose gate is not GO, independent of whatever the caller claims. Returns the version_id.
list() -> list[RegistryEntry]
All archived versions, newest first (by published_at, then id).
diff(version_a: str, version_b: str) -> dict[str, object]
Compare two versions field by field.
Every field maps to an (a_value, b_value) tuple (equal fields
included), so a reviewer sees the full promotion-time evidence — gate
decision, PnL, trades, allowed segments — side by side. Compare the two
elements of each tuple to spot what actually changed.
rollback(
version_id: str, output_dir: Path | str
) -> ChampionManifest
Restore an archived version as the live champion in output_dir.
Rewrites champion.json / champion.pkl (and calibrator.pkl if present) in output_dir to point at the archived artifacts, so the standard ChampionManifest.load(output_dir / "champion.json") flow serves the rolled-back model. Does not delete newer versions — history is preserved.
dataclass
¶
RegistryEntry(
version_id: str,
model_name: str,
plugin_family: str,
published_at: str,
gate_decision: str,
aggregate_pnl: float,
aggregate_trades: int,
)
Lightweight metadata for one archived champion (what list() returns).