Spaces:
Sleeping
Sleeping
| """ | |
| KNN KALİBRASYON v2 — rastgele arama + genişleyen pencereli walk-forward + gürültü (null) testi. | |
| Kullanım (app.py ile aynı klasörde, internet gerekir): | |
| python knn_calibrate.py # BTC-USDT 1H, 4000 bar, 120 örnek, 6 dilim | |
| python knn_calibrate.py BTC-USDT 1H 6000 200 8 # sembol bar toplam_bar örnek dilim | |
| python knn_calibrate.py BTC-USDT 1H 4000 120 6 --no-null --force-save | |
| Dinamik SL/TP (dyn, tpq, slq) de aranan parametrelerdir; sabit SL/TP ile aynı walk-forward'da yarışır. | |
| Neden bu kadar titiz: | |
| * KNN eğitim gerektirmez (nedensel) -> tek bir geçmiş geçişi tüm dilimler için geçerlidir. | |
| * WALK-FORWARD: dilim k için parametreler yalnızca k'dan ÖNCE tamamen sonuçlanmış işlemlere bakılarak seçilir, | |
| sonra dilim k'da test edilir. Birleşik test (OOS) işlemleri "seçim PROSEDÜRÜNÜN" dürüst performansıdır. | |
| * Seçim ölçütü kâr faktörü tepe değeri değil, ortalama R'nin ALT GÜVEN SINIRI (ort − hata): az işlemli şanslı | |
| kombinasyonlar cezalandırılır. Win rate hedef değil; başa baş win rate her satırda gösterilir. | |
| * NULL TESTİ: aynı prosedür, bar blokları karıştırılmış (zaman yapısı yok edilmiş) fiyat üzerinde çalıştırılır. | |
| Gürültüde de "iyi" görünüyorsa prosedür kendini kandırıyor demektir. | |
| * knn_params.json YALNIZCA test (OOS) ortalama R'si için %95 bootstrap alt sınırı > 0 ise yazılır (--force-save ile zorlanır). | |
| """ | |
| import datetime as dt | |
| import json | |
| import os | |
| import sys | |
| import time | |
| os.environ["APP_NO_LAUNCH"] = "1" | |
| import numpy as np | |
| import pandas as pd | |
| import app | |
| MIN_TRAIN_TRADES = 15 | |
| BOOT = 2000 | |
| SPACE = { | |
| "K": (7, 11, 15, 21, 31, 45), "edge": (0.10, 0.15, 0.20, 0.30, 0.40, 0.50), | |
| "fwd": (3, 5, 8, 12), "move": (0.10, 0.15, 0.25, 0.35), "pool": (400, 800, 1500), | |
| "sl": (1.0, 1.5, 2.0, 2.5), "tp": (1.5, 2.0, 2.5, 3.5, 5.0), "thr": (2, 3, 4), | |
| "dyn": (0, 1), "tpq": (35, 50, 65), "slq": (60, 75, 90), # dyn=1: KNN dinamik SL/TP (sl/tp = yedek sabit değerler) | |
| # --- v3: giriş filtreleri + zaman çıkışı (skorlar önbellekten gelir, bunlar KNN taramasını yeniden çalıştırmaz) --- | |
| "align": (0, 1), # 1: piyasa yapısı sinyal yönüne ters ise işlem yok | |
| "adx": (0, 15, 20, 25), # >0: ADX alt sınırı | |
| "fee_r": (0.0, 0.30, 0.20, 0.12), # >0: komisyonun R maliyeti üst sınırı (düşük volatilite = pahalı işlem) | |
| "hold": (12, 24, 48), # zaman çıkışı (bar) | |
| # --- v4: formasyon tanıma (KNN özelliği ve/veya teyit oyu). Önbellek anahtarına girer → KNN taraması yeniden çalışır --- | |
| "pat": (0, 1), # 1: KNN özellik vektörüne mum formasyonları + mum şekli eklenir | |
| "pvote": (0, 1), # 1: en güçlü mum formasyonu ±1 teyit oyu | |
| } | |
| BASELINE = {"K": 15, "edge": 0.15, "fwd": 5, "move": 0.15, "pool": 1500, "sl": 1.5, "tp": 2.5, "thr": 2, | |
| "dyn": 1, "tpq": 50, "slq": 75, "align": 0, "adx": 0, "fee_r": 0.0, "hold": 24, "pat": 0, "pvote": 0} | |
| # --------------------------------------------------------------------------- | |
| def fetch_history(inst_id="BTC-USDT", bar="1H", total=4000): | |
| rows, after = [], None | |
| while len(rows) < total: | |
| p = {"instId": inst_id, "bar": bar, "limit": 100} | |
| if after: | |
| p["after"] = after | |
| raw = app._get("/api/v5/market/history-candles", p) | |
| if not raw: | |
| break | |
| rows += raw | |
| after = raw[-1][0] # en eski mum -> sonraki sayfa daha eskiyi getirir | |
| time.sleep(0.12) | |
| cols = ["ts", "o", "h", "l", "c", "vol", "volCcy", "volCcyQuote", "confirm"] | |
| df = pd.DataFrame(rows, columns=cols).drop_duplicates("ts") | |
| for c in ["o", "h", "l", "c", "vol"]: | |
| df[c] = df[c].astype(float) | |
| df["confirm"] = df["confirm"].astype(int) | |
| df["ts"] = pd.to_datetime(df["ts"].astype(np.int64), unit="ms") | |
| df = df[df["confirm"] == 1].sort_values("ts").reset_index(drop=True) | |
| return df.tail(total).reset_index(drop=True) | |
| def shuffle_blocks(df, block=24, seed=0): | |
| """NULL verisi: bar getirilerini bloklar halinde karıştırır (dağılım korunur, zamansal yapı yok olur).""" | |
| rng = np.random.default_rng(seed) | |
| prev = df["c"].shift(1) | |
| rel = pd.DataFrame({k: df[k] / prev for k in ("o", "h", "l", "c")}).iloc[1:].reset_index(drop=True) | |
| vol = df["vol"].iloc[1:].reset_index(drop=True) | |
| idx = np.arange(len(rel)) | |
| blocks = [idx[i:i + block] for i in range(0, len(idx), block)] | |
| order = np.concatenate([blocks[j] for j in rng.permutation(len(blocks))]) | |
| rel, vol = rel.iloc[order].reset_index(drop=True), vol.iloc[order].reset_index(drop=True) | |
| c = float(df["c"].iloc[0]) * rel["c"].cumprod() | |
| pc = np.r_[float(df["c"].iloc[0]), c.to_numpy()[:-1]] | |
| out = pd.DataFrame({"ts": df["ts"].iloc[1:].reset_index(drop=True), "o": rel["o"] * pc, "h": rel["h"] * pc, | |
| "l": rel["l"] * pc, "c": c, "vol": vol}) | |
| out["h"] = out[["o", "h", "l", "c"]].max(axis=1) | |
| out["l"] = out[["o", "h", "l", "c"]].min(axis=1) | |
| return out | |
| def sample_params(rng, n): | |
| out = [dict(BASELINE)] # 0. örnek = mevcut varsayılanlar (kıyas için) | |
| seen = {tuple(BASELINE.values())} | |
| while len(out) < n: | |
| p = {k: (type(v[0])(rng.choice(v))) for k, v in SPACE.items()} | |
| if not p["dyn"]: | |
| p["tpq"], p["slq"] = BASELINE["tpq"], BASELINE["slq"] # dyn=0 iken yüzdelikler anlamsız -> tekrarı önle | |
| if p["tp"] <= p["sl"] * 0.5 or tuple(p.values()) in seen: | |
| continue | |
| seen.add(tuple(p.values())) | |
| out.append(p) | |
| return out | |
| def _set_knn(p): | |
| app.KNN_K, app.KNN_MIN_EDGE, app.KNN_FORWARD = p["K"], p["edge"], p["fwd"] | |
| app.KNN_MOVE_ATR, app.KNN_POOL_MAX = p["move"], p["pool"] | |
| app.KNN_TIERS = app._knn_tiers(p["edge"]) | |
| app.DYNAMIC_EXITS, app.DYN_TP_PCT, app.DYN_SL_PCT = bool(p["dyn"]), p["tpq"], p["slq"] | |
| app.KNN_PATTERN_FEATURES, app.PATTERN_VOTES = bool(p["pat"]), bool(p["pvote"]) | |
| app.ENTRY_REQUIRE_ALIGN, app.ENTRY_MIN_ADX, app.ENTRY_MAX_FEE_R = False, 0.0, 0.0 # önbellek = FİLTRESİZ skorlar | |
| def _set_exec(p): | |
| """Skor önbelleğinden bağımsız parametreler: SL/TP, zaman çıkışı, giriş filtreleri.""" | |
| app.SL_ATR, app.TP_ATR, app.MAX_HOLD_BARS = p["sl"], p["tp"], int(p["hold"]) | |
| app.ENTRY_REQUIRE_ALIGN, app.ENTRY_MIN_ADX, app.ENTRY_MAX_FEE_R = bool(p["align"]), float(p["adx"]), float(p["fee_r"]) | |
| def evaluate_all(df, params, start0, log=True): | |
| """Her parametre kümesi için TÜM zamanın işlemlerini üretir (skorlar KNN anahtarına göre önbelleğe alınır).""" | |
| n = len(df) | |
| ind = app.compute_indicators(df) | |
| cache, results = {}, [] | |
| t0 = time.time() | |
| for k, p in enumerate(params): | |
| key = (p["K"], p["edge"], p["fwd"], p["move"], p["pool"], p["dyn"], p["tpq"], p["slq"], p["pat"], p["pvote"]) | |
| if key not in cache: | |
| _set_knn(p) | |
| ind["_arr"]["knn_F"] = app.knn_features(df, ind, pattern=bool(p["pat"])) | |
| ind["_arr"]["knn_y"] = app.knn_labels(df, ind, forward=p["fwd"], move=p["move"]) | |
| cache[key] = [app._score_at(ind, i) for i in range(start0, n)] | |
| _set_exec(p) | |
| raw = cache[key] | |
| if p["align"] or p["adx"] or p["fee_r"]: | |
| raw = [app.apply_entry_filters(sc, ind["_arr"], start0 + j) for j, sc in enumerate(raw)] | |
| trades = app.simulate_trades(df, ind, start0, raw, threshold=p["thr"]) | |
| results.append((p, trades)) | |
| if log and (k + 1) % 20 == 0: | |
| print(f" {k + 1}/{len(params)} kombinasyon ({time.time() - t0:.0f}s)", flush=True) | |
| return results | |
| def objective(trades): | |
| """Ortalama R'nin alt güven sınırı (ort − hata). İşlem azsa None.""" | |
| if len(trades) < MIN_TRAIN_TRADES: | |
| return None | |
| r = np.array([t["r"] for t in trades]) | |
| return float(r.mean() - r.std(ddof=1) / np.sqrt(len(r))) | |
| def summarize(trades, seed=1): | |
| if not trades: | |
| return None | |
| r = np.array([t["r"] for t in trades]) | |
| pnl = np.array([t["pnl"] for t in trades]) | |
| gl = abs(pnl[pnl <= 0].sum()) | |
| rng = np.random.default_rng(seed) | |
| bm = rng.choice(r, size=(BOOT, len(r)), replace=True).mean(axis=1) | |
| win_r, loss_r = r[pnl > 0], r[pnl <= 0] | |
| be = float(abs(loss_r.mean()) / (win_r.mean() + abs(loss_r.mean()))) if len(win_r) and len(loss_r) else float("nan") | |
| return {"n": len(r), "wr": float((pnl > 0).mean()), "pf": float(pnl[pnl > 0].sum() / gl) if gl > 0 else float("inf"), | |
| "mean_r": float(r.mean()), "ci_lo": float(np.percentile(bm, 2.5)), "ci_hi": float(np.percentile(bm, 97.5)), | |
| "ret": float(np.prod(1 + pnl) - 1), "be_wr": be} | |
| def walk_forward(results, start0, n, folds): | |
| edges = np.linspace(start0, n, folds + 1).astype(int) | |
| oos, picks = [], [] | |
| for k in range(1, folds): | |
| cut = edges[k] | |
| best = None | |
| for p, tr in results: | |
| sc = objective([t for t in tr if t["exit_i"] < cut]) # yalnızca tamamen sonuçlanmış geçmiş | |
| if sc is not None and (best is None or sc > best[0]): | |
| best = (sc, p, tr) | |
| if best is None: | |
| continue | |
| seg = [t for t in best[2] if edges[k] <= t["signal_i"] < edges[k + 1]] | |
| oos += seg | |
| picks.append((k, best[1], len(seg), summarize(seg))) | |
| base = [t for t in results[0][1] if t["signal_i"] >= edges[1]] | |
| return oos, picks, base | |
| def final_pick(results): | |
| best = None | |
| for p, tr in results: | |
| sc = objective(tr) | |
| if sc is not None and (best is None or sc > best[0]): | |
| best = (sc, p, tr) | |
| return best | |
| def procedure(df, n_samples, folds, seed=7, log=True): | |
| n = len(df) | |
| start0 = app.WARMUP_BARS + app.KNN_MIN_POOL + max(SPACE["fwd"]) | |
| params = sample_params(np.random.default_rng(seed), n_samples) | |
| results = evaluate_all(df, params, start0, log) | |
| oos, picks, base = walk_forward(results, start0, n, folds) | |
| return {"results": results, "oos": oos, "picks": picks, "base": base, "final": final_pick(results), "n": n, "start0": start0} | |
| def _pdesc(p): | |
| ex = f"DİNAMİK TP%{p['tpq']}/SL%{p['slq']} (yedek {p['sl']}/{p['tp']})" if p["dyn"] else f"SABİT SL/TP={p['sl']}/{p['tp']}" | |
| pt = f"formasyon[KNN={'+' if p['pat'] else '-'} oy={'+' if p['pvote'] else '-'}] " | |
| flt = pt + f"filtre[yapı={'uyumlu' if p['align'] else '-'} ADX≥{p['adx']} fee≤{p['fee_r']}R] tutuş={p['hold']}" | |
| return f"K={p['K']} edge={p['edge']} fwd={p['fwd']} move={p['move']} pool={p['pool']} {ex} eşik={p['thr']} {flt}" | |
| def _fmt(s): | |
| if not s: | |
| return "—" | |
| return (f"n={s['n']:<3d} WR=%{s['wr'] * 100:4.1f} (başa baş %{s['be_wr'] * 100:4.1f}) PF={min(s['pf'], 99):4.2f} " | |
| f"getiri={s['ret'] * 100:+.1f}% ort.R={s['mean_r']:+.3f} [%95 GA {s['ci_lo']:+.3f}…{s['ci_hi']:+.3f}]") | |
| def main(): | |
| args = [a for a in sys.argv[1:] if not a.startswith("--")] | |
| flags = {a for a in sys.argv[1:] if a.startswith("--")} | |
| inst = args[0] if len(args) > 0 else "BTC-USDT" | |
| bar = args[1] if len(args) > 1 else "1H" | |
| total = int(args[2]) if len(args) > 2 else 4000 | |
| n_samples = int(args[3]) if len(args) > 3 else 200 | |
| folds = int(args[4]) if len(args) > 4 else 6 | |
| print(f"{inst} {bar}: {total} bar indiriliyor...") | |
| df = fetch_history(inst, bar, total) | |
| print(f"{len(df)} bar: {df['ts'].iloc[0]} → {df['ts'].iloc[-1]}") | |
| print(f"\n[1/2] GERÇEK veri: {n_samples} kombinasyon, {folds} dilimli walk-forward") | |
| real = procedure(df, n_samples, folds) | |
| oos, base = summarize(real["oos"]), summarize(real["base"]) | |
| print("\n── Dilim dilim seçilen parametreler (her biri yalnızca geçmişe bakarak) ──") | |
| for k, p, m, s in real["picks"]: | |
| print(f" dilim {k}: {_pdesc(p)} → test {_fmt(s)}") | |
| print("\n── BİRLEŞİK TEST (OOS) ──") | |
| print(" Kalibre prosedür :", _fmt(oos)) | |
| print(" Sabit varsayılan :", _fmt(base)) | |
| sz = app.apply_position_sizing(real["oos"]) | |
| if sz: | |
| print(f" Boyutlandırma : çeyrek-Kelly+DD → getiri {sz['total_return'] * 100:+.1f}% maks.DD {sz['max_dd'] * 100:.1f}% PF={min(sz['pf'], 99):.2f} " | |
| f"({sz['n_active']}/{sz['n']} işlem açık) | sabit %{app.SIZE_RISK_BASE * 100:.0f} risk → getiri {sz['flat_return'] * 100:+.1f}% " | |
| f"maks.DD {sz['flat_max_dd'] * 100:.1f}% PF={min(sz['flat_pf'], 99):.2f}") | |
| print(f" Risk limitleri : VaR95={sz['var95_r'] if sz['var95_r'] is None else round(sz['var95_r'], 2)}R " | |
| f"CVaR95={sz['cvar95_r'] if sz['cvar95_r'] is None else round(sz['cvar95_r'], 2)}R · vetolar: {sz['vetoes'] or 'yok'} " | |
| f"· sıradaki işlem: {sz['next']['veto'] or 'AÇIK (risk %%%.2f)' % (sz['next']['risk'] * 100)}") | |
| null = None | |
| if "--no-null" not in flags: | |
| print(f"\n[2/2] NULL testi (bar blokları karıştırılmış fiyat; aynı prosedür)") | |
| nl = procedure(shuffle_blocks(df), n_samples, folds, log=False) | |
| null = summarize(nl["oos"]) | |
| print(" Null OOS :", _fmt(null)) | |
| fin = real["final"] | |
| print("\n── DÜRÜST HÜKÜM ──") | |
| evidence = bool(oos and oos["ci_lo"] > 0 and oos["n"] >= 30 and oos["pf"] > 1.0 and oos["ret"] > 0) | |
| pos = [s for _, _, _, s in real["picks"] if s] | |
| if pos and sum(1 for s in pos if s["mean_r"] > 0) < len(pos) / 2: | |
| print(f" ⚠ Test dilimlerinin yarısından fazlasında ort. R ≤ 0 ({sum(1 for s in pos if s['mean_r'] > 0)}/{len(pos)} pozitif): kâr birkaç dilime sıkışmış.") | |
| evidence = False | |
| if null and null["mean_r"] > 0.05: | |
| print(f" ⚠ Null testinde bile ort. R {null['mean_r']:+.3f}: prosedür gürültüden 'kâr' üretiyor, sonuçlara temkinli yaklaş.") | |
| evidence = evidence and oos["mean_r"] > null["mean_r"] + 0.05 | |
| if evidence: | |
| print(f" ✔ OOS ort. R %95 alt sınırı > 0 ({oos['ci_lo']:+.3f}) ve ≥30 işlem: istatistiksel kanıt VAR (garanti değil, rejim değişebilir).") | |
| else: | |
| print(" ✘ Kalibre edilmiş sistemin OOS kârlılığı için yeterli istatistiksel kanıt YOK." + (" (--force-save: yine de yazılıyor)" if "--force-save" in flags else " Parametreler dosyaya yazılmaz.")) | |
| if fin: | |
| p = fin[1] | |
| print(f" Tüm veride en iyi (iyimser, in-sample): {_pdesc(p)} → {_fmt(summarize(fin[2]))}") | |
| if evidence or "--force-save" in flags: | |
| out = {"instrument": inst, "bar": bar, "bars": len(df), "created": dt.datetime.now().isoformat(timespec="seconds"), | |
| "params": {"KNN_K": p["K"], "KNN_MIN_EDGE": p["edge"], "KNN_FORWARD": p["fwd"], "KNN_MOVE_ATR": p["move"], | |
| "KNN_POOL_MAX": p["pool"], "SL_ATR": p["sl"], "TP_ATR": p["tp"], "SIGNAL_THRESHOLD": float(p["thr"]), | |
| "DYNAMIC_EXITS": bool(p["dyn"]), "DYN_TP_PCT": p["tpq"], "DYN_SL_PCT": p["slq"], | |
| "MAX_HOLD_BARS": int(p["hold"]), "ENTRY_REQUIRE_ALIGN": bool(p["align"]), | |
| "ENTRY_MIN_ADX": float(p["adx"]), "ENTRY_MAX_FEE_R": float(p["fee_r"]), | |
| "KNN_PATTERN_FEATURES": bool(p["pat"]), "PATTERN_VOTES": bool(p["pvote"])}, | |
| "oos": oos, "null_oos": null} | |
| path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "knn_params.json") | |
| with open(path, "w", encoding="utf-8") as f: | |
| json.dump(out, f, indent=2, ensure_ascii=False) | |
| print(f" → {path} yazıldı. app.py bir sonraki açılışta bu parametreleri otomatik yükler.") | |
| if __name__ == "__main__": | |
| main() | |