#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ analyze_l0.py — วิเคราะห์ผล backtest ระดับ 0 ของระบบแม่ปลา อ่าน CSV ที่ MaePla_L0.mq5 เขียนออกมา แล้วตัดสินตามเกณฑ์ EA-SPEC §4.1 (A-F) ใช้: python3 analyze_l0.py [data/maepla_l0_all.csv] หลักการที่ต้องเข้าใจก่อนอ่านผล ------------------------------ * แถว kind=CTRL = "เข้าไม้ที่ทุกแท่ง H4 ด้วย SL/TP เดียวกัน" = baseline สุ่มเชิงประจักษ์ (เส้นที่สัญญาณต้องชนะ — ไม่ใช่แค่ชนะ 0) * แถว kind=SIG = PA ที่ตรวจเจอ · status=0 คือผ่านด่านไส้หลังซิกครบ (= Sig สมบูรณ์) status 1/2/3/4 = ตกด่าน W1/W2/W3/W5 (เก็บไว้ดู funnel) * tp1_before_sl = การแข่งเรขาคณิตคงที่ TP +1500 / SL -300 จาก entry → เกณฑ์ B ของ EA-SPEC * sys_tp1_before_sl* = การแข่งของ "ระบบจริง" ท่า A2: TP = ไส้หลังซิก+1500 · SL = ไส้ของ Sig * null เชิงทฤษฎีของการแข่ง (random walk ไม่มี drift) = S/(S+T) """ import sys, math import numpy as np import pandas as pd PT = 0.01 CSV = sys.argv[1] if len(sys.argv) > 1 else "data/maepla_l0_all.csv" RNG = np.random.default_rng(20260830) df = pd.read_csv(CSV) df["datetime"] = pd.to_datetime(df["datetime"], format="%Y.%m.%d %H:%M") df["fwd6_atr"] = (df["fwd6_points"] * PT) / df["atr14"].replace(0, np.nan) df["fwd20_atr"] = (df["fwd20_points"] * PT) / df["atr14"].replace(0, np.nan) ctrl = df[df["kind"] == "CTRL"].copy() sig = df[(df["kind"] == "SIG") & (df["status"] == 0)].copy() # Sig สมบูรณ์เท่านั้น rej = df[(df["kind"] == "SIG") & (df["status"] != 0)].copy() # ระยะ SL/TP จริงของแต่ละสัญญาณ (ท่า A2) d = np.where(sig["direction"] == "BUY", 1, -1) sig["sl_dist_pts"] = d * (sig["entry_price"] - sig["sl_price"]) / PT sig["tp_dist_pts"] = d * (sig["tp1_price"] - sig["entry_price"]) / PT sig["null_sys"] = sig["sl_dist_pts"] / (sig["sl_dist_pts"] + sig["tp_dist_pts"]) WINDOWS = { "FULL 2019.01–2026.08": ("2019-01-01", "2026-08-01"), "IN-SAMPLE 2019.01–2024.08":("2019-01-01", "2024-08-01"), "HOLDOUT 2024.08–2026.08": ("2024-08-01", "2026-08-01"), "TASK 2024.01–2026.08": ("2024-01-01", "2026-08-01"), } def win(x, a, b): return x[(x["datetime"] >= a) & (x["datetime"] < b)] def boot_ci(v, n=10000, lo=2.5, hi=97.5): v = np.asarray(v, dtype=float) if len(v) == 0: return (np.nan, np.nan) idx = RNG.integers(0, len(v), size=(n, len(v))) m = v[idx].mean(axis=1) return (np.percentile(m, lo), np.percentile(m, hi)) def prop_z(k1, n1, k2, n2): """z ของผลต่างสัดส่วน (สัญญาณ vs baseline)""" if n1 == 0 or n2 == 0: return np.nan p1, p2 = k1/n1, k2/n2 p = (k1+k2)/(n1+n2) se = math.sqrt(p*(1-p)*(1/n1+1/n2)) return (p1-p2)/se if se > 0 else np.nan def welch_t(a, b): a = np.asarray(a, float); a = a[~np.isnan(a)] b = np.asarray(b, float); b = b[~np.isnan(b)] if len(a) < 3 or len(b) < 3: return np.nan, np.nan diff = a.mean() - b.mean() se = math.sqrt(a.var(ddof=1)/len(a) + b.var(ddof=1)/len(b)) return diff, (diff/se if se > 0 else np.nan) def line(c="-"): print(c*78) OUT = [] def say(s=""): print(s); OUT.append(s) say("="*78) say("MaePla ระดับ 0 — baseline เปลือย : ตัวสัญญาณมีเปรียบเชิงสถิติจริงไหม") say("="*78) say(f"ไฟล์ : {CSV}") say(f"ช่วงข้อมูล : {df['datetime'].min()} → {df['datetime'].max()}") say(f"แถวทั้งหมด : {len(df):,} (CTRL {len(ctrl):,} / SIG {len(sig)+len(rej):,})") say(f"race ตัดสินด้วย M1 : {(df['race_src']=='M1').mean()*100:.1f}%") say("") # ---------------------------------------------------------------- funnel say("## 1. FUNNEL — จาก PA ดิบ เหลือ Sig สมบูรณ์เท่าไหร่") allsig = df[df["kind"] == "SIG"] smap = {0:"VALID (ผ่านครบ)",1:"W1 ปิดผิดทาง",2:"W2 บอดี้เกินครึ่ง (PA ถูกทำลาย)", 3:"W3 ไส้เกินแท่งก่อน (PA ถูกทำลาย)",4:"W5 ไส้ทะลุชุด PA"} for s_, name in smap.items(): n = (allsig["status"] == s_).sum() say(f" status {s_} {name:<34} : {n:6,} ({n/len(allsig)*100:5.1f}%)") say(f" → PA candidates (หลัง dedupe) {len(allsig):,} → Sig สมบูรณ์ {len(sig):,}") say("") # ---------------------------------------------------------------- N say("## 2. จำนวนสัญญาณ (เกณฑ์ F: N >= 300)") say(f" รวม : {len(sig)} BUY {int((sig['direction']=='BUY').sum())} / SELL {int((sig['direction']=='SELL').sum())}") say(" แยกตาม pattern:") for p, g in sig.groupby("pattern_type"): nb = int((g["direction"]=="BUY").sum()); ns = int((g["direction"]=="SELL").sum()) say(f" {p:<7} : {len(g):4d} (BUY {nb:3d} / SELL {ns:3d})") say(" ต่อปี:") yr = sig.groupby(sig["datetime"].dt.year).size() say(" " + " ".join(f"{y}:{n}" for y, n in yr.items())) say(f" เกณฑ์ F : {'ผ่าน' if len(sig)>=300 else 'ไม่ผ่าน'} (N={len(sig)})") say("") # ------------------------------------------------- criterion B (fixed geom) say("## 3. เกณฑ์ B — การแข่ง TP/SL เรขาคณิตคงที่ P(MFE ถึง 1500 ก่อน MAE ถึง 300)") say(" null ทฤษฎี (random walk) = 300/(300+1500) = 16.67%") say(" เกณฑ์ผ่าน EA-SPEC: rate >= 22% และ bootstrap 95%CI ขอบล่าง > 18%") say("") say(f" {'ชุด':<34}{'N':>6}{'rate':>9}{'95%CI':>18}{'vs CTRL z':>11}") line() def rate_block(label, s_, c_, key="tp1_before_sl"): n1, k1 = len(s_), int(s_[key].sum()) n2, k2 = len(c_), int(c_[key].sum()) r1 = k1/n1 if n1 else np.nan lo, hi = boot_ci(s_[key].values) if n1 else (np.nan, np.nan) z = prop_z(k1, n1, k2, n2) say(f" {label:<34}{n1:>6}{r1*100:>8.1f}% [{lo*100:5.1f}%,{hi*100:5.1f}%]{z:>11.2f}") return dict(label=label, n=n1, rate=r1, lo=lo, hi=hi, z=z, base=(k2/n2 if n2 else np.nan), base_n=n2) res = {} for wname, (a, b) in WINDOWS.items(): s_ = win(sig, a, b); c_ = win(ctrl, a, b) res[wname] = rate_block(wname, s_, c_) line() c_all = ctrl say(f" {'BASELINE สุ่ม (ทุกแท่ง H4) FULL':<34}{len(c_all):>6}{c_all['tp1_before_sl'].mean()*100:>8.1f}%" f" [{boot_ci(c_all['tp1_before_sl'].values)[0]*100:5.1f}%,{boot_ci(c_all['tp1_before_sl'].values)[1]*100:5.1f}%]") say("") say(" แยกฝั่ง (เกณฑ์ E — ทองขาขึ้น ถ้าไม่แยกจะเข้าใจผิด):") say(f" {'':<34}{'N':>6}{'SIG rate':>10}{'CTRL rate':>11}{'z':>8}") line() for dname in ["BUY", "SELL"]: s_ = sig[sig["direction"] == dname]; c_ = ctrl[ctrl["direction"] == dname] z = prop_z(int(s_["tp1_before_sl"].sum()), len(s_), int(c_["tp1_before_sl"].sum()), len(c_)) say(f" {dname:<34}{len(s_):>6}{s_['tp1_before_sl'].mean()*100:>9.1f}%{c_['tp1_before_sl'].mean()*100:>10.1f}%{z:>8.2f}") line(); say("") say(" แยกตาม pattern (FULL):") say(f" {'pattern':<12}{'N':>5}{'rate':>9}{'95%CI ขอบล่าง':>16}{'vs CTRL z':>11}") line() cb = int(ctrl["tp1_before_sl"].sum()); cn = len(ctrl) pat_rows = [] for p, g in sig.groupby("pattern_type"): lo, hi = boot_ci(g["tp1_before_sl"].values) z = prop_z(int(g["tp1_before_sl"].sum()), len(g), cb, cn) say(f" {p:<12}{len(g):>5}{g['tp1_before_sl'].mean()*100:>8.1f}%{lo*100:>15.1f}%{z:>11.2f}") pat_rows.append((p, len(g), g["tp1_before_sl"].mean(), lo, z)) line(); say("") # --------------------------------------------- criterion B on real A2 geometry say("## 4. การแข่งของ 'ระบบจริง' ท่า A2 — TP = ไส้หลังซิก+1500 · SL = ไส้ของ Sig") say(" (คนละอันกับข้อ 3 — ข้อ 3 คือหัววัดคงที่ของ EA-SPEC, ข้อนี้คือเรขาคณิตที่โค้ชสอนจริง)") say(f" ระยะ SL เฉลี่ย {sig['sl_dist_pts'].mean():.0f} จุด (มัธยฐาน {sig['sl_dist_pts'].median():.0f}) · " f"TP เฉลี่ย {sig['tp_dist_pts'].mean():.0f} จุด") say(f" null ทฤษฎีเฉลี่ยต่อสัญญาณ S/(S+T) = {sig['null_sys'].mean()*100:.1f}%") say("") say(f" {'สำนัก (กติกาไส้ถูกทำลาย §4.1)':<40}{'rate':>9}{'null':>8}{'ส่วนต่าง':>10}") line() for col, nm in [("sys_tp1_before_sl_pla", "แม่ปลา price > ไส้"), ("sys_tp1_before_sl_kaew", "ลูกแก้ว price >= ไส้ (สัมผัสก็ตาย)"), ("sys_tp1_before_sl_moja", "โมจา price > ไส้ + 300 จุด")]: r = sig[col].mean() nu = sig["null_sys"].mean() if "moja" not in col else \ ((sig["sl_dist_pts"]+300)/((sig["sl_dist_pts"]+300)+sig["tp_dist_pts"])).mean() say(f" {nm:<40}{r*100:>8.1f}%{nu*100:>7.1f}%{(r-nu)*100:>9.1f}pp") line() say(" baseline สุ่มเชิงประจักษ์ (โค้ง CTRL: rate จริง เทียบ null ทฤษฎี ที่ 4 เรขาคณิต):") for sl_col, tp_col, S, T in [("min_to_sl300","min_to_tp1",300,1500), ("min_to_sl500","min_to_tp1",500,1500), ("min_to_sl300","min_to_tp2",300,3000), ("min_to_sl500","min_to_tp2",500,3000)]: hit_tp = ctrl[tp_col] >= 0 hit_sl = ctrl[sl_col] >= 0 first = hit_tp & (~hit_sl | (ctrl[tp_col] < ctrl[sl_col])) nu = S/(S+T) say(f" SL {S:4d} / TP {T:4d} → null {nu*100:5.1f}% จริง {first.mean()*100:5.1f}%" f" (จริง/null = {first.mean()/nu:.2f}x)") say("") # ---------------------------------------------------------- criterion A say("## 5. เกณฑ์ A — Forward excursion dir x (Close[t+6] - entry) / ATR หัก drift ของ 'ทุกแท่ง'") say(" เกณฑ์ผ่าน EA-SPEC: ส่วนต่าง >= +0.20 ATR และ t >= 2.5") say("") say(f" {'ชุด':<24}{'SIG mean':>10}{'CTRL mean':>11}{'ส่วนต่าง':>10}{'t':>8}") line() critA = {} for dname in ["ALL", "BUY", "SELL"]: s_ = sig if dname == "ALL" else sig[sig["direction"] == dname] c_ = ctrl if dname == "ALL" else ctrl[ctrl["direction"] == dname] diff, t = welch_t(s_["fwd6_atr"], c_["fwd6_atr"]) say(f" fwd6 {dname:<18}{s_['fwd6_atr'].mean():>10.3f}{c_['fwd6_atr'].mean():>11.3f}{diff:>10.3f}{t:>8.2f}") critA[dname] = (diff, t) for dname in ["ALL"]: s_ = sig; c_ = ctrl diff, t = welch_t(s_["fwd20_atr"], c_["fwd20_atr"]) say(f" fwd20 {dname:<18}{s_['fwd20_atr'].mean():>10.3f}{c_['fwd20_atr'].mean():>11.3f}{diff:>10.3f}{t:>8.2f}") line(); say("") # ---------------------------------------------------------- MFE / MAE say("## 6. MFE / MAE (20 แท่ง H4) — หน่วยจุด") say(f" {'ชุด':<16}{'MFE mean':>10}{'MFE med':>9}{'MAE mean':>10}{'MAE med':>9}{'MFE/MAE':>9}{'bars→max':>10}") line() for nm, g in [("SIG ทั้งหมด", sig), ("SIG BUY", sig[sig.direction=="BUY"]), ("SIG SELL", sig[sig.direction=="SELL"]), ("CTRL ทั้งหมด", ctrl), ("CTRL BUY", ctrl[ctrl.direction=="BUY"]), ("CTRL SELL", ctrl[ctrl.direction=="SELL"]), ("PA ตกด่าน", rej)]: say(f" {nm:<16}{g['mfe_points_20bars'].mean():>10.0f}{g['mfe_points_20bars'].median():>9.0f}" f"{g['mae_points_20bars'].mean():>10.0f}{g['mae_points_20bars'].median():>9.0f}" f"{g['mfe_points_20bars'].mean()/max(g['mae_points_20bars'].mean(),1):>9.2f}" f"{g['bars_to_max'].mean():>10.1f}") line(); say("") # ---------------------------------------------------------- criterion C say("## 7. เกณฑ์ C — Expectancy จริง (R multiple, เรขาคณิตคงที่ 1500/300, ต้นทุน 40 จุด)") say(" ชนะ = +5R (1500/300) · แพ้ = -1R · ไม่ถึงทั้งคู่ใน 20 แท่ง = ปิดที่ราคาแท่งที่ 20") def expectancy(g, cost_pts=40.0): S, T = 300.0, 1500.0 winf = g["tp1_before_sl"] == 1 losef = (g["min_to_sl300"] >= 0) & (~winf) openf = ~(winf | losef) r = np.where(winf, (T-cost_pts)/S, np.where(losef, -(S+cost_pts)/S, (g["fwd20_points"]-cost_pts)/S)) return r for nm, g in [("SIG ทั้งหมด", sig), ("SIG BUY", sig[sig.direction=="BUY"]), ("SIG SELL", sig[sig.direction=="SELL"]), ("CTRL ทั้งหมด", ctrl)]: r = expectancy(g) gross_w = r[r > 0].sum(); gross_l = -r[r < 0].sum() pf = gross_w/gross_l if gross_l > 0 else np.inf lo, hi = boot_ci(r) say(f" {nm:<16} E = {r.mean():+.3f}R [{lo:+.3f},{hi:+.3f}] PF = {pf:.2f} N={len(g)}") say("") # ---------------------------------------------------------- criterion D say("## 8. เกณฑ์ D — ความสม่ำเสมอรายปี (rate เกณฑ์ B ต่อปี)") say(f" {'ปี':<7}{'N':>5}{'SIG rate':>10}{'CTRL rate':>11}{'ชนะ?':>7}") line() wins = 0; years = 0 for y in sorted(sig["datetime"].dt.year.unique()): s_ = sig[sig["datetime"].dt.year == y]; c_ = ctrl[ctrl["datetime"].dt.year == y] ok = s_["tp1_before_sl"].mean() > c_["tp1_before_sl"].mean() wins += int(ok); years += 1 say(f" {y:<7}{len(s_):>5}{s_['tp1_before_sl'].mean()*100:>9.1f}%{c_['tp1_before_sl'].mean()*100:>10.1f}%{('ชนะ' if ok else 'แพ้'):>7}") line(); say(f" ชนะ baseline {wins}/{years} ปี"); say("") # ---------------------------------------------------------- funnel value say("## 9. ด่านไส้หลังซิกเพิ่มค่าไหม (VALID เทียบ PA ที่ตกด่าน)") say(f" {'ชุด':<32}{'N':>6}{'rate B':>9}{'fwd6 ATR':>10}") line() say(f" {'Sig สมบูรณ์ (VALID)':<32}{len(sig):>6}{sig['tp1_before_sl'].mean()*100:>8.1f}%{sig['fwd6_atr'].mean():>10.3f}") for s_, nm in [(1,"ตก W1 ปิดผิดทาง"),(2,"ตก W2 บอดี้เกิน"),(3,"ตก W3 ไส้เกิน"),(4,"ตก W5 ทะลุชุด PA")]: g = rej[rej["status"] == s_] if len(g) == 0: continue say(f" {nm:<32}{len(g):>6}{g['tp1_before_sl'].mean()*100:>8.1f}%{g['fwd6_atr'].mean():>10.3f}") say(f" {'baseline สุ่ม (CTRL)':<32}{len(ctrl):>6}{ctrl['tp1_before_sl'].mean()*100:>8.1f}%{ctrl['fwd6_atr'].mean():>10.3f}") line(); say("") # ---------------------------------------------------------- verdict say("## 10. คำตัดสิน") line("=") full = res["FULL 2019.01–2026.08"] A_diff, A_t = critA["ALL"] A_pass = (A_diff >= 0.20) and (A_t >= 2.5) B_pass = (full["rate"] >= 0.22) and (full["lo"] > 0.18) E_buy = sig[sig.direction=="BUY"]["tp1_before_sl"].mean() E_sell = sig[sig.direction=="SELL"]["tp1_before_sl"].mean() E_pass = (E_buy >= 0.167-0.01) and (E_sell >= 0.167-0.01) F_pass = len(sig) >= 300 D_pass = wins >= 4 r_ = expectancy(sig); C_pass = (r_.mean() >= 0.05) say(f" A Forward excursion ต้อง >= +0.20 ATR, t >= 2.5 → {A_diff:+.3f} ATR, t={A_t:.2f} {'ผ่าน' if A_pass else 'ไม่ผ่าน'}") say(f" B การแข่ง TP/SL ต้อง >= 22%, CI ขอบล่าง > 18% → {full['rate']*100:.1f}%, ขอบล่าง {full['lo']*100:.1f}% {'ผ่าน' if B_pass else 'ไม่ผ่าน'}") say(f" C Expectancy ต้อง >= +0.05R → {r_.mean():+.3f}R {'ผ่าน' if C_pass else 'ไม่ผ่าน'}") say(f" D สม่ำเสมอ ต้องชนะ >= 4 ปี → {wins}/{years} ปี {'ผ่าน' if D_pass else 'ไม่ผ่าน'}") say(f" E แยกฝั่ง BUY/SELL ต้องผ่านทั้งคู่ → BUY {E_buy*100:.1f}% / SELL {E_sell*100:.1f}% {'ผ่าน' if E_pass else 'ไม่ผ่าน'}") say(f" F N ต้อง >= 300 → {len(sig)} {'ผ่าน' if F_pass else 'ไม่ผ่าน'}") line("=") if A_pass and B_pass: say(" ➜ ผ่านระดับ 0 (A และ B ผ่านทั้งคู่) — ไปต่อระดับ 1 ได้") elif A_pass and not B_pass: say(" ➜ A ผ่าน B ไม่ผ่าน — มีเปรียบเชิงทิศทาง แต่ TP/SL ตั้งผิด") say(" ต้องแก้ TP/SL ก่อน (ลอง ATR_SCALED) ห้ามข้ามไปเติมตัวกรอง") else: say(" ➜ ไม่ผ่านระดับ 0 — สัญญาณเปล่าไม่มีเปรียบเหนือการเข้าไม้สุ่ม") say(" EA-SPEC §4.1 สั่งไว้: หยุด · ห้ามหวังว่าตัวกรองจะช่วย") say(" (ใส่ตัวกรองบนสัญญาณที่เปรียบเป็นศูนย์ = ได้ subset ที่ดูดีเสมอ = overfitting)") line("=") with open("RESULT-L0.txt", "w", encoding="utf-8") as f: f.write("\n".join(OUT)) print("\n[เขียนผลดิบไว้ที่ RESULT-L0.txt]")