Delete knn_calibrate.py

knn_calibrate.py

@@ -1,128 +0,0 @@
-"""
-KNN KALİBRASYON — walk-forward (seçim / test ayrımlı) parametre taraması.
-
-Kullanım (app.py ile aynı klasörde, internet gerekir):
-    python knn_calibrate.py                      # BTC-USDT 1H, son 4000 bar
-    python knn_calibrate.py BTC-USDT 1H 6000
-
-Mantık:
-  1) OKX history-candles ile uzun geçmiş indirilir (yalnızca kapanmış mumlar).
-  2) Parametre ızgarası: K, edge eşiği, ileri bar, SL/TP (ATR) çarpanları.
-  3) Skor aralığının İLK %60'ı = SEÇİM, SON %40'ı = TEST (hiç görülmemiş veri).
-     Parametreler yalnızca seçim dilimindeki sonuca göre sıralanır; TEST dilimi sadece raporlanır.
-  4) Her satırda "başa baş win rate" gösterilir: SL/(SL+TP). Win rate bunun üstünde değilse yüksek win rate kâr demek DEĞİLDİR.
-"""
-import itertools
-import os
-import sys
-import time
-
-os.environ["APP_NO_LAUNCH"] = "1"
-import numpy as np
-import pandas as pd
-
-import app
-
-GRID_K = (9, 15, 25)
-GRID_EDGE = (0.15, 0.30, 0.50)
-GRID_FWD = (3, 5, 8)
-GRID_SLTP = ((1.5, 2.5), (2.0, 2.0), (2.5, 1.5), (3.0, 1.0))   # (SL, TP) ATR
-MIN_TRADES = 20          # seçim diliminde en az bu kadar işlem yoksa satır sıralamaya girmez
-SPLIT = 0.60
-
-
-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 mumun zamanı -> bir 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 _stats(trades, sl, tp):
-    if not trades:
-        return None
-    pnl = np.array([t["pnl"] for t in trades])
-    r = np.array([t["r"] for t in trades])
-    gl = abs(pnl[pnl <= 0].sum())
-    return {"n": len(trades), "wr": float((pnl > 0).mean()),
-            "pf": float(pnl[pnl > 0].sum() / gl) if gl > 0 else float("inf"),
-            "exp_r": float(r.mean()), "be_wr": sl / (sl + tp)}
-
-
-def run(df):
-    n = len(df)
-    results = []
-    for fwd in GRID_FWD:
-        app.KNN_FORWARD = fwd
-        for K, edge in itertools.product(GRID_K, GRID_EDGE):
-            app.KNN_K = K
-            app.KNN_MIN_EDGE = edge
-            app.KNN_TIERS = (edge, min(edge + 0.15, 0.85), min(edge + 0.35, 0.90), min(edge + 0.55, 0.95))
-            ind = app.compute_indicators(df)
-            ind["_arr"]["knn_y"] = app.knn_labels(df, ind, forward=fwd)
-            start = app.WARMUP_BARS + app.KNN_MIN_POOL + fwd
-            scores = [app._score_at(ind, i) for i in range(start, n)]
-            cut = start + int((n - start) * SPLIT)
-            for sl, tp in GRID_SLTP:
-                app.SL_ATR, app.TP_ATR = sl, tp
-                trades = app.simulate_trades(df, ind, start, scores)
-                sel = _stats([t for t in trades if t["signal_i"] < cut], sl, tp)
-                tst = _stats([t for t in trades if t["signal_i"] >= cut], sl, tp)
-                results.append({"K": K, "edge": edge, "fwd": fwd, "sl": sl, "tp": tp, "sel": sel, "test": tst})
-    return results
-
-
-def _row(r):
-    f = lambda s: "   —   " if not s else f"n={s['n']:<3d} WR=%{s['wr'] * 100:4.1f} PF={min(s['pf'], 99):4.2f} R={s['exp_r']:+.2f}"
-    return (f"K={r['K']:<2d} edge={r['edge']:.2f} fwd={r['fwd']} SL/TP={r['sl']}/{r['tp']} (başa baş WR %{r['sel']['be_wr'] * 100:.0f}) | "
-            f"SEÇİM {f(r['sel'])} | TEST {f(r['test'])}")
-
-
-def report(results):
-    ok = [r for r in results if r["sel"] and r["sel"]["n"] >= MIN_TRADES]
-    print(f"\nToplam kombinasyon: {len(results)} · seçim diliminde ≥{MIN_TRADES} işlemli: {len(ok)}")
-    if not ok:
-        print("Yeterli işlem üreten kombinasyon yok — daha çok bar indirin.")
-        return
-    print("\n── Seçim diliminde PROFIT FACTOR'e göre ilk 5 (test dilimi hiç görülmedi) ──")
-    for r in sorted(ok, key=lambda r: -r["sel"]["pf"])[:5]:
-        print(" ", _row(r))
-    hi = [r for r in ok if r["sel"]["pf"] >= 1.0]
-    print("\n── Seçimde PF ≥ 1 olanlar içinde WIN RATE'e göre ilk 5 ──")
-    for r in sorted(hi, key=lambda r: -r["sel"]["wr"])[:5]:
-        print(" ", _row(r))
-    tests = [r for r in ok if r["test"] and r["test"]["n"] >= 10]
-    best = max(tests, key=lambda r: r["test"]["wr"], default=None)
-    print("\n── DÜRÜST ÖZET ──")
-    if best:
-        print(f"Test diliminde (≥10 işlem) görülen en yüksek win rate: %{best['test']['wr'] * 100:.1f}  (bu satır test sonucuna bakılarak seçildi → iyimser)")
-    top = sorted(ok, key=lambda r: -r["sel"]["pf"])[0]
-    if top["test"]:
-        t = top["test"]
-        print(f"Seçimde en iyi parametrenin TEST sonucu: WR %{t['wr'] * 100:.1f}, PF {min(t['pf'], 99):.2f}, ort. R {t['exp_r']:+.2f} ({t['n']} işlem)")
-        print("→ %80 hedefi tutuyorsa hem test win rate'i hem PF > 1 olmalı; yalnızca win rate yüksek ama PF < 1 ise parametre zarar ettiriyor demektir.")
-
-
-if __name__ == "__main__":
-    inst = sys.argv[1] if len(sys.argv) > 1 else "BTC-USDT"
-    bar = sys.argv[2] if len(sys.argv) > 2 else "1H"
-    total = int(sys.argv[3]) if len(sys.argv) > 3 else 4000
-    print(f"{inst} {bar} için {total} bar indiriliyor...")
-    data = fetch_history(inst, bar, total)
-    print(f"{len(data)} bar: {data['ts'].iloc[0]} → {data['ts'].iloc[-1]}")
-    report(run(data))