""" EURUSD D1 → secp256k1 → CatBoost v2 (2000 свечей) ===================================================== Улучшенная версия: - Множественные lookback окна (5, 10, 20, 50) - Расширенные EC-фичи (энтропия nonce, фрактальные x-биты) - Тюнинг гиперпараметров - Ансамбль из 3 моделей """ import numpy as np import hashlib import json from datetime import datetime, timedelta from catboost import CatBoostClassifier, Pool from sklearn.metrics import ( accuracy_score, precision_score, recall_score, f1_score, roc_auc_score, log_loss ) from sklearn.model_selection import TimeSeriesSplit from collections import Counter import warnings warnings.filterwarnings('ignore') # ============================================================ # secp256k1 # ============================================================ P = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEFFFFFC2F A, B = 0, 7 def extended_gcd(a, b): if a == 0: return b, 0, 1 g, x, y = extended_gcd(b % a, a) return g, y - (b // a) * x, x def mod_inverse(a, m=P): if a < 0: a = a % m g, x, _ = extended_gcd(a, m) return x % m def point_add(p1, p2): if p1 is None: return p2 if p2 is None: return p1 x1, y1 = p1; x2, y2 = p2 if x1 == x2 and y1 != y2: return None if x1 == x2: lam = (3 * x1 * x1) * mod_inverse(2 * y1) % P else: lam = (y2 - y1) * mod_inverse(x2 - x1) % P x3 = (lam * lam - x1 - x2) % P y3 = (lam * (x1 - x3) - y1) % P return (x3, y3) def tonelli_shanks(n, p): if pow(n, (p - 1) // 2, p) != 1: return None Q, S = p - 1, 0 while Q % 2 == 0: Q //= 2; S += 1 if S == 1: return pow(n, (p + 1) // 4, p) z = 2 while pow(z, (p - 1) // 2, p) != p - 1: z += 1 M, c, t, R = S, pow(z, Q, p), pow(n, Q, p), pow(n, (Q + 1) // 2, p) while True: if t == 1: return R i, temp = 0, t while temp != 1: temp = pow(temp, 2, p); i += 1 b = pow(c, 1 << (M - i - 1), p) M, c, t, R = i, pow(b, 2, p), (t * pow(b, 2, p)) % P, (R * b) % P def price_to_curve_point(value, salt="eurusd"): value_str = f"{salt}:{value:.10f}" for nonce in range(256): data = f"{value_str}:{nonce}".encode('utf-8') h = hashlib.sha256(data).digest() x = int.from_bytes(h, 'big') % P y_sq = (pow(x, 3, P) + B) % P y = tonelli_shanks(y_sq, P) if y is not None: if y % 2 != 0: y = P - y return (x, y, nonce) return None # ============================================================ # Data generation (2000 свечей, regime-switching) # ============================================================ def generate_eurusd_data(n_candles=2000, seed=42): np.random.seed(seed) base = 1.0850 vol = 0.0012 prices = [base] regime, dur = 0, 0 for i in range(n_candles * 4 + 200): dur += 1 if dur > np.random.geometric(0.015): regime = np.random.choice([0, 1, 2], p=[0.4, 0.3, 0.3]) dur = 0 drift = {0: 0.0, 1: 0.0004, 2: -0.0004}[regime] mr = -0.012 * (prices[-1] - base) noise = np.random.normal(0, vol) if len(prices) > 1: vol_adj = 1.0 + 2.5 * abs(prices[-1] - prices[-2]) / prices[-2] else: vol_adj = 1.0 prices.append(prices[-1] + drift + mr + noise * vol_adj) ohlc = [] date = datetime(2016, 1, 4) for i in range(n_candles): idx = i * 4 chunk = prices[idx:idx+4] o, c = chunk[0], chunk[-1] h = max(chunk) + abs(np.random.normal(0, 0.0005)) l = min(chunk) - abs(np.random.normal(0, 0.0005)) ohlc.append({ 'date': date.strftime('%Y-%m-%d'), 'open': round(o, 5), 'high': round(h, 5), 'low': round(l, 5), 'close': round(c, 5), }) date += timedelta(days=1) while date.weekday() >= 5: date += timedelta(days=1) return ohlc # ============================================================ # Feature Engineering v2 — Multi-Lookback + Expanded EC # ============================================================ def extract_features_v2(ohlc, max_lookback=50): print(f"[FEATURES v2] Маппинг {len(ohlc)} цен на secp256k1...") # Pre-compute all curve points pts = {} for i, c in enumerate(ohlc): if i % 100 == 0: print(f" Маппинг {i}/{len(ohlc)}...") for t in ['open', 'high', 'low', 'close']: pts[f"{t}:{i}"] = price_to_curve_point(c[t], f"{t}:{i}") # Pre-compute returns returns = [0.0] for i in range(1, len(ohlc)): returns.append((ohlc[i]['close'] - ohlc[i-1]['close']) / ohlc[i-1]['close']) # Pre-compute nonces and x-bits for close nonces = [] x_high = [] x_mid = [] x_low_bits = [] for i in range(len(ohlc)): pt = pts[f'close:{i}'] nonces.append(pt[2] if pt else 0) if pt: xb = pt[0].to_bytes(32, 'big') x_high.append(int.from_bytes(xb[:8], 'big') / (2**64)) x_mid.append(int.from_bytes(xb[8:16], 'big') / (2**64)) x_low_bits.append(int.from_bytes(xb[16:24], 'big') / (2**64)) else: x_high.append(0); x_mid.append(0); x_low_bits.append(0) print("[FEATURES v2] Извлечение фичей...") features = [] lookbacks = [5, 10, 20, 50] for i in range(max_lookback, len(ohlc)): row = {} c = ohlc[i] o, h, l, cl = c['open'], c['high'], c['low'], c['close'] # === Current candle EC === pt_o = pts[f'open:{i}'] pt_h = pts[f'high:{i}'] pt_l = pts[f'low:{i}'] pt_c = pts[f'close:{i}'] row['nonce_o'] = pt_o[2] if pt_o else 0 row['nonce_h'] = pt_h[2] if pt_h else 0 row['nonce_l'] = pt_l[2] if pt_l else 0 row['nonce_c'] = pt_c[2] if pt_c else 0 row['nonce_sum'] = row['nonce_o'] + row['nonce_h'] + row['nonce_l'] + row['nonce_c'] row['nonce_range'] = max(row['nonce_o'], row['nonce_h'], row['nonce_l'], row['nonce_c']) - \ min(row['nonce_o'], row['nonce_h'], row['nonce_l'], row['nonce_c']) # x-bits текущей свечи row['x_high'] = x_high[i] row['x_mid'] = x_mid[i] row['x_low'] = x_low_bits[i] row['x_parity'] = pt_c[0] % 2 if pt_c else 0 row['x_quadrant'] = (pt_c[0] >> 254) & 3 if pt_c else 0 row['x_byte0'] = (pt_c[0] >> 248) & 0xFF if pt_c else 0 # первый байт # Delta x (close - open) if pt_c and pt_o: dx = (pt_c[0] - pt_o[0]) % P row['dx_norm'] = dx / P row['dx_sign'] = 1 if dx < P // 2 else -1 else: row['dx_norm'] = 0; row['dx_sign'] = 0 # Point addition O+C, H+L if pt_o and pt_c: comb = point_add((pt_o[0], pt_o[1]), (pt_c[0], pt_c[1])) if comb: cb = comb[0].to_bytes(32, 'big') row['oc_x'] = int.from_bytes(cb[:8], 'big') / (2**64) row['oc_parity'] = comb[0] % 2 else: row['oc_x'] = 0; row['oc_parity'] = 0 else: row['oc_x'] = 0; row['oc_parity'] = 0 if pt_h and pt_l: comb_hl = point_add((pt_h[0], pt_h[1]), (pt_l[0], pt_l[1])) row['hl_x'] = int.from_bytes(comb_hl[0].to_bytes(32, 'big')[:8], 'big') / (2**64) if comb_hl else 0 else: row['hl_x'] = 0 # === Multi-lookback features === for lb in lookbacks: sl = slice(max(0, i - lb), i) n_arr = nonces[sl] x_arr = x_high[sl] xm_arr = x_mid[sl] ret_arr = returns[max(1, i - lb):i + 1] # Nonce stats row[f'nonce_mean_{lb}'] = np.mean(n_arr) row[f'nonce_std_{lb}'] = np.std(n_arr) if len(n_arr) > 1 else 0 row[f'nonce_max_{lb}'] = max(n_arr) if n_arr else 0 row[f'nonce_zeros_{lb}'] = sum(1 for n in n_arr if n == 0) / max(len(n_arr), 1) # Nonce entropy cnt = Counter(n_arr) total = sum(cnt.values()) ent = -sum((v/total) * np.log2(v/total + 1e-10) for v in cnt.values()) row[f'nonce_entropy_{lb}'] = ent # x-bits stats row[f'x_mean_{lb}'] = np.mean(x_arr) row[f'x_std_{lb}'] = np.std(x_arr) if len(x_arr) > 1 else 0 row[f'xm_mean_{lb}'] = np.mean(xm_arr) row[f'xm_std_{lb}'] = np.std(xm_arr) if len(xm_arr) > 1 else 0 # x-bits trend (linear slope) if len(x_arr) > 2: x_idx = np.arange(len(x_arr)) row[f'x_slope_{lb}'] = np.polyfit(x_idx, x_arr, 1)[0] else: row[f'x_slope_{lb}'] = 0 # Returns stats row[f'ret_mean_{lb}'] = np.mean(ret_arr) if ret_arr else 0 row[f'ret_std_{lb}'] = np.std(ret_arr) if len(ret_arr) > 1 else 0 row[f'sharpe_{lb}'] = row[f'ret_mean_{lb}'] / (row[f'ret_std_{lb}'] + 1e-10) # Volatility highs = [ohlc[j]['high'] for j in range(max(0, i-lb), i+1)] lows = [ohlc[j]['low'] for j in range(max(0, i-lb), i+1)] row[f'atr_{lb}'] = np.mean([(h-l)/ohlc[max(0,i-lb)+k]['open'] for k, (h,l) in enumerate(zip(highs, lows))]) # Cross: nonce × return row[f'nonce_x_ret_{lb}'] = row[f'nonce_mean_{lb}'] * row[f'ret_mean_{lb}'] # Cross: x_bits × volatility row[f'x_x_vol_{lb}'] = row[f'x_mean_{lb}'] * row[f'ret_std_{lb}'] # === Classical TA === row['return_1'] = returns[i] row['return_2'] = returns[i-1] if i > 0 else 0 row['return_3'] = returns[i-2] if i > 1 else 0 row['volatility'] = (h - l) / o row['body_ratio'] = abs(cl - o) / (h - l) if (h - l) > 0 else 0 row['upper_shadow'] = (h - max(o, cl)) / (h - l) if (h - l) > 0 else 0 row['lower_shadow'] = (min(o, cl) - l) / (h - l) if (h - l) > 0 else 0 row['direction'] = 1 if cl >= o else 0 # MA ratios for lb in [5, 10, 20, 50]: ma = np.mean([ohlc[j]['close'] for j in range(max(0, i-lb), i+1)]) row[f'ma_ratio_{lb}'] = cl / ma # RSI gains = [r for r in returns[max(1,i-14):i+1] if r > 0] losses = [-r for r in returns[max(1,i-14):i+1] if r < 0] ag = np.mean(gains) if gains else 0 al = np.mean(losses) if losses else 1e-8 row['rsi'] = ag / (ag + al) # Consecutive direction consec = 0 for j in range(i, max(i-20, 0), -1): if (ohlc[j]['close'] >= ohlc[j]['open']) == (cl >= o): consec += 1 else: break row['consec_dir'] = consec # Day of week row['dow'] = datetime.strptime(c['date'], '%Y-%m-%d').weekday() features.append(row) names = list(features[0].keys()) X = np.array([[r[f] for f in names] for r in features]) print(f"[FEATURES v2] Готово: {X.shape[0]} × {X.shape[1]} features") return X, names def create_target(ohlc, offset=50): targets = [] for i in range(offset, len(ohlc) - 1): nxt = ohlc[i + 1] targets.append(1 if nxt['close'] >= nxt['open'] else 0) return np.array(targets) # ============================================================ # MAIN # ============================================================ if __name__ == "__main__": print("=" * 70) print(" EURUSD D1 → secp256k1 → CatBoost v2 (2000 свечей)") print("=" * 70) N = 2000 LB = 50 # Data try: import MetaTrader5 as mt5 if mt5.initialize(): rates = mt5.copy_rates_from_pos("EURUSD", mt5.TIMEFRAME_D1, 0, N) mt5.shutdown() ohlc = [{'date': datetime.fromtimestamp(r['time']).strftime('%Y-%m-%d'), 'open': round(float(r['open']),5), 'high': round(float(r['high']),5), 'low': round(float(r['low']),5), 'close': round(float(r['close']),5)} for r in rates] print(f"\n[MT5] Загружено {len(ohlc)} реальных свечей") else: raise Exception() except: print(f"\n[DEMO] Генерация {N} свечей с regime-switching...") ohlc = generate_eurusd_data(N) print(f" Период: {ohlc[0]['date']} — {ohlc[-1]['date']}") # Features X, names = extract_features_v2(ohlc, LB) y = create_target(ohlc, LB) X = X[:len(y)] print(f"\n Samples: {len(y)} | Features: {len(names)}") print(f" Balance: BULL={y.sum()} ({100*y.mean():.1f}%) | BEAR={len(y)-y.sum()} ({100*(1-y.mean()):.1f}%)") # Split n = len(X) tr_end = int(n * 0.70) val_end = int(n * 0.85) X_tr, y_tr = X[:tr_end], y[:tr_end] X_val, y_val = X[tr_end:val_end], y[tr_end:val_end] X_te, y_te = X[val_end:], y[val_end:] print(f" Train: {len(X_tr)} | Val: {len(X_val)} | Test: {len(X_te)}") # ============================================================ # Identify feature groups # ============================================================ ec_feats = [n for n in names if any(n.startswith(p) for p in ['nonce_', 'x_', 'xm_', 'dx_', 'oc_', 'hl_', 'x_x_'])] ta_feats = [n for n in names if n not in ec_feats] ec_idx = [names.index(f) for f in ec_feats] ta_idx = [names.index(f) for f in ta_feats] print(f"\n EC features: {len(ec_feats)} | TA features: {len(ta_feats)}") # ============================================================ # Hyperparameter configs # ============================================================ configs = [ {'depth': 4, 'lr': 0.02, 'l2': 3, 'iters': 1000, 'sub': 0.8, 'rsm': 0.8}, {'depth': 6, 'lr': 0.01, 'l2': 1, 'iters': 1500, 'sub': 0.7, 'rsm': 0.7}, {'depth': 3, 'lr': 0.05, 'l2': 5, 'iters': 800, 'sub': 0.9, 'rsm': 0.9}, ] def train_model(Xtr, ytr, Xv, yv, cfg, feat_names, label=""): m = CatBoostClassifier( iterations=cfg['iters'], depth=cfg['depth'], learning_rate=cfg['lr'], l2_leaf_reg=cfg['l2'], subsample=cfg['sub'], rsm=cfg['rsm'], random_seed=42, verbose=0, eval_metric='AUC', early_stopping_rounds=100, use_best_model=True, bootstrap_type='Bernoulli', ) tp = Pool(Xtr, ytr, feature_names=feat_names) vp = Pool(Xv, yv, feature_names=feat_names) m.fit(tp, eval_set=vp) return m def evaluate(m, X, y, label=""): pred = m.predict(X) prob = m.predict_proba(X)[:, 1] acc = accuracy_score(y, pred) f1 = f1_score(y, pred, zero_division=0) try: auc = roc_auc_score(y, prob) except: auc = 0.5 prec = precision_score(y, pred, zero_division=0) rec = recall_score(y, pred, zero_division=0) return {'acc': acc, 'f1': f1, 'auc': auc, 'prec': prec, 'rec': rec, 'prob': prob} # ============================================================ # Train all models # ============================================================ results = {} # --- FULL ENSEMBLE (EC + TA) --- print("\n" + "=" * 70) print(" FULL MODEL ENSEMBLE (EC + TA)") print("=" * 70) full_models = [] for ci, cfg in enumerate(configs): print(f" Training config {ci+1}: depth={cfg['depth']}, lr={cfg['lr']}, l2={cfg['l2']}...") m = train_model(X_tr, y_tr, X_val, y_val, cfg, names) full_models.append(m) r = evaluate(m, X_val, y_val) print(f" Val AUC={r['auc']:.4f} Acc={r['acc']:.4f} F1={r['f1']:.4f} (iters={m.best_iteration_})") # Ensemble: average probabilities val_probs_full = np.mean([m.predict_proba(X_val)[:, 1] for m in full_models], axis=0) test_probs_full = np.mean([m.predict_proba(X_te)[:, 1] for m in full_models], axis=0) val_pred_full = (val_probs_full > 0.5).astype(int) test_pred_full = (test_probs_full > 0.5).astype(int) results['Full Ensemble'] = { 'val': { 'acc': accuracy_score(y_val, val_pred_full), 'f1': f1_score(y_val, val_pred_full, zero_division=0), 'auc': roc_auc_score(y_val, val_probs_full), 'prec': precision_score(y_val, val_pred_full, zero_division=0), 'rec': recall_score(y_val, val_pred_full, zero_division=0), }, 'test': { 'acc': accuracy_score(y_te, test_pred_full), 'f1': f1_score(y_te, test_pred_full, zero_division=0), 'auc': roc_auc_score(y_te, test_probs_full), 'prec': precision_score(y_te, test_pred_full, zero_division=0), 'rec': recall_score(y_te, test_pred_full, zero_division=0), } } print(f"\n ENSEMBLE Val: Acc={results['Full Ensemble']['val']['acc']:.4f} " f"F1={results['Full Ensemble']['val']['f1']:.4f} " f"AUC={results['Full Ensemble']['val']['auc']:.4f}") print(f" ENSEMBLE Test: Acc={results['Full Ensemble']['test']['acc']:.4f} " f"F1={results['Full Ensemble']['test']['f1']:.4f} " f"AUC={results['Full Ensemble']['test']['auc']:.4f}") # Best single full model best_full = full_models[np.argmax([roc_auc_score(y_val, m.predict_proba(X_val)[:,1]) for m in full_models])] fi = best_full.get_feature_importance() fi_sorted = sorted(zip(names, fi), key=lambda x: -x[1]) print(f"\n Top-20 Feature Importance (best single):") for nm, imp in fi_sorted[:20]: tag = "EC" if nm in ec_feats else "TA" bar = "█" * int(imp) print(f" [{tag:2s}] {nm:25s} {imp:6.2f} {bar}") # --- TA BASELINE --- print("\n" + "=" * 70) print(" TA BASELINE ENSEMBLE") print("=" * 70) ta_models = [] for ci, cfg in enumerate(configs): m = train_model(X_tr[:, ta_idx], y_tr, X_val[:, ta_idx], y_val, cfg, ta_feats) ta_models.append(m) val_probs_ta = np.mean([m.predict_proba(X_val[:, ta_idx])[:, 1] for m in ta_models], axis=0) test_probs_ta = np.mean([m.predict_proba(X_te[:, ta_idx])[:, 1] for m in ta_models], axis=0) results['TA Baseline'] = { 'val': { 'acc': accuracy_score(y_val, (val_probs_ta > 0.5).astype(int)), 'f1': f1_score(y_val, (val_probs_ta > 0.5).astype(int), zero_division=0), 'auc': roc_auc_score(y_val, val_probs_ta), }, 'test': { 'acc': accuracy_score(y_te, (test_probs_ta > 0.5).astype(int)), 'f1': f1_score(y_te, (test_probs_ta > 0.5).astype(int), zero_division=0), 'auc': roc_auc_score(y_te, test_probs_ta), } } print(f" Val: Acc={results['TA Baseline']['val']['acc']:.4f} AUC={results['TA Baseline']['val']['auc']:.4f}") print(f" Test: Acc={results['TA Baseline']['test']['acc']:.4f} AUC={results['TA Baseline']['test']['auc']:.4f}") # --- EC ONLY --- print("\n" + "=" * 70) print(" EC ONLY ENSEMBLE") print("=" * 70) ec_models = [] for ci, cfg in enumerate(configs): m = train_model(X_tr[:, ec_idx], y_tr, X_val[:, ec_idx], y_val, cfg, ec_feats) ec_models.append(m) val_probs_ec = np.mean([m.predict_proba(X_val[:, ec_idx])[:, 1] for m in ec_models], axis=0) test_probs_ec = np.mean([m.predict_proba(X_te[:, ec_idx])[:, 1] for m in ec_models], axis=0) results['EC Only'] = { 'val': { 'acc': accuracy_score(y_val, (val_probs_ec > 0.5).astype(int)), 'f1': f1_score(y_val, (val_probs_ec > 0.5).astype(int), zero_division=0), 'auc': roc_auc_score(y_val, val_probs_ec), }, 'test': { 'acc': accuracy_score(y_te, (test_probs_ec > 0.5).astype(int)), 'f1': f1_score(y_te, (test_probs_ec > 0.5).astype(int), zero_division=0), 'auc': roc_auc_score(y_te, test_probs_ec), } } print(f" Val: Acc={results['EC Only']['val']['acc']:.4f} AUC={results['EC Only']['val']['auc']:.4f}") print(f" Test: Acc={results['EC Only']['test']['acc']:.4f} AUC={results['EC Only']['test']['auc']:.4f}") # ============================================================ # TIME SERIES CV (5-fold) # ============================================================ print("\n" + "=" * 70) print(" TIME SERIES CV (5-fold)") print("=" * 70) tscv = TimeSeriesSplit(n_splits=5) cv = {'full': [], 'ta': [], 'ec': []} best_cfg = configs[1] # deepest model for CV for fold, (tri, tei) in enumerate(tscv.split(X)): # Use last 20% of train as internal val split = int(len(tri) * 0.8) tr_i, vi = tri[:split], tri[split:] # Full m = CatBoostClassifier(iterations=800, depth=best_cfg['depth'], learning_rate=best_cfg['lr'], l2_leaf_reg=best_cfg['l2'], subsample=0.8, rsm=0.8, random_seed=42, verbose=0, eval_metric='AUC', early_stopping_rounds=80, use_best_model=True, bootstrap_type='Bernoulli') m.fit(Pool(X[tr_i], y[tr_i], feature_names=names), eval_set=Pool(X[vi], y[vi], feature_names=names)) try: auc_f = roc_auc_score(y[tei], m.predict_proba(X[tei])[:, 1]) except: auc_f = 0.5 cv['full'].append(auc_f) # TA m_ta = CatBoostClassifier(iterations=800, depth=best_cfg['depth'], learning_rate=best_cfg['lr'], l2_leaf_reg=best_cfg['l2'], subsample=0.8, rsm=0.8, random_seed=42, verbose=0, eval_metric='AUC', early_stopping_rounds=80, use_best_model=True, bootstrap_type='Bernoulli') m_ta.fit(Pool(X[tr_i][:, ta_idx], y[tr_i], feature_names=ta_feats), eval_set=Pool(X[vi][:, ta_idx], y[vi], feature_names=ta_feats)) try: auc_ta = roc_auc_score(y[tei], m_ta.predict_proba(X[tei][:, ta_idx])[:, 1]) except: auc_ta = 0.5 cv['ta'].append(auc_ta) # EC m_ec = CatBoostClassifier(iterations=800, depth=best_cfg['depth'], learning_rate=best_cfg['lr'], l2_leaf_reg=best_cfg['l2'], subsample=0.8, rsm=0.8, random_seed=42, verbose=0, eval_metric='AUC', early_stopping_rounds=80, use_best_model=True, bootstrap_type='Bernoulli') m_ec.fit(Pool(X[tr_i][:, ec_idx], y[tr_i], feature_names=ec_feats), eval_set=Pool(X[vi][:, ec_idx], y[vi], feature_names=ec_feats)) try: auc_ec = roc_auc_score(y[tei], m_ec.predict_proba(X[tei][:, ec_idx])[:, 1]) except: auc_ec = 0.5 cv['ec'].append(auc_ec) print(f" Fold {fold+1}: Full={auc_f:.4f} | TA={auc_ta:.4f} | EC={auc_ec:.4f}") print(f"\n Mean AUC Full: {np.mean(cv['full']):.4f} ± {np.std(cv['full']):.4f}") print(f" Mean AUC TA: {np.mean(cv['ta']):.4f} ± {np.std(cv['ta']):.4f}") print(f" Mean AUC EC: {np.mean(cv['ec']):.4f} ± {np.std(cv['ec']):.4f}") print(f" Δ (Full - TA): {np.mean(cv['full']) - np.mean(cv['ta']):+.4f}") # ============================================================ # SUMMARY # ============================================================ print("\n" + "=" * 70) print(" ИТОГОВАЯ СВОДКА (2000 свечей, ансамбль 3 моделей)") print("=" * 70) print(f"\n {'Model':20s} | {'Val Acc':8s} | {'Val AUC':8s} | {'Test Acc':8s} | {'Test AUC':8s} | {'CV AUC':10s}") print(f" {'-'*20}-+-{'-'*8}-+-{'-'*8}-+-{'-'*8}-+-{'-'*8}-+-{'-'*10}") for name, r in results.items(): cv_key = 'full' if 'Full' in name else ('ta' if 'TA' in name else 'ec') cv_mean = np.mean(cv[cv_key]) cv_std = np.std(cv[cv_key]) print(f" {name:20s} | {r['val']['acc']:8.4f} | {r['val']['auc']:8.4f} | " f"{r['test']['acc']:8.4f} | {r['test']['auc']:8.4f} | {cv_mean:.4f}±{cv_std:.4f}") # EC contribution ec_in_top10 = sum(1 for nm, _ in fi_sorted[:10] if nm in ec_feats) ec_imp_total = sum(imp for nm, imp in fi_sorted if nm in ec_feats) print(f"\n EC фичей в Top-10: {ec_in_top10}/10") print(f" Суммарная важность EC фичей: {ec_imp_total:.1f}%") print(f" Суммарная важность TA фичей: {100-ec_imp_total:.1f}%") # Save save_data = { 'results': {k: {kk: {kkk: round(vvv, 4) for kkk, vvv in vv.items() if kkk != 'prob'} for kk, vv in v.items()} for k, v in results.items()}, 'cv': {k: [round(v, 4) for v in vals] for k, vals in cv.items()}, 'fi_top20': [(n, round(v, 2)) for n, v in fi_sorted[:20]], 'config': {'n_candles': N, 'lookback': LB, 'n_features': len(names), 'n_ec': len(ec_feats), 'n_ta': len(ta_feats)}, } with open('/home/claude/catboost_v2_results.json', 'w') as f: json.dump(save_data, f, indent=2) print(f"\n Saved: catboost_v2_results.json") print("=" * 70)