edgekit.ml

The meta-labelling / ML framework — the Lopez de Prado stack. Label each candidate with a triple barrier, weight by uniqueness, validate with purged walk-forward (no leakage), fit a gradient-boosted meta-label, and freeze it to a cloud-safe tree blob a cTrader cBot can evaluate with no scikit-learn.

What's inside. Config dataclasses (LabelConfig, WalkForwardConfig, SessionConfig, ModelConfig, ExperimentConfig); labelling (triple_barrier, uniqueness_weights); leakage-safe splits (walk_forward_windows, time_inner_split, PurgedKFold); features (build_features); models (make_model, ModelWrapper); metrics (ev_at, best_threshold, trade_metrics, confusion, safe_auc); and export (export_trees, emit_python, emit_csharp, MetaLabeler, shuffle_label_gate).

!Lean import, lazy backends

import edgekit.ml needs only numpy + pandas. Heavy backends are lazy-imported inside the functions that fit models — you only pay when you actually fit:

make_model("rf") and make_model("hgb") need scikit-learn.
make_model("xgb") needs xgboost.
make_model("lgbm") needs lightgbm.
safe_auc lazy-imports sklearn and returns NaN if it is absent.
ExperimentConfig.from_yaml lazy-imports yaml.

Each missing backend raises an actionable ImportError at make_model time (an unknown name raises ValueError).

Config dataclasses (ml.config)#

LabelConfig#

Triple-barrier geometry in price units.

LabelConfig(rr: float = 2.0, stop_atr_mult: float = 1.0, atr_period: int = 14,
            min_stop: float | None = None, max_stop: float | None = None,
            horizon_bars: int = 48, eod_flatten: bool = False,
            eod_hour_utc: int = 21, cost_in_label: bool = False)

rr — sets the take-profit as a multiple of the stop distance.
stop_atr_mult, atr_period — stop = stop_atr_mult × ATR(atr_period).
min_stop / max_stop — clamp the ATR stop.
horizon_bars — the vertical (time) barrier.
eod_flatten — add an intraday vertical barrier at eod_hour_utc (leave False for 24/7 assets).
cost_in_label — widen the TP by cost so a labelled win must clear costs.

WalkForwardConfig#

Rolling out-of-sample walk geometry.

WalkForwardConfig(train_days: float = 7.0, test_hours: float = 4.0,
                  embargo_hours: float = 4.0, dev_lookback_days: float = 365,
                  inner_val_frac: float = 0.3)

!The embargo is the firewall

embargo_hoursmust be ≥ the label horizon — it gaps train from test so no train label's outcome can resolve inside the test window (the whole point of purging). dev_lookback_days=0 walks all data.

SessionConfig, ModelConfig, ExperimentConfig#

SessionConfig(london_start: int = 7, london_end: int = 16, overlap_start: int = 12,
              overlap_end: int = 16, mode: str = "london_and_overlap")

ModelConfig(name: str = "hgb", params: dict = {})   # name in "rf"|"xgb"|"lgbm"|"hgb"

ExperimentConfig(experiment_id="exp", data_path="", bar_minutes=5, holdout_frac=0.2,
                 feature_families=[...], label=LabelConfig(), session=SessionConfig(),
                 wf=WalkForwardConfig(), models=[ModelConfig("hgb")], seed=7,
                 shuffle_labels=False, out_root=".")

ExperimentConfig is the single source of truth for a run. Methods: logs_dir() -> Path, to_dict() -> dict, and classmethods from_yaml(path) (lazy yaml) and from_dict(raw).

Labelling (ml.labeling)#

triple_barrier#

Cost-aware triple-barrier labelling whose fills mirror the engine, so labels and backtest agree: entry at the next bar's open; TP = entry + dir·rr·stop, SL = entry − dir·stop; a bar touching both barriers counts as SL (pessimistic); vertical barrier at horizon_bars; y=1 iff TP hit first.

triple_barrier(df, cand_mask, cfg: LabelConfig, cost: float = 0.0,
               horizon_bars: int | None = None, both_directions: bool = True) -> pd.DataFrame

Param	Type	Default	Meaning
`df`	pd.DataFrame	—	OHLC frame.
`cand_mask`	bool array	—	Which bars are labelling candidates (over df rows).
`cfg`	LabelConfig	—	Barrier geometry.
`cost`	float	0.0	Round-trip cost in price units, charged in each label's net R.
`horizon_bars`	int \| None	None	Override cfg's vertical barrier.
`both_directions`	bool	True	Label both a long and a short per candidate.

Returns a long DataFrame, one row per (candidate, direction), columns: entry_index, entry_time, exit_index, exit_time, dir, y, r, net, stop, outcome, weight.

labeling.py

import numpy as np
from edgekit.ml import triple_barrier, LabelConfig

cand_mask = np.zeros(len(bars), dtype=bool)
cand_mask[::5] = True                              # every 5th bar is a candidate
cfg = LabelConfig(rr=2.0, stop_atr_mult=1.0, atr_period=14, horizon_bars=48)
labels = triple_barrier(bars, cand_mask, cfg, cost=0.0)
labels[["dir", "y", "r", "outcome"]].head()

uniqueness_weights#

Average-uniqueness weight per label = mean over its lifespan of 1/concurrency — down-weighting overlapping labels so the model does not over-count redundant samples.

uniqueness_weights(entry_index, exit_index, n_bars: int) -> np.ndarray

w = uniqueness_weights(labels["entry_index"], labels["exit_index"], len(bars))
model.fit(X, y, sample_weight=w)

Splits (ml.splits)#

walk_forward_windows#

The out-of-sample backtest generator: train on the rolling past, test on the next block, with an embargo gap. Yields dicts of .iloc positions.

walk_forward_windows(labels: pd.DataFrame, wf: WalkForwardConfig)   # generator -> {t0, t1, train_pos, test_pos}

labels must carry entry_time / exit_time. A train label is kept only if its exit_time ≤ the embargoed train-window end (the purge); windows need ≥ 40 train labels.

from edgekit.ml import walk_forward_windows, WalkForwardConfig
wf = WalkForwardConfig(train_days=7, test_hours=8, embargo_hours=6)
for w in walk_forward_windows(labels, wf):
    tr, te = labels.iloc[w["train_pos"]], labels.iloc[w["test_pos"]]
    # fit on tr, predict te — no train outcome resolves inside the embargo

time_inner_split#

Chronological inner split for early-stop / threshold tuning: the earliest (1−val_frac) is train, the latest val_frac is val, purging train labels whose horizon runs into the val block.

time_inner_split(train_labels: pd.DataFrame, val_frac: float) -> (inner_tr, inner_val)   # position arrays

PurgedKFold#

K contiguous-in-time folds with purge + embargo, for hyperparameter selection inside a training window.

PurgedKFold(n_splits: int = 4, embargo_frac: float = 0.02)
.split(labels)   # yields (train, val) index arrays

Purges train labels whose horizon overlaps the val fold's time span, plus an embargo tail.

Features (ml.features)#

build_features#

Build the selected causal feature families, aligned to df (inf → NaN). Every feature at bar i uses only bars <= i.

build_features(df, families: list[str] | None = None, mtf_rule: str = "1h") -> pd.DataFrame

families — defaults to all of FAMILIES = ["price", "atr", "trend", "macd", "structure", "volume", "time", "mtf", "reversal", "extra"].
mtf_rule — pandas offset for the higher-timeframe context (shifted one completed bar so it never peeks).

The volume family is skipped unless a tick_volume column is present.

from edgekit.ml import build_features
X = build_features(bars, families=["price", "trend", "macd", "structure"])

Models (ml.models)#

make_model#

Build a ModelWrapper for a backend — RF / XGBoost / LightGBM / HistGBM behind one fit/predict_proba API.

make_model(name: str, task: str = "classification", params: dict | None = None,
           seed: int = 7) -> ModelWrapper

Param	Type	Default	Meaning
`name`	str	—	"rf" \| "xgb" \| "lgbm" \| "hgb" (unknown → ValueError).
`task`	str	"classification"	"classification" or "regression".
`params`	dict \| None	None	Overrides the tuned defaults.
`seed`	int	7	Random seed.

ModelWrapper#

ModelWrapper(name: str, estimator, task: str = "classification")
.fit(X, y, sample_weight=None) -> self
.predict_proba(X) -> np.ndarray      # P(class 1); constant for a degenerate single-class window
.predict(X)

from edgekit.ml import make_model
m = make_model("hgb", params={"max_iter": 30})
m.fit(X, y)                        # needs scikit-learn (lazy import)
p = m.predict_proba(X)             # P(win) per row

Metrics (ml.metrics)#

ML/trading metrics for meta-labelling — the objective is EV per trade in R, after costs.

threshold_grid() -> np.ndarray                          # candidate cut-offs 0.30..0.79 step 0.01
ev_at(proba, r, thr) -> (mean_R, count)                 # of the trades a cut-off would take
best_threshold(proba, r, min_trades: int = 10) -> (best_t, best_ev)   # tune on VAL only
trade_metrics(r_taken) -> dict                          # {n, ev_r, win_rate, pf, sharpe, total_r}
confusion(y_true, y_pred) -> dict                       # {tp, tn, fp, fn, precision, recall}
safe_auc(y_true, proba) -> float                        # ROC-AUC; NaN if sklearn missing / one class

thresholding.py

from edgekit.ml import best_threshold, ev_at, trade_metrics
thr, ev = best_threshold(val_proba, val_r, min_trades=10)   # tune on VAL, never the test block
mean_r, n = ev_at(test_proba, test_r, thr)                  # apply on the held-out block
metrics = trade_metrics(test_r[test_proba >= thr])

!Tune the threshold on validation only

best_threshold picks the EV-maximising cut-off — do that on the validation split, then apply it unchanged on the test block. Choosing the threshold on the same data you report EV on is a leak. safe_auc degrades to NaN if sklearn is absent or only one class is present.

Export (ml.export)#

The cloud-safe round-trip: serialise a fitted HistGradientBoosting meta-label to a plain tree-array string, and generate pure-Python / pure-C# inference for a cBot. Blob format: "<baseline>#<tree>|<tree>...", where tree = "<node>;<node>..." and node = "feat,thr,left,right,isleaf,missleft,value"; P(class 1) = sigmoid(baseline + sum of hit-leaf values). Module constant THRESHOLD = 0.55.

export_trees(clf) -> str — serialise a fitted HistGradientBoostingClassifier to the blob (reads sklearn's private node arrays). Raises TypeError if not a fitted HistGradientBoosting estimator.
parse_trees(blob) -> (baseline, trees) — parse a blob into the baseline float and node-tuple lists.
tree_predict_proba(blob, X) -> np.ndarray — the reference pure-python forward pass (no sklearn); must match sklearn to float tolerance.
shuffle_label_gate(make_model_fn, X, y, r=None, n=10, test_frac=0.3, seed=0) -> dict — leakage null test: permute labels, refit, confirm OOS AUC ~ 0.5. Returns {mean_auc, aucs, mean_ev}.

MetaLabeler#

A frozen meta-label — take a trade only when P(win) ≥ threshold. Built from a fitted estimator (clf=, exported immediately) or a raw blob=; needs either. Carries no sklearn dependency once constructed.

MetaLabeler(blob: str | None = None, clf=None, threshold: float = 0.55,
            feature_names: list[str] | None = None)
.proba(X) -> np.ndarray
.take(X) -> np.ndarray        # boolean take/skip at the threshold

emit_python, emit_csharp#

emit_python(models, feature_names=None, threshold=0.55) -> str
emit_csharp(models, feature_names=None, threshold=0.55,
            class_name="Models", namespace="cAlgo.Robots") -> str

emit_python generates a self-contained pure-Python inference module (no numpy/sklearn) defining BLOBS, THRESHOLD, FEATURES and predict(name, x) / take(name, x). emit_csharp generates a C# static class embedding the blob(s) plus pure-C# tree inference (Predict(name, x), Threshold) for a cTrader cBot. In both, models is a blob string or a {name: blob} dict.

✦The cloud-safe guarantee

The whole point of the export is that a cBot with no scikit-learn agrees with your trained model. The round-trip is guaranteed to float tolerance: |tree_predict_proba(blob, X) − clf.predict_proba(X)[:,1]| < 1e-6.

export.py

import numpy as np
from sklearn.ensemble import HistGradientBoostingClassifier
from edgekit.ml import export_trees, tree_predict_proba, MetaLabeler, emit_csharp

clf = HistGradientBoostingClassifier(max_depth=3, max_iter=40).fit(X, y)
blob = export_trees(clf)

# the round-trip: pure-python inference matches sklearn to 1e-6
assert np.abs(tree_predict_proba(blob, X) - clf.predict_proba(X)[:, 1]).max() < 1e-6

meta = MetaLabeler(clf=clf, threshold=0.55)
take = meta.take(X_live)                                   # boolean take/skip, no sklearn needed
cs_source = emit_csharp({"orb": blob}, feature_names=feat_names)   # drop into a cTrader cBot