""" Compare two Kronos advisor configurations via backtest. Config A: "conservative" — текущие параметры - pause при range > 5% - step = max(base*0.7, min(base*1.5, advice)) - center_offset clamp ±0.2% Config B: "adjusted" — мягкие - pause при range > 8% (или bias>0.7) - step = max(base*0.85, min(base*1.3, advice)) # меньше снижение - center_offset clamp ±0.15% """ from __future__ import annotations import argparse import json import logging import math import sys import time from dataclasses import dataclass from pathlib import Path from typing import Optional import numpy as np import pandas as pd logger = logging.getLogger(__name__) def simulate_grid(df, start, horizon, step_pct, levels, qty=0.001, center_offset_pct=0.0, fee_pct=0.001): if start >= len(df) - 1: return 0.0, 0, 0 center = float(df["close"].iloc[start]) * (1.0 + center_offset_pct) step = center * step_pct if step <= 0: return 0.0, 0, 0 buy_levels = sorted([center - n * step for n in range(1, levels + 1) if center - n * step > 0]) sell_levels = sorted([center + n * step for n in range(1, levels + 1)]) end = min(start + horizon, len(df)) pnl, n_roundtrips, n_wins = 0.0, 0, 0 pos_open = None for i in range(start + 1, end): hi, lo = float(df["high"].iloc[i]), float(df["low"].iloc[i]) if pos_open is None: for bl in buy_levels: if lo <= bl: pnl -= bl * qty * fee_pct pos_open = bl break else: for sl in sell_levels: if hi >= sl: gross = (sl - pos_open) * qty fee = sl * qty * fee_pct + pos_open * qty * fee_pct net = gross - fee pnl += net if net > 0: n_wins += 1 n_roundtrips += 1 pos_open = None break pnl_pct = pnl / (center * qty) if center > 0 else 0.0 return pnl_pct, n_roundtrips, n_wins @dataclass class BTResult: name: str total_pnl_pct: float n_trades: int winrate: float max_dd_pct: float sharpe: float def run_one(df, tf_min, lookback, pred_len, static_step_pct, levels, advisor, cfg: dict, stride=12, regime_window_hours=720): n = len(df) pnls, ns, ws, regimes = [], [], [], [] pause_hits = 0 peak, eq, max_dd = 1.0, 1.0, 0.0 for t in range(lookback, n - pred_len, stride): horizon = min(pred_len, stride * 2) # Regime win_start = max(0, t - regime_window_hours) win = df["close"].iloc[win_start:t + 1] ret = (win.iloc[-1] - win.iloc[0]) / win.iloc[0] if len(win) > 1 and win.iloc[0] > 0 else 0.0 regime = "trending" if abs(ret) > 0.05 else "sideways" regimes.append(regime) # PAUSE check advice = advisor.advise(df.iloc[:t + 1], lookback=lookback, pred_len=pred_len, tf_min=tf_min) if advice.pause_grid and advice.confidence >= cfg["pause_min_conf"]: if advice.expected_range_pct > cfg["pause_range_thr"] or ( abs(advice.center_offset_pct) / max(advice.expected_range_pct, 1e-4) > cfg["pause_bias_thr"] ): pause_hits += 1 pnls.append(0.0); ns.append(0); ws.append(0) continue # Step kronos_step = float(advice.step_percent) eff_step = max(static_step_pct * cfg["step_floor"], min(static_step_pct * cfg["step_ceil"], kronos_step)) # Center eff_off = max(-cfg["center_clamp"], min(cfg["center_clamp"], advice.center_offset_pct)) pnl_pct, nt, nw = simulate_grid(df, t, horizon, eff_step, levels, center_offset_pct=eff_off) pnls.append(pnl_pct); ns.append(nt); ws.append(nw) eq *= (1.0 + pnl_pct) peak = max(peak, eq) max_dd = min(max_dd, (eq - peak) / peak) if not pnls: return BTResult("?", 0.0, 0, 0.0, 0.0, 0.0), [] total = float(np.prod(1.0 + np.array(pnls)) - 1.0) nt, nw = sum(ns), sum(ws) wr = nw / nt if nt else 0.0 sharpe = float(np.mean(pnls) / (np.std(pnls) + 1e-9) * math.sqrt(252 / max(1, len(pnls) // 24))) if len(pnls) > 1 else 0.0 return BTResult(cfg["name"], total, nt, wr, max_dd, sharpe), regimes def main(): ap = argparse.ArgumentParser() ap.add_argument("--csv", required=True) ap.add_argument("--tf", type=int, default=60) ap.add_argument("--lookback", type=int, default=400) ap.add_argument("--pred-len", type=int, default=24) ap.add_argument("--step", type=float, default=0.005) ap.add_argument("--levels", type=int, default=10) ap.add_argument("--stride", type=int, default=12) ap.add_argument("--model", default="kronos-mini") ap.add_argument("--device", default="cpu") args = ap.parse_args() logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") log = logging.getLogger("compare") df = pd.read_csv(args.csv) df.columns = [c.strip().lower() for c in df.columns] if "timestamps" in df.columns: df["timestamps"] = pd.to_datetime(df["timestamps"]) df = df.set_index("timestamps") elif "timestamp" in df.columns: df["timestamps"] = pd.to_datetime(df["timestamp"]) df = df.set_index("timestamps") df = df[["open", "high", "low", "close", "volume"]].astype(float).sort_index() log.info(f"Loaded {len(df)} candles") # Static baseline (no advisor) log.info("=== STATIC baseline (no advisor) ===") static_pnls, static_n, static_w = [], [], [] peak, eq, max_dd = 1.0, 1.0, 0.0 n = len(df) for t in range(args.lookback, n - args.pred_len, args.stride): horizon = min(args.pred_len, args.stride * 2) pnl, nt, nw = simulate_grid(df, t, horizon, args.step, args.levels) static_pnls.append(pnl); static_n.append(nt); static_w.append(nw) eq *= (1.0 + pnl) peak = max(peak, eq) max_dd = min(max_dd, (eq - peak) / peak) s_total = float(np.prod(1.0 + np.array(static_pnls)) - 1.0) s_sharpe = float(np.mean(static_pnls) / (np.std(static_pnls) + 1e-9) * math.sqrt(252 / max(1, len(static_pnls) // 24))) static_res = BTResult("static", s_total, sum(static_n), sum(static_w) / sum(static_n) if sum(static_n) else 0.0, max_dd, s_sharpe) from kronos import KronosAdvisor advisor = KronosAdvisor(model_name=args.model, device=args.device) log.info("Loaded advisor") cfgs = [ {"name": "conservative", "pause_min_conf": 0.6, "pause_range_thr": 0.05, "pause_bias_thr": 0.6, "step_floor": 0.7, "step_ceil": 1.5, "center_clamp": 0.002}, {"name": "adjusted", "pause_min_conf": 0.7, "pause_range_thr": 0.08, "pause_bias_thr": 0.75, "step_floor": 0.85, "step_ceil": 1.3, "center_clamp": 0.0015}, {"name": "minimal", "pause_min_conf": 0.85, "pause_range_thr": 0.10, "pause_bias_thr": 0.9, "step_floor": 0.9, "step_ceil": 1.15, "center_clamp": 0.001}, ] results = [static_res] for cfg in cfgs: log.info(f"=== {cfg['name']} ===") t0 = time.time() r, regimes = run_one(df, args.tf, args.lookback, args.pred_len, args.step, args.levels, advisor, cfg, args.stride) log.info(f" {cfg['name']} done in {time.time()-t0:.1f}s") # Per-regime breakdown results.append(r) print() print("=" * 88) print(f"{'MODE':<15} {'PnL %':>10} {'#Trades':>10} {'WinRate':>10} {'MaxDD %':>10} {'Sharpe':>10}") print("-" * 88) for r in results: print(f"{r.name:<15} {r.total_pnl_pct*100:>9.2f}% {r.n_trades:>10d} " f"{r.winrate*100:>9.1f}% {r.max_dd_pct*100:>9.2f}% {r.sharpe:>9.2f}") print("=" * 88) if __name__ == "__main__": main()