commit 13ec1f0204ac92297a9678abc697918dc51857a4 Author: belikovme Date: Fri Jun 26 14:23:12 2026 +0700 init diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b7a0841 --- /dev/null +++ b/.gitignore @@ -0,0 +1,15 @@ +.venv +venv/ +__pycache__/ +*.pyc +*.pyo +*.pyd +*.pyw +*.pyz +*.pywz +*.pyzw +*.pyzwz +*.pyzwzw +*.pyzwzwz +*.parquet +*.csv \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..989c6c4 --- /dev/null +++ b/README.md @@ -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// +``` + +В ней будут: + +```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-у +с какими параметрами +была обучена +``` diff --git a/configs/experiment.yaml b/configs/experiment.yaml new file mode 100644 index 0000000..b5b50a6 --- /dev/null +++ b/configs/experiment.yaml @@ -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 diff --git a/notebooks/00_project_pipeline_as_cells.py b/notebooks/00_project_pipeline_as_cells.py new file mode 100644 index 0000000..bec007e --- /dev/null +++ b/notebooks/00_project_pipeline_as_cells.py @@ -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) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..5947bc3 --- /dev/null +++ b/pyproject.toml @@ -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"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..54dc60e --- /dev/null +++ b/requirements.txt @@ -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 diff --git a/scripts/predict_saved.py b/scripts/predict_saved.py new file mode 100644 index 0000000..3c726f3 --- /dev/null +++ b/scripts/predict_saved.py @@ -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() diff --git a/scripts/run_full_experiment.py b/scripts/run_full_experiment.py new file mode 100644 index 0000000..3bfd1b2 --- /dev/null +++ b/scripts/run_full_experiment.py @@ -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() diff --git a/src/ml_crypto_lab/__init__.py b/src/ml_crypto_lab/__init__.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/src/ml_crypto_lab/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/src/ml_crypto_lab/core/artifacts.py b/src/ml_crypto_lab/core/artifacts.py new file mode 100644 index 0000000..99a4359 --- /dev/null +++ b/src/ml_crypto_lab/core/artifacts.py @@ -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) diff --git a/src/ml_crypto_lab/core/config.py b/src/ml_crypto_lab/core/config.py new file mode 100644 index 0000000..3e02ce4 --- /dev/null +++ b/src/ml_crypto_lab/core/config.py @@ -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) diff --git a/src/ml_crypto_lab/core/registry.py b/src/ml_crypto_lab/core/registry.py new file mode 100644 index 0000000..fd4da57 --- /dev/null +++ b/src/ml_crypto_lab/core/registry.py @@ -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") diff --git a/src/ml_crypto_lab/core/types.py b/src/ml_crypto_lab/core/types.py new file mode 100644 index 0000000..573f628 --- /dev/null +++ b/src/ml_crypto_lab/core/types.py @@ -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] diff --git a/src/ml_crypto_lab/data/loading.py b/src/ml_crypto_lab/data/loading.py new file mode 100644 index 0000000..f63aa3a --- /dev/null +++ b/src/ml_crypto_lab/data/loading.py @@ -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"]) diff --git a/src/ml_crypto_lab/ensembles/pipeline.py b/src/ml_crypto_lab/ensembles/pipeline.py new file mode 100644 index 0000000..66a8885 --- /dev/null +++ b/src/ml_crypto_lab/ensembles/pipeline.py @@ -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 {}) diff --git a/src/ml_crypto_lab/ensembles/voting.py b/src/ml_crypto_lab/ensembles/voting.py new file mode 100644 index 0000000..1608b30 --- /dev/null +++ b/src/ml_crypto_lab/ensembles/voting.py @@ -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) diff --git a/src/ml_crypto_lab/evaluation/backtest.py b/src/ml_crypto_lab/evaluation/backtest.py new file mode 100644 index 0000000..de680cb --- /dev/null +++ b/src/ml_crypto_lab/evaluation/backtest.py @@ -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), + } diff --git a/src/ml_crypto_lab/evaluation/metrics.py b/src/ml_crypto_lab/evaluation/metrics.py new file mode 100644 index 0000000..b108228 --- /dev/null +++ b/src/ml_crypto_lab/evaluation/metrics.py @@ -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()), + } diff --git a/src/ml_crypto_lab/features/builders.py b/src/ml_crypto_lab/features/builders.py new file mode 100644 index 0000000..ad19e4a --- /dev/null +++ b/src/ml_crypto_lab/features/builders.py @@ -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) diff --git a/src/ml_crypto_lab/features/pipeline.py b/src/ml_crypto_lab/features/pipeline.py new file mode 100644 index 0000000..748c73f --- /dev/null +++ b/src/ml_crypto_lab/features/pipeline.py @@ -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 diff --git a/src/ml_crypto_lab/features/utils.py b/src/ml_crypto_lab/features/utils.py new file mode 100644 index 0000000..da729e7 --- /dev/null +++ b/src/ml_crypto_lab/features/utils.py @@ -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 diff --git a/src/ml_crypto_lab/inference/predict.py b/src/ml_crypto_lab/inference/predict.py new file mode 100644 index 0000000..ea84003 --- /dev/null +++ b/src/ml_crypto_lab/inference/predict.py @@ -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 diff --git a/src/ml_crypto_lab/models/pipeline.py b/src/ml_crypto_lab/models/pipeline.py new file mode 100644 index 0000000..050b10d --- /dev/null +++ b/src/ml_crypto_lab/models/pipeline.py @@ -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 {}) diff --git a/src/ml_crypto_lab/models/sklearn_models.py b/src/ml_crypto_lab/models/sklearn_models.py new file mode 100644 index 0000000..278cdfa --- /dev/null +++ b/src/ml_crypto_lab/models/sklearn_models.py @@ -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) diff --git a/src/ml_crypto_lab/models/torch_lstm.py b/src/ml_crypto_lab/models/torch_lstm.py new file mode 100644 index 0000000..ccabd0e --- /dev/null +++ b/src/ml_crypto_lab/models/torch_lstm.py @@ -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) diff --git a/src/ml_crypto_lab/targets/builders.py b/src/ml_crypto_lab/targets/builders.py new file mode 100644 index 0000000..07943a2 --- /dev/null +++ b/src/ml_crypto_lab/targets/builders.py @@ -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) diff --git a/src/ml_crypto_lab/targets/pipeline.py b/src/ml_crypto_lab/targets/pipeline.py new file mode 100644 index 0000000..9adee2e --- /dev/null +++ b/src/ml_crypto_lab/targets/pipeline.py @@ -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 diff --git a/src/ml_crypto_lab/targets/utils.py b/src/ml_crypto_lab/targets/utils.py new file mode 100644 index 0000000..6bf39b8 --- /dev/null +++ b/src/ml_crypto_lab/targets/utils.py @@ -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") diff --git a/src/ml_crypto_lab/train/runner.py b/src/ml_crypto_lab/train/runner.py new file mode 100644 index 0000000..fe280a9 --- /dev/null +++ b/src/ml_crypto_lab/train/runner.py @@ -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} diff --git a/src/ml_crypto_lab/train/split.py b/src/ml_crypto_lab/train/split.py new file mode 100644 index 0000000..3ceac66 --- /dev/null +++ b/src/ml_crypto_lab/train/split.py @@ -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} diff --git a/tests/test_imports.py b/tests/test_imports.py new file mode 100644 index 0000000..b194c7c --- /dev/null +++ b/tests/test_imports.py @@ -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()