#!/usr/bin/env python3 """Банк методов на 5минутных свечах + централизованный протокол честности. Все методы принимают ОДИН аргумент — словарь рынка из harness.build_matrix() (grid, C float32, syms, liquid, hours) и возвращают сырые сделки (time, symbol, side, gross_bp) БЕЗ издержек. Издержки, плацебо, рандом-контроль, OOS, block-bootstrap (по дням), разбивка по годам и просадка считаются в harness.py одинаково для всех — чтобы их нельзя было «подкрутить» внутри метода. Память: полная вселенная (250 пар × ~210 тыс. баров) не влезает в RAM, поэтому кросс-секционные методы работают по матрице float32, а временные ряды — потоково, по одной паре за раз. Правила каждого метода зафиксированы ДО прогона и после просмотра результата не меняются. Запуск: .venv/bin/python tools/methods.py [--maker] [имена методов...] """ from __future__ import annotations import json import os import sys import numpy as np import pandas as pd ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) from tools import harness as H # noqa: E402 BAR_MS = 300_000 MAX_REBALANCES = 20_000 EMPTY = ["time", "symbol", "side", "gross_bp"] # --------------------------------------------------------------- утилиты ---- def fwd_extremes(high: np.ndarray, low: np.ndarray, hold: int): n = len(high) hi_f = np.full(n, np.nan) lo_f = np.full(n, np.nan) if n > hold: hm = np.lib.stride_tricks.sliding_window_view(high, hold) lm = np.lib.stride_tricks.sliding_window_view(low, hold) hi_f[: n - hold] = hm[1: n - hold + 1].max(axis=1) lo_f[: n - hold] = lm[1: n - hold + 1].min(axis=1) return hi_f, lo_f def nonoverlap(candidates: np.ndarray, hold: int) -> list[int]: out, next_free = [], -1 for i in candidates: if i >= next_free: out.append(int(i)) next_free = i + hold + 1 return out def trades_from_signals(ts, close, high, low, entries, hold, sl=None, tp=None, symbol=""): if not entries: return pd.DataFrame(columns=EMPTY) hi_f, lo_f = fwd_extremes(high, low, hold) n = len(close) rows = [] for i, side in entries: if i + 1 >= n: continue e = close[i] if e <= 0: continue j = min(i + hold, n - 1) d = 1.0 if side == "long" else -1.0 pnl = None if sl and i in sl and not np.isnan(lo_f[i]): if side == "long" and lo_f[i] <= sl[i]: pnl = (sl[i] - e) / e elif side == "short" and hi_f[i] >= sl[i]: pnl = (e - sl[i]) / e if pnl is None and tp and i in tp and not np.isnan(hi_f[i]): if side == "long" and hi_f[i] >= tp[i]: pnl = (tp[i] - e) / e elif side == "short" and lo_f[i] <= tp[i]: pnl = (e - tp[i]) / e if pnl is None: pnl = (close[j] - e) / e rows.append({"time": int(ts[i]), "symbol": symbol, "side": side, "gross_bp": float(d * pnl * 1e4)}) return pd.DataFrame(rows) def ts_trades(sym, df, mask, sides, hold, sl_map=None, tp_map=None): entries = [(i, sides[i]) for i in nonoverlap(np.flatnonzero(mask), hold)] return trades_from_signals(df["ts"].to_numpy(), df["close"].to_numpy(float), df["high"].to_numpy(float), df["low"].to_numpy(float), entries, hold, sl=sl_map, tp=tp_map, symbol=sym) def _run_ts(fn) -> pd.DataFrame: out = [] for sym, df in H.iter_pairs(): tr = fn(sym, df) if tr is not None and not tr.empty: out.append(tr) return pd.concat(out, ignore_index=True) if out else pd.DataFrame(columns=EMPTY) # ------------------------------------------------- кросс-секционные методы ---- def xs_book(mkt, name, k, hold, mode="reversal", quantile=0.1, hours_filter=None, shift=0, random_pick=False, seed=3, min_dispersion=0.0) -> pd.DataFrame: """Книга: шорт верхнего / лонг нижнего квантиля по доходности за k баров. shift>0 — плацебо: сигнал сдвинут вперёд на горизонт (то же правило на «будущих» данных). min_dispersion>0 — cost-aware гейт: торгуем только если разброс доходностей в вселенной превышает заданный порог (в долях), иначе оборот съедает всё. """ C, grid, syms, hours = mkt["C"], mkt["grid"], mkt["syms"], mkt["hours"] n_cols = C.shape[1] k_names = max(1, int(round(n_cols * quantile))) rng = np.random.default_rng(seed) rows = [] n_reb = 0 for pos in range(k + 1, len(grid) - hold - shift, hold): if hours_filter is not None and hours[pos] not in hours_filter: continue if n_reb >= MAX_REBALANCES: break n_reb += 1 with np.errstate(invalid="ignore", divide="ignore"): sc = C[pos] / C[pos - k] - 1.0 fw = C[pos + hold + shift] / C[pos + shift] - 1.0 ok = np.isfinite(sc) & np.isfinite(fw) if ok.sum() < max(6, 2 * k_names): continue idxs = np.flatnonzero(ok) scv = sc[idxs] if min_dispersion > 0 and float(np.std(scv)) < min_dispersion: continue if random_pick: pick = rng.permutation(len(idxs))[: 2 * k_names] longs, shorts = idxs[pick[:k_names]], idxs[pick[k_names:]] else: order = np.argsort(scv) longs, shorts = idxs[order[:k_names]], idxs[order[-k_names:]] if mode == "momentum": longs, shorts = shorts, longs t = int(grid[pos]) for j in longs: rows.append((t, syms[j], "long", float(fw[j]) * 1e4)) for j in shorts: rows.append((t, syms[j], "short", -float(fw[j]) * 1e4)) return pd.DataFrame(rows, columns=EMPTY) def m_xs_reversal(mkt) -> pd.DataFrame: return xs_book(mkt, "xs_reversal_15m", k=3, hold=12, mode="reversal") def m_xs_momentum(mkt) -> pd.DataFrame: return xs_book(mkt, "xs_momentum_15m", k=3, hold=12, mode="momentum") def m_hour_xs_reversal(mkt) -> pd.DataFrame: return xs_book(mkt, "hour_xs_reversal", k=3, hold=12, mode="reversal", hours_filter={0, 1, 2, 3, 4, 5, 6, 7}) def m_xs_reversal_costaware(mkt) -> pd.DataFrame: """Cost-aware вариант: торгуем только когда разброс доходностей больше 2 кругов издержек.""" return xs_book(mkt, "xs_reversal_costaware", k=3, hold=12, mode="reversal", min_dispersion=2 * (H.FEE_BP_TAKER + H.SLIP_BP_OTHER) / 1e4) def m_xs_reversal_placebo(mkt) -> pd.DataFrame: return xs_book(mkt, "placebo_xs_reversal", k=3, hold=12, mode="reversal", shift=12) def m_xs_reversal_random(mkt) -> pd.DataFrame: return xs_book(mkt, "random_xs_reversal", k=3, hold=12, random_pick=True) def m_intraday_momentum(mkt) -> pd.DataFrame: """Импульс первой половины суток (00:00–10:00 UTC) → позиция на вторую (10:00–23:00).""" C, grid, syms, hours = mkt["C"], mkt["grid"], mkt["syms"], mkt["hours"] dt = pd.to_datetime(grid, unit="ms", utc=True) day = dt.strftime("%Y-%m-%d").to_numpy() minute = (dt.hour * 60 + dt.minute).to_numpy() i_open = np.flatnonzero(minute == 0) i_cut = np.flatnonzero(minute == 600) i_last = np.flatnonzero(minute == 1380) if not (len(i_open) and len(i_cut) and len(i_last)): return pd.DataFrame(columns=EMPTY) n = min(len(i_open), len(i_cut), len(i_last)) rows = [] for a, b, c in zip(i_open[:n], i_cut[:n], i_last[:n]): s1 = C[b] / C[a] - 1.0 seg = C[c] / C[b] - 1.0 ok = np.isfinite(s1) & np.isfinite(seg) if ok.sum() < 6: continue idxs = np.flatnonzero(ok) med = np.median(s1[idxs]) longs = idxs[s1[idxs] >= med] shorts = idxs[s1[idxs] < med] t = int(grid[b]) for j in longs: rows.append((t, syms[j], "long", float(seg[j]) * 1e4)) for j in shorts: rows.append((t, syms[j], "short", -float(seg[j]) * 1e4)) return pd.DataFrame(rows, columns=EMPTY) def m_lead_lag_btc(mkt) -> pd.DataFrame: """Импульс BTC за 2 бара → позиция в альте на 12 баров; доход — альфа сверх хода BTC.""" C, grid, syms = mkt["C"], mkt["grid"], mkt["syms"] if "BTCUSDT" not in syms: return pd.DataFrame(columns=EMPTY) jb = syms.index("BTCUSDT") btc = C[:, jb] with np.errstate(invalid="ignore", divide="ignore"): r2 = btc / np.roll(btc, 2) - 1.0 fwd_btc = np.roll(btc, -12) / btc - 1.0 r2[:3] = np.nan fwd_btc[-12:] = np.nan thr = pd.Series(r2).rolling(288).quantile(0.8).to_numpy() mask = np.isfinite(r2) & (np.abs(r2) > thr) rows = [] for pos in nonoverlap(np.flatnonzero(mask), 12): side = "long" if r2[pos] > 0 else "short" t = int(grid[pos]) fb = fwd_btc[pos] with np.errstate(invalid="ignore", divide="ignore"): fw_all = C[pos + 12] / C[pos] - 1.0 for j in range(len(syms)): if j == jb: continue fw = fw_all[j] if not np.isfinite(fw) or not np.isfinite(fb): continue alpha = fw - fb rows.append((t, syms[j], side, float(alpha * 1e4 if side == "long" else -alpha * 1e4))) return pd.DataFrame(rows, columns=EMPTY) def m_pairs_btc_eth(mkt) -> pd.DataFrame: """Спред log(ETH/BTC): z-скор за 288 баров, вход |z|>2, выход |z|<0.5 или 288 баров.""" C, grid, syms = mkt["C"], mkt["grid"], mkt["syms"] if not {"BTCUSDT", "ETHUSDT"}.issubset(set(syms)): return pd.DataFrame(columns=EMPTY) jb, je = syms.index("BTCUSDT"), syms.index("ETHUSDT") with np.errstate(invalid="ignore", divide="ignore"): spread = np.log(C[:, je]) - np.log(C[:, jb]) s = pd.Series(spread) z = ((s - s.rolling(288).mean()) / s.rolling(288).std(ddof=0)).to_numpy() rows, pos = [], 288 while pos < len(grid) - 1: zi = z[pos] if not np.isfinite(zi) or abs(zi) <= 2: pos += 1 continue side = "short" if zi > 0 else "long" end = min(pos + 288, len(grid) - 1) for k in range(pos + 1, end): if np.isfinite(z[k]) and abs(z[k]) < 0.5: end = k break t = int(grid[pos]) eth_ret = C[end, je] / C[pos, je] - 1 btc_ret = C[end, jb] / C[pos, jb] - 1 if side == "short": rows += [(t, "ETHUSDT", "short", float(-eth_ret * 1e4)), (t, "BTCUSDT", "long", float(btc_ret * 1e4))] else: rows += [(t, "ETHUSDT", "long", float(eth_ret * 1e4)), (t, "BTCUSDT", "short", float(-btc_ret * 1e4))] pos = end + 1 return pd.DataFrame(rows, columns=EMPTY) # ------------------------------------------------ временные ряды по парам ---- def m_ou_ema(mkt, costaware: bool = False) -> pd.DataFrame: """Возврат к EMA(20): вход |z|>2, выход через 12 баров; cost-aware — только если ожидаемое движение до средней >= 2 кругов издержек (приём из arXiv 2606.00060).""" liquid = mkt["liquid"] def fn(sym, d): c = d["close"].to_numpy(float) ema = pd.Series(c).ewm(span=20, adjust=False).mean().to_numpy() sd = pd.Series(c).rolling(20).std(ddof=0).to_numpy() with np.errstate(invalid="ignore", divide="ignore"): z = (c - ema) / np.where(sd > 0, sd, np.nan) mask = (np.abs(z) > 2) & np.isfinite(z) if costaware: cost_rt = H.cost_bp_per_turn(sym, liquid) / 1e4 mask &= (np.abs(c - ema) / c) >= 2 * cost_rt return ts_trades(sym, d, mask, np.where(z < 0, "long", "short"), 12) return _run_ts(fn) def m_ou_ema_costaware(mkt) -> pd.DataFrame: return m_ou_ema(mkt, costaware=True) def m_vwap_revert(mkt) -> pd.DataFrame: """Откат к VWAP дня: вход при уходе за 2σ от VWAP, выход через 12 баров.""" def fn(sym, d): c = d["close"].to_numpy(float) v = d["volume"].to_numpy(float) with np.errstate(invalid="ignore", divide="ignore"): vwap = np.cumsum(c * v) / np.cumsum(v) dev = c / vwap - 1 sd = pd.Series(dev).rolling(288).std(ddof=0).to_numpy() with np.errstate(invalid="ignore", divide="ignore"): z = dev / np.where(sd > 0, sd, np.nan) mask = np.abs(z) > 2 return ts_trades(sym, d, mask, np.where(z < 0, "long", "short"), 12) return _run_ts(fn) def m_clv_flow(mkt) -> pd.DataFrame: """Ордер-флоу прокси: CLV на краю + всплеск объёма → вход против импульса, выход 6 баров.""" def fn(sym, d): c, h, l = d["close"].to_numpy(float), d["high"].to_numpy(float), d["low"].to_numpy(float) v = d["volume"].to_numpy(float) with np.errstate(invalid="ignore", divide="ignore"): clv = (2 * c - h - l) / np.where((h - l) > 0, h - l, np.nan) vz = (v - pd.Series(v).rolling(288).mean().to_numpy()) / pd.Series(v).rolling(288).std(ddof=0).to_numpy() mask = (np.abs(clv) > 0.8) & (vz > 2) return ts_trades(sym, d, mask, np.where(clv < 0, "long", "short"), 6) return _run_ts(fn) def m_sweep(mkt) -> pd.DataFrame: """Liquidity sweep: прокол экстремума за 48 баров и возврат внутрь диапазона, выход 6 баров.""" def fn(sym, d): c, h, l = d["close"].to_numpy(float), d["high"].to_numpy(float), d["low"].to_numpy(float) lo_min = pd.Series(l).rolling(48).min().shift(1).to_numpy() hi_max = pd.Series(h).rolling(48).max().shift(1).to_numpy() long_m = (l < lo_min) & (c > lo_min) short_m = (h > hi_max) & (c < hi_max) mask = np.nan_to_num(long_m | short_m, nan=False).astype(bool) sides = np.where(np.nan_to_num(long_m, nan=False), "long", "short") return ts_trades(sym, d, mask, sides, 6) return _run_ts(fn) def m_vol_breakout(mkt) -> pd.DataFrame: """Пробой диапазона 48 баров при ATR ниже 80-го персентиля, выход 12 баров.""" def fn(sym, d): c, h, l = d["close"].to_numpy(float), d["high"].to_numpy(float), d["low"].to_numpy(float) hi_max = pd.Series(h).rolling(48).max().shift(1).to_numpy() lo_min = pd.Series(l).rolling(48).min().shift(1).to_numpy() prev = pd.Series(c).shift(1) tr = pd.concat([pd.Series(h) - pd.Series(l), (pd.Series(h) - prev).abs(), (pd.Series(l) - prev).abs()], axis=1).max(axis=1) atr_pct = tr.rolling(14).mean().rolling(288).rank(pct=True).to_numpy() long_m = np.nan_to_num((c > hi_max) & (atr_pct < 0.8), nan=False) short_m = np.nan_to_num((c < lo_min) & (atr_pct < 0.8), nan=False) mask = (long_m | short_m).astype(bool) return ts_trades(sym, d, mask, np.where(long_m, "long", "short"), 12) return _run_ts(fn) def m_prod_mr_5m(mkt) -> pd.DataFrame: """Прод-правила MR TON-бота, применённые НАПРЯМУЮ к 5м-барам (боковик ADX, полосы, RSI, SL/TP).""" from tools.mr_engine import MRParamsStand, bollinger, rsi, adx as adx_fn, atr as atr_fn p = MRParamsStand() def fn(sym, d): f = d.copy() f["atr"] = atr_fn(f) f["mr_rsi"] = rsi(f["close"]) f["mr_adx"] = adx_fn(f) f = pd.concat([f, bollinger(f["close"])], axis=1) c = f["close"].to_numpy(float) mid, up, lo = (f["bb_middle"].to_numpy(float), f["bb_upper"].to_numpy(float), f["bb_lower"].to_numpy(float)) with np.errstate(invalid="ignore", divide="ignore"): bn, rs, ax, at = (f["bb_norm"].to_numpy(float), f["mr_rsi"].to_numpy(float), f["mr_adx"].to_numpy(float), f["atr"].to_numpy(float)) expected = np.abs(c - mid) / c base = (ax <= p.adx_max) & (at > 0) & (expected >= 2 * 0.0011) long_m = np.nan_to_num((bn <= -p.bb_extreme) & (rs <= p.rsi_os) & base, nan=False) short_m = np.nan_to_num((bn >= p.bb_extreme) & (rs >= p.rsi_ob) & base, nan=False) mask = (long_m | short_m).astype(bool) sides = np.where(long_m, "long", "short") sl_map, tp_map = {}, {} for i in np.flatnonzero(mask): if sides[i] == "long": sl_map[int(i)] = min(lo[i] - p.sl_buffer_atr * at[i], c[i] * (1 - p.sl_min_pct / 100)) else: sl_map[int(i)] = max(up[i] + p.sl_buffer_atr * at[i], c[i] * (1 + p.sl_min_pct / 100)) tp_map[int(i)] = mid[i] return ts_trades(sym, d, mask, sides, 8, sl_map=sl_map, tp_map=tp_map) return _run_ts(fn) METHODS = { "xs_reversal_15m": m_xs_reversal, "xs_momentum_15m": m_xs_momentum, "hour_xs_reversal": m_hour_xs_reversal, "xs_reversal_costaware": m_xs_reversal_costaware, "intraday_momentum": m_intraday_momentum, "lead_lag_btc": m_lead_lag_btc, "pairs_btc_eth": m_pairs_btc_eth, "ou_ema_revert": m_ou_ema, "ou_ema_costaware": m_ou_ema_costaware, "vwap_revert": m_vwap_revert, "clv_flow": m_clv_flow, "liquidity_sweep": m_sweep, "vol_breakout": m_vol_breakout, "prod_mr_5m": m_prod_mr_5m, } def main() -> int: import argparse ap = argparse.ArgumentParser() ap.add_argument("--maker", action="store_true", help="считать издержки по мейкеру (0,02% на ногу)") ap.add_argument("methods", nargs="*") args = ap.parse_args() want = args.methods or list(METHODS) tag = "_maker" if args.maker else "" print("сканирую вселенную...", flush=True) mkt = H.build_matrix() liquid = mkt["liquid"] print(f"пар {len(mkt['syms'])} | 5м-сетка {len(mkt['grid'])} интервалов | ликвидных (топ-30): {len(liquid)}", flush=True) rows, tests = [], len(want) for name in want: raw = METHODS[name](mkt) if name == "xs_reversal_15m": # контроли привязаны к базовому методу place_raw = m_xs_reversal_placebo(mkt) rand_raw = m_xs_reversal_random(mkt) tests += 2 else: place_raw = rand_raw = pd.DataFrame(columns=EMPTY) tr = H.apply_costs(raw, liquid, maker=args.maker) pl = H.apply_costs(place_raw, liquid, maker=args.maker) rn = H.apply_costs(rand_raw, liquid, maker=args.maker) H.save_trades(name + tag, tr) row = H.summarize(name, tr, pl, rn) rows.append(row) os.makedirs(os.path.join(ROOT, "results", "rows"), exist_ok=True) pd.DataFrame([row])[["method", "n_trades", "ev_bp", "ev_bp_oos", "t_boot", "placebo_ev_bp", "rand_ev_bp", "sharpe", "max_dd_pct", "by_year_ok"]].to_csv( os.path.join(ROOT, "results", "rows", f"{name}{tag}.csv"), index=False) print(f"{name}: сделок {row['n_trades']}, EV {row['ev_bp']} б.п., OOS {row['ev_bp_oos']}, " f"t {row['t_boot']}, плацебо {row['placebo_ev_bp']}, рандом {row['rand_ev_bp']}, " f"Sharpe {row['sharpe']}, by_year {row['by_year_ok']}", flush=True) del raw, tr, pl, rn if place_raw is not None and not place_raw.empty: del place_raw, rand_raw cols = ["method", "n_trades", "ev_bp", "ev_bp_oos", "t_boot", "placebo_ev_bp", "rand_ev_bp", "sharpe", "max_dd_pct", "by_year_ok"] os.makedirs(os.path.join(ROOT, "results"), exist_ok=True) pd.DataFrame(rows)[cols].to_csv(os.path.join(ROOT, "results", f"summary_5m{tag}.csv"), index=False) H.save_by_year(rows, tag=tag) H.save_counts(tests, tag=tag) with open(os.path.join(ROOT, "results", f"summary_5m{tag}.json"), "w") as fh: json.dump(rows, fh, ensure_ascii=False, indent=1) print(f"записано results/summary_5m{tag}.csv, reports/by_year{tag}.csv, reports/count_tests{tag}.json") return 0 if __name__ == "__main__": sys.exit(main())