danil_belikov/ml_crypto_lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
belikovme 2026-06-26 14:23:12 +07:00
commit 13ec1f0204
31 changed files with 2032 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
.gitignore vendored Normal file
View File

@ -0,0 +1,15 @@
.venv
venv/
__pycache__/
*.pyc
*.pyo
*.pyd
*.pyw
*.pyz
*.pywz
*.pyzw
*.pyzwz
*.pyzwzw
*.pyzwzwz
*.parquet
*.csv

185
README.md Normal file
View File

@ -0,0 +1,185 @@
# ml_crypto_lab
Удобный проектный каркас для ML-исследований торговых режимов:
- проектирование признаков;
- проектирование целевых параметров;
- обучение моделей;
- сохранение полного описания эксперимента;
- создание ансамблей;
- повторное использование сохранённых моделей на новых данных.
Главная идея: **каждый артефакт знает, из чего он был создан**.
Модель сохраняется не просто как `model.pkl`, а вместе с:
- списком входных признаков;
- именем target;
- параметрами признаков;
- параметрами target-а;
- параметрами модели;
- результатами train/test/valid;
- backtest-метриками;
- версией кода/конфига.
## Быстрый запуск
```bash
pip install -r requirements.txt
python scripts/run_full_experiment.py --config configs/experiment.yaml
```
После запуска появится папка:
```text
artifacts/runs/<run_id>/
```
В ней будут:
```text
models/ сохранённые модели
predictions/ предсказания моделей
reports/ таблицы результатов
ensembles/ ансамбли
run_config.yaml фактический конфиг запуска
registry_snapshot.json что было доступно в registry
```
## Основные сущности
### Feature Builder
Любая функция, которая принимает OHLCV-данные и возвращает DataFrame признаков.
Пример:
```python
@FEATURE_REGISTRY.register("market_basic")
def build_market_basic(df, cfg):
...
return features_df
```
### Target Builder
Любая функция, которая принимает OHLCV-данные и возвращает Series/DataFrame target-ов.
```python
@TARGET_REGISTRY.register("zigzag")
def build_zigzag_targets(df, cfg):
...
return target_df
```
### Model Factory
Фабрика модели. Возвращает объект с `.fit()` и `.predict()`.
```python
@MODEL_REGISTRY.register("logreg")
def make_logreg(cfg):
return SklearnBinaryModel(...)
```
### Experiment
Комбинация:
```text
feature_set + target + model
```
Каждая такая комбинация сохраняется отдельно.
## Как добавить новый признак
Создай файл, например:
```text
src/ml_crypto_lab/features/my_features.py
```
Добавь:
```python
from ml_crypto_lab.core.registry import FEATURE_REGISTRY
@FEATURE_REGISTRY.register("my_feature_block")
def build_my_features(df, cfg):
out = ...
return out
```
Потом добавь имя в `configs/experiment.yaml`:
```yaml
feature_sets:
my_set:
builders:
- name: my_feature_block
params: {}
```
## Как добавить новый target
Аналогично:
```python
from ml_crypto_lab.core.registry import TARGET_REGISTRY
@TARGET_REGISTRY.register("my_target")
def build_my_target(df, cfg):
return target_df
```
И в конфиг:
```yaml
target_sets:
my_targets:
builders:
- name: my_target
params: {}
```
## Как добавить новую модель
```python
from ml_crypto_lab.core.registry import MODEL_REGISTRY
@MODEL_REGISTRY.register("my_model")
def make_my_model(cfg):
return MyModelWrapper(...)
```
## Важная логика
Проект специально разделяет:
```text
features
отдельно
targets
отдельно
models
отдельно
experiments
отдельно
ensembles
отдельно
```
Так не возникает ситуации, когда непонятно:
```text
какая модель
на каких признаках
по какому target-у
с какими параметрами
была обучена
```

189
configs/experiment.yaml Normal file
View File

@ -0,0 +1,189 @@
run:
name: baseline_feature_target_model_registry
seed: 42
output_dir: artifacts/runs
# ------------------------------------------------------------
# DATA
# ------------------------------------------------------------
data:
path: bars_bybit_1min_2026-03-01_2026-05-11.parquet
time_col: time
symbol_col: symbol
symbol_candle: INDEX
candle_rule: 1min
min_coverage: 0.98
max_symbols: 120
required_cols:
- open
- high
- low
- close
- buy_volume
- sell_volume
# ------------------------------------------------------------
# SPLIT
# ------------------------------------------------------------
split:
train_size: 0.70
test_size: 0.15
valid_size: 0.15
# ------------------------------------------------------------
# BACKTEST
# ------------------------------------------------------------
backtest:
fee_rate: 0.0005
initial_state: 1
# ------------------------------------------------------------
# FEATURE SETS
# ------------------------------------------------------------
feature_sets:
A_market_basic:
description: "returns, range, ATR, volatility, volume imbalance"
builders:
- name: market_basic
params:
return_lags: [1, 3, 5, 15, 30, 60, 120]
range_windows: [14, 30, 60, 120]
atr_windows: [14, 60, 120]
vol_windows: [30, 60, 120, 360]
volume_windows: [30, 60, 120]
B_primitive_pressure:
description: "primitive continuous pressure parameters from OHLCV"
builders:
- name: primitive_pressure
params:
price_lags: [13, 34, 55, 144, 233]
force_lag: 34
volume_norm_win: 1440
C_mode_values:
description: "continuous mode values: level, speed, accel, cycle, persistence"
builders:
- name: mode_values
params:
speed_ema: 45
speed_lag: 5
cycle_win: 360
persist_win: 120
D_state_features:
description: "discrete state features derived from mode values"
builders:
- name: state_features
params:
deadband: 0.05
cycle_deadband: 0.15
persist_deadband: 0.20
E_agreement_features:
description: "agreement features derived from state features"
builders:
- name: agreement_features
params:
state:
deadband: 0.05
cycle_deadband: 0.15
persist_deadband: 0.20
F_long_context_features:
description: "long-context features derived from primitive pressures"
builders:
- name: long_context
params:
ema_spans: [120, 240, 360, 720, 1440]
trend_lags: [120, 240, 360, 720]
cycle_wins: [360, 720, 1440]
persist_wins: [240, 480, 720, 1440]
fast_span: 45
# ------------------------------------------------------------
# TARGET SETS
# ------------------------------------------------------------
target_sets:
zz_long:
builders:
- name: zigzag
params:
pct_list: [0.018, 0.022, 0.026, 0.030, 0.035, 0.040]
atr_mult_list: [1.5, 2.0, 2.5]
min_bars_list: [180, 240, 360, 480, 720]
atr_window: 60
initial_state: 1
future_mean:
builders:
- name: future_mean
params:
horizon_list: [360, 720, 1080, 1440]
min_move_pct_list: [0.004, 0.0065, 0.010]
atr_mult_list: [0.75, 1.25, 1.75]
fee_rate: 0.0005
fee_safety_rate: 0.00075
atr_window: 60
initial_state: 1
future_return:
builders:
- name: future_return
params:
horizon_list: [240, 360, 720, 1080, 1440]
min_move_pct_list: [0.005, 0.008, 0.012]
atr_mult_list: [1.0, 1.5, 2.0]
fee_rate: 0.0005
fee_safety_rate: 0.00075
atr_window: 60
initial_state: 1
# ------------------------------------------------------------
# MODELS
# ------------------------------------------------------------
models:
logreg:
name: sklearn_logreg
params:
C: 0.5
max_iter: 2000
class_weight: balanced
extra_trees:
name: sklearn_extra_trees
params:
n_estimators: 400
max_depth: 12
min_samples_leaf: 60
max_features: sqrt
n_jobs: -1
random_state: 42
hist_gb:
name: sklearn_hist_gradient_boosting
params:
max_iter: 250
learning_rate: 0.05
max_leaf_nodes: 31
l2_regularization: 0.01
random_state: 42
# ------------------------------------------------------------
# EXPERIMENT MATRIX
# ------------------------------------------------------------
experiment_matrix:
feature_sets: [A_market_basic, B_primitive_pressure, C_mode_values, D_state_features, E_agreement_features, F_long_context_features]
target_sets: [zz_long, future_mean, future_return]
models: [logreg, extra_trees, hist_gb]
training:
max_targets_per_set: null
save_predictions: true
save_models: true
ensemble:
enabled: true
methods:
- majority_vote
- score_average

20
notebooks/00_project_pipeline_as_cells.py Normal file
View File

@ -0,0 +1,20 @@
# %% [0] IMPORTS
from pathlib import Path
import sys
PROJECT_ROOT = Path.cwd()
sys.path.insert(0, str(PROJECT_ROOT / "src"))
from ml_crypto_lab.core.config import load_yaml
from ml_crypto_lab.train.runner import run_full_experiment
# %% [1] LOAD CONFIG
cfg = load_yaml("configs/experiment.yaml")
# %% [2] RUN FULL EXPERIMENT
result = run_full_experiment(cfg)
# %% [3] SHOW SUMMARY
summary_df = result["summary"]
print(result["run_dir"])
summary_df.head(30)

8
pyproject.toml Normal file
View File

@ -0,0 +1,8 @@
[project]
name = "ml-crypto-lab"
version = "0.1.0"
description = "Flexible ML crypto research framework"
requires-python = ">=3.10"
[tool.setuptools.packages.find]
where = ["src"]

9
requirements.txt Normal file
View File

@ -0,0 +1,9 @@
numpy>=1.24
pandas>=2.0
pyarrow>=14.0
scikit-learn>=1.3
joblib>=1.3
PyYAML>=6.0
matplotlib>=3.8
plotly>=5.18
torch>=2.0

40
scripts/predict_saved.py Normal file
View File

@ -0,0 +1,40 @@
from __future__ import annotations
import argparse
from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT_ROOT / "src"))
from ml_crypto_lab.core.config import load_yaml
from ml_crypto_lab.data.loading import load_raw_table, build_index_ohlc_and_matrices
from ml_crypto_lab.inference.predict import predict_with_saved_model
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--model-pack", required=True)
parser.add_argument("--config", required=True)
parser.add_argument("--output", required=True)
args = parser.parse_args()
cfg = load_yaml(args.config)
raw = load_raw_table(cfg["data"]["path"])
built = build_index_ohlc_and_matrices(raw, cfg["data"])
base = built["ohlc"].copy()
symbol = cfg["data"].get("symbol_candle", "INDEX")
if symbol in built["buy"].columns:
base["buy_volume"] = built["buy"][symbol]
base["sell_volume"] = built["sell"][symbol]
else:
base["buy_volume"] = built["buy"].sum(axis=1)
base["sell_volume"] = built["sell"].sum(axis=1)
pred = predict_with_saved_model(args.model_pack, base, args.output)
print(pred.tail())
print("saved:", args.output)
if __name__ == "__main__":
main()

29
scripts/run_full_experiment.py Normal file
View File

@ -0,0 +1,29 @@
from __future__ import annotations
import argparse
from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT_ROOT / "src"))
from ml_crypto_lab.core.config import load_yaml
from ml_crypto_lab.train.runner import run_full_experiment
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--config", required=True, help="Path to experiment yaml")
args = parser.parse_args()
cfg = load_yaml(args.config)
result = run_full_experiment(cfg)
print("=" * 120)
print("DONE")
print("run_id:", result["run_id"])
print("run_dir:", result["run_dir"])
print("best rows:")
print(result["summary"].head(20))
if __name__ == "__main__":
main()

1
src/ml_crypto_lab/__init__.py Normal file
View File

@ -0,0 +1 @@
__version__ = "0.1.0"

56
src/ml_crypto_lab/core/artifacts.py Normal file
View File

@ -0,0 +1,56 @@
from __future__ import annotations
from dataclasses import asdict
from pathlib import Path
from typing import Any
import json
import joblib
import pandas as pd
def ensure_dir(path: str | Path) -> Path:
p = Path(path)
p.mkdir(parents=True, exist_ok=True)
return p
def save_json(obj: Any, path: str | Path) -> None:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False, indent=2, default=str)
def load_json(path: str | Path) -> Any:
with Path(path).open("r", encoding="utf-8") as f:
return json.load(f)
def save_table(df: pd.DataFrame, path: str | Path) -> None:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
if path.suffix.lower() == ".parquet":
df.to_parquet(path)
elif path.suffix.lower() == ".csv":
df.to_csv(path, index=True)
else:
raise ValueError(f"Unsupported table extension: {path.suffix}")
def load_table(path: str | Path) -> pd.DataFrame:
path = Path(path)
if path.suffix.lower() == ".parquet":
return pd.read_parquet(path)
if path.suffix.lower() == ".csv":
return pd.read_csv(path, index_col=0, parse_dates=True)
raise ValueError(f"Unsupported table extension: {path.suffix}")
def save_model_pack(pack: dict[str, Any], path: str | Path) -> None:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
joblib.dump(pack, path)
def load_model_pack(path: str | Path) -> dict[str, Any]:
return joblib.load(path)

21
src/ml_crypto_lab/core/config.py Normal file
View File

@ -0,0 +1,21 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml
def load_yaml(path: str | Path) -> dict[str, Any]:
path = Path(path)
with path.open("r", encoding="utf-8") as f:
data = yaml.safe_load(f)
if data is None:
data = {}
return data
def save_yaml(obj: dict[str, Any], path: str | Path) -> None:
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as f:
yaml.safe_dump(obj, f, allow_unicode=True, sort_keys=False)

38
src/ml_crypto_lab/core/registry.py Normal file
View File

@ -0,0 +1,38 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, Iterable
@dataclass
class Registry:
"""Simple name -> callable registry."""
kind: str
items: Dict[str, Callable[..., Any]] = field(default_factory=dict)
def register(self, name: str):
def deco(fn: Callable[..., Any]):
if name in self.items:
raise KeyError(f"{self.kind} already registered: {name}")
self.items[name] = fn
return fn
return deco
def get(self, name: str) -> Callable[..., Any]:
if name not in self.items:
known = ", ".join(sorted(self.items))
raise KeyError(f"Unknown {self.kind}: {name}. Known: {known}")
return self.items[name]
def names(self) -> list[str]:
return sorted(self.items)
def snapshot(self) -> dict[str, list[str]]:
return {self.kind: self.names()}
FEATURE_REGISTRY = Registry("feature_builder")
TARGET_REGISTRY = Registry("target_builder")
MODEL_REGISTRY = Registry("model_factory")
ENSEMBLE_REGISTRY = Registry("ensemble_builder")

62
src/ml_crypto_lab/core/types.py Normal file
View File

@ -0,0 +1,62 @@
from __future__ import annotations
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Any, Optional
import time
import uuid
def make_run_id(prefix: str = "run") -> str:
ts = time.strftime("%Y%m%d_%H%M%S")
short = uuid.uuid4().hex[:8]
return f"{prefix}_{ts}_{short}"
@dataclass
class DatasetBundle:
ohlc: Any
features: Any
targets: Any
full_df: Any
@dataclass
class SplitIndex:
train_index: Any
test_index: Any
valid_index: Any
@dataclass
class ExperimentSpec:
experiment_id: str
feature_set_name: str
target_set_name: str
target_name: str
model_alias: str
model_name: str
feature_columns: list[str]
model_params: dict[str, Any]
feature_config: dict[str, Any]
target_config: dict[str, Any]
def to_dict(self) -> dict[str, Any]:
return asdict(self)
@dataclass
class TrainedArtifact:
spec: ExperimentSpec
model: Any
scaler: Any
metrics: dict[str, Any]
backtest: dict[str, Any]
paths: dict[str, str]
@dataclass
class PredictionFrameInfo:
prediction_path: str
rows: int
columns: list[str]

102
src/ml_crypto_lab/data/loading.py Normal file
View File

@ -0,0 +1,102 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import numpy as np
import pandas as pd
def load_raw_table(path: str | Path) -> pd.DataFrame:
path = Path(path)
if path.suffix.lower() == ".parquet":
return pd.read_parquet(path)
if path.suffix.lower() == ".csv":
return pd.read_csv(path)
raise ValueError(f"Unsupported file extension: {path.suffix}")
def build_index_ohlc_and_matrices(raw: pd.DataFrame, cfg: dict[str, Any]) -> dict[str, pd.DataFrame]:
time_col = cfg.get("time_col", "time")
symbol_col = cfg.get("symbol_col", "symbol")
symbol_candle = cfg.get("symbol_candle", "INDEX")
candle_rule = cfg.get("candle_rule", "1min")
min_coverage = float(cfg.get("min_coverage", 0.98))
max_symbols = cfg.get("max_symbols", None)
df = raw.copy()
df[time_col] = pd.to_datetime(df[time_col])
df = df.sort_values([symbol_col, time_col]).reset_index(drop=True)
required = cfg.get("required_cols", ["open", "high", "low", "close", "buy_volume", "sell_volume"])
missing = [c for c in [symbol_col, time_col] + required if c not in df.columns]
if missing:
raise ValueError(f"Missing columns in raw data: {missing}")
r = (
df.set_index(time_col)
.groupby(symbol_col)
.resample(candle_rule)
.agg(
open=("open", "first"),
high=("high", "max"),
low=("low", "min"),
close=("close", "last"),
buy_volume=("buy_volume", "sum"),
sell_volume=("sell_volume", "sum"),
)
.reset_index()
)
px = r.pivot(index=time_col, columns=symbol_col, values="close").sort_index()
buy = r.pivot(index=time_col, columns=symbol_col, values="buy_volume").sort_index()
sell = r.pivot(index=time_col, columns=symbol_col, values="sell_volume").sort_index()
coverage = px.notna().mean()
keep = coverage[coverage >= min_coverage].index.tolist()
if max_symbols is not None:
keep_no_index = [s for s in keep if s != symbol_candle]
keep = ([symbol_candle] if symbol_candle in keep else []) + keep_no_index[: int(max_symbols)]
px = px[keep].ffill().dropna()
buy = buy.reindex(px.index)[keep].fillna(0.0)
sell = sell.reindex(px.index)[keep].fillna(0.0)
idx_src = df[df[symbol_col] == symbol_candle].copy()
if idx_src.empty:
raise ValueError(f"Symbol for candles not found: {symbol_candle}")
ohlc = (
idx_src.set_index(time_col)
.sort_index()
.resample(candle_rule)
.agg(open=("open", "first"), high=("high", "max"), low=("low", "min"), close=("close", "last"))
.dropna()
)
common = ohlc.index.intersection(px.index).sort_values()
ohlc = ohlc.loc[common]
px = px.loc[common].ffill().dropna()
buy = buy.loc[px.index].fillna(0.0)
sell = sell.loc[px.index].fillna(0.0)
ohlc = ohlc.loc[px.index]
return {
"ohlc": ohlc,
"px": px,
"buy": buy,
"sell": sell,
"raw_resampled": r,
}
def merge_existing_context(ohlc: pd.DataFrame, context_frames: list[pd.DataFrame] | None = None) -> pd.DataFrame:
"""Optional helper for notebook use: merge already engineered tables into one base frame."""
out = ohlc.copy()
if context_frames:
for frame in context_frames:
if frame is None or frame.empty:
continue
aligned = frame.reindex(out.index).ffill().fillna(0.0)
out = pd.concat([out, aligned], axis=1)
out = out.loc[:, ~out.columns.duplicated()]
return out.replace([np.inf, -np.inf], np.nan).ffill().dropna(subset=["open", "high", "low", "close"])

12
src/ml_crypto_lab/ensembles/pipeline.py Normal file
View File

@ -0,0 +1,12 @@
from __future__ import annotations
from typing import Any
import pandas as pd
from ml_crypto_lab.core.registry import ENSEMBLE_REGISTRY
import ml_crypto_lab.ensembles.voting # noqa: F401
def build_ensemble(method: str, prediction_frames: dict[str, pd.DataFrame], cfg: dict[str, Any] | None = None) -> pd.DataFrame:
fn = ENSEMBLE_REGISTRY.get(method)
return fn(prediction_frames, cfg or {})

43
src/ml_crypto_lab/ensembles/voting.py Normal file
View File

@ -0,0 +1,43 @@
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
from ml_crypto_lab.core.registry import ENSEMBLE_REGISTRY
@ENSEMBLE_REGISTRY.register("majority_vote")
def majority_vote(prediction_frames: dict[str, pd.DataFrame], cfg: dict[str, Any] | None = None) -> pd.DataFrame:
cfg = cfg or {}
state_cols = []
score_cols = []
for model_id, df in prediction_frames.items():
if "pred_state" in df.columns:
state_cols.append(df["pred_state"].rename(model_id))
if "pred_score" in df.columns:
score_cols.append(df["pred_score"].rename(model_id))
if not state_cols:
raise ValueError("No pred_state columns for majority_vote")
state_df = pd.concat(state_cols, axis=1).ffill().fillna(1).astype(int)
vote_sum = state_df.sum(axis=1)
out = pd.DataFrame(index=state_df.index)
out["ensemble_vote_sum"] = vote_sum
out["ensemble_state"] = np.where(vote_sum >= 0, 1, -1).astype(np.int8)
if score_cols:
score_df = pd.concat(score_cols, axis=1).ffill().fillna(0.0)
out["ensemble_score"] = score_df.mean(axis=1)
return out
@ENSEMBLE_REGISTRY.register("score_average")
def score_average(prediction_frames: dict[str, pd.DataFrame], cfg: dict[str, Any] | None = None) -> pd.DataFrame:
score_cols = []
for model_id, df in prediction_frames.items():
if "pred_score" in df.columns:
score_cols.append(df["pred_score"].rename(model_id))
if not score_cols:
raise ValueError("No pred_score columns for score_average")
score_df = pd.concat(score_cols, axis=1).ffill().fillna(0.0)
score = score_df.mean(axis=1)
return pd.DataFrame({"ensemble_score": score, "ensemble_state": np.where(score >= 0, 1, -1).astype(np.int8)}, index=score.index)

93
src/ml_crypto_lab/evaluation/backtest.py Normal file
View File

@ -0,0 +1,93 @@
from __future__ import annotations
import numpy as np
import pandas as pd
def backtest_state_index_points(close, state, fee_rate=0.0005) -> dict:
close = pd.Series(close).astype(float)
state = pd.Series(state).reindex(close.index).ffill().fillna(0).astype(float)
position = state.shift(1).fillna(0)
position = pd.Series(np.where(position > 0, 1.0, np.where(position < 0, -1.0, 0.0)), index=close.index)
prev_position = position.shift(1).fillna(0)
exec_price = close.shift(1).fillna(close.iloc[0])
gross_pnl = pd.Series(0.0, index=close.index)
fee_pnl = pd.Series(0.0, index=close.index)
turnover_sides = (position - prev_position).abs().astype(float)
unrealized = pd.Series(0.0, index=close.index)
trades = []
entry_price = None
entry_time = None
entry_i = None
entry_fee = 0.0
for i in range(1, len(close)):
prev_pos = float(position.iloc[i - 1])
cur_pos = float(position.iloc[i])
if cur_pos != prev_pos:
price = float(exec_price.iloc[i])
event_time = close.index[i - 1]
fee_pnl.iloc[i] += abs(cur_pos - prev_pos) * fee_rate * price
if prev_pos != 0:
if entry_price is None:
entry_price = price; entry_time = event_time; entry_i = i - 1; entry_fee = abs(prev_pos) * fee_rate * price
trade_gross = prev_pos * (price - entry_price)
exit_fee = abs(prev_pos) * fee_rate * price
gross_pnl.iloc[i] += trade_gross
trades.append({"entry_time": entry_time, "exit_time": event_time, "side": prev_pos, "entry_price": entry_price, "exit_price": price, "gross_pnl": trade_gross, "entry_fee": entry_fee, "exit_fee": exit_fee, "net_pnl": trade_gross - entry_fee - exit_fee, "entry_i": entry_i, "exit_i": i})
entry_price = None; entry_time = None; entry_i = None; entry_fee = 0.0
if cur_pos != 0:
entry_price = price; entry_time = event_time; entry_i = i; entry_fee = abs(cur_pos) * fee_rate * price
if cur_pos != 0 and entry_price is not None:
unrealized.iloc[i] = cur_pos * (float(close.iloc[i]) - float(entry_price))
last_pos = float(position.iloc[-1])
if last_pos != 0 and entry_price is not None:
price = float(close.iloc[-1])
event_time = close.index[-1]
exit_fee = abs(last_pos) * fee_rate * price
fee_pnl.iloc[-1] += exit_fee
turnover_sides.iloc[-1] += abs(last_pos)
trade_gross = last_pos * (price - entry_price)
gross_pnl.iloc[-1] += trade_gross
trades.append({"entry_time": entry_time, "exit_time": event_time, "side": last_pos, "entry_price": entry_price, "exit_price": price, "gross_pnl": trade_gross, "entry_fee": entry_fee, "exit_fee": exit_fee, "net_pnl": trade_gross - entry_fee - exit_fee, "entry_i": entry_i, "exit_i": len(close)-1})
unrealized.iloc[-1] = 0.0
net_pnl = gross_pnl - fee_pnl
realized_gross = gross_pnl.cumsum()
realized_net = net_pnl.cumsum()
fee_cum = fee_pnl.cumsum()
equity_gross = close.iloc[0] + realized_gross + unrealized
equity_net = close.iloc[0] + realized_gross + unrealized - fee_cum
running_max = equity_net.cummax()
drawdown = equity_net - running_max
max_dd_abs = abs(float(drawdown.min()))
final_net = float(net_pnl.sum())
flips = int(((position.shift(1).fillna(0) * position) < 0).sum())
entries = int(((prev_position == 0) & (position != 0)).sum())
exits = int(((prev_position != 0) & (position == 0)).sum())
if last_pos != 0:
exits += 1
return {
"final_net_pnl": final_net,
"final_gross_pnl": float(gross_pnl.sum()),
"max_dd_abs": max_dd_abs,
"max_dd_pct": float(max_dd_abs / running_max.max()) if running_max.max() != 0 else np.nan,
"return_to_dd": float(final_net / max_dd_abs) if max_dd_abs > 0 else np.nan,
"turnover_sides_sum": float(turnover_sides.sum()),
"flips": flips,
"entries": entries,
"exits": exits,
"equity_net": equity_net,
"equity_gross": equity_gross,
"position": position,
"turnover_sides": turnover_sides,
"gross_pnl": gross_pnl,
"net_pnl": net_pnl,
"fee_pnl": fee_pnl,
"trades_df": pd.DataFrame(trades),
}

19
src/ml_crypto_lab/evaluation/metrics.py Normal file
View File

@ -0,0 +1,19 @@
from __future__ import annotations
import numpy as np
import pandas as pd
from sklearn.metrics import accuracy_score, balanced_accuracy_score, f1_score
def binary_classification_metrics(y_true, y_pred) -> dict:
y_true = pd.Series(y_true).astype(int)
y_pred = pd.Series(y_pred).astype(int).reindex(y_true.index).fillna(1).astype(int)
return {
"accuracy": float(accuracy_score(y_true, y_pred)),
"balanced_accuracy": float(balanced_accuracy_score(y_true, y_pred)),
"f1_macro": float(f1_score(y_true, y_pred, average="macro")),
"long_share_pred": float((y_pred == 1).mean()),
"long_share_true": float((y_true == 1).mean()),
"flips_pred": int(y_pred.shift(1).fillna(y_pred.iloc[0]).ne(y_pred).sum()),
"flips_true": int(y_true.shift(1).fillna(y_true.iloc[0]).ne(y_true).sum()),
}

276
src/ml_crypto_lab/features/builders.py Normal file
View File

@ -0,0 +1,276 @@
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
from ml_crypto_lab.core.registry import FEATURE_REGISTRY
from ml_crypto_lab.features.utils import calc_atr, causal_zscore, tanh_normalize_causal, clean_numeric, select_columns_by_patterns
@FEATURE_REGISTRY.register("market_basic")
def build_market_basic(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
"""Group A: pure market features from OHLCV-like columns."""
required = ["open", "high", "low", "close"]
missing = [c for c in required if c not in base_df.columns]
if missing:
raise ValueError(f"market_basic requires columns: {missing}")
close = base_df["close"].astype(float)
high = base_df["high"].astype(float)
low = base_df["low"].astype(float)
open_ = base_df["open"].astype(float)
out = pd.DataFrame(index=base_df.index)
return_lags = cfg.get("return_lags", [1, 3, 5, 15, 30, 60, 120])
for lag in return_lags:
lag = int(lag)
ret = close.pct_change(lag).replace([np.inf, -np.inf], np.nan).fillna(0.0)
out[f"a_ret_{lag}"] = ret
out[f"a_logret_{lag}"] = np.log(close.clip(lower=1e-9)).diff(lag).fillna(0.0)
out["a_body_pct"] = ((close - open_) / (open_.abs() + 1e-9)).replace([np.inf, -np.inf], np.nan).fillna(0.0)
out["a_range_pct"] = ((high - low) / (close.abs() + 1e-9)).replace([np.inf, -np.inf], np.nan).fillna(0.0)
out["a_close_pos_in_bar"] = ((close - low) / ((high - low) + 1e-9) * 2.0 - 1.0).clip(-1, 1).fillna(0.0)
for win in cfg.get("range_windows", [14, 30, 60, 120]):
win = int(win)
hist_hi = high.shift(1).rolling(win, min_periods=max(5, win // 5)).max()
hist_lo = low.shift(1).rolling(win, min_periods=max(5, win // 5)).min()
rng = (hist_hi - hist_lo).replace(0, np.nan)
out[f"a_pos_in_range_{win}"] = (2.0 * ((close - hist_lo) / (rng + 1e-9)) - 1.0).clip(-1, 1).fillna(0.0)
out[f"a_range_z_{win}"] = causal_zscore(high - low, win=win)
for win in cfg.get("atr_windows", [14, 60, 120]):
win = int(win)
atr = calc_atr(base_df, win)
out[f"a_atr_{win}_pct"] = (atr / (close.abs() + 1e-9)).replace([np.inf, -np.inf], np.nan).fillna(0.0)
out[f"a_atr_{win}_z"] = causal_zscore(atr, win=max(30, win * 3))
ret1 = close.pct_change(1).replace([np.inf, -np.inf], np.nan).fillna(0.0)
for win in cfg.get("vol_windows", [30, 60, 120, 360]):
win = int(win)
out[f"a_volatility_{win}"] = ret1.rolling(win, min_periods=max(5, win // 5)).std(ddof=0).fillna(0.0)
out[f"a_absret_mean_{win}"] = ret1.abs().rolling(win, min_periods=max(5, win // 5)).mean().fillna(0.0)
buy_col = "buy_volume" if "buy_volume" in base_df.columns else None
sell_col = "sell_volume" if "sell_volume" in base_df.columns else None
if buy_col and sell_col:
buy = base_df[buy_col].astype(float).fillna(0.0)
sell = base_df[sell_col].astype(float).fillna(0.0)
imb = (buy - sell) / (buy + sell + 1e-9)
out["a_volume_imbalance"] = imb.fillna(0.0)
out["a_log_volume_total"] = np.log1p(buy + sell)
for win in cfg.get("volume_windows", [30, 60, 120]):
win = int(win)
out[f"a_volume_imbalance_mean_{win}"] = imb.rolling(win, min_periods=max(5, win // 5)).mean().fillna(0.0)
out[f"a_volume_total_z_{win}"] = causal_zscore(np.log1p(buy + sell), win=win)
return clean_numeric(out)
@FEATURE_REGISTRY.register("primitive_pressure_from_existing")
def build_primitive_pressure_from_existing(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
"""Group B: existing primitive pressure columns only."""
wanted = cfg.get("columns", [])
out = pd.DataFrame(index=base_df.index)
for c in wanted:
if c in base_df.columns:
out[f"b_{c}"] = base_df[c]
elif f"sig__{c}" in base_df.columns:
out[f"b_{c}"] = base_df[f"sig__{c}"]
if out.empty:
# fallback: columns with pressure-ish names but not states/targets
patterns = ["pressure", "fusion_score", "fusion_force"]
cols = select_columns_by_patterns(base_df, include_patterns=patterns, exclude_patterns=["_state", "target_"])
for c in cols:
out[f"b_{c}"] = base_df[c]
return clean_numeric(out)
@FEATURE_REGISTRY.register("select_existing_by_patterns")
def build_select_existing_by_patterns(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
include = cfg.get("include_patterns", [])
exclude = cfg.get("exclude_patterns", [])
cols = select_columns_by_patterns(base_df, include_patterns=include, exclude_patterns=exclude)
out = base_df[cols].copy() if cols else pd.DataFrame(index=base_df.index)
prefix = cfg.get("prefix", "x_")
out.columns = [str(c) if str(c).startswith(prefix) else f"{prefix}{c}" for c in out.columns]
return clean_numeric(out)
def _primitive_pressure_core(base_df: pd.DataFrame, cfg: dict[str, Any] | None = None) -> pd.DataFrame:
"""Causal primitive pressures built only from OHLCV.
This is intentionally simple and extensible: these are not final trading rules,
but normalized continuous market-pressure parameters that can be reused by
mode/state/agreement/long-context feature groups.
"""
cfg = cfg or {}
close = base_df["close"].astype(float)
high = base_df["high"].astype(float)
low = base_df["low"].astype(float)
open_ = base_df["open"].astype(float)
buy = base_df["buy_volume"].astype(float).fillna(0.0) if "buy_volume" in base_df.columns else pd.Series(0.0, index=base_df.index)
sell = base_df["sell_volume"].astype(float).fillna(0.0) if "sell_volume" in base_df.columns else pd.Series(0.0, index=base_df.index)
out = pd.DataFrame(index=base_df.index)
# fusion_score proxy: normalized multi-lag return agreement.
lags = cfg.get("price_lags", [13, 34, 55, 144, 233])
ret_votes = []
for lag in lags:
ret_votes.append(np.sign(close.diff(int(lag)).fillna(0.0)))
out["fusion_score"] = pd.concat(ret_votes, axis=1).mean(axis=1).clip(-1, 1)
# fusion_force proxy: normalized directional impulse.
impulse = close.diff(int(cfg.get("force_lag", 34))).fillna(0.0)
out["fusion_force"] = tanh_normalize_causal(impulse, span=int(cfg.get("norm_span", 300)), k=1.5)
# volume_pressure: buy/sell pressure.
flow = np.log1p(buy) - np.log1p(sell)
out["volume_pressure"] = pd.Series(np.tanh(causal_zscore(flow, int(cfg.get("volume_norm_win", 1440))) / 2.0), index=base_df.index)
# index_pressure: trend + momentum + candle location.
ema_fast = close.ewm(span=int(cfg.get("ema_fast", 34)), adjust=False, min_periods=1).mean()
ema_slow = close.ewm(span=int(cfg.get("ema_slow", 233)), adjust=False, min_periods=1).mean()
trend = tanh_normalize_causal(ema_fast - ema_slow, span=300, k=1.5)
mom = tanh_normalize_causal(close.diff(int(cfg.get("momentum_lag", 55))).fillna(0.0), span=300, k=1.5)
rng = (high - low).replace(0, np.nan)
candle = ((close - open_) / (rng + 1e-9)).clip(-1, 1).fillna(0.0)
out["index_pressure"] = (0.45 * trend + 0.35 * mom + 0.20 * candle).clip(-1, 1)
# trend efficiency.
tep_win = int(cfg.get("trend_efficiency_win", 144))
path = close.diff().abs().rolling(tep_win, min_periods=max(20, tep_win // 5)).sum()
net = close - close.shift(tep_win)
out["trend_efficiency_pressure"] = (net / (path + 1e-9)).clip(-1, 1).fillna(0.0)
# range breakout pressure.
br_win = int(cfg.get("breakout_win", 360))
hist_hi = high.shift(1).rolling(br_win, min_periods=max(20, br_win // 10)).max()
hist_lo = low.shift(1).rolling(br_win, min_periods=max(20, br_win // 10)).min()
mid = (hist_hi + hist_lo) / 2.0
half = (hist_hi - hist_lo).replace(0, np.nan) / 2.0
out["index_breakout_pressure"] = ((close - mid) / (half + 1e-9)).clip(-1, 1).fillna(0.0)
# volatility direction pressure.
ret1 = close.diff().fillna(0.0)
vol_fast = ret1.abs().ewm(span=34, adjust=False, min_periods=1).mean()
vol_slow = ret1.abs().shift(1).ewm(span=233, adjust=False, min_periods=1).mean()
vol_expansion = np.tanh(((vol_fast / (vol_slow + 1e-9)) - 1.0) / 0.5)
direction = np.sign(close.diff(21).ewm(span=13, adjust=False, min_periods=1).mean().fillna(0.0))
out["volatility_direction_pressure"] = pd.Series(direction * vol_expansion, index=base_df.index).clip(-1, 1).fillna(0.0)
return clean_numeric(out.clip(-1.0, 1.0))
@FEATURE_REGISTRY.register("primitive_pressure")
def build_primitive_pressure(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
out = _primitive_pressure_core(base_df, cfg)
out.columns = [f"b_{c}" for c in out.columns]
return clean_numeric(out)
def _mode_values_core(base_df: pd.DataFrame, cfg: dict[str, Any] | None = None) -> pd.DataFrame:
cfg = cfg or {}
params = _primitive_pressure_core(base_df, cfg.get("primitive", {}))
out = pd.DataFrame(index=base_df.index)
for name in params.columns:
s = params[name].astype(float).clip(-1, 1)
out[f"{name}_level"] = s
smooth = s.ewm(span=int(cfg.get("speed_ema", 45)), adjust=False, min_periods=1).mean()
speed_raw = smooth.diff(int(cfg.get("speed_lag", 5))).fillna(0.0)
out[f"{name}_speed"] = tanh_normalize_causal(speed_raw, span=100, k=1.25)
accel_raw = speed_raw.ewm(span=13, adjust=False, min_periods=1).mean().diff(5).fillna(0.0)
out[f"{name}_accel"] = tanh_normalize_causal(accel_raw, span=180, k=1.25)
win = int(cfg.get("cycle_win", 360))
ref = s.shift(1)
lo = ref.rolling(win, min_periods=max(20, win // 10)).min()
hi = ref.rolling(win, min_periods=max(20, win // 10)).max()
cycle = (2.0 * ((s - lo) / ((hi - lo) + 1e-9)) - 1.0).clip(-1, 1).fillna(0.0)
out[f"{name}_cycle"] = cycle.ewm(span=9, adjust=False, min_periods=1).mean().clip(-1, 1)
pwin = int(cfg.get("persist_win", 120))
out[f"{name}_persistence"] = np.sign(s).rolling(pwin, min_periods=max(5, pwin // 10)).mean().fillna(0.0).clip(-1, 1)
return clean_numeric(out)
@FEATURE_REGISTRY.register("mode_values")
def build_mode_values(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
out = _mode_values_core(base_df, cfg)
out.columns = [f"c_{c}" for c in out.columns]
return clean_numeric(out)
def _state_features_core(base_df: pd.DataFrame, cfg: dict[str, Any] | None = None) -> pd.DataFrame:
cfg = cfg or {}
values = _mode_values_core(base_df, cfg.get("mode", {}))
deadband = float(cfg.get("deadband", 0.05))
out = pd.DataFrame(index=base_df.index)
for c in values.columns:
s = values[c].astype(float)
db = deadband
if c.endswith("_cycle"):
db = float(cfg.get("cycle_deadband", 0.15))
elif c.endswith("_persistence"):
db = float(cfg.get("persist_deadband", 0.20))
out[f"{c}_state"] = np.where(s > db, 1, np.where(s < -db, -1, 0))
return clean_numeric(out)
@FEATURE_REGISTRY.register("state_features")
def build_state_features(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
out = _state_features_core(base_df, cfg)
out.columns = [f"d_{c}" for c in out.columns]
return clean_numeric(out)
@FEATURE_REGISTRY.register("agreement_features")
def build_agreement_features(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
states = _state_features_core(base_df, cfg.get("state", {}))
out = pd.DataFrame(index=base_df.index)
mode_names = ["level", "speed", "accel", "cycle", "persistence"]
for mode in mode_names:
cols = [c for c in states.columns if f"_{mode}_state" in c]
if not cols:
continue
score = states[cols].sum(axis=1)
out[f"agreement_{mode}_score"] = score
out[f"agreement_{mode}_state"] = np.where(score > 0, 1, np.where(score < 0, -1, 0))
all_score = states.sum(axis=1)
out["agreement_all_score"] = all_score
out["agreement_all_state"] = np.where(all_score > 0, 1, np.where(all_score < 0, -1, 0))
out.columns = [f"e_{c}" for c in out.columns]
return clean_numeric(out)
@FEATURE_REGISTRY.register("long_context")
def build_long_context(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
params = _primitive_pressure_core(base_df, cfg.get("primitive", {}))
ema_spans = cfg.get("ema_spans", [120, 240, 360, 720, 1440])
trend_lags = cfg.get("trend_lags", [120, 240, 360, 720])
cycle_wins = cfg.get("cycle_wins", [360, 720, 1440])
persist_wins = cfg.get("persist_wins", [240, 480, 720, 1440])
fast_span = int(cfg.get("fast_span", 45))
out = pd.DataFrame(index=base_df.index)
for name in params.columns:
s = params[name].astype(float).clip(-1, 1)
for span in ema_spans:
out[f"{name}_lctx_ema_{int(span)}"] = s.ewm(span=int(span), adjust=False, min_periods=1).mean().clip(-1, 1)
for lag in trend_lags:
slow = s.ewm(span=max(2, int(lag) // 2), adjust=False, min_periods=1).mean()
out[f"{name}_lctx_trend_{int(lag)}"] = tanh_normalize_causal(slow.diff(int(lag)).fillna(0.0), span=max(100, int(lag)), k=1.5)
for win in cycle_wins:
ref = s.shift(1)
lo = ref.rolling(int(win), min_periods=max(20, int(win) // 10)).min()
hi = ref.rolling(int(win), min_periods=max(20, int(win) // 10)).max()
out[f"{name}_lctx_cycle_{int(win)}"] = (2 * ((s - lo) / ((hi - lo) + 1e-9)) - 1).clip(-1, 1).fillna(0.0)
for win in persist_wins:
out[f"{name}_lctx_persist_{int(win)}"] = np.sign(s).rolling(int(win), min_periods=max(10, int(win) // 10)).mean().fillna(0.0).clip(-1, 1)
fast = s.ewm(span=fast_span, adjust=False, min_periods=1).mean()
for span in [240, 720, 1440]:
long = s.ewm(span=span, adjust=False, min_periods=1).mean()
out[f"{name}_lctx_fast_vs_long_{span}"] = tanh_normalize_causal(fast - long, span=span, k=1.5)
out.columns = [f"f_{c}" for c in out.columns]
return clean_numeric(out)

40
src/ml_crypto_lab/features/pipeline.py Normal file
View File

@ -0,0 +1,40 @@
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
from ml_crypto_lab.core.registry import FEATURE_REGISTRY
from ml_crypto_lab.features.utils import clean_numeric
# Import registers builders
import ml_crypto_lab.features.builders # noqa: F401
def build_feature_set(base_df: pd.DataFrame, feature_set_cfg: dict[str, Any]) -> pd.DataFrame:
parts = []
for item in feature_set_cfg.get("builders", []):
name = item["name"]
params = item.get("params", {}) or {}
fn = FEATURE_REGISTRY.get(name)
part = fn(base_df, params)
if part is None or part.empty:
continue
part = part.copy()
part.columns = [str(c) for c in part.columns]
parts.append(part)
if not parts:
raise ValueError(f"Feature set produced no columns: {feature_set_cfg}")
out = pd.concat(parts, axis=1)
out = out.loc[:, ~out.columns.duplicated()]
out = clean_numeric(out)
return out
def build_all_feature_sets(base_df: pd.DataFrame, cfg: dict[str, Any]) -> dict[str, pd.DataFrame]:
result = {}
for name, fs_cfg in cfg.get("feature_sets", {}).items():
result[name] = build_feature_set(base_df, fs_cfg)
return result

73
src/ml_crypto_lab/features/utils.py Normal file
View File

@ -0,0 +1,73 @@
from __future__ import annotations
import re
import numpy as np
import pandas as pd
EPS = 1e-9
def clean_numeric(df: pd.DataFrame) -> pd.DataFrame:
out = df.replace([np.inf, -np.inf], np.nan).ffill().fillna(0.0)
out = out.select_dtypes(include=[np.number]).copy()
return out.astype(np.float32)
def add_prefix(df: pd.DataFrame, prefix: str) -> pd.DataFrame:
out = df.copy()
out.columns = [c if str(c).startswith(prefix) else f"{prefix}{c}" for c in out.columns]
return out
def calc_atr(ohlc: pd.DataFrame, window: int = 60) -> pd.Series:
high = ohlc["high"].astype(float)
low = ohlc["low"].astype(float)
close = ohlc["close"].astype(float)
prev_close = close.shift(1)
tr = pd.concat([
high - low,
(high - prev_close).abs(),
(low - prev_close).abs(),
], axis=1).max(axis=1)
return tr.rolling(window, min_periods=1).mean().replace([np.inf, -np.inf], np.nan).ffill().fillna(0.0)
def causal_zscore(s: pd.Series, win: int, min_periods: int | None = None) -> pd.Series:
s = pd.Series(s).astype(float)
win = int(win)
if min_periods is None:
min_periods = min(win, max(5, win // 10))
else:
min_periods = min(win, int(min_periods))
mu = s.rolling(win, min_periods=min_periods).mean().shift(1)
sd = s.rolling(win, min_periods=min_periods).std(ddof=0).shift(1)
z = (s - mu) / (sd + EPS)
return z.replace([np.inf, -np.inf], np.nan).fillna(0.0)
def tanh_normalize_causal(s: pd.Series, span: int = 300, k: float = 1.5) -> pd.Series:
s = pd.Series(s).astype(float)
scale = s.abs().shift(1).ewm(span=int(span), adjust=False, min_periods=1).mean()
fallback = float(s.abs().median())
if not np.isfinite(fallback) or fallback <= 0:
fallback = 1.0
scale = scale.replace(0, np.nan).ffill().fillna(fallback).replace(0, EPS)
return pd.Series(np.tanh(s / (scale * float(k) + EPS)), index=s.index).clip(-1.0, 1.0)
def select_columns_by_patterns(
df: pd.DataFrame,
include_patterns: list[str] | None = None,
exclude_patterns: list[str] | None = None,
) -> list[str]:
include_patterns = include_patterns or []
exclude_patterns = exclude_patterns or []
cols = []
for c in df.columns:
cs = str(c)
if include_patterns and not any(p in cs for p in include_patterns):
continue
if exclude_patterns and any(p in cs for p in exclude_patterns):
continue
cols.append(c)
return cols

28
src/ml_crypto_lab/inference/predict.py Normal file
View File

@ -0,0 +1,28 @@
from __future__ import annotations
from pathlib import Path
import pandas as pd
from ml_crypto_lab.core.artifacts import load_model_pack, save_table
from ml_crypto_lab.features.pipeline import build_feature_set
def predict_with_saved_model(model_pack_path: str | Path, base_df: pd.DataFrame, output_path: str | Path | None = None) -> pd.DataFrame:
pack = load_model_pack(model_pack_path)
spec = pack["spec"]
model = pack["model"]
feature_cfg = spec["feature_config"]
features = build_feature_set(base_df, feature_cfg)
cols = spec["feature_columns"]
missing = [c for c in cols if c not in features.columns]
if missing:
raise ValueError(f"Missing features for saved model: {missing[:20]} ... total={len(missing)}")
X = features[cols]
score = model.predict_score(X)
state = model.predict_state(X)
out = pd.DataFrame({"pred_score": score, "pred_state": state}, index=base_df.index)
if "close" in base_df.columns:
out.insert(0, "close", base_df["close"])
if output_path is not None:
save_table(out, output_path)
return out

13
src/ml_crypto_lab/models/pipeline.py Normal file
View File

@ -0,0 +1,13 @@
from __future__ import annotations
from typing import Any
from ml_crypto_lab.core.registry import MODEL_REGISTRY
# Import registers model factories
import ml_crypto_lab.models.sklearn_models # noqa: F401
import ml_crypto_lab.models.torch_lstm # noqa: F401
def make_model(model_name: str, params: dict[str, Any]):
fn = MODEL_REGISTRY.get(model_name)
return fn(params or {})

99
src/ml_crypto_lab/models/sklearn_models.py Normal file
View File

@ -0,0 +1,99 @@
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import RobustScaler, StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import ExtraTreesClassifier, HistGradientBoostingClassifier, RandomForestClassifier, GradientBoostingClassifier
from ml_crypto_lab.core.registry import MODEL_REGISTRY
class SklearnBinaryModel:
def __init__(self, estimator, scaler: str = "robust"):
if scaler == "robust":
scale_step = RobustScaler()
elif scaler == "standard":
scale_step = StandardScaler()
elif scaler in (None, "none"):
scale_step = "passthrough"
else:
raise ValueError(f"Unknown scaler: {scaler}")
self.pipeline = Pipeline([
("imputer", SimpleImputer(strategy="median")),
("scaler", scale_step),
("model", estimator),
])
self.feature_columns_: list[str] | None = None
def fit(self, X: pd.DataFrame, y: pd.Series):
self.feature_columns_ = list(X.columns)
y01 = pd.Series(y).astype(int).replace({-1: 0, 1: 1})
self.pipeline.fit(X, y01)
return self
def predict_score(self, X: pd.DataFrame) -> pd.Series:
X = X[self.feature_columns_]
model = self.pipeline.named_steps["model"]
if hasattr(self.pipeline, "predict_proba"):
proba = self.pipeline.predict_proba(X)
classes = list(model.classes_) if hasattr(model, "classes_") else [0, 1]
if 1 in classes and 0 in classes:
p1 = proba[:, classes.index(1)]
p0 = proba[:, classes.index(0)]
score = p1 - p0
else:
score = proba[:, -1] * 2.0 - 1.0
elif hasattr(self.pipeline, "decision_function"):
score = self.pipeline.decision_function(X)
else:
pred = self.pipeline.predict(X)
score = np.where(pred > 0, 1.0, -1.0)
return pd.Series(score, index=X.index, dtype=float)
def predict_state(self, X: pd.DataFrame) -> pd.Series:
score = self.predict_score(X)
return pd.Series(np.where(score >= 0, 1, -1), index=X.index, dtype=np.int8)
@MODEL_REGISTRY.register("sklearn_logreg")
def make_logreg(cfg: dict[str, Any]) -> SklearnBinaryModel:
params = dict(cfg)
scaler = params.pop("scaler", "robust")
est = LogisticRegression(**params)
return SklearnBinaryModel(est, scaler=scaler)
@MODEL_REGISTRY.register("sklearn_extra_trees")
def make_extra_trees(cfg: dict[str, Any]) -> SklearnBinaryModel:
params = dict(cfg)
scaler = params.pop("scaler", "none")
est = ExtraTreesClassifier(**params)
return SklearnBinaryModel(est, scaler=scaler)
@MODEL_REGISTRY.register("sklearn_random_forest")
def make_random_forest(cfg: dict[str, Any]) -> SklearnBinaryModel:
params = dict(cfg)
scaler = params.pop("scaler", "none")
est = RandomForestClassifier(**params)
return SklearnBinaryModel(est, scaler=scaler)
@MODEL_REGISTRY.register("sklearn_gradient_boosting")
def make_gradient_boosting(cfg: dict[str, Any]) -> SklearnBinaryModel:
params = dict(cfg)
scaler = params.pop("scaler", "none")
est = GradientBoostingClassifier(**params)
return SklearnBinaryModel(est, scaler=scaler)
@MODEL_REGISTRY.register("sklearn_hist_gradient_boosting")
def make_hist_gb(cfg: dict[str, Any]) -> SklearnBinaryModel:
params = dict(cfg)
scaler = params.pop("scaler", "none")
est = HistGradientBoostingClassifier(**params)
return SklearnBinaryModel(est, scaler=scaler)

117
src/ml_crypto_lab/models/torch_lstm.py Normal file
View File

@ -0,0 +1,117 @@
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
try:
import torch
from torch import nn
from torch.utils.data import Dataset, DataLoader
except Exception: # pragma: no cover
torch = None
nn = None
Dataset = object
DataLoader = None
from ml_crypto_lab.core.registry import MODEL_REGISTRY
class SequenceDataset(Dataset):
def __init__(self, X: np.ndarray, y: np.ndarray, seq_len: int):
self.X = X.astype(np.float32)
self.y = y.astype(np.float32)
self.seq_len = int(seq_len)
def __len__(self):
return max(0, len(self.X) - self.seq_len + 1)
def __getitem__(self, i):
j = i + self.seq_len - 1
return self.X[i:j+1], self.y[j]
class TinyLSTMNet(nn.Module):
def __init__(self, n_features: int, hidden: int = 64, layers: int = 1, dropout: float = 0.0):
super().__init__()
self.norm = nn.LayerNorm(n_features)
self.lstm = nn.LSTM(n_features, hidden, num_layers=layers, batch_first=True, dropout=dropout if layers > 1 else 0.0)
self.head = nn.Sequential(nn.Linear(hidden, hidden), nn.ReLU(), nn.Linear(hidden, 1))
def forward(self, x):
x = self.norm(x)
out, _ = self.lstm(x)
last = out[:, -1, :]
return self.head(last).squeeze(-1)
class TorchLSTMBinaryModel:
def __init__(self, seq_len=120, hidden=64, layers=1, dropout=0.0, lr=1e-3, batch_size=256, epochs=5, device="auto"):
if torch is None:
raise ImportError("torch is not installed")
self.seq_len = int(seq_len)
self.hidden = int(hidden)
self.layers = int(layers)
self.dropout = float(dropout)
self.lr = float(lr)
self.batch_size = int(batch_size)
self.epochs = int(epochs)
self.device = "cuda" if device == "auto" and torch.cuda.is_available() else ("cpu" if device == "auto" else device)
self.feature_columns_: list[str] | None = None
self.mean_: np.ndarray | None = None
self.std_: np.ndarray | None = None
self.net: TinyLSTMNet | None = None
def fit(self, X: pd.DataFrame, y: pd.Series):
self.feature_columns_ = list(X.columns)
Xv = X.to_numpy(dtype=np.float32)
self.mean_ = np.nanmean(Xv, axis=0)
self.std_ = np.nanstd(Xv, axis=0)
self.std_[self.std_ == 0] = 1.0
Xn = np.nan_to_num((Xv - self.mean_) / self.std_, nan=0.0, posinf=0.0, neginf=0.0)
yv = pd.Series(y).astype(int).replace({-1: 0, 1: 1}).to_numpy(dtype=np.float32)
ds = SequenceDataset(Xn, yv, self.seq_len)
dl = DataLoader(ds, batch_size=self.batch_size, shuffle=True, drop_last=False)
self.net = TinyLSTMNet(Xn.shape[1], hidden=self.hidden, layers=self.layers, dropout=self.dropout).to(self.device)
opt = torch.optim.AdamW(self.net.parameters(), lr=self.lr)
loss_fn = nn.BCEWithLogitsLoss()
self.net.train()
for _ in range(self.epochs):
for xb, yb in dl:
xb = xb.to(self.device); yb = yb.to(self.device)
opt.zero_grad()
logits = self.net(xb)
loss = loss_fn(logits, yb)
loss.backward()
opt.step()
return self
def _normalized(self, X: pd.DataFrame) -> np.ndarray:
X = X[self.feature_columns_]
Xv = X.to_numpy(dtype=np.float32)
return np.nan_to_num((Xv - self.mean_) / self.std_, nan=0.0, posinf=0.0, neginf=0.0)
def predict_score(self, X: pd.DataFrame) -> pd.Series:
Xn = self._normalized(X)
scores = np.full(len(Xn), np.nan, dtype=np.float32)
self.net.eval()
with torch.no_grad():
for start in range(0, max(0, len(Xn) - self.seq_len + 1), self.batch_size):
idxs = list(range(start, min(start + self.batch_size, len(Xn) - self.seq_len + 1)))
xb = np.stack([Xn[i:i+self.seq_len] for i in idxs]).astype(np.float32)
logits = self.net(torch.from_numpy(xb).to(self.device)).detach().cpu().numpy()
score = 1.0 / (1.0 + np.exp(-logits)) * 2.0 - 1.0
for k, i in enumerate(idxs):
scores[i + self.seq_len - 1] = score[k]
s = pd.Series(scores, index=X.index).ffill().fillna(0.0)
return s
def predict_state(self, X: pd.DataFrame) -> pd.Series:
score = self.predict_score(X)
return pd.Series(np.where(score >= 0, 1, -1), index=X.index, dtype=np.int8)
@MODEL_REGISTRY.register("torch_lstm")
def make_torch_lstm(cfg: dict[str, Any]) -> TorchLSTMBinaryModel:
return TorchLSTMBinaryModel(**cfg)

136
src/ml_crypto_lab/targets/builders.py Normal file
View File

@ -0,0 +1,136 @@
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
from ml_crypto_lab.core.registry import TARGET_REGISTRY
from ml_crypto_lab.features.utils import calc_atr
from ml_crypto_lab.targets.utils import make_binary_no_flat, required_move_vector, future_rolling_mean, param_name
@TARGET_REGISTRY.register("zigzag")
def build_zigzag_targets(ohlc: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
pct_list = cfg.get("pct_list", [0.018, 0.022])
atr_mult_list = cfg.get("atr_mult_list", [1.5, 2.0])
min_bars_list = cfg.get("min_bars_list", [180, 360])
atr_window = int(cfg.get("atr_window", 60))
initial_state = int(cfg.get("initial_state", 1))
atr = calc_atr(ohlc, atr_window).to_numpy(dtype=float)
out = {}
for pct in pct_list:
for atr_mult in atr_mult_list:
for mbp in min_bars_list:
name = f"target_zz_pct{int(round(float(pct)*10000)):04d}_atr{param_name(atr_mult)}_mbp{int(mbp):03d}"
out[name] = _zigzag_state(ohlc, float(pct), float(atr_mult), int(mbp), atr, initial_state)
return pd.DataFrame(out, index=ohlc.index).astype(np.int8)
def _zigzag_state(ohlc: pd.DataFrame, pct: float, atr_mult: float, min_bars: int, atr_values: np.ndarray, initial_state: int) -> pd.Series:
high = ohlc["high"].astype(float).to_numpy()
low = ohlc["low"].astype(float).to_numpy()
close = ohlc["close"].astype(float).to_numpy()
n = len(ohlc)
if n == 0:
return pd.Series(dtype=np.int8, index=ohlc.index)
def threshold(anchor: float, i: int) -> float:
return max(abs(anchor) * pct, float(atr_values[i]) * atr_mult)
pivots: list[tuple[int, float, int]] = []
def add_pivot(i: int, price: float, typ: int):
if not pivots:
pivots.append((i, price, typ)); return
li, lp, lt = pivots[-1]
if lt == typ:
if (typ == 1 and price < lp) or (typ == -1 and price > lp):
pivots[-1] = (i, price, typ)
return
if i - li < min_bars:
return
pivots.append((i, price, typ))
trend = 0
chi, chp = 0, high[0]
cli, clp = 0, low[0]
for i in range(1, n):
hi, lo = float(high[i]), float(low[i])
if trend == 0:
if hi > chp: chi, chp = i, hi
if lo < clp: cli, clp = i, lo
if hi - clp >= threshold(clp, i):
add_pivot(cli, clp, 1); trend = 1; chi, chp = i, hi
elif chp - lo >= threshold(chp, i):
add_pivot(chi, chp, -1); trend = -1; cli, clp = i, lo
elif trend == 1:
if hi >= chp: chi, chp = i, hi
if chp - lo >= threshold(chp, i):
add_pivot(chi, chp, -1); trend = -1; cli, clp = i, lo
else:
if lo <= clp: cli, clp = i, lo
if hi - clp >= threshold(clp, i):
add_pivot(cli, clp, 1); trend = 1; chi, chp = i, hi
if trend == 1:
add_pivot(chi, chp, -1)
elif trend == -1:
add_pivot(cli, clp, 1)
state = np.zeros(n, dtype=np.int8)
if len(pivots) < 2:
state[:] = 1 if close[-1] >= close[0] else -1
else:
for k in range(1, len(pivots)):
i1, p1, _ = pivots[k - 1]
i2, p2, _ = pivots[k]
if i2 > i1:
state[i1:i2 + 1] = 1 if p2 > p1 else -1
first, last = pivots[0][0], pivots[-1][0]
if first > 0: state[:first] = state[first]
if last < n - 1: state[last:] = state[last]
return make_binary_no_flat(pd.Series(state, index=ohlc.index), initial_state)
@TARGET_REGISTRY.register("future_return")
def build_future_return_targets(ohlc: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
close_s = ohlc["close"].astype(float).ffill().bfill()
close = close_s.to_numpy(dtype=float)
atr = calc_atr(ohlc, int(cfg.get("atr_window", 60))).to_numpy(dtype=float)
fee_rate = float(cfg.get("fee_rate", 0.0005))
safety = float(cfg.get("fee_safety_rate", 0.00075))
initial = int(cfg.get("initial_state", 1))
out = {}
for h in cfg.get("horizon_list", [240, 360, 720]):
future_close = close_s.shift(-int(h)).fillna(close_s.iloc[-1]).to_numpy(dtype=float)
for min_move in cfg.get("min_move_pct_list", [0.005, 0.008]):
for atr_mult in cfg.get("atr_mult_list", [1.0, 1.5]):
req = required_move_vector(close, atr, fee_rate, safety, float(min_move), float(atr_mult))
move = future_close - close
raw = np.where(move > req, 1, np.where(move < -req, -1, 0))
name = f"target_fret_h{int(h):04d}_move{param_name(min_move)}_atr{param_name(atr_mult)}"
out[name] = make_binary_no_flat(pd.Series(raw, index=ohlc.index), initial)
return pd.DataFrame(out, index=ohlc.index).astype(np.int8)
@TARGET_REGISTRY.register("future_mean")
def build_future_mean_targets(ohlc: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame:
close_s = ohlc["close"].astype(float).ffill().bfill()
close = close_s.to_numpy(dtype=float)
atr = calc_atr(ohlc, int(cfg.get("atr_window", 60))).to_numpy(dtype=float)
fee_rate = float(cfg.get("fee_rate", 0.0005))
safety = float(cfg.get("fee_safety_rate", 0.00075))
initial = int(cfg.get("initial_state", 1))
out = {}
for h in cfg.get("horizon_list", [360, 720]):
future_mean = future_rolling_mean(close_s, int(h)).fillna(close_s.iloc[-1]).to_numpy(dtype=float)
future_close = close_s.shift(-int(h)).fillna(close_s.iloc[-1]).to_numpy(dtype=float)
terminal_state = np.where(future_close - close >= 0, 1, -1)
for min_move in cfg.get("min_move_pct_list", [0.004, 0.0065]):
for atr_mult in cfg.get("atr_mult_list", [0.75, 1.25]):
req = required_move_vector(close, atr, fee_rate, safety, float(min_move), float(atr_mult))
move = future_mean - close
raw = np.where(move > req, 1, np.where(move < -req, -1, terminal_state))
name = f"target_fmean_h{int(h):04d}_move{param_name(min_move)}_atr{param_name(atr_mult)}"
out[name] = make_binary_no_flat(pd.Series(raw, index=ohlc.index), initial)
return pd.DataFrame(out, index=ohlc.index).astype(np.int8)

33
src/ml_crypto_lab/targets/pipeline.py Normal file
View File

@ -0,0 +1,33 @@
from __future__ import annotations
from typing import Any
import pandas as pd
from ml_crypto_lab.core.registry import TARGET_REGISTRY
# Import registers builders
import ml_crypto_lab.targets.builders # noqa: F401
def build_target_set(ohlc: pd.DataFrame, target_set_cfg: dict[str, Any]) -> pd.DataFrame:
parts = []
for item in target_set_cfg.get("builders", []):
name = item["name"]
params = item.get("params", {}) or {}
fn = TARGET_REGISTRY.get(name)
part = fn(ohlc, params)
if part is None or part.empty:
continue
parts.append(part)
if not parts:
raise ValueError(f"Target set produced no columns: {target_set_cfg}")
out = pd.concat(parts, axis=1)
out = out.loc[:, ~out.columns.duplicated()]
return out.astype("int8")
def build_all_target_sets(ohlc: pd.DataFrame, cfg: dict[str, Any]) -> dict[str, pd.DataFrame]:
result = {}
for name, ts_cfg in cfg.get("target_sets", {}).items():
result[name] = build_target_set(ohlc, ts_cfg)
return result

40
src/ml_crypto_lab/targets/utils.py Normal file
View File

@ -0,0 +1,40 @@
from __future__ import annotations
import numpy as np
import pandas as pd
from ml_crypto_lab.features.utils import calc_atr
def make_binary_no_flat(s: pd.Series, initial_state: int = 1) -> pd.Series:
s = pd.Series(s).replace(0, np.nan).ffill().bfill().fillna(initial_state)
return pd.Series(np.where(s.astype(float) > 0, 1, -1), index=s.index, dtype=np.int8)
def target_flips_count(s: pd.Series) -> int:
s = pd.Series(s).astype(np.int8)
if len(s) == 0:
return 0
return int(s.shift(1).fillna(s.iloc[0]).ne(s).sum())
def required_move_vector(close: np.ndarray, atr: np.ndarray, fee_rate: float, fee_safety_rate: float, min_move_pct: float, atr_mult: float) -> np.ndarray:
pct_req = max(2.0 * float(fee_rate) + float(fee_safety_rate), float(min_move_pct))
price_part = np.abs(close) * pct_req
atr_part = np.nan_to_num(atr, nan=0.0, posinf=0.0, neginf=0.0) * float(atr_mult)
return np.maximum(price_part, np.maximum(atr_part, 0.0))
def future_rolling_max(s: pd.Series, horizon: int) -> pd.Series:
return s.shift(-1).iloc[::-1].rolling(int(horizon), min_periods=1).max().iloc[::-1].reindex(s.index)
def future_rolling_min(s: pd.Series, horizon: int) -> pd.Series:
return s.shift(-1).iloc[::-1].rolling(int(horizon), min_periods=1).min().iloc[::-1].reindex(s.index)
def future_rolling_mean(s: pd.Series, horizon: int) -> pd.Series:
return s.shift(-1).iloc[::-1].rolling(int(horizon), min_periods=1).mean().iloc[::-1].reindex(s.index)
def param_name(x) -> str:
return str(x).replace(".", "p").replace("-", "m")

210
src/ml_crypto_lab/train/runner.py Normal file
View File

@ -0,0 +1,210 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
import hashlib
import itertools
import json
import numpy as np
import pandas as pd
from ml_crypto_lab.core.types import ExperimentSpec, make_run_id
from ml_crypto_lab.core.artifacts import ensure_dir, save_json, save_model_pack, save_table
from ml_crypto_lab.core.config import save_yaml
from ml_crypto_lab.core.registry import FEATURE_REGISTRY, TARGET_REGISTRY, MODEL_REGISTRY, ENSEMBLE_REGISTRY
from ml_crypto_lab.data.loading import load_raw_table, build_index_ohlc_and_matrices
from ml_crypto_lab.features.pipeline import build_all_feature_sets
from ml_crypto_lab.targets.pipeline import build_all_target_sets
from ml_crypto_lab.models.pipeline import make_model
from ml_crypto_lab.train.split import make_time_split
from ml_crypto_lab.evaluation.metrics import binary_classification_metrics
from ml_crypto_lab.evaluation.backtest import backtest_state_index_points
from ml_crypto_lab.ensembles.pipeline import build_ensemble
def stable_hash(obj: Any, n: int = 12) -> str:
s = json.dumps(obj, sort_keys=True, default=str, ensure_ascii=False)
return hashlib.md5(s.encode("utf-8")).hexdigest()[:n]
def build_base_frame_from_config(cfg: dict[str, Any]) -> pd.DataFrame:
raw = load_raw_table(cfg["data"]["path"])
built = build_index_ohlc_and_matrices(raw, cfg["data"])
base = built["ohlc"].copy()
# Add index buy/sell volume when available from matrices.
symbol = cfg["data"].get("symbol_candle", "INDEX")
if symbol in built["buy"].columns:
base["buy_volume"] = built["buy"][symbol]
base["sell_volume"] = built["sell"][symbol]
else:
base["buy_volume"] = built["buy"].sum(axis=1)
base["sell_volume"] = built["sell"].sum(axis=1)
return base.replace([np.inf, -np.inf], np.nan).ffill().dropna(subset=["open", "high", "low", "close"])
def train_one_experiment(
base_df: pd.DataFrame,
feature_df: pd.DataFrame,
target_s: pd.Series,
close_s: pd.Series,
split_idx: dict[str, pd.Index],
spec: ExperimentSpec,
run_dir: Path,
cfg: dict[str, Any],
) -> dict[str, Any]:
X = feature_df.reindex(base_df.index).replace([np.inf, -np.inf], np.nan).ffill().fillna(0.0)
y = target_s.reindex(base_df.index).ffill().bfill().astype(int)
train_idx = split_idx["train"].intersection(X.index).intersection(y.index)
test_idx = split_idx["test"].intersection(X.index).intersection(y.index)
valid_idx = split_idx["valid"].intersection(X.index).intersection(y.index)
model = make_model(spec.model_name, spec.model_params)
model.fit(X.loc[train_idx, spec.feature_columns], y.loc[train_idx])
pred_frames = {}
metrics = {}
bt_metrics = {}
for part_name, idx in [("train", train_idx), ("test", test_idx), ("valid", valid_idx)]:
Xp = X.loc[idx, spec.feature_columns]
yp = y.loc[idx]
score = model.predict_score(Xp)
state = model.predict_state(Xp)
pred_df = pd.DataFrame({
"close": close_s.reindex(idx),
"target": yp,
"pred_score": score,
"pred_state": state,
}, index=idx)
pred_frames[part_name] = pred_df
metrics[part_name] = binary_classification_metrics(yp, state)
bt = backtest_state_index_points(pred_df["close"], pred_df["pred_state"], fee_rate=float(cfg.get("backtest", {}).get("fee_rate", 0.0005)))
bt_metrics[part_name] = {k: v for k, v in bt.items() if not isinstance(v, (pd.Series, pd.DataFrame))}
model_path = run_dir / "models" / f"{spec.experiment_id}.joblib"
pred_path = run_dir / "predictions" / f"{spec.experiment_id}_valid.parquet"
spec_path = run_dir / "specs" / f"{spec.experiment_id}.json"
if cfg.get("training", {}).get("save_models", True):
save_model_pack({
"spec": spec.to_dict(),
"model": model,
"metrics": metrics,
"backtest": bt_metrics,
}, model_path)
if cfg.get("training", {}).get("save_predictions", True):
save_table(pred_frames["valid"], pred_path)
save_json(spec.to_dict(), spec_path)
row = {
"experiment_id": spec.experiment_id,
"feature_set": spec.feature_set_name,
"target_set": spec.target_set_name,
"target_name": spec.target_name,
"model_alias": spec.model_alias,
"model_name": spec.model_name,
"n_features": len(spec.feature_columns),
"model_path": str(model_path),
"valid_prediction_path": str(pred_path),
**{f"train_{k}": v for k, v in metrics["train"].items()},
**{f"test_{k}": v for k, v in metrics["test"].items()},
**{f"valid_{k}": v for k, v in metrics["valid"].items()},
**{f"bt_train_{k}": v for k, v in bt_metrics["train"].items()},
**{f"bt_test_{k}": v for k, v in bt_metrics["test"].items()},
**{f"bt_valid_{k}": v for k, v in bt_metrics["valid"].items()},
}
return {"row": row, "model": model, "valid_pred": pred_frames["valid"]}
def run_full_experiment(cfg: dict[str, Any]) -> dict[str, Any]:
run_id = make_run_id(cfg.get("run", {}).get("name", "run"))
output_root = Path(cfg.get("run", {}).get("output_dir", "artifacts/runs"))
run_dir = ensure_dir(output_root / run_id)
ensure_dir(run_dir / "models")
ensure_dir(run_dir / "predictions")
ensure_dir(run_dir / "reports")
ensure_dir(run_dir / "ensembles")
ensure_dir(run_dir / "specs")
save_yaml(cfg, run_dir / "run_config.yaml")
save_json({
"features": FEATURE_REGISTRY.names(),
"targets": TARGET_REGISTRY.names(),
"models": MODEL_REGISTRY.names(),
"ensembles": ENSEMBLE_REGISTRY.names(),
}, run_dir / "registry_snapshot.json")
base_df = build_base_frame_from_config(cfg)
feature_sets = build_all_feature_sets(base_df, cfg)
target_sets = build_all_target_sets(base_df[["open", "high", "low", "close"]], cfg)
split_idx = make_time_split(base_df.index, **cfg.get("split", {}))
rows = []
valid_prediction_frames_for_ensemble: dict[str, pd.DataFrame] = {}
matrix = cfg.get("experiment_matrix", {})
fs_names = matrix.get("feature_sets", list(feature_sets.keys()))
ts_names = matrix.get("target_sets", list(target_sets.keys()))
model_aliases = matrix.get("models", list(cfg.get("models", {}).keys()))
max_targets = cfg.get("training", {}).get("max_targets_per_set", None)
for fs_name, ts_name, model_alias in itertools.product(fs_names, ts_names, model_aliases):
feature_df = feature_sets[fs_name]
target_df = target_sets[ts_name]
target_cols = list(target_df.columns)
if max_targets is not None:
target_cols = target_cols[: int(max_targets)]
model_cfg = cfg["models"][model_alias]
model_name = model_cfg["name"]
model_params = model_cfg.get("params", {}) or {}
for target_name in target_cols:
spec_dict = {
"feature_set": fs_name,
"target_set": ts_name,
"target_name": target_name,
"model_alias": model_alias,
"model_name": model_name,
"model_params": model_params,
"n_features": feature_df.shape[1],
}
exp_id = f"exp_{stable_hash(spec_dict)}"
spec = ExperimentSpec(
experiment_id=exp_id,
feature_set_name=fs_name,
target_set_name=ts_name,
target_name=target_name,
model_alias=model_alias,
model_name=model_name,
feature_columns=list(feature_df.columns),
model_params=model_params,
feature_config=cfg.get("feature_sets", {}).get(fs_name, {}),
target_config=cfg.get("target_sets", {}).get(ts_name, {}),
)
print(f"TRAIN {exp_id}: {fs_name} | {ts_name}:{target_name} | {model_alias}")
res = train_one_experiment(base_df, feature_df, target_df[target_name], base_df["close"], split_idx, spec, run_dir, cfg)
rows.append(res["row"])
valid_prediction_frames_for_ensemble[exp_id] = res["valid_pred"]
summary_df = pd.DataFrame(rows).sort_values("bt_valid_return_to_dd", ascending=False)
save_table(summary_df, run_dir / "reports" / "experiment_summary.parquet")
summary_df.to_csv(run_dir / "reports" / "experiment_summary.csv", index=False)
if cfg.get("ensemble", {}).get("enabled", True) and valid_prediction_frames_for_ensemble:
methods = cfg.get("ensemble", {}).get("methods", ["majority_vote"])
ens_rows = []
for method in methods:
ens = build_ensemble(method, valid_prediction_frames_for_ensemble)
close = base_df["close"].reindex(ens.index)
state_col = "ensemble_state"
bt = backtest_state_index_points(close, ens[state_col], fee_rate=float(cfg.get("backtest", {}).get("fee_rate", 0.0005)))
ens_path = run_dir / "ensembles" / f"{method}_valid.parquet"
save_table(ens, ens_path)
ens_rows.append({"ensemble_method": method, "path": str(ens_path), **{k: v for k, v in bt.items() if not isinstance(v, (pd.Series, pd.DataFrame))}})
ens_df = pd.DataFrame(ens_rows).sort_values("return_to_dd", ascending=False)
save_table(ens_df, run_dir / "reports" / "ensemble_summary.parquet")
ens_df.to_csv(run_dir / "reports" / "ensemble_summary.csv", index=False)
return {"run_id": run_id, "run_dir": str(run_dir), "summary": summary_df}

16
src/ml_crypto_lab/train/split.py Normal file
View File

@ -0,0 +1,16 @@
from __future__ import annotations
import pandas as pd
def make_time_split(index: pd.Index, train_size: float = 0.70, test_size: float = 0.15, valid_size: float = 0.15) -> dict[str, pd.Index]:
if abs((train_size + test_size + valid_size) - 1.0) > 1e-6:
raise ValueError("train_size + test_size + valid_size must equal 1.0")
idx = pd.Index(index).sort_values()
n = len(idx)
n_train = int(n * train_size)
n_test = int(n * test_size)
train_idx = idx[:n_train]
test_idx = idx[n_train:n_train + n_test]
valid_idx = idx[n_train + n_test:]
return {"train": train_idx, "test": test_idx, "valid": valid_idx}

9
tests/test_imports.py Normal file
View File

@ -0,0 +1,9 @@
def test_imports():
import ml_crypto_lab
from ml_crypto_lab.core.registry import FEATURE_REGISTRY, TARGET_REGISTRY, MODEL_REGISTRY
import ml_crypto_lab.features.builders
import ml_crypto_lab.targets.builders
import ml_crypto_lab.models.sklearn_models
assert "market_basic" in FEATURE_REGISTRY.names()
assert "zigzag" in TARGET_REGISTRY.names()
assert "sklearn_logreg" in MODEL_REGISTRY.names()