diff --git a/configs/experiment.yaml b/configs/experiment.yaml index b5b50a6..8336508 100644 --- a/configs/experiment.yaml +++ b/configs/experiment.yaml @@ -101,6 +101,123 @@ feature_sets: persist_wins: [240, 480, 720, 1440] fast_span: 45 + + + # ---------------------------------------------------------- + # BIG NOTEBOOK-LIKE FEATURE SETS + # ---------------------------------------------------------- + # These builders use full market matrices px/buy/sell, not only INDEX OHLC. + # They are implemented in: + # src/ml_crypto_lab/features/fusion_factory.py + # src/ml_crypto_lab/features/fusion_builders.py + + B_big_primitive_pressure: + description: "large primitive pressure table: fusion_score, fusion_force, volume_pressure, index_pressure and additional pressures" + builders: + - name: advanced_fusion_params + params: + prefix: "b_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + include_index_in_cross_section: false + vol_use_mean: true + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_p_list: [1] + vol_roll_sets: [[1600], [2000], [2400]] + force_winsor_q_list: [0.15] + force_base_weight_list: [2.0] + q_weight_scale_list: [0.70] + force_w_price_list: [0.75] + use_base_imbalance_list: [true] + z_roll_wins: [400, 900, 1400, 1900] + breadth_wins: [100, 170, 240, 310, 380] + pct_up_norms: [50] + pct_down_norms: [50] + score_w_list: [60] + score_p_list: [2] + score_roll_sets: [[1400]] + + C_big_mode_values: + description: "large continuous mode values derived from B_big primitive pressures" + builders: + - name: advanced_fusion_mode_values + params: + prefix: "c_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + + D_big_state_features: + description: "large discrete *_state features derived from B_big/C_big" + builders: + - name: advanced_fusion_states + params: + prefix: "d_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + + E_big_agreement_features: + description: "large agreement features by mode and by parameter" + builders: + - name: advanced_fusion_agreements + params: + prefix: "e_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + + F_big_long_context_features: + description: "large long-context feature set derived from primitive pressures" + builders: + - name: advanced_fusion_long_context + params: + prefix: "f_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + lctx_ema_spans: [120, 240, 360, 720, 1440] + lctx_trend_lags: [120, 240, 360, 720] + lctx_cycle_wins: [360, 720, 1440] + lctx_persist_wins: [240, 480, 720, 1440] + lctx_breakout_wins: [360, 720, 1440] + lctx_impulse_lags: [120, 240, 360, 720] + lctx_max_output_cols: 10000 + + ALL_big_designed_features: + description: "full big designed table: params + modes + states + agreements + long_context" + builders: + - name: advanced_fusion_full + params: + prefix: "sig_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + lctx_max_output_cols: 10000 + # ------------------------------------------------------------ # TARGET SETS # ------------------------------------------------------------ @@ -173,7 +290,8 @@ models: # 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] + # Old compact sets are still available. For big notebook-like experiments use the *_big sets. + feature_sets: [A_market_basic, B_big_primitive_pressure, C_big_mode_values, D_big_state_features, E_big_agreement_features, F_big_long_context_features, ALL_big_designed_features] target_sets: [zz_long, future_mean, future_return] models: [logreg, extra_trees, hist_gb] diff --git a/configs/experiment_advanced_fusion.yaml b/configs/experiment_advanced_fusion.yaml new file mode 100644 index 0000000..a71f10c --- /dev/null +++ b/configs/experiment_advanced_fusion.yaml @@ -0,0 +1,307 @@ +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 + + + + # ---------------------------------------------------------- + # BIG NOTEBOOK-LIKE FEATURE SETS + # ---------------------------------------------------------- + # These builders use full market matrices px/buy/sell, not only INDEX OHLC. + # They are implemented in: + # src/ml_crypto_lab/features/fusion_factory.py + # src/ml_crypto_lab/features/fusion_builders.py + + B_big_primitive_pressure: + description: "large primitive pressure table: fusion_score, fusion_force, volume_pressure, index_pressure and additional pressures" + builders: + - name: advanced_fusion_params + params: + prefix: "b_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + include_index_in_cross_section: false + vol_use_mean: true + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_p_list: [1] + vol_roll_sets: [[1600], [2000], [2400]] + force_winsor_q_list: [0.15] + force_base_weight_list: [2.0] + q_weight_scale_list: [0.70] + force_w_price_list: [0.75] + use_base_imbalance_list: [true] + z_roll_wins: [400, 900, 1400, 1900] + breadth_wins: [100, 170, 240, 310, 380] + pct_up_norms: [50] + pct_down_norms: [50] + score_w_list: [60] + score_p_list: [2] + score_roll_sets: [[1400]] + + C_big_mode_values: + description: "large continuous mode values derived from B_big primitive pressures" + builders: + - name: advanced_fusion_mode_values + params: + prefix: "c_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + + D_big_state_features: + description: "large discrete *_state features derived from B_big/C_big" + builders: + - name: advanced_fusion_states + params: + prefix: "d_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + + E_big_agreement_features: + description: "large agreement features by mode and by parameter" + builders: + - name: advanced_fusion_agreements + params: + prefix: "e_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + + F_big_long_context_features: + description: "large long-context feature set derived from primitive pressures" + builders: + - name: advanced_fusion_long_context + params: + prefix: "f_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + lctx_ema_spans: [120, 240, 360, 720, 1440] + lctx_trend_lags: [120, 240, 360, 720] + lctx_cycle_wins: [360, 720, 1440] + lctx_persist_wins: [240, 480, 720, 1440] + lctx_breakout_wins: [360, 720, 1440] + lctx_impulse_lags: [120, 240, 360, 720] + lctx_max_output_cols: 10000 + + ALL_big_designed_features: + description: "full big designed table: params + modes + states + agreements + long_context" + builders: + - name: advanced_fusion_full + params: + prefix: "sig_big__" + factory: + symbol_candle: INDEX + max_symbols: 120 + price_z_wins: [1600] + price_breadth_wins: [500, 900] + vol_w_list: [120, 180] + vol_roll_sets: [[1600], [2000], [2400]] + lctx_max_output_cols: 10000 + +# ------------------------------------------------------------ +# 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: + # Old compact sets are still available. For big notebook-like experiments use the *_big sets. + feature_sets: [B_big_primitive_pressure, C_big_mode_values, D_big_state_features, E_big_agreement_features, F_big_long_context_features, ALL_big_designed_features] + target_sets: [future_mean] + models: [logreg] + +training: + max_targets_per_set: 3 + save_predictions: true + save_models: true + +ensemble: + enabled: true + methods: + - majority_vote + - score_average diff --git a/src/ml_crypto_lab/features/fusion_builders.py b/src/ml_crypto_lab/features/fusion_builders.py new file mode 100644 index 0000000..ac0b287 --- /dev/null +++ b/src/ml_crypto_lab/features/fusion_builders.py @@ -0,0 +1,93 @@ + +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.fusion_factory import FusionFeatureFactory, FusionFeatureBundle, make_fusion_cache_key +from ml_crypto_lab.features.utils import clean_numeric + + +def _extract_market_data(base_df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame | None, pd.DataFrame | None, pd.DataFrame | None]: + """Read market matrices from base_df.attrs. + + train.runner attaches these attrs from data.loading.build_index_ohlc_and_matrices. + If attrs are absent, the factory falls back to single-symbol INDEX data. + """ + md = base_df.attrs.get("market_data", {}) or {} + ohlc = md.get("ohlc", base_df[["open", "high", "low", "close"]].copy()) + px = md.get("px", None) + buy = md.get("buy", None) + sell = md.get("sell", None) + return ohlc, px, buy, sell + + +def _get_bundle(base_df: pd.DataFrame, cfg: dict[str, Any]) -> FusionFeatureBundle: + cache_key = "advanced_fusion__" + make_fusion_cache_key(cfg) + cache = base_df.attrs.setdefault("feature_cache", {}) + if cache_key in cache: + return cache[cache_key] + ohlc, px, buy, sell = _extract_market_data(base_df) + bundle = FusionFeatureFactory(ohlc=ohlc, px=px, buy=buy, sell=sell, cfg=cfg).build() + cache[cache_key] = bundle + return bundle + + +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 clean_numeric(out) + + +@FEATURE_REGISTRY.register("advanced_fusion_params") +def build_advanced_fusion_params(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame: + """Primitive pressure parameters: fusion_score, fusion_force, volume_pressure, index_pressure, etc.""" + prefix = cfg.get("prefix", "b_big__") + bundle = _get_bundle(base_df, cfg.get("factory", cfg)) + return _add_prefix(bundle.params_df.reindex(base_df.index).ffill().fillna(0.0), prefix) + + +@FEATURE_REGISTRY.register("advanced_fusion_mode_values") +def build_advanced_fusion_mode_values(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame: + """Continuous mode values: level, speed, accel, cycle, persistence, etc.""" + prefix = cfg.get("prefix", "c_big__") + bundle = _get_bundle(base_df, cfg.get("factory", cfg)) + parts = [bundle.extended_mode_raw_df, bundle.extended_mode_values_df] + df = pd.concat(parts, axis=1).reindex(base_df.index).ffill().fillna(0.0) + return _add_prefix(df, prefix) + + +@FEATURE_REGISTRY.register("advanced_fusion_states") +def build_advanced_fusion_states(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame: + """Discrete state features derived from the primitive pressure mode values.""" + prefix = cfg.get("prefix", "d_big__") + bundle = _get_bundle(base_df, cfg.get("factory", cfg)) + return _add_prefix(bundle.extended_mode_states_df.reindex(base_df.index).ffill().fillna(0.0), prefix) + + +@FEATURE_REGISTRY.register("advanced_fusion_agreements") +def build_advanced_fusion_agreements(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame: + """Agreement features across mode states and across parameters.""" + prefix = cfg.get("prefix", "e_big__") + bundle = _get_bundle(base_df, cfg.get("factory", cfg)) + df = pd.concat([bundle.agreement_by_mode_df, bundle.agreement_by_param_df], axis=1) + df = df.reindex(base_df.index).ffill().fillna(0.0) + return _add_prefix(df, prefix) + + +@FEATURE_REGISTRY.register("advanced_fusion_long_context") +def build_advanced_fusion_long_context(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame: + """Long-context features built on primitive pressure parameters.""" + prefix = cfg.get("prefix", "f_big__") + bundle = _get_bundle(base_df, cfg.get("factory", cfg)) + return _add_prefix(bundle.long_context_all_df.reindex(base_df.index).ffill().fillna(0.0), prefix) + + +@FEATURE_REGISTRY.register("advanced_fusion_full") +def build_advanced_fusion_full(base_df: pd.DataFrame, cfg: dict[str, Any]) -> pd.DataFrame: + """Full big designed feature table: primitive params + modes + states + agreements + long context.""" + prefix = cfg.get("prefix", "sig_big__") + bundle = _get_bundle(base_df, cfg.get("factory", cfg)) + return _add_prefix(bundle.state_details_df.reindex(base_df.index).ffill().fillna(0.0), prefix) diff --git a/src/ml_crypto_lab/features/fusion_factory.py b/src/ml_crypto_lab/features/fusion_factory.py new file mode 100644 index 0000000..0e57f0a --- /dev/null +++ b/src/ml_crypto_lab/features/fusion_factory.py @@ -0,0 +1,1050 @@ + +from __future__ import annotations + +from dataclasses import dataclass +from itertools import product +from typing import Any +import hashlib +import json + +import numpy as np +import pandas as pd +from numpy.lib.stride_tricks import sliding_window_view + +EPS = 1e-9 + + +@dataclass +class FusionFeatureConfig: + """Configuration for large cross-sectional market feature engineering.""" + + symbol_candle: str = "INDEX" + max_symbols: int | None = 120 + include_index_in_cross_section: bool = False + vol_use_mean: bool = True + + # fusion_force + q_levels: tuple[float, ...] = (0.25, 0.50, 0.70, 0.85, 0.93, 0.97) + q_weights_base: tuple[float, ...] = (0.7, 0.9, 1.0, 1.2, 1.4, 1.7) + price_z_wins: tuple[int, ...] = (1600,) + price_breadth_wins: tuple[int, ...] = (500, 900) + vol_w_list: tuple[int, ...] = (120, 180) + vol_p_list: tuple[int, ...] = (1,) + vol_roll_sets: tuple[tuple[int, ...], ...] = ((1600,), (2000,), (2400,)) + force_winsor_q_list: tuple[float, ...] = (0.15,) + force_base_weight_list: tuple[float, ...] = (2.0,) + q_weight_scale_list: tuple[float, ...] = (0.70,) + force_w_price_list: tuple[float, ...] = (0.75,) + use_base_imbalance_list: tuple[bool, ...] = (True,) + force_norm_span: int = 300 + force_norm_k: float = 1.50 + + # fusion_score + z_roll_wins: tuple[int, ...] = (400, 900, 1400, 1900) + breadth_wins: tuple[int, ...] = (100, 170, 240, 310, 380) + pct_up_norms: tuple[float, ...] = (50.0,) + pct_down_norms: tuple[float, ...] = (50.0,) + score_w_list: tuple[int, ...] = (60,) + score_p_list: tuple[int, ...] = (2,) + score_roll_sets: tuple[tuple[int, ...], ...] = ((1400,),) + vol_strength_norm_win: int = 1984 + fusion_w_price: float = 1.0 + fusion_w_vol: float = 0.0 + fusion_agree_bonus: float = 0.0 + + # primitive pressure windows + volume_pressure_norm_win: int = 1440 + volume_pressure_smooth: int = 30 + volume_pressure_agg_method: str = "median" + vp_norm_span: int = 300 + vp_norm_k: float = 1.50 + + ipp_fast_ema: int = 34 + ipp_slow_ema: int = 233 + ipp_mom_lag: int = 21 + ipp_vol_span: int = 300 + ipp_candle_smooth: int = 12 + ipp_w_trend: float = 0.35 + ipp_w_momentum: float = 0.35 + ipp_w_location: float = 0.20 + ipp_w_candle: float = 0.10 + ipp_norm_span: int = 300 + ipp_norm_k: float = 1.50 + + btp_ema_span: int = 233 + btp_ret_lag: int = 34 + btp_smooth: int = 21 + btp_w_location: float = 0.60 + btp_w_momentum: float = 0.40 + btp_norm_span: int = 300 + btp_norm_k: float = 1.50 + + tep_win: int = 144 + tep_smooth: int = 13 + tep_norm_span: int = 300 + tep_norm_k: float = 1.50 + + rsp_lag: int = 55 + rsp_smooth: int = 21 + rsp_norm_span: int = 300 + rsp_norm_k: float = 1.50 + + fpp_price_lag: int = 21 + fpp_price_norm_span: int = 300 + fpp_price_norm_k: float = 1.50 + fpp_smooth: int = 13 + fpp_w_flow: float = 0.60 + fpp_w_price: float = 0.40 + fpp_agree_boost: float = 0.25 + + # additional pressures + ibp_win: int = 360 + ibp_smooth: int = 13 + iip_fast_lag: int = 13 + iip_slow_lag: int = 55 + vdp_fast_vol: int = 34 + vdp_slow_vol: int = 233 + vdp_dir_lag: int = 21 + vdp_smooth: int = 13 + mfp_norm_win: int = 1440 + mfp_smooth: int = 21 + bmp_lag: int = 34 + bmp_smooth: int = 21 + eap_ema_spans: tuple[int, ...] = (21, 55, 144, 233, 377) + eap_smooth: int = 13 + pp_dir_lag: int = 21 + pp_vol_win: int = 1440 + pp_smooth: int = 13 + cap_lag: int = 34 + cap_smooth: int = 21 + dsp_ret_lag: int = 34 + dsp_win: int = 720 + dsp_dir_lag: int = 34 + dsp_smooth: int = 21 + bap_lag: int = 34 + bap_diff_lag: int = 13 + bap_smooth: int = 13 + prp_range_win: int = 720 + prp_smooth: int = 13 + + # mode/state features + mode_level_deadband: float = 0.00 + mode_speed_ema_span: int = 45 + mode_speed_diff_lag: int = 5 + mode_speed_norm_span: int = 100 + mode_speed_norm_k: float = 1.25 + mode_speed_deadband: float = 0.10 + mode_accel_ema_span: int = 13 + mode_accel_diff_lag: int = 5 + mode_accel_norm_span: int = 180 + mode_accel_norm_k: float = 1.25 + mode_accel_deadband: float = 0.10 + mode_cycle_roll_win: int = 360 + mode_cycle_smooth: int = 9 + mode_cycle_deadband: float = 0.10 + mode_reversion_cycle_deadband: float = 0.70 + mode_breakout_win: int = 240 + mode_persist_win: int = 120 + mode_persist_deadband: float = 0.20 + mode_strength_ema_span: int = 21 + mode_strength_diff_lag: int = 5 + mode_strength_norm_span: int = 180 + mode_strength_norm_k: float = 1.25 + mode_strength_deadband: float = 0.10 + mode_pullback_trend_deadband: float = 0.10 + mode_pullback_cycle_deadband: float = 0.50 + + # long context + lctx_ema_spans: tuple[int, ...] = (120, 240, 360, 720, 1440) + lctx_trend_lags: tuple[int, ...] = (120, 240, 360, 720) + lctx_cycle_wins: tuple[int, ...] = (360, 720, 1440) + lctx_persist_wins: tuple[int, ...] = (240, 480, 720, 1440) + lctx_breakout_wins: tuple[int, ...] = (360, 720, 1440) + lctx_impulse_lags: tuple[int, ...] = (120, 240, 360, 720) + lctx_fast_ema_for_spread: int = 45 + lctx_deadband_level: float = 0.05 + lctx_deadband_trend: float = 0.08 + lctx_deadband_cycle: float = 0.15 + lctx_deadband_persist: float = 0.20 + lctx_deadband_impulse: float = 0.10 + lctx_deadband_spread: float = 0.08 + lctx_max_output_cols: int | None = 10000 + + @staticmethod + def from_dict(cfg: dict[str, Any] | None) -> "FusionFeatureConfig": + cfg = cfg or {} + field_names = set(FusionFeatureConfig.__dataclass_fields__.keys()) + clean = {k: v for k, v in cfg.items() if k in field_names} + # Convert list-like values into tuples for hashability/reproducibility. + for k, v in list(clean.items()): + if isinstance(v, list): + if v and all(isinstance(x, list) for x in v): + clean[k] = tuple(tuple(int(y) for y in x) for x in v) + else: + clean[k] = tuple(v) + return FusionFeatureConfig(**clean) + + +@dataclass +class FusionFeatureBundle: + params_df: pd.DataFrame + extended_mode_raw_df: pd.DataFrame + extended_mode_values_df: pd.DataFrame + extended_mode_states_df: pd.DataFrame + agreement_by_mode_df: pd.DataFrame + agreement_by_param_df: pd.DataFrame + long_context_values_df: pd.DataFrame + long_context_states_df: pd.DataFrame + long_context_agreement_df: pd.DataFrame + long_context_all_df: pd.DataFrame + state_details_df: pd.DataFrame + + +def _clean_numeric(df: pd.DataFrame, dtype=np.float32) -> 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(dtype) + + +def _safe_col_name(x: str) -> str: + return ( + str(x) + .replace(" ", "_") + .replace("/", "_") + .replace("|", "_") + .replace(":", "_") + .replace("#", "n") + .replace(".", "p") + .replace("-", "m") + ) + + +def _sign_np(x): + x = np.asarray(x, dtype=float) + return np.where(x > 0.0, 1.0, np.where(x < 0.0, -1.0, 0.0)) + + +def _make_weights(W: int, P: int) -> np.ndarray: + w = np.arange(1, int(W) + 1, dtype=float) ** int(P) + w /= w.sum() + return w + + +def _weighted_roll_valid(arr: np.ndarray, W: int, P: int) -> np.ndarray: + W = int(W) + if W <= 1: + return np.asarray(arr, dtype=np.float64) + if arr.shape[0] < W: + return np.empty((0, arr.shape[1]), dtype=np.float64) + w = _make_weights(W, P) + sw = sliding_window_view(np.asarray(arr, dtype=np.float64), window_shape=W, axis=0) + return np.tensordot(sw, w, axes=([-1], [0])) + + +def _prefix_sums(arr: np.ndarray): + arr = np.asarray(arr, dtype=np.float64) + cs = np.empty((arr.shape[0] + 1, arr.shape[1]), dtype=np.float64) + cs[0] = 0.0 + np.cumsum(arr, axis=0, out=cs[1:]) + cs2 = np.empty((arr.shape[0] + 1, arr.shape[1]), dtype=np.float64) + cs2[0] = 0.0 + np.cumsum(arr * arr, axis=0, out=cs2[1:]) + return cs, cs2 + + +def _rolling_mean_std_from_prefix(cs: np.ndarray, cs2: np.ndarray, rw: int): + rw = int(rw) + s = cs[rw:] - cs[:-rw] + s2 = cs2[rw:] - cs2[:-rw] + mu = s / rw + var = np.maximum((s2 / rw) - mu * mu, 0.0) + sd = np.sqrt(var) + return mu, sd + + +def _causal_zscore(s: pd.Series, win: int, min_periods: int | None = None) -> pd.Series: + s = pd.Series(s).astype(float) + win = int(max(2, win)) + if min_periods is None: + min_periods = max(20, win // 10) + 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(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) + out = np.tanh(s / (scale * float(k) + EPS)) + return pd.Series(out, index=s.index, dtype=float).clip(-1.0, 1.0) + + +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(int(window), min_periods=1).mean().replace([np.inf, -np.inf], np.nan).ffill().fillna(0.0) + + +def _rowwise_winsorize(arr2d: np.ndarray, q: float = 0.05) -> np.ndarray: + arr = np.asarray(arr2d, dtype=float) + if arr.ndim != 2: + raise ValueError("rowwise_winsorize expects 2D array") + if arr.shape[1] == 0: + return arr.copy() + lo = np.quantile(arr, q, axis=1, keepdims=True) + hi = np.quantile(arr, 1.0 - q, axis=1, keepdims=True) + return np.clip(arr, lo, hi) + + +def _cross_section_agg(arr2d: np.ndarray, use_mean: bool = False) -> np.ndarray: + if arr2d.size == 0: + return np.zeros(arr2d.shape[0], dtype=float) + return arr2d.mean(axis=1) if use_mean else np.median(arr2d, axis=1) + + +class FusionFeatureFactory: + """Builds a large family of market-regime features from market matrices. + + Inputs are the same structural objects that exist in the notebook: + - ``ohlc``: INDEX open/high/low/close indexed by time + - ``px``: close matrix, rows=time, columns=symbol + - ``buy``: buy-volume matrix + - ``sell``: sell-volume matrix + + The factory builds primitive continuous parameters first + (fusion_score, fusion_force, volume_pressure, index_pressure, etc.), + then derives mode values, state features, agreement features, and + long-context features. + """ + + def __init__( + self, + ohlc: pd.DataFrame, + px: pd.DataFrame | None = None, + buy: pd.DataFrame | None = None, + sell: pd.DataFrame | None = None, + cfg: FusionFeatureConfig | dict[str, Any] | None = None, + ): + self.cfg = cfg if isinstance(cfg, FusionFeatureConfig) else FusionFeatureConfig.from_dict(cfg) + self.ohlc = self._prepare_ohlc(ohlc) + self.px = self._prepare_matrix(px, fallback_col="close") + self.buy = self._prepare_matrix(buy, fallback_col=None).reindex(self.px.index).fillna(0.0) + self.sell = self._prepare_matrix(sell, fallback_col=None).reindex(self.px.index).fillna(0.0) + self.index = self.ohlc.index.intersection(self.px.index).sort_values() + self.ohlc = self.ohlc.reindex(self.index).ffill().dropna() + self.px = self.px.reindex(self.index).ffill().dropna(how="all") + self.buy = self.buy.reindex(self.index).fillna(0.0) + self.sell = self.sell.reindex(self.index).fillna(0.0) + self.index = self.ohlc.index.intersection(self.px.index).sort_values() + + def _prepare_ohlc(self, ohlc: pd.DataFrame) -> pd.DataFrame: + required = ["open", "high", "low", "close"] + missing = [c for c in required if c not in ohlc.columns] + if missing: + raise ValueError(f"FusionFeatureFactory ohlc missing columns: {missing}") + out = ohlc[required].copy() + out.index = pd.to_datetime(out.index) + out = out[~out.index.duplicated(keep="last")].sort_index() + return out.replace([np.inf, -np.inf], np.nan).ffill().dropna() + + def _prepare_matrix(self, matrix: pd.DataFrame | None, fallback_col: str | None) -> pd.DataFrame: + if matrix is not None and not matrix.empty: + out = matrix.copy() + out.index = pd.to_datetime(out.index) + out = out[~out.index.duplicated(keep="last")].sort_index() + return out.replace([np.inf, -np.inf], np.nan).ffill().fillna(0.0) + if fallback_col is not None: + return self.ohlc[[fallback_col]].rename(columns={fallback_col: self.cfg.symbol_candle}).copy() + return pd.DataFrame(0.0, index=self.ohlc.index, columns=[self.cfg.symbol_candle]) + + def build(self) -> FusionFeatureBundle: + params_df = self.build_params() + ( + mode_raw_df, + mode_values_df, + mode_states_df, + agreement_by_mode_df, + agreement_by_param_df, + ) = self.build_modes_states_agreements(params_df) + ( + long_values_df, + long_states_df, + long_agreement_df, + long_all_df, + ) = self.build_long_context(params_df) + + state_details_df = pd.concat( + [ + params_df, + mode_raw_df, + mode_values_df, + mode_states_df, + agreement_by_mode_df, + agreement_by_param_df, + long_all_df, + ], + axis=1, + ).reindex(self.index).ffill().fillna(0.0) + state_details_df = state_details_df.loc[:, ~state_details_df.columns.duplicated()] + state_details_df = _clean_numeric(state_details_df) + + return FusionFeatureBundle( + params_df=params_df, + extended_mode_raw_df=mode_raw_df, + extended_mode_values_df=mode_values_df, + extended_mode_states_df=mode_states_df, + agreement_by_mode_df=agreement_by_mode_df, + agreement_by_param_df=agreement_by_param_df, + long_context_values_df=long_values_df, + long_context_states_df=long_states_df, + long_context_agreement_df=long_agreement_df, + long_context_all_df=long_all_df, + state_details_df=state_details_df, + ) + + def _symbols(self, include_index: bool = False) -> list[str]: + cols = list(self.px.columns) + if not include_index: + cols = [c for c in cols if c != self.cfg.symbol_candle] + if self.cfg.max_symbols is not None: + cols = cols[: int(self.cfg.max_symbols)] + if not cols: + cols = list(self.px.columns)[:1] + return cols + + def build_params(self) -> pd.DataFrame: + parts = [ + self._build_fusion_force(), + self._build_fusion_score(), + self._build_volume_pressure(), + self._build_index_pressure(), + self._build_breadth_trend_pressure(), + self._build_trend_efficiency_pressure(), + self._build_relative_strength_pressure(), + self._build_flow_price_pressure(), + self._build_additional_pressures_v1(), + self._build_additional_pressures_v2(), + ] + out = pd.concat(parts, axis=1).reindex(self.index).ffill().fillna(0.0).clip(-1.0, 1.0) + out = out.loc[:, ~out.columns.duplicated()] + return _clean_numeric(out) + + def _quantile_pressure_parts(self, arr2d: np.ndarray, winsor_q: float, use_base_imbalance: bool): + arr = np.asarray(arr2d, dtype=np.float32) + arr = np.nan_to_num(arr, nan=0.0, posinf=0.0, neginf=0.0) + arr = _rowwise_winsorize(arr, q=float(winsor_q)).astype(np.float32, copy=False) + pos = np.maximum(arr, 0.0) + neg = np.maximum(-arr, 0.0) + if use_base_imbalance: + base_part = (pos.mean(axis=1) - neg.mean(axis=1)).astype(np.float32) + else: + base_part = np.zeros(arr.shape[0], dtype=np.float32) + tail_part = np.zeros(arr.shape[0], dtype=np.float64) + for q, w in zip(self.cfg.q_levels, self.cfg.q_weights_base): + thr_up = np.quantile(pos, float(q), axis=1, keepdims=True) + thr_dn = np.quantile(neg, float(q), axis=1, keepdims=True) + up_excess = np.maximum(pos - thr_up, 0.0) + dn_excess = np.maximum(neg - thr_dn, 0.0) + tail_part += float(w) * (up_excess.mean(axis=1) - dn_excess.mean(axis=1)) + return base_part, tail_part.astype(np.float32) + + def _build_fusion_force(self) -> pd.DataFrame: + cfg = self.cfg + syms_price = self._symbols(include_index=cfg.include_index_in_cross_section) + syms_vol = self._symbols(include_index=False) + n = len(self.index) + out_idx = self.index + + # Price-stage variants. + logp = np.log(self.px[syms_price].clip(lower=EPS).to_numpy(dtype=np.float64, copy=False)) + price_cache: dict[tuple, dict[str, np.ndarray]] = {} + for z in cfg.price_z_wins: + z = int(max(2, z)) + if n < z: + continue + cs, cs2 = _prefix_sums(logp) + mu, sd = _rolling_mean_std_from_prefix(cs, cs2, z) + pos = np.full(logp.shape, np.nan, dtype=np.float32) + pos[z - 1:] = ((logp[z - 1:] - mu) / (sd + EPS)).astype(np.float32) + for b in cfg.price_breadth_wins: + b = int(max(1, b)) + delta = np.full_like(pos, np.nan, dtype=np.float32) + if b < n: + delta[b:] = pos[b:] - pos[:-b] + force_mat = np.nan_to_num(delta, nan=0.0, posinf=0.0, neginf=0.0) + for winsor_q in cfg.force_winsor_q_list: + for ubi in cfg.use_base_imbalance_list: + base, tail = self._quantile_pressure_parts(force_mat, float(winsor_q), bool(ubi)) + price_cache[(z, b, float(winsor_q), bool(ubi))] = {"base": base, "tail": tail} + + # Volume-stage variants. + buy_arr = self.buy[syms_vol].to_numpy(dtype=np.float64, copy=False) + sell_arr = self.sell[syms_vol].to_numpy(dtype=np.float64, copy=False) + volume_cache: dict[tuple, dict[str, np.ndarray]] = {} + unique_rws = sorted({int(rw) for roll_set in cfg.vol_roll_sets for rw in roll_set}) + for W in cfg.vol_w_list: + W = int(max(2, W)) + if n < W: + continue + for P in cfg.vol_p_list: + buy_s = _weighted_roll_valid(buy_arr, W=W, P=int(P)) + sell_s = _weighted_roll_valid(sell_arr, W=W, P=int(P)) + if buy_s.shape[0] == 0: + continue + lbuy = np.log1p(buy_s).astype(np.float64, copy=False) + lsell = np.log1p(sell_s).astype(np.float64, copy=False) + cs_b, cs2_b = _prefix_sums(lbuy) + cs_s, cs2_s = _prefix_sums(lsell) + n_valid = lbuy.shape[0] + for rw in unique_rws: + rw = int(max(2, rw)) + if n_valid < rw: + continue + mu_b, sd_b = _rolling_mean_std_from_prefix(cs_b, cs2_b, rw) + mu_s, sd_s = _rolling_mean_std_from_prefix(cs_s, cs2_s, rw) + out_start = rw - 1 + buy_est = (lbuy[out_start:] - mu_b) / (sd_b + EPS) + sell_est = (lsell[out_start:] - mu_s) / (sd_s + EPS) + flow_strength_short = (buy_est - sell_est).astype(np.float32) + flow_strength = np.zeros((n, len(syms_vol)), dtype=np.float32) + start_i = (W - 1) + out_start + if start_i < n: + flow_strength[start_i:start_i + flow_strength_short.shape[0]] = flow_strength_short[: max(0, n - start_i)] + for winsor_q in cfg.force_winsor_q_list: + for ubi in cfg.use_base_imbalance_list: + base, tail = self._quantile_pressure_parts(flow_strength, float(winsor_q), bool(ubi)) + volume_cache[(W, int(P), rw, float(winsor_q), bool(ubi))] = {"base": base, "tail": tail} + + variant_series = {} + i = 0 + for z, b, W, P, roll_set, winsor_q, base_weight, q_scale, w_price, ubi in product( + cfg.price_z_wins, + cfg.price_breadth_wins, + cfg.vol_w_list, + cfg.vol_p_list, + cfg.vol_roll_sets, + cfg.force_winsor_q_list, + cfg.force_base_weight_list, + cfg.q_weight_scale_list, + cfg.force_w_price_list, + cfg.use_base_imbalance_list, + ): + p_key = (int(max(2, z)), int(max(1, b)), float(winsor_q), bool(ubi)) + if p_key in price_cache: + comp = price_cache[p_key] + price_raw = float(base_weight) * comp["base"] + float(q_scale) * comp["tail"] + else: + price_raw = np.zeros(n, dtype=np.float32) + vol_sum = np.zeros(n, dtype=np.float32) + vol_tail = np.zeros(n, dtype=np.float32) + ok = False + for rw in roll_set: + v_key = (int(max(2, W)), int(P), int(max(2, rw)), float(winsor_q), bool(ubi)) + if v_key in volume_cache: + ok = True + vol_sum += volume_cache[v_key]["base"] + vol_tail += volume_cache[v_key]["tail"] + if ok: + vol_raw = float(base_weight) * vol_sum + float(q_scale) * vol_tail + else: + vol_raw = np.zeros(n, dtype=np.float32) + raw = float(w_price) * price_raw + (1.0 - float(w_price)) * vol_raw + variant_series[f"ff_variant_{i:05d}"] = pd.Series(raw, index=out_idx, dtype=float) + i += 1 + + if not variant_series: + force_raw = pd.Series(0.0, index=out_idx, dtype=float) + else: + df = pd.DataFrame(variant_series, index=out_idx).astype(np.float32) + sum_pos = df.where(df > 0, 0.0).sum(axis=1) + sum_neg = df.where(df < 0, 0.0).sum(axis=1) + force_raw = abs(sum_pos) - abs(sum_neg) + force = _tanh_normalize(force_raw, span=cfg.force_norm_span, k=cfg.force_norm_k).rename("fusion_force") + return force.to_frame() + + def _build_fusion_score(self) -> pd.DataFrame: + cfg = self.cfg + syms_price = self._symbols(include_index=cfg.include_index_in_cross_section) + syms_vol = self._symbols(include_index=False) + idx = self.index + logp = np.log(self.px[syms_price].clip(lower=EPS)) + signal_cols = {} + combo_id = 0 + for z in cfg.z_roll_wins: + z = int(max(2, z)) + mu = logp.rolling(z, min_periods=max(20, z // 10)).mean() + sd = logp.rolling(z, min_periods=max(20, z // 10)).std(ddof=0) + pos = (logp - mu) / (sd + EPS) + for bw in cfg.breadth_wins: + bw = int(max(1, bw)) + delta = pos - pos.shift(bw) + up_cnt = delta.gt(0).sum(axis=1) + down_cnt = delta.lt(0).sum(axis=1) + active_cnt = (up_cnt + down_cnt).replace(0, np.nan) + up_pct = (up_cnt / active_cnt * 100.0).fillna(0.0) + down_pct = (down_cnt / active_cnt * 100.0).fillna(0.0) + for up_thr in cfg.pct_up_norms: + for down_thr in cfg.pct_down_norms: + signal_cols[f"price_{combo_id:04d}"] = np.where(up_pct > up_thr, 1, np.where(down_pct > down_thr, -1, 0)) + combo_id += 1 + if signal_cols: + price_signal_df = pd.DataFrame(signal_cols, index=idx) + price_score = ((price_signal_df.eq(1).mean(axis=1) - price_signal_df.eq(-1).mean(axis=1))).clip(-1, 1) + price_bull = price_signal_df.eq(1).mean(axis=1) * 100.0 + price_bear = price_signal_df.eq(-1).mean(axis=1) * 100.0 + else: + price_score = pd.Series(0.0, index=idx) + price_bull = pd.Series(0.0, index=idx) + price_bear = pd.Series(0.0, index=idx) + + buy_arr = self.buy[syms_vol].to_numpy(dtype=np.float64, copy=False) + sell_arr = self.sell[syms_vol].to_numpy(dtype=np.float64, copy=False) + vol_signal_cols = {} + vol_buy_cols = {} + vol_sell_cols = {} + combo_id = 0 + all_rws = sorted({int(rw) for rs in cfg.score_roll_sets for rw in rs}) + for W in cfg.score_w_list: + W = int(max(2, W)) + if len(idx) < W: + continue + for P in cfg.score_p_list: + buy_s = _weighted_roll_valid(buy_arr, W=W, P=int(P)) + sell_s = _weighted_roll_valid(sell_arr, W=W, P=int(P)) + if buy_s.shape[0] == 0: + continue + lbuy = np.log1p(buy_s) + lsell = np.log1p(sell_s) + cs_b, cs2_b = _prefix_sums(lbuy) + cs_s, cs2_s = _prefix_sums(lsell) + sig_by_rw = {} + buy_by_rw = {} + sell_by_rw = {} + for rw in all_rws: + rw = int(max(2, rw)) + if lbuy.shape[0] < rw: + continue + mu_b, sd_b = _rolling_mean_std_from_prefix(cs_b, cs2_b, rw) + mu_s, sd_s = _rolling_mean_std_from_prefix(cs_s, cs2_s, rw) + out_start = rw - 1 + buy_est = (lbuy[out_start:] - mu_b) / (sd_b + EPS) + sell_est = (lsell[out_start:] - mu_s) / (sd_s + EPS) + buy_agg = _cross_section_agg(buy_est, use_mean=cfg.vol_use_mean) + sell_agg = _cross_section_agg(sell_est, use_mean=cfg.vol_use_mean) + full = np.zeros(len(idx), dtype=float) + full_buy = np.zeros(len(idx), dtype=float) + full_sell = np.zeros(len(idx), dtype=float) + start_i = (W - 1) + out_start + length = max(0, min(len(idx) - start_i, len(buy_agg))) + if length > 0: + full[start_i:start_i + length] = _sign_np(buy_agg[:length] - sell_agg[:length]) + full_buy[start_i:start_i + length] = buy_agg[:length] + full_sell[start_i:start_i + length] = sell_agg[:length] + sig_by_rw[rw] = full + buy_by_rw[rw] = full_buy + sell_by_rw[rw] = full_sell + for roll_set_id, rws in enumerate(cfg.score_roll_sets): + ok = [int(rw) for rw in rws if int(rw) in sig_by_rw] + if not ok: + continue + sig_stack = np.column_stack([sig_by_rw[rw] for rw in ok]) + buy_stack = np.column_stack([buy_by_rw[rw] for rw in ok]) + sell_stack = np.column_stack([sell_by_rw[rw] for rw in ok]) + vol_signal_cols[f"vol_{combo_id:04d}"] = _sign_np(np.median(sig_stack, axis=1)) + vol_buy_cols[f"vol_{combo_id:04d}"] = np.median(buy_stack, axis=1) + vol_sell_cols[f"vol_{combo_id:04d}"] = np.median(sell_stack, axis=1) + combo_id += 1 + if vol_signal_cols: + vol_signal_df = pd.DataFrame(vol_signal_cols, index=idx) + vol_combo_score = (vol_signal_df.eq(1).mean(axis=1) - vol_signal_df.eq(-1).mean(axis=1)).clip(-1, 1) + buy_med = pd.DataFrame(vol_buy_cols, index=idx).median(axis=1) + sell_med = pd.DataFrame(vol_sell_cols, index=idx).median(axis=1) + spread_z = _causal_zscore(buy_med - sell_med, cfg.vol_strength_norm_win) + sp_score = pd.Series(np.tanh(spread_z / 2.0), index=idx) + vol_score = (0.7 * vol_combo_score + 0.3 * sp_score).clip(-1, 1) + vol_bull = vol_signal_df.eq(1).mean(axis=1) * 100.0 + vol_bear = vol_signal_df.eq(-1).mean(axis=1) * 100.0 + else: + vol_score = pd.Series(0.0, index=idx) + vol_bull = pd.Series(0.0, index=idx) + vol_bear = pd.Series(0.0, index=idx) + + price_conf = ((price_bull - price_bear).abs() / 100.0).clip(0, 1) + vol_conf = ((vol_bull - vol_bear).abs() / 100.0).clip(0, 1) + ps = pd.Series(_sign_np(price_score), index=idx) + vs = pd.Series(_sign_np(vol_score), index=idx) + agree = pd.Series(np.where((ps == vs) & (ps != 0), ps * np.minimum(price_conf, vol_conf), 0.0), index=idx) + fusion_score = (cfg.fusion_w_price * price_score + cfg.fusion_w_vol * vol_score + cfg.fusion_agree_bonus * agree).clip(-1, 1) + return fusion_score.rename("fusion_score").to_frame() + + def _build_volume_pressure(self) -> pd.DataFrame: + cfg = self.cfg + syms = self._symbols(include_index=False) + buy = self.buy[syms].fillna(0.0) + sell = self.sell[syms].fillna(0.0) + x_buy = np.log1p(buy.astype(float)) + x_sell = np.log1p(sell.astype(float)) + minp = max(20, cfg.volume_pressure_norm_win // 10) + buy_z = ((x_buy - x_buy.rolling(cfg.volume_pressure_norm_win, min_periods=minp).mean().shift(1)) / + (x_buy.rolling(cfg.volume_pressure_norm_win, min_periods=minp).std(ddof=0).shift(1) + EPS)).fillna(0.0).clip(-5, 5) + sell_z = ((x_sell - x_sell.rolling(cfg.volume_pressure_norm_win, min_periods=minp).mean().shift(1)) / + (x_sell.rolling(cfg.volume_pressure_norm_win, min_periods=minp).std(ddof=0).shift(1) + EPS)).fillna(0.0).clip(-5, 5) + pressure = (buy_z - sell_z).clip(-5, 5) + raw = pressure.mean(axis=1) if cfg.volume_pressure_agg_method == "mean" else pressure.median(axis=1) + if cfg.volume_pressure_smooth > 1: + raw = raw.rolling(cfg.volume_pressure_smooth, min_periods=1).mean() + vp = _tanh_normalize(raw, span=cfg.vp_norm_span, k=cfg.vp_norm_k).rename("volume_pressure") + return vp.to_frame() + + def _build_index_pressure(self) -> pd.DataFrame: + cfg = self.cfg + o = self.ohlc["open"].astype(float) + h = self.ohlc["high"].astype(float) + l = self.ohlc["low"].astype(float) + c = self.ohlc["close"].astype(float) + ret1 = c.diff().fillna(0.0) + vol = ret1.abs().shift(1).ewm(span=cfg.ipp_vol_span, adjust=False, min_periods=1).mean() + fallback = float(ret1.abs().median()) or 1.0 + vol = vol.replace(0, np.nan).ffill().fillna(fallback).replace(0, EPS) + ema_fast = c.ewm(span=cfg.ipp_fast_ema, adjust=False, min_periods=1).mean() + ema_slow = c.ewm(span=cfg.ipp_slow_ema, adjust=False, min_periods=1).mean() + trend = np.tanh(((ema_fast - ema_slow) / (vol * np.sqrt(cfg.ipp_slow_ema / 4.0) + EPS)).fillna(0.0) / 1.5) + mom = np.tanh((c.diff(cfg.ipp_mom_lag).fillna(0.0) / (vol * np.sqrt(cfg.ipp_mom_lag) + EPS)).fillna(0.0) / 1.5) + loc = np.tanh(((c - ema_slow) / (vol * np.sqrt(cfg.ipp_slow_ema / 2.0) + EPS)).fillna(0.0) / 1.5) + rng = (h - l).replace(0, np.nan) + candle = ((c - o) / (rng + EPS)).clip(-1, 1).fillna(0.0).rolling(cfg.ipp_candle_smooth, min_periods=1).mean() + raw = cfg.ipp_w_trend * trend + cfg.ipp_w_momentum * mom + cfg.ipp_w_location * loc + cfg.ipp_w_candle * candle + ip = _tanh_normalize(pd.Series(raw, index=self.index), span=cfg.ipp_norm_span, k=cfg.ipp_norm_k).rename("index_pressure") + return ip.to_frame() + + def _build_breadth_trend_pressure(self) -> pd.DataFrame: + cfg = self.cfg + syms = self._symbols(include_index=False) + px = self.px[syms].ffill() + ema = px.ewm(span=cfg.btp_ema_span, adjust=False, min_periods=1).mean() + location = ((px > ema).sum(axis=1) - (px < ema).sum(axis=1)) / max(1, len(syms)) + ret = px.pct_change(cfg.btp_ret_lag).replace([np.inf, -np.inf], np.nan).fillna(0.0) + momentum = ((ret > 0).sum(axis=1) - (ret < 0).sum(axis=1)) / max(1, len(syms)) + raw = (cfg.btp_w_location * location + cfg.btp_w_momentum * momentum).rolling(cfg.btp_smooth, min_periods=1).mean() + out = _tanh_normalize(raw, span=cfg.btp_norm_span, k=cfg.btp_norm_k).rename("breadth_trend_pressure") + return out.to_frame() + + def _build_trend_efficiency_pressure(self) -> pd.DataFrame: + cfg = self.cfg + c = self.ohlc["close"].astype(float) + diff = c.diff().fillna(0.0) + net = c - c.shift(cfg.tep_win) + path = diff.abs().rolling(cfg.tep_win, min_periods=max(20, cfg.tep_win // 5)).sum() + raw = (net / (path + EPS)).clip(-1, 1).fillna(0.0).ewm(span=cfg.tep_smooth, adjust=False, min_periods=1).mean() + out = _tanh_normalize(raw, span=cfg.tep_norm_span, k=cfg.tep_norm_k).rename("trend_efficiency_pressure") + return out.to_frame() + + def _build_relative_strength_pressure(self) -> pd.DataFrame: + cfg = self.cfg + c = self.ohlc["close"].astype(float) + syms = self._symbols(include_index=False) + idx_ret = c.pct_change(cfg.rsp_lag).replace([np.inf, -np.inf], np.nan).fillna(0.0) + mkt_ret = self.px[syms].pct_change(cfg.rsp_lag).replace([np.inf, -np.inf], np.nan).fillna(0.0).median(axis=1) + raw = (idx_ret - mkt_ret).rolling(cfg.rsp_smooth, min_periods=1).mean() + out = _tanh_normalize(raw, span=cfg.rsp_norm_span, k=cfg.rsp_norm_k).rename("relative_strength_pressure") + return out.to_frame() + + def _build_flow_price_pressure(self) -> pd.DataFrame: + cfg = self.cfg + vp = self._build_volume_pressure()["volume_pressure"].reindex(self.index).ffill().fillna(0.0).clip(-1, 1) + c = self.ohlc["close"].astype(float) + price_comp = _tanh_normalize(c.diff(cfg.fpp_price_lag).fillna(0.0), span=cfg.fpp_price_norm_span, k=cfg.fpp_price_norm_k).clip(-1, 1) + same = (np.sign(vp) == np.sign(price_comp)) & (np.sign(vp) != 0) + agree = pd.Series(np.where(same, np.sign(vp) * np.minimum(vp.abs(), price_comp.abs()), 0.0), index=self.index) + raw = (cfg.fpp_w_flow * vp + cfg.fpp_w_price * price_comp + cfg.fpp_agree_boost * agree).rolling(cfg.fpp_smooth, min_periods=1).mean().clip(-1, 1) + return raw.rename("flow_price_pressure").to_frame() + + def _build_additional_pressures_v1(self) -> pd.DataFrame: + cfg = self.cfg + h = self.ohlc["high"].astype(float) + l = self.ohlc["low"].astype(float) + o = self.ohlc["open"].astype(float) + c = self.ohlc["close"].astype(float) + ret1 = c.diff().fillna(0.0) + vol_pts = ret1.abs().shift(1).ewm(span=300, adjust=False, min_periods=1).mean() + vol_pts = vol_pts.replace(0, np.nan).ffill().fillna(float(ret1.abs().median()) or 1.0).replace(0, EPS) + out = pd.DataFrame(index=self.index) + + hist_hi = h.shift(1).rolling(cfg.ibp_win, min_periods=max(20, cfg.ibp_win // 10)).max() + hist_lo = l.shift(1).rolling(cfg.ibp_win, min_periods=max(20, cfg.ibp_win // 10)).min() + mid = (hist_hi + hist_lo) / 2.0 + half_rng = (hist_hi - hist_lo).replace(0, np.nan) / 2.0 + out["index_breakout_pressure"] = ((c - mid) / (half_rng + EPS)).clip(-2, 2).fillna(0.0).ewm(span=cfg.ibp_smooth, adjust=False, min_periods=1).mean().clip(-1, 1) + + fast_z = c.diff(cfg.iip_fast_lag).fillna(0.0) / (vol_pts * np.sqrt(cfg.iip_fast_lag) + EPS) + slow_z = c.diff(cfg.iip_slow_lag).fillna(0.0) / (vol_pts * np.sqrt(cfg.iip_slow_lag) + EPS) + out["index_impulse_pressure"] = (0.6 * np.tanh(fast_z / 1.5) + 0.4 * np.tanh(slow_z / 1.5)).clip(-1, 1).fillna(0.0) + + vf = ret1.abs().ewm(span=cfg.vdp_fast_vol, adjust=False, min_periods=1).mean() + vs = ret1.abs().shift(1).ewm(span=cfg.vdp_slow_vol, adjust=False, min_periods=1).mean() + vexp = np.tanh(((vf / (vs + EPS)) - 1.0).fillna(0.0) / 0.5) + direction = np.sign(c.diff(cfg.vdp_dir_lag).ewm(span=cfg.vdp_smooth, adjust=False, min_periods=1).mean().fillna(0.0)) + out["volatility_direction_pressure"] = pd.Series(direction * vexp, index=self.index).ewm(span=cfg.vdp_smooth, adjust=False, min_periods=1).mean().clip(-1, 1) + + syms = self._symbols(include_index=False) + buy_total = self.buy[syms].sum(axis=1) + sell_total = self.sell[syms].sum(axis=1) + market_flow_raw = np.log1p(buy_total) - np.log1p(sell_total) + out["market_flow_pressure"] = _tanh_normalize(_causal_zscore(market_flow_raw, cfg.mfp_norm_win).rolling(cfg.mfp_smooth, min_periods=1).mean(), span=300, k=1.5) + + px = self.px[syms].ffill() + bmp_ret = px.pct_change(cfg.bmp_lag).replace([np.inf, -np.inf], np.nan).fillna(0.0) + out["breadth_momentum_pressure"] = (((bmp_ret > 0).sum(axis=1) - (bmp_ret < 0).sum(axis=1)) / max(1, len(syms))).clip(-1, 1).rolling(cfg.bmp_smooth, min_periods=1).mean() + + votes = [] + for span in cfg.eap_ema_spans: + ema = c.ewm(span=int(span), adjust=False, min_periods=1).mean() + votes.append(pd.Series(np.sign(c - ema), index=self.index)) + out["ema_alignment_pressure"] = pd.concat(votes, axis=1).mean(axis=1).rolling(cfg.eap_smooth, min_periods=1).mean().clip(-1, 1) + return _clean_numeric(out.clip(-1, 1)) + + def _build_additional_pressures_v2(self) -> pd.DataFrame: + cfg = self.cfg + o = self.ohlc["open"].astype(float) + h = self.ohlc["high"].astype(float) + l = self.ohlc["low"].astype(float) + c = self.ohlc["close"].astype(float) + syms = self._symbols(include_index=False) + px = self.px[syms].ffill() + buy = self.buy[syms].fillna(0.0) + sell = self.sell[syms].fillna(0.0) + out = pd.DataFrame(index=self.index) + + total_vol = buy.sum(axis=1) + sell.sum(axis=1) + vol_z = _causal_zscore(np.log1p(total_vol), cfg.pp_vol_win) + pp_dir = np.sign(c.diff(cfg.pp_dir_lag).ewm(span=cfg.pp_smooth, adjust=False, min_periods=1).mean().fillna(0.0)) + out["participation_pressure"] = pd.Series(pp_dir * np.tanh(vol_z / 2.0), index=self.index).ewm(span=cfg.pp_smooth, adjust=False, min_periods=1).mean().clip(-1, 1) + + idx_ret = c.pct_change(cfg.cap_lag).replace([np.inf, -np.inf], np.nan).fillna(0.0) + px_ret = px.pct_change(cfg.cap_lag).replace([np.inf, -np.inf], np.nan).fillna(0.0) + idx_sign = np.sign(idx_ret) + px_sign = np.sign(px_ret) + same = px_sign.eq(idx_sign, axis=0) & (idx_sign != 0).to_numpy().reshape(-1, 1) + align_strength = (2.0 * same.mean(axis=1) - 1.0).clip(-1, 1) + out["correlation_alignment_pressure"] = (pd.Series(idx_sign, index=self.index) * pd.Series(align_strength, index=self.index)).rolling(cfg.cap_smooth, min_periods=1).mean().clip(-1, 1) + + dsp_ret = px.pct_change(cfg.dsp_ret_lag).replace([np.inf, -np.inf], np.nan).fillna(0.0) + cross_std = dsp_ret.std(axis=1).fillna(0.0) + dsp_z = _causal_zscore(cross_std, cfg.dsp_win) + dsp_dir = np.sign(c.diff(cfg.dsp_dir_lag).ewm(span=cfg.dsp_smooth, adjust=False, min_periods=1).mean().fillna(0.0)) + out["dispersion_pressure"] = pd.Series(dsp_dir * np.tanh(dsp_z / 2.0), index=self.index).ewm(span=cfg.dsp_smooth, adjust=False, min_periods=1).mean().clip(-1, 1) + + bap_ret = px.pct_change(cfg.bap_lag).replace([np.inf, -np.inf], np.nan).fillna(0.0) + breadth = (((bap_ret > 0).sum(axis=1) - (bap_ret < 0).sum(axis=1)) / max(1, len(syms))).clip(-1, 1) + breadth_smooth = breadth.ewm(span=cfg.bap_smooth, adjust=False, min_periods=1).mean() + out["breadth_accel_pressure"] = _tanh_normalize(breadth_smooth.diff(cfg.bap_diff_lag).fillna(0.0), span=300, k=1.5) + + rng = (h - l).replace(0, np.nan) + body_pos = ((c - o) / (rng + EPS)).clip(-1, 1).fillna(0.0) + range_z = _causal_zscore(np.log1p(rng.fillna(0.0)), cfg.prp_range_win) + out["price_range_pressure"] = (body_pos * np.tanh(range_z / 2.0)).rolling(cfg.prp_smooth, min_periods=1).mean().clip(-1, 1) + return _clean_numeric(out.clip(-1, 1)) + + def _make_state(self, s: pd.Series, deadband: float) -> pd.Series: + s = pd.Series(s).astype(float) + return pd.Series(np.where(s > deadband, 1, np.where(s < -deadband, -1, 0)), index=s.index, dtype=np.float32) + + def _cycle_position(self, s: pd.Series, win: int, smooth: int = 1) -> pd.Series: + s = pd.Series(s).astype(float) + ref = s.shift(1) + minp = max(20, int(win) // 10) + lo = ref.rolling(int(win), min_periods=minp).min() + hi = ref.rolling(int(win), min_periods=minp).max() + cycle = (2.0 * ((s - lo) / ((hi - lo) + EPS)) - 1.0).replace([np.inf, -np.inf], np.nan).fillna(0.0).clip(-1, 1) + if smooth and smooth > 1: + cycle = cycle.ewm(span=int(smooth), adjust=False, min_periods=1).mean() + return cycle.clip(-1, 1) + + def build_modes_states_agreements(self, params_df: pd.DataFrame): + cfg = self.cfg + raw_cols = {} + val_cols = {} + state_cols = {} + for param_name in params_df.columns: + s = params_df[param_name].astype(float).reindex(self.index).ffill().fillna(0.0).clip(-1, 1) + level = s.rename(f"{param_name}_level") + val_cols[level.name] = level + state_cols[f"{param_name}_level_state"] = self._make_state(level, cfg.mode_level_deadband) + + smooth = s.ewm(span=cfg.mode_speed_ema_span, adjust=False, min_periods=1).mean() + speed_raw = smooth.diff(cfg.mode_speed_diff_lag).fillna(0.0).rename(f"{param_name}_speed_raw") + speed = _tanh_normalize(speed_raw, span=cfg.mode_speed_norm_span, k=cfg.mode_speed_norm_k).rename(f"{param_name}_speed") + raw_cols[speed_raw.name] = speed_raw + val_cols[speed.name] = speed + state_cols[f"{param_name}_speed_state"] = self._make_state(speed, cfg.mode_speed_deadband) + + accel_raw = speed_raw.ewm(span=cfg.mode_accel_ema_span, adjust=False, min_periods=1).mean().diff(cfg.mode_accel_diff_lag).fillna(0.0).rename(f"{param_name}_accel_raw") + accel = _tanh_normalize(accel_raw, span=cfg.mode_accel_norm_span, k=cfg.mode_accel_norm_k).rename(f"{param_name}_accel") + raw_cols[accel_raw.name] = accel_raw + val_cols[accel.name] = accel + state_cols[f"{param_name}_accel_state"] = self._make_state(accel, cfg.mode_accel_deadband) + + cycle = self._cycle_position(s, win=cfg.mode_cycle_roll_win, smooth=cfg.mode_cycle_smooth).rename(f"{param_name}_cycle") + val_cols[cycle.name] = cycle + state_cols[f"{param_name}_cycle_state"] = self._make_state(cycle, cfg.mode_cycle_deadband) + + rev = (-cycle).rename(f"{param_name}_reversion_cycle") + val_cols[rev.name] = rev + state_cols[f"{param_name}_reversion_cycle_state"] = self._make_state(rev, cfg.mode_reversion_cycle_deadband) + + hist_hi = s.shift(1).rolling(cfg.mode_breakout_win, min_periods=max(20, cfg.mode_breakout_win // 10)).max() + hist_lo = s.shift(1).rolling(cfg.mode_breakout_win, min_periods=max(20, cfg.mode_breakout_win // 10)).min() + state_cols[f"{param_name}_breakout_state"] = pd.Series(np.where(s > hist_hi, 1, np.where(s < hist_lo, -1, 0)), index=self.index, dtype=np.float32) + + sign_s = pd.Series(np.sign(s), index=self.index, dtype=float) + persist = sign_s.rolling(cfg.mode_persist_win, min_periods=max(5, cfg.mode_persist_win // 10)).mean().fillna(0.0).clip(-1, 1).rename(f"{param_name}_persistence") + val_cols[persist.name] = persist + state_cols[f"{param_name}_persistence_state"] = self._make_state(persist, cfg.mode_persist_deadband) + + smooth2 = s.ewm(span=cfg.mode_strength_ema_span, adjust=False, min_periods=1).mean() + strength_delta = smooth2.abs().diff(cfg.mode_strength_diff_lag).fillna(0.0) + strength = _tanh_normalize(np.sign(smooth2) * strength_delta, span=cfg.mode_strength_norm_span, k=cfg.mode_strength_norm_k).rename(f"{param_name}_strength_expansion") + val_cols[strength.name] = strength + state_cols[f"{param_name}_strength_expansion_state"] = self._make_state(strength, cfg.mode_strength_deadband) + + pullback = pd.Series( + np.where( + (s > cfg.mode_pullback_trend_deadband) & (cycle < -cfg.mode_pullback_cycle_deadband), + 1, + np.where((s < -cfg.mode_pullback_trend_deadband) & (cycle > cfg.mode_pullback_cycle_deadband), -1, 0), + ), + index=self.index, + dtype=np.float32, + ) + state_cols[f"{param_name}_pullback_state"] = pullback + + mode_raw_df = _clean_numeric(pd.DataFrame(raw_cols, index=self.index)) + mode_values_df = _clean_numeric(pd.DataFrame(val_cols, index=self.index).clip(-1, 1)) + mode_states_df = _clean_numeric(pd.DataFrame(state_cols, index=self.index)) + + mode_groups = { + "level": [c for c in mode_states_df.columns if c.endswith("_level_state")], + "speed": [c for c in mode_states_df.columns if c.endswith("_speed_state")], + "accel": [c for c in mode_states_df.columns if c.endswith("_accel_state")], + "cycle": [c for c in mode_states_df.columns if c.endswith("_cycle_state") and not c.endswith("_reversion_cycle_state")], + "reversion_cycle": [c for c in mode_states_df.columns if c.endswith("_reversion_cycle_state")], + "breakout": [c for c in mode_states_df.columns if c.endswith("_breakout_state")], + "persistence": [c for c in mode_states_df.columns if c.endswith("_persistence_state")], + "strength_expansion": [c for c in mode_states_df.columns if c.endswith("_strength_expansion_state")], + "pullback": [c for c in mode_states_df.columns if c.endswith("_pullback_state")], + } + agreement_by_mode_parts = [] + for mode_name, cols in mode_groups.items(): + if not cols: + continue + score = mode_states_df[cols].sum(axis=1).rename(f"agreement_{mode_name}_score") + state = self._make_state(score, 0.0).rename(f"agreement_{mode_name}_state") + agreement_by_mode_parts.extend([score, state]) + agreement_by_mode_df = _clean_numeric(pd.concat(agreement_by_mode_parts, axis=1) if agreement_by_mode_parts else pd.DataFrame(index=self.index)) + + agreement_by_param_parts = [] + for param_name in params_df.columns: + cols = [c for c in mode_states_df.columns if c.startswith(f"{param_name}_")] + if not cols: + continue + score = mode_states_df[cols].sum(axis=1).rename(f"agreement_{param_name}_score") + state = self._make_state(score, 0.0).rename(f"agreement_{param_name}_state") + agreement_by_param_parts.extend([score, state]) + agreement_by_param_df = _clean_numeric(pd.concat(agreement_by_param_parts, axis=1) if agreement_by_param_parts else pd.DataFrame(index=self.index)) + return mode_raw_df, mode_values_df, mode_states_df, agreement_by_mode_df, agreement_by_param_df + + def build_long_context(self, params_df: pd.DataFrame): + cfg = self.cfg + value_parts = {} + state_parts = {} + for param_name in params_df.columns: + p_name = _safe_col_name(param_name) + s = params_df[param_name].astype(float).reindex(self.index).ffill().fillna(0.0).clip(-1, 1) + for span in cfg.lctx_ema_spans: + ema = s.ewm(span=int(span), adjust=False, min_periods=1).mean().clip(-1, 1) + col = f"{p_name}_lctx_ema_{int(span)}" + value_parts[col] = ema + state_parts[f"{col}_state"] = self._make_state(ema, cfg.lctx_deadband_level) + for lag in cfg.lctx_trend_lags: + slow = s.ewm(span=max(2, int(lag) // 2), adjust=False, min_periods=1).mean() + slope = _tanh_normalize(slow.diff(int(lag)).fillna(0.0), span=max(100, int(lag)), k=1.5) + col = f"{p_name}_lctx_trend_{int(lag)}" + value_parts[col] = slope + state_parts[f"{col}_state"] = self._make_state(slope, cfg.lctx_deadband_trend) + for win in cfg.lctx_cycle_wins: + cycle = self._cycle_position(s, win=int(win), smooth=21) + col = f"{p_name}_lctx_cycle_{int(win)}" + value_parts[col] = cycle + state_parts[f"{col}_state"] = self._make_state(cycle, cfg.lctx_deadband_cycle) + for win in cfg.lctx_persist_wins: + persist = pd.Series(np.sign(s), index=self.index).rolling(int(win), min_periods=max(10, int(win) // 10)).mean().fillna(0.0).clip(-1, 1) + col = f"{p_name}_lctx_persist_{int(win)}" + value_parts[col] = persist + state_parts[f"{col}_state"] = self._make_state(persist, cfg.lctx_deadband_persist) + for win in cfg.lctx_breakout_wins: + hist_hi = s.shift(1).rolling(int(win), min_periods=max(20, int(win) // 10)).max() + hist_lo = s.shift(1).rolling(int(win), min_periods=max(20, int(win) // 10)).min() + state_parts[f"{p_name}_lctx_breakout_{int(win)}_state"] = pd.Series(np.where(s > hist_hi, 1, np.where(s < hist_lo, -1, 0)), index=self.index, dtype=np.float32) + for lag in cfg.lctx_impulse_lags: + z = _causal_zscore(s.diff(int(lag)).fillna(0.0), max(240, int(lag) * 2)).clip(-6, 6) + impulse = pd.Series(np.tanh(z / 2.0), index=self.index).clip(-1, 1) + col = f"{p_name}_lctx_impulse_{int(lag)}" + value_parts[col] = impulse + state_parts[f"{col}_state"] = self._make_state(impulse, cfg.lctx_deadband_impulse) + fast = s.ewm(span=cfg.lctx_fast_ema_for_spread, adjust=False, min_periods=1).mean() + for span in (240, 720, 1440): + long = s.ewm(span=span, adjust=False, min_periods=1).mean() + spread = _tanh_normalize(fast - long, span=span, k=1.5) + col = f"{p_name}_lctx_fast_vs_long_{span}" + value_parts[col] = spread + state_parts[f"{col}_state"] = self._make_state(spread, cfg.lctx_deadband_spread) + + values_df = _clean_numeric(pd.DataFrame(value_parts, index=self.index).clip(-1, 1)) + states_df = _clean_numeric(pd.DataFrame(state_parts, index=self.index)) + + agreement_parts = [] + patterns = { + "lctx_ema": "_lctx_ema_", + "lctx_trend": "_lctx_trend_", + "lctx_cycle": "_lctx_cycle_", + "lctx_persist": "_lctx_persist_", + "lctx_breakout": "_lctx_breakout_", + "lctx_impulse": "_lctx_impulse_", + "lctx_fast_vs_long": "_lctx_fast_vs_long_", + } + for mode_name, pattern in patterns.items(): + cols = [c for c in states_df.columns if pattern in c] + if not cols: + continue + score = states_df[cols].sum(axis=1).rename(f"agreement_{mode_name}_score") + state = self._make_state(score, 0.0).rename(f"agreement_{mode_name}_state") + agreement_parts.extend([score, state]) + for param_name in params_df.columns: + p_name = _safe_col_name(param_name) + cols = [c for c in states_df.columns if c.startswith(f"{p_name}_lctx_")] + if not cols: + continue + score = states_df[cols].sum(axis=1).rename(f"agreement_{p_name}_lctx_score") + state = self._make_state(score, 0.0).rename(f"agreement_{p_name}_lctx_state") + agreement_parts.extend([score, state]) + agreement_df = _clean_numeric(pd.concat(agreement_parts, axis=1) if agreement_parts else pd.DataFrame(index=self.index)) + all_df = pd.concat([values_df, states_df, agreement_df], axis=1).reindex(self.index).ffill().fillna(0.0) + all_df = all_df.loc[:, ~all_df.columns.duplicated()] + all_df = _clean_numeric(all_df) + if cfg.lctx_max_output_cols is not None and all_df.shape[1] > int(cfg.lctx_max_output_cols): + var_s = all_df.var(ddof=0).replace([np.inf, -np.inf], np.nan).fillna(0.0).sort_values(ascending=False) + all_df = all_df[var_s.head(int(cfg.lctx_max_output_cols)).index.tolist()].copy() + return values_df, states_df, agreement_df, all_df + + +def make_fusion_cache_key(cfg: dict[str, Any] | None) -> str: + s = json.dumps(cfg or {}, sort_keys=True, default=str, ensure_ascii=False) + return hashlib.md5(s.encode("utf-8")).hexdigest()[:16] diff --git a/src/ml_crypto_lab/features/pipeline.py b/src/ml_crypto_lab/features/pipeline.py index 748c73f..9468134 100644 --- a/src/ml_crypto_lab/features/pipeline.py +++ b/src/ml_crypto_lab/features/pipeline.py @@ -9,6 +9,7 @@ from ml_crypto_lab.features.utils import clean_numeric # Import registers builders import ml_crypto_lab.features.builders # noqa: F401 +import ml_crypto_lab.features.fusion_builders # noqa: F401 def build_feature_set(base_df: pd.DataFrame, feature_set_cfg: dict[str, Any]) -> pd.DataFrame: diff --git a/src/ml_crypto_lab/train/runner.py b/src/ml_crypto_lab/train/runner.py index fe280a9..e71d8ec 100644 --- a/src/ml_crypto_lab/train/runner.py +++ b/src/ml_crypto_lab/train/runner.py @@ -39,7 +39,20 @@ def build_base_frame_from_config(cfg: dict[str, Any]) -> pd.DataFrame: 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"]) + + base = base.replace([np.inf, -np.inf], np.nan).ffill().dropna(subset=["open", "high", "low", "close"]) + + # Important for advanced cross-sectional feature engineering. + # The large fusion factory needs full market matrices, not only INDEX OHLC. + base.attrs["market_data"] = { + "ohlc": built["ohlc"], + "px": built["px"], + "buy": built["buy"], + "sell": built["sell"], + "raw_resampled": built.get("raw_resampled"), + } + base.attrs["feature_cache"] = {} + return base def train_one_experiment(