#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ analyze_l1_filters.py — ทดสอบตัวกรองระดับ 1 แบบ leave-one-in (ทีละตัว) อ่าน CSV จาก MaePla_L1.mq5 หลักการที่ห้ามลืม ----------------- * แถว CTRL กับ SIG ที่ "แท่งเดียวกัน + ทิศเดียวกัน" มีผลลัพธ์ตรงกันเป๊ะ (entry เดียวกัน · เรขาคณิต TP1500/SL300 เดียวกัน) → ระบบนี้อ้างได้อย่างเดียวคือ "เลือกแท่งเก่ง" · ตัวกรองก็เช่นกัน * ดังนั้นทุกตัวกรอง f ต้องเทียบ: P(ชนะ | มี SIG และผ่าน f) vs P(ชนะ | ผ่าน f เฉยๆ) ไม่ใช่เทียบกับ baseline รวม — ถ้า f ทำให้ทั้งสองฝั่งดีขึ้นเท่ากัน = f ไม่มีค่า * CTRL ถ่วงน้ำหนักตามสัดส่วน BUY/SELL ของ SIG ที่ผ่านตัวกรองนั้น (ทองขาขึ้น BUY ได้เปรียบฟรี) """ import csv, math, random, sys from collections import Counter PT = 0.01 GRID = 5.00 # เส้นสถิติทุก 500 จุด ลงท้าย 0/5 [ปลา3 p.20] ล็อก TOUCH_IN = 0.50 # TouchInPts = 50 จุด [SPEC-เดา G4 default ของ EA-SPEC §3] OVL = 3.00 # OverlapPts = 300 จุด [ปลา3 p.2] ล็อก COST = 40.0 # ต้นทุน (จุด) — ตัวเดียวกับ analyze_l0.py SL_PTS, TP_PTS = 300.0, 1500.0 random.seed(20260830) CSV = sys.argv[1] if len(sys.argv) > 1 else "data/maepla_l1_all.csv" # ---------- โหลด ---------- raw = list(csv.DictReader(open(CSV))) def mk(r): d = 1 if r["direction"] == "BUY" else -1 o = { "kind": r["kind"], "dt": r["datetime"], "year": int(r["year"]), "d": d, "n": int(r["pa_bars"]) if r["pa_bars"] not in ("", "0") else 1, "status": int(r["status"]), "win": int(r["tp1_before_sl"]), "sl": int(r["min_to_sl300"]), "fwd20": float(r["fwd20_points"]), "fwd6": float(r["fwd6_points"]), "atr": float(r["atr14"]), "lo": [float(r["lo1"]), float(r["lo2"]), float(r["lo3"])], "hi": [float(r["hi1"]), float(r["hi2"]), float(r["hi3"])], "bbot": float(r["bbot"]), "btop": float(r["btop"]), "a8": float(r["anch8"]), "a4": float(r["anch4"]), "d1b": int(r["d1b"]), "h4b": int(r["h4b"]), "sw": int(r["swmask"]), } # ผลตอบแทนเป็น R (R = SL 300 จุด) — สูตรเดียวกับ analyze_l0.py if o["win"] == 1: o["R"] = (TP_PTS - COST) / SL_PTS elif o["sl"] >= 0: o["R"] = -(SL_PTS + COST) / SL_PTS else: o["R"] = (o["fwd20"] - COST) / SL_PTS o["fwd6atr"] = (o["fwd6"] * PT / o["atr"]) if o["atr"] > 0 else 0.0 return o rows = [mk(r) for r in raw] SIG = [r for r in rows if r["kind"] == "SIG" and r["status"] == 0] sigkey = set((r["dt"], r["d"]) for r in SIG) CTRL = [r for r in rows if r["kind"] == "CTRL" and (r["dt"], r["d"]) not in sigkey] # ---------- ตัวกรอง ---------- def frame_hit(r, span, lines=None): """ไส้แตะเส้น (ทะลุได้ไม่เกิน OVL) + เนื้อของแท่งปิดกลับฝั่งถูกต้อง""" if r["d"] > 0: lo = r["lo"][span - 1] if lines is None: k0 = math.ceil((lo - TOUCH_IN) / GRID - 1e-9) k1 = math.floor((lo + OVL) / GRID + 1e-9) cand = [k * GRID for k in range(k0, k1 + 1)] else: cand = [f for f in lines if lo - TOUCH_IN - 1e-9 <= f <= lo + OVL + 1e-9] return any(f < r["bbot"] - 1e-9 for f in cand) else: hi = r["hi"][span - 1] if lines is None: k0 = math.ceil((hi - OVL) / GRID - 1e-9) k1 = math.floor((hi + TOUCH_IN) / GRID + 1e-9) cand = [k * GRID for k in range(k0, k1 + 1)] else: cand = [f for f in lines if hi - OVL - 1e-9 <= f <= hi + TOUCH_IN + 1e-9] return any(f > r["btop"] + 1e-9 for f in cand) def day_lines(r, which="a8"): a = r[which] return None if a <= 0 else [a - GRID, a, a + GRID] def popcount(x): return bin(x).count("1") FILTERS = [ ("F1 เกิดที่กรอบ (เส้นสถิติ 0/5)", lambda r: frame_hit(r, 1)), ("F2 เทรนด์ Day (สีแท่ง D1)", lambda r: r["d1b"] == r["d"]), ("F3 เทรนด์ H4 (โครงสร้าง swing)", lambda r: r["h4b"] == r["d"]), ("F4 ไม่อยู่ใน SW (ครบ 4 ข้อ = SW)", lambda r: r["sw"] != 15), ("F5 กรอบวัน 1,000 จุด (anchor 08:00)", lambda r: (day_lines(r) is not None) and frame_hit(r, 1, day_lines(r))), ] VARIANTS = [ ("F1b เกิดที่กรอบ — span = ความยาว PA", lambda r: frame_hit(r, min(r["n"], 3))), ("F1c เกิดที่กรอบ — เกณฑ์อ่อนตาม §1.4.3 (close แทน body)", lambda r: frame_hit_soft(r, 1)), ("F4b ไม่อยู่ใน SW (>=3 จาก 4 ข้อ = SW)", lambda r: popcount(r["sw"]) < 3), ("F4c ไม่อยู่ใน SW (inside range = SW)", lambda r: (r["sw"] & 3) != 3), ("F5b กรอบวัน — anchor 04:00", lambda r: (day_lines(r,'a4') is not None) and frame_hit(r, 1, day_lines(r,'a4'))), ] def frame_hit_soft(r, span): """เกณฑ์ตามตัวหนังสือ §1.4.3 เป๊ะ: เงื่อนไข ③ ใช้ close > f - OVL (อ่อนมาก)""" if r["d"] > 0: lo = r["lo"][span - 1] k0 = math.ceil((lo - TOUCH_IN) / GRID - 1e-9); k1 = math.floor((lo + OVL) / GRID + 1e-9) return any((k * GRID) - OVL < r["btop"] for k in range(k0, k1 + 1)) and k1 >= k0 hi = r["hi"][span - 1] k0 = math.ceil((hi - OVL) / GRID - 1e-9); k1 = math.floor((hi + TOUCH_IN) / GRID + 1e-9) return any((k * GRID) + OVL > r["bbot"] for k in range(k0, k1 + 1)) and k1 >= k0 # ---------- สถิติ ---------- def mean(v): return sum(v) / len(v) if v else float("nan") def varm(v): n = len(v) if n < 2: return float("nan") m = mean(v); return sum((x - m) ** 2 for x in v) / (n - 1) def std_ctrl(sig_sub, ctrl_sub, key): """CTRL ถ่วงน้ำหนักตามสัดส่วนทิศทางของ SIG ที่ผ่านตัวกรอง""" if not sig_sub: return (float("nan"), float("nan")) w = Counter(r["d"] for r in sig_sub); tot = len(sig_sub) val = 0.0; vsum = 0.0 for d, cnt in w.items(): sub = [key(r) for r in ctrl_sub if r["d"] == d] if len(sub) < 2: return (float("nan"), float("nan")) val += (cnt / tot) * mean(sub) vsum += (cnt / tot) ** 2 * varm(sub) / len(sub) return val, math.sqrt(vsum) def edge(sig_sub, ctrl_sub, key): """เปรียบ = SIG − CTRL(ถ่วงทิศ) ภายใน subset เดียวกัน""" if len(sig_sub) < 2: return (float("nan"),) * 4 s = [key(r) for r in sig_sub] ms, vs = mean(s), varm(s) / len(s) mc, sc = std_ctrl(sig_sub, ctrl_sub, key) e = ms - mc; se = math.sqrt(vs + sc * sc) return e, se, ms, mc def boot_ci(v, n=5000): if len(v) < 2: return (float("nan"), float("nan")) N = len(v); out = [] for _ in range(n): out.append(sum(v[random.randrange(N)] for _ in range(N)) / N) out.sort() return out[int(.025 * n)], out[int(.975 * n)] OUT = [] def say(s=""): print(s); OUT.append(s) kW = lambda r: float(r["win"]) kR = lambda r: r["R"] kF = lambda r: r["fwd6atr"] N0 = len(SIG) say("=" * 96) say(f"ระดับ 1 — leave-one-in ทีละตัว | SIG สมบูรณ์ = {N0} · CTRL (ตัดแท่งที่มี SIG ออก) = {len(CTRL)}") say(f"ไฟล์: {CSV}") say("=" * 96) def report(name, f, tag=""): sp = [r for r in SIG if f(r)]; sn = [r for r in SIG if not f(r)] cp = [r for r in CTRL if f(r)]; cn = [r for r in CTRL if not f(r)] n = len(sp) say("") say("-" * 96) say(f"### {name} {tag}") say(f"N เหลือ = {n} ({100*n/N0:.1f}% ของ {N0}) | CTRL ที่ผ่านตัวกรองเดียวกัน = {len(cp)} ({100*len(cp)/len(CTRL):.1f}%)") if n == 0: say(" → ไม่เหลือไม้เลย สรุปไม่ได้"); return None dmix = Counter('BUY' if r['d']>0 else 'SELL' for r in sp) say(f"สัดส่วนทิศทางของ SIG ที่ผ่าน: BUY {dmix.get('BUY',0)} / SELL {dmix.get('SELL',0)}") res = {} for label, key, fmt in [("tp1_before_sl", kW, "pct"), ("expectancy (R)", kR, "R"), ("fwd6 (ATR)", kF, "atr")]: eF, seF, msF, mcF = edge(sp, cp, key) eN, seN, msN, mcN = edge(sn, cn, key) if len(sn) >= 2 else (float("nan"),)*4 eA, seA, msA, mcA = edge(SIG, CTRL, key) dd = eF - eN sdd = math.sqrt(seF**2 + seN**2) if seN == seN else float("nan") t = dd / sdd if sdd == sdd and sdd > 0 else float("nan") lo, hi = boot_ci([key(r) for r in sp]) if fmt == "pct": say(f" {label:<16} SIG {100*msF:5.1f}% [CI {100*lo:4.1f}, {100*hi:4.1f}] " f"CTRL(ผ่านตัวกรองเดียวกัน) {100*mcF:5.1f}% เปรียบ {100*eF:+5.2f}pp (t={eF/seF:+.2f})") elif fmt == "R": say(f" {label:<16} SIG {msF:+.3f}R [CI {lo:+.3f},{hi:+.3f}] CTRL {mcF:+.3f}R " f"เปรียบ {eF:+.3f}R (t={eF/seF:+.2f})") else: say(f" {label:<16} SIG {msF:+.3f} [CI {lo:+.3f},{hi:+.3f}] CTRL {mcF:+.3f} " f"เปรียบ {eF:+.3f} (t={eF/seF:+.2f})") say(f" {'':<16} เปรียบตอนไม่มีตัวกรอง = {eA:+.4f} · เปรียบของไม้ที่ตัวกรองทิ้ง = {eN:+.4f}" f" → Δเปรียบ = {eF-eA:+.4f} (DiD ผ่าน/ตก = {dd:+.4f}, t = {t:+.2f})") res[label] = dict(n=n, eF=eF, eA=eA, dd=dd, t=t, msF=msF, mcF=mcF) # รายปี (เกณฑ์ 4) yrs = sorted(set(r["year"] for r in SIG)) helped = 0; counted = 0; ydet = [] for y in yrs: spy = [r for r in sp if r["year"] == y]; cpy = [r for r in cp if r["year"] == y] sy = [r for r in SIG if r["year"] == y]; cy = [r for r in CTRL if r["year"] == y] if len(spy) < 5: ydet.append(f"{y}:N{len(spy)}(น้อย)"); continue ey, _, _, _ = edge(spy, cpy, kR); ea, _, _, _ = edge(sy, cy, kR) counted += 1 if ey > ea: helped += 1 ydet.append(f"{y}:{ey-ea:+.2f}R(N{len(spy)})") say(f" รายปี (Δเปรียบ R เทียบไม่มีตัวกรอง): {' '.join(ydet)}") say(f" → ช่วย {helped}/{counted} ปีที่นับได้") # เกณฑ์ EA-SPEC §4.2 r_ = res["expectancy (R)"] c1 = r_["eF"] - r_["eA"] >= 0.05 c2 = n >= 0.35 * N0 c3 = (r_["t"] == r_["t"]) and r_["t"] >= 1.5 c4 = helped >= 4 say(f" เกณฑ์ §4.2 : 1) Δexpectancy(เปรียบ) >= +0.05R → {r_['eF']-r_['eA']:+.3f}R {'ผ่าน' if c1 else 'ไม่ผ่าน'}" f" · 2) N >= 35% ({int(0.35*N0)}) → {n} {'ผ่าน' if c2 else 'ไม่ผ่าน'}" f" · 3) t >= 1.5 → {r_['t']:+.2f} {'ผ่าน' if c3 else 'ไม่ผ่าน'}" f" · 4) ช่วย >= 4 ปี → {helped} {'ผ่าน' if c4 else 'ไม่ผ่าน'}") say(f" ⇒ {'เก็บไว้' if (c1 and c2 and c3 and c4) else '🔴 ทิ้ง'}" + (" ⚠️ N < 80 → สรุปไม่ได้" if n < 80 else "")) return res say("") say("##### ชุดหลัก (ทีละตัว — ห้ามรวมกันก่อน) #####") for name, f in FILTERS: report(name, f) say("") say("##### ชุดตรวจความไว (นิยามทางเลือกของตัวเดียวกัน — รายงานคู่กัน ไม่ได้เลือกอันที่สวย) #####") for name, f in VARIANTS: report(name, f) open(CSV.rsplit("/", 1)[0] + "/l1_filters_" + CSV.rsplit("_", 1)[-1].replace(".csv", "") + ".txt", "w", encoding="utf-8").write("\n".join(OUT))