250 lines
10 KiB
Python
250 lines
10 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""P3-S.2 SPEC TESTS — LIQUIDITY SWEEP (synthetic, dari PROJECT SEMANTIC SPECIFICATION v1).
|
||
|
|
|
||
|
|
Truth : docs/SMC_LIQUIDITY_SWEEP_SPEC_v1.md -> spec_oracle() di file ini
|
||
|
|
(expected result di spec_test_cases_liquidity_sweep.json diturunkan
|
||
|
|
dari SPEC, BUKAN dari kode).
|
||
|
|
Code under : DetectLiquidityGrabs (EA AlgoForge_Backtest_Baseline.mq5) + emission
|
||
|
|
audit g_feat[7] (P3-S.0) -> code_port() — hasilnya DILAPORKAN sebagai
|
||
|
|
observasi konformansi, BUKAN definisi truth.
|
||
|
|
|
||
|
|
Disiplin P3-S.2:
|
||
|
|
- spec oracle vs expected : ASSERT (spec = truth)
|
||
|
|
- code port vs spec : REPORT (differential conformance observation)
|
||
|
|
- tidak ada AUC/PF/backtest/human annotation/ML di file ini.
|
||
|
|
|
||
|
|
Usage: python spec_tests_liquidity_sweep.py
|
||
|
|
Output: output/spec_tests_liquidity_sweep_report.json
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
|
||
|
|
import numpy as np
|
||
|
|
|
||
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
|
|
|
||
|
|
# ---- konstanta dari SPEC (mirror smc_semantic_common; sengaja lokal agar oracle
|
||
|
|
# independen dari kode yang sedang diaudit) ----
|
||
|
|
GRAB_WINDOW = 8
|
||
|
|
SEQ_WINDOW = 40
|
||
|
|
INTERNAL_LEN = 5
|
||
|
|
W_LO = 604 # batas bawah window referensi (kasus cache penuh) -> p >= max(95, t-604)
|
||
|
|
EPS = 1e-9
|
||
|
|
|
||
|
|
|
||
|
|
# =====================================================================
|
||
|
|
# SPEC ORACLE (truth) — diturunkan langsung dari SMC_LIQUIDITY_SWEEP_SPEC_v1.md
|
||
|
|
# =====================================================================
|
||
|
|
def spec_oracle(h, l, c, decision_bar, refs, grab_window=GRAB_WINDOW,
|
||
|
|
seq_window=SEQ_WINDOW, window=None):
|
||
|
|
"""Evaluator SPEC (S3-S9).
|
||
|
|
|
||
|
|
refs : [(p, lvl, is_high)] — daftar referensi (internal swing pivot)
|
||
|
|
yang SUDAH valid pada decision_bar (hasil pembentukan referensi).
|
||
|
|
window : opsional [lo, hi] filter bar referensi (SPEC S3).
|
||
|
|
Return : (onsets, f7) ; onsets = [(b, dir, lvl, p)] kronologis;
|
||
|
|
f7 = array int[decision_bar+1] (lifecycle NO_SWEEP/ONSET/ACTIVE/EXPIRED).
|
||
|
|
"""
|
||
|
|
if window is not None:
|
||
|
|
lo, hi = window
|
||
|
|
refs = [(p, lvl, ih) for (p, lvl, ih) in refs if lo <= p <= hi]
|
||
|
|
|
||
|
|
onsets = []
|
||
|
|
for (p, lvl, ih) in refs: # kronologis (p naik)
|
||
|
|
last = min(decision_bar, p + grab_window) # SPEC S4: b in (p, p+GRAB_WINDOW]
|
||
|
|
for b in range(p + 1, last + 1):
|
||
|
|
if ih and h[b] > lvl and c[b] < lvl: # buy-side swept -> dir -1 (SPEC S5)
|
||
|
|
onsets.append((b, -1, lvl, p))
|
||
|
|
break # satu onset per referensi (first-match)
|
||
|
|
if (not ih) and l[b] < lvl and c[b] > lvl: # sell-side swept -> dir +1
|
||
|
|
onsets.append((b, 1, lvl, p))
|
||
|
|
break
|
||
|
|
|
||
|
|
# lifecycle (SPEC S8/S9): f7[r] = dir onset terbaru dgn b<=r dan r-b<=SEQ_WINDOW
|
||
|
|
f7 = np.zeros(decision_bar + 1, dtype=int)
|
||
|
|
for (b, d, lvl, p) in onsets: # urutan kronologis; onset lebih baru
|
||
|
|
upto = min(decision_bar, b + seq_window) # menimpa (supersession)
|
||
|
|
f7[b:upto + 1] = d
|
||
|
|
return onsets, f7
|
||
|
|
|
||
|
|
|
||
|
|
# =====================================================================
|
||
|
|
# CODE PORT (implementasi yang diaudit) — semantik EA DetectLiquidityGrabs
|
||
|
|
# =====================================================================
|
||
|
|
def code_port(h, l, c, decision_bar, refs, grab_window=GRAB_WINDOW,
|
||
|
|
seq_window=SEQ_WINDOW, window=None):
|
||
|
|
"""Port setia EA AlgoForge_Backtest_Baseline.mq5 DetectLiquidityGrabs +
|
||
|
|
emission ComputeMLFeatures g_feat[7] (P3-S.0 fix, AF_BT_SEQ_WINDOW).
|
||
|
|
|
||
|
|
refs : daftar pivot internal g_ip[] pada decision_bar (sudah windowed EA).
|
||
|
|
Return: (cur_sb, cur_sd, f7) — bar onset terakhir, arah state, f7 final.
|
||
|
|
Perbedaan vs spec_oracle:
|
||
|
|
- guard `b > cur_sb` STRICT -> pada bar yang sama, pivot pertama (terlama)
|
||
|
|
yang menang (tie-break berbeda dgn oracle); unreachable dgn data konsisten.
|
||
|
|
- state g_swpDir/g_swpBar persist (bukan event); expiration hanya di emission.
|
||
|
|
"""
|
||
|
|
if window is not None:
|
||
|
|
lo, hi = window
|
||
|
|
refs = [(p, lvl, ih) for (p, lvl, ih) in refs if lo <= p <= hi]
|
||
|
|
cur_sb, cur_sd = -1, 0
|
||
|
|
for (p, lvl, ih) in refs:
|
||
|
|
last = min(decision_bar, p + grab_window)
|
||
|
|
for b in range(p + 1, last + 1):
|
||
|
|
if ih and h[b] > lvl and c[b] < lvl:
|
||
|
|
if b > cur_sb:
|
||
|
|
cur_sb, cur_sd = b, -1
|
||
|
|
break
|
||
|
|
if (not ih) and l[b] < lvl and c[b] > lvl:
|
||
|
|
if b > cur_sb:
|
||
|
|
cur_sb, cur_sd = b, 1
|
||
|
|
break
|
||
|
|
# emission f7 (EA): (tc-1 - g_swpBar) <= SEQ_WINDOW ? g_swpDir : 0
|
||
|
|
f7 = 0 if (cur_sb < 0 or (decision_bar - cur_sb) > seq_window) else cur_sd
|
||
|
|
return cur_sb, cur_sd, f7
|
||
|
|
|
||
|
|
|
||
|
|
# =====================================================================
|
||
|
|
# HELPERS
|
||
|
|
# =====================================================================
|
||
|
|
def materialize(case):
|
||
|
|
"""Baris OHLC -> arrays o,h,l,c ; dukungan bar_spec (generator + patches)."""
|
||
|
|
if "bar_spec" in case:
|
||
|
|
bs = case["bar_spec"]
|
||
|
|
bars = [list(bs["neutral"]) for _ in range(bs["count"])]
|
||
|
|
for (idx, ohlc) in bs["patches"]:
|
||
|
|
bars[idx] = list(ohlc)
|
||
|
|
else:
|
||
|
|
bars = [list(b) for b in case["bars"]]
|
||
|
|
arr = np.asarray(bars, dtype=float)
|
||
|
|
return arr[:, 0], arr[:, 1], arr[:, 2], arr[:, 3]
|
||
|
|
|
||
|
|
|
||
|
|
def apply_transform(o, h, l, c, refs, transform):
|
||
|
|
if transform is None:
|
||
|
|
return o, h, l, c, refs
|
||
|
|
kind = transform["kind"]
|
||
|
|
if kind == "scale":
|
||
|
|
k = transform["factor"]
|
||
|
|
return o * k, h * k, l * k, c * k, [(p, lvl * k, ih) for (p, lvl, ih) in refs]
|
||
|
|
if kind == "translate":
|
||
|
|
v = transform["value"]
|
||
|
|
return o + v, h + v, l + v, c + v, [(p, lvl + v, ih) for (p, lvl, ih) in refs]
|
||
|
|
raise ValueError("unknown transform " + kind)
|
||
|
|
|
||
|
|
|
||
|
|
def apply_mutations(o, h, l, c, mutations):
|
||
|
|
if not mutations:
|
||
|
|
return o, h, l, c
|
||
|
|
o = o.copy(); h = h.copy(); l = l.copy(); c = c.copy()
|
||
|
|
for (idx, ohlc) in mutations:
|
||
|
|
o[idx], h[idx], l[idx], c[idx] = ohlc
|
||
|
|
return o, h, l, c
|
||
|
|
|
||
|
|
|
||
|
|
def onsets_equal(a, b):
|
||
|
|
if len(a) != len(b):
|
||
|
|
return False
|
||
|
|
for x, y in zip(a, b):
|
||
|
|
if x[0] != y[0] or x[1] != y[1] or abs(x[2] - y[2]) > EPS or x[3] != y[3]:
|
||
|
|
return False
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
def run_case(case):
|
||
|
|
o, h, l, c = materialize(case)
|
||
|
|
refs = [tuple(r) for r in case["given_references"]]
|
||
|
|
db = case["decision_bar"]
|
||
|
|
window = case.get("window")
|
||
|
|
transform = case.get("transform")
|
||
|
|
mutations = case.get("mutations")
|
||
|
|
|
||
|
|
# --- run spec oracle pada data dasar ---
|
||
|
|
onsets, f7 = spec_oracle(h, l, c, db, refs, window=window)
|
||
|
|
|
||
|
|
# --- invariant: transform (scale/translate) ---
|
||
|
|
o2, h2, l2, c2, refs2 = apply_transform(o, h, l, c, refs, transform)
|
||
|
|
onsets2, f7_2 = spec_oracle(h2, l2, c2, db, refs2, window=window)
|
||
|
|
|
||
|
|
# --- invariant: future mutation (T13) ---
|
||
|
|
o3, h3, l3, c3 = apply_mutations(o, h, l, c, mutations)
|
||
|
|
onsets3, f7_3 = spec_oracle(h3, l3, c3, db, refs, window=window)
|
||
|
|
|
||
|
|
# --- code port (implementasi diaudit) ---
|
||
|
|
cur_sb, cur_sd, f7_code = code_port(h, l, c, db, refs, window=window)
|
||
|
|
|
||
|
|
exp = case["expected"]
|
||
|
|
checks = {}
|
||
|
|
checks["onsets"] = onsets_equal(onsets, [tuple(e) for e in exp["onsets"]])
|
||
|
|
checks["f7_at_decision"] = bool(int(f7[db]) == int(exp["f7_at_decision"]))
|
||
|
|
lc = exp.get("lifecycle")
|
||
|
|
if lc:
|
||
|
|
if lc["active_until"] <= db:
|
||
|
|
checks["lifecycle_active_until"] = bool(int(f7[lc["active_until"]]) == int(lc["dir"]))
|
||
|
|
if lc["expired_at"] <= db:
|
||
|
|
checks["lifecycle_expired_at"] = bool(int(f7[lc["expired_at"]]) == 0)
|
||
|
|
# invariance transform: struktural (b, dir, p) identik; level ikut skala/translasi
|
||
|
|
struct = lambda os_: [(b, d, p) for (b, d, lv, p) in os_]
|
||
|
|
checks["invariant_transform"] = bool(struct(onsets2) == struct(onsets) and
|
||
|
|
int(f7_2[db]) == int(f7[db]))
|
||
|
|
checks["invariant_mutation"] = bool(onsets_equal(onsets3, onsets) and
|
||
|
|
int(f7_3[db]) == int(f7[db]))
|
||
|
|
|
||
|
|
return {
|
||
|
|
"id": case["id"],
|
||
|
|
"title": case["title"],
|
||
|
|
"expected_onsets": exp["onsets"],
|
||
|
|
"expected_f7": exp["f7_at_decision"],
|
||
|
|
"spec_oracle_onsets": [[b, d, round(float(lv), 6), p] for (b, d, lv, p) in onsets],
|
||
|
|
"spec_oracle_f7": int(f7[db]),
|
||
|
|
"code_port_onset_bar": cur_sb,
|
||
|
|
"code_port_dir": cur_sd,
|
||
|
|
"code_port_f7": int(f7_code),
|
||
|
|
"code_matches_spec": bool(int(f7_code) == int(f7[db])),
|
||
|
|
"checks": checks,
|
||
|
|
"pass": all(checks.values()),
|
||
|
|
"note": case.get("note"),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
cases_path = os.path.join(HERE, "spec_test_cases_liquidity_sweep.json")
|
||
|
|
cases = json.load(open(cases_path, encoding="utf-8"))["cases"]
|
||
|
|
results = [run_case(c) for c in cases]
|
||
|
|
|
||
|
|
n_pass = sum(1 for r in results if r["pass"])
|
||
|
|
n_total = len(results)
|
||
|
|
print(f"=== P3-S.2 SPEC TESTS — LIQUIDITY SWEEP ({n_pass}/{n_total} PASS) ===")
|
||
|
|
print(f"{'ID':<8} {'PASS':<6} {'f7 spec':<8} {'f7 code':<8} {'match':<6} notes")
|
||
|
|
for r in results:
|
||
|
|
print(f"{r['id']:<8} {str(r['pass']):<6} {r['spec_oracle_f7']:<8} "
|
||
|
|
f"{r['code_port_f7']:<8} {str(r['code_matches_spec']):<6} {r['title'][:44]}")
|
||
|
|
if n_pass != n_total:
|
||
|
|
print("\nFAILED:")
|
||
|
|
for r in results:
|
||
|
|
if not r["pass"]:
|
||
|
|
print(" ", r["id"], r["checks"])
|
||
|
|
sys.exit(1)
|
||
|
|
print("\nAll spec assertions PASS — expected results are consistent with "
|
||
|
|
"PROJECT SEMANTIC SPECIFICATION v1.")
|
||
|
|
|
||
|
|
report = {
|
||
|
|
"spec_doc": "docs/SMC_LIQUIDITY_SWEEP_SPEC_v1.md",
|
||
|
|
"phase": "P3-S.2",
|
||
|
|
"generated_utc": __import__("datetime").datetime.now(
|
||
|
|
__import__("datetime").timezone.utc).isoformat(),
|
||
|
|
"constants": {"GRAB_WINDOW": GRAB_WINDOW, "SEQ_WINDOW": SEQ_WINDOW,
|
||
|
|
"INTERNAL_LEN": INTERNAL_LEN, "W_LO": W_LO},
|
||
|
|
"summary": {"total": n_total, "passed": n_pass, "failed": n_total - n_pass},
|
||
|
|
"cases": results,
|
||
|
|
}
|
||
|
|
outdir = os.path.join(HERE, "output")
|
||
|
|
os.makedirs(outdir, exist_ok=True)
|
||
|
|
out_path = os.path.join(outdir, "spec_tests_liquidity_sweep_report.json")
|
||
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
||
|
|
json.dump(report, f, indent=2, default=str)
|
||
|
|
print(f"[saved] {out_path}")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|