Upload 2 files

knn_calibrate.py

@@ -0,0 +1,128 @@
+"""
+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.30, 0.50, 0.70)
+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.2, 0.90), min(edge + 0.4, 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))