298 lines
12 KiB
Python
 | """VtCore.mqh / VtExposure.mqh / VtJournal.mqh against tests/reference.py.
| ||
| |||
python3 tests/check_core.py write <dir> # cases for the dump tool (files folder, sub-folder vttest) + journal paths
| |||
python3 tests/check_core.py check <dir> # compare out_<k>.txt, out_pip.txt, journal_out.txt with the twin
| |||
Cases (CASES) cover hedging and netting (reversals), partial closes, cash flow and credit, commissions/swap/fee, the
| |||
filters (symbol, magic, comment, start date), 'now' on a Monday, the 1st of a month, 1 January and a Sunday, an empty
| |||
history and a 3,000-position one. Every number of the card is compared with 1e-9 relative tolerance."""
| |||
# SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0
| |||
# Copyright 2026 Allan Niñal (https://www.allanninal.dev/mql5)
| |||
import csv
| |||
import random
| |||
import sys
| |||
from pathlib import Path
| |||
| |||
sys.path.insert(0, str(Path(__file__).parent))
| |||
from reference import DAY, Engine, days_from_civil, pip_multiplier, synth_case # noqa: E402
| |||
| |||
COLS = ["time", "pos", "magic", "type", "entry", "volume", "profit", "swap", "commission", "fee", "symbol", "comment"]
| |||
| |||
| |||
def at(y, m, d, h=0, mi=0, s=0):
| |||
return days_from_civil(y, m, d) * DAY + h * 3600 + mi * 60 + s
| |||
| |||
| |||
CASES = [
| |||
dict(name="hedging, plain", kw=dict(seed=1)),
| |||
dict(name="netting reversals", kw=dict(seed=2, n_pos=800, netting=True)),
| |||
dict(name="filter symbol", kw=dict(seed=3, n_pos=500), f=dict(symbol="XAUUSD")),
| |||
dict(name="filter magic+comment", kw=dict(seed=4, n_pos=900), f=dict(magic=11, comment="SCALP")),
| |||
dict(name="now Monday 00:00:05", kw=dict(seed=5, now=at(2026, 10, 19, 0, 0, 5), n_pos=600)),
| |||
dict(name="now 1 Nov (month start)", kw=dict(seed=6, now=at(2026, 11, 1), n_pos=600, netting=True)),
| |||
dict(name="now 1 Jan (year start)", kw=dict(seed=7, now=at(2027, 1, 1, 0, 0, 1), n_pos=700)),
| |||
dict(name="start-date filter", kw=dict(seed=8, n_pos=700), f=dict(start=at(2026, 3, 1))),
| |||
dict(name="empty history", kw=dict(seed=9, n_pos=0, cash=False)),
| |||
dict(name="Sunday, 3000 positions", kw=dict(seed=10, n_pos=3000, now=at(2026, 10, 18, 22), netting=True,
| |||
magics=tuple(range(30)))),
| |||
dict(name="no cash flow", kw=dict(seed=11, n_pos=400, cash=False)),
| |||
]
| |||
PIPS = [("EURUSD", 5), ("EURUSD", 4), ("USDJPY", 3), ("USDJPY", 2), ("XAUUSD", 2), ("XAUUSD", 3), ("GOLD", 2),
| |||
("XAGUSD", 3), ("XAGUSD", 4), ("SILVER", 3), ("US30", 1), ("US30", 2), ("GER40", 1), ("BTCUSD", 2),
| |||
("BTCUSD", 0), ("XAUEUR", 2), ("AUDNZD", 6), ("EURUSD.m", 5), ("GBPJPY", 4)]
| |||
| |||
| |||
def build(k):
| |||
return synth_case(**CASES[k]["kw"])
| |||
| |||
| |||
def run_twin(deals, now, bal, journal, positions, f):
| |||
e = Engine(symbol=f.get("symbol", ""), magic=f.get("magic", -1), comment=f.get("comment", ""),
| |||
start=f.get("start", 0))
| |||
for d in deals:
| |||
e.add(d)
| |||
o = e.report(now, bal, journal)
| |||
o.update(e.positions(positions, bal))
| |||
return o
| |||
| |||
| |||
def write(out: Path):
| |||
out.mkdir(parents=True, exist_ok=True)
| |||
for k, c in enumerate(CASES):
| |||
deals, now, bal, journal, positions = build(k)
| |||
f = c.get("f", {})
| |||
with open(out / f"deals_{k}.csv", "w", newline="") as fh:
| |||
w = csv.writer(fh)
| |||
w.writerow(COLS)
| |||
for d in deals:
| |||
w.writerow([repr(d[x]) if isinstance(d[x], float) else d[x] for x in COLS])
| |||
with open(out / f"meta_{k}.csv", "w", newline="") as fh:
| |||
fh.write("key,value\n")
| |||
fh.write(f"now,{now}\nbalance,{bal!r}\nf_symbol,{f.get('symbol', '')}\nf_magic,{f.get('magic', -1)}\n")
| |||
fh.write(f"f_comment,{f.get('comment', '')}\nf_start,{f.get('start', 0)}\n")
| |||
with open(out / f"jr_{k}.csv", "w", newline="") as fh:
| |||
fh.write("day,money,pct\n")
| |||
for dn in sorted(journal):
| |||
fh.write(f"{dn},{journal[dn][0]!r},{journal[dn][1]!r}\n")
| |||
with open(out / f"pos_{k}.csv", "w", newline="") as fh:
| |||
fh.write("symbol,magic,comment,dir,volume,profit,pips,sl,risk,value,base,ccy\n")
| |||
for p in positions:
| |||
fh.write(",".join(str(p[x]) if not isinstance(p[x], float) else repr(p[x]) for x in
| |||
("symbol", "magic", "comment", "dir", "volume", "profit", "pips", "sl", "risk", "value",
| |||
"base", "ccy")) + "\n")
| |||
with open(out / "pip_test.csv", "w") as fh:
| |||
fh.write("symbol,digits\n")
| |||
for s, dg in PIPS:
| |||
fh.write(f"{s},{dg}\n")
| |||
write_journal(out)
| |||
print(f"wrote {len(CASES)} cases to {out}")
| |||
| |||
| |||
# ---- compare ------------------------------------------------------------------------------------------------------------
| |||
def parse_out(p: Path):
| |||
r = {}
| |||
for line in p.read_text().splitlines():
| |||
if "\t" in line:
| |||
k, v = line.split("\t", 1)
| |||
r[k] = v
| |||
return r
| |||
| |||
| |||
def same(a, b):
| |||
try:
| |||
x, y = float(a), float(b)
| |||
except (TypeError, ValueError):
| |||
return str(a) == str(b)
| |||
return abs(x - y) <= 1e-9 * max(1.0, abs(x), abs(y))
| |||
| |||
| |||
def diff(mt, ref):
| |||
bad = []
| |||
for k in sorted(set(mt) | set(ref)):
| |||
if k not in mt:
| |||
bad.append(f"{k}: missing in the terminal (twin {ref[k]})")
| |||
elif k not in ref:
| |||
bad.append(f"{k}: missing in twin (the terminal {mt[k]})")
| |||
elif not same(mt[k], ref[k]):
| |||
bad.append(f"{k}: the terminal {mt[k]} vs twin {ref[k]}")
| |||
return bad
| |||
| |||
| |||
def check(d: Path):
| |||
bad_total = 0
| |||
for k, c in enumerate(CASES):
| |||
p = d / f"out_{k}.txt"
| |||
if not p.exists():
| |||
print(f"case {k} ({c['name']}): out file missing")
| |||
bad_total += 1
| |||
continue
| |||
deals, now, bal, journal, positions = build(k)
| |||
ref = run_twin(deals, now, bal, journal, positions, c.get("f", {}))
| |||
mt = parse_out(p)
| |||
bad = diff(mt, ref)
| |||
print(f"case {k:2} {c['name']:26} deals {len(deals):5} trades {ref['ntrades']:5} values compared {len(ref):6} "
| |||
f"{'OK' if not bad else 'MISMATCH ' + str(len(bad))}")
| |||
for b in bad[:6]:
| |||
print(" ", b)
| |||
bad_total += len(bad)
| |||
pip = parse_out(d / "out_pip.txt")
| |||
pbad = [f"pip.{i}: {pip.get(f'pip.{i}')} vs {pip_multiplier(s, dg)} ({s},{dg})" for i, (s, dg) in enumerate(PIPS)
| |||
if not same(pip.get(f"pip.{i}"), pip_multiplier(s, dg))]
| |||
print(f"pip table {len(PIPS)} symbols: {'OK' if not pbad else 'MISMATCH'}")
| |||
for b in pbad:
| |||
print(" ", b)
| |||
bad_total += len(pbad)
| |||
print("ALL PASS: 0 mismatches" if not bad_total else f"FAILURES: {bad_total} mismatches")
| |||
return bad_total == 0
| |||
| |||
| |||
# ---- the journal ---------------------------------------------------------------------------------------------------------
| |||
class JModel:
| |||
"""Python twin of CVtJournal (and the file): memory dict day -> (money, pct); 'file' = list of lines."""
| |||
def __init__(self, files, name, ms):
| |||
self.files, self.name = files, name
| |||
self.mem = {}
| |||
self.dirty = False
| |||
self.last = ms
| |||
for dn, m, p in self.read():
| |||
self.put(dn, m, p)
| |||
self.dirty = False
| |||
| |||
def read(self):
| |||
out = []
| |||
for line in self.files.get(self.name, []):
| |||
v = line.split(",")
| |||
if len(v) >= 3 and int(v[0]) > 0:
| |||
out.append((int(v[0]), min(float(v[1]), 0.0), min(float(v[2]), 0.0)))
| |||
return out
| |||
| |||
def put(self, dn, m, p):
| |||
if dn in self.mem:
| |||
if m < self.mem[dn][0]:
| |||
self.mem[dn] = (m, p)
| |||
return True
| |||
return False
| |||
self.mem[dn] = (m, p)
| |||
return True
| |||
| |||
def sample(self, t, eq, bal, ms):
| |||
day = t // DAY
| |||
money = round(min(eq - bal, 0.0), 2)
| |||
pct = round(money / bal * 100.0 if bal > 0 else 0.0, 4)
| |||
if self.put(day, money, pct):
| |||
self.dirty = True
| |||
if self.dirty and ms - self.last >= 10000:
| |||
self.flush(False, ms)
| |||
| |||
def flush(self, force, ms):
| |||
if not self.dirty:
| |||
return
| |||
if not force and ms - self.last < 10000:
| |||
return
| |||
self.last = ms
| |||
for dn, m, p in self.read():
| |||
self.put(dn, m, p)
| |||
self.files[self.name] = ["%d,%.2f,%.4f" % (dn, self.mem[dn][0], self.mem[dn][1]) for dn in sorted(self.mem)]
| |||
self.dirty = False
| |||
| |||
| |||
def journal_script():
| |||
"""(seed file lines, [(kind, t, eq, bal, ms)]): midnight crossings, clean and crash restarts, a restart across
| |||
midnight, a three-day gap, a flat day, another chart having written a deeper low."""
| |||
rnd = random.Random(77)
| |||
d0 = days_from_civil(2026, 10, 19)
| |||
seed = ["%d,%.2f,%.4f" % (d0 - 2, -120.5, -1.205), "%d,%.2f,%.4f" % (d0 - 1, 0.0, 0.0),
| |||
"%d,%.2f,%.4f" % (d0, -35.0, -0.35)]
| |||
steps = []
| |||
t = d0 * DAY + 23 * 3600 + 50 * 60
| |||
ms = 5000
| |||
bal = 10000.0
| |||
eq = bal
| |||
for i in range(900): # 15 minutes across midnight
| |||
eq = round(bal + rnd.gauss(-20, 60) if rnd.random() < 0.7 else bal + rnd.uniform(0, 80), 2)
| |||
steps.append(("S", t, eq, bal, ms))
| |||
t += 1
| |||
ms += 1000
| |||
if i in (200, 640):
| |||
steps.append(("R", t, 0, 0, ms)) # clean restart
| |||
if i in (400, 700):
| |||
steps.append(("C", t, 0, 0, ms)) # crash restart (no flush)
| |||
if i == 450:
| |||
bal = 10110.33 # a deposit/close changed the balance
| |||
# restart across midnight and a three-day gap
| |||
steps.append(("C", t, 0, 0, ms))
| |||
t += 3 * DAY + 4000
| |||
ms += 7000
| |||
for i in range(60):
| |||
eq = round(bal - rnd.uniform(0, 300), 2)
| |||
steps.append(("S", t, eq, bal, ms))
| |||
t += 1
| |||
ms += 500 # faster than 1 s: throttle matters
| |||
steps.append(("R", t, 0, 0, ms))
| |||
t += 100
| |||
for i in range(30): # a flat day: equity never under the balance
| |||
steps.append(("S", t + DAY, bal + rnd.uniform(0, 50), bal, ms))
| |||
t += 1
| |||
ms += 1000
| |||
return seed, steps
| |||
| |||
| |||
def write_journal(out: Path):
| |||
seed, steps = journal_script()
| |||
(out / "jt_seed.csv").write_text("day,money,pct\n" + "\n".join(seed) + "\n")
| |||
with open(out / "jt_paths.csv", "w", newline="") as fh:
| |||
fh.write("kind,time,equity,balance,ms\n")
| |||
for k, t, eq, bal, ms in steps:
| |||
fh.write(f"{k},{t},{eq!r},{bal!r},{ms}\n")
| |||
| |||
| |||
def check_journal(d: Path):
| |||
p = d / "journal_out.txt"
| |||
if not p.exists():
| |||
print("journal: journal_out.txt missing")
| |||
return False
| |||
seed, steps = journal_script()
| |||
files = {"f": list(seed)}
| |||
j = JModel(files, "f", 0)
| |||
snaps, n = [], 0
| |||
for k, t, eq, bal, ms in steps:
| |||
if k == "S":
| |||
j.sample(t, eq, bal, ms)
| |||
else:
| |||
if k == "R":
| |||
j.flush(True, ms)
| |||
j = JModel(files, "f", ms)
| |||
snaps.append(("%d" % n, list(files["f"])))
| |||
n += 1
| |||
j.flush(True, 4000000000)
| |||
snaps.append(("final", list(files["f"])))
| |||
mem = ["%d,%.2f,%.4f" % (dn, j.mem[dn][0], j.mem[dn][1]) for dn in sorted(j.mem)]
| |||
exp = []
| |||
for tag, lines in snaps:
| |||
exp += ["SNAP " + tag] + lines
| |||
exp += ["MEM"] + mem
| |||
got = [x for x in p.read_text().splitlines() if x]
| |||
ok = got == exp
| |||
nsamples = sum(1 for s in steps if s[0] == "S")
| |||
nrestarts = len(steps) - nsamples
| |||
print(f"journal: {nsamples} samples, {nrestarts} restarts, {len(mem)} days; compared {len(exp)} lines: "
| |||
f"{'OK, 0 mismatches' if ok else 'MISMATCH'}")
| |||
if not ok:
| |||
for i, (a, b) in enumerate(zip(got, exp)):
| |||
if a != b:
| |||
print(" first difference at line", i, ": the terminal", a, "| twin", b)
| |||
break
| |||
if len(got) != len(exp):
| |||
print(" line counts: the terminal", len(got), "twin", len(exp))
| |||
return ok
| |||
| |||
| |||
if __name__ == "__main__":
| |||
mode, d = sys.argv[1], Path(sys.argv[2])
| |||
if mode == "write":
| |||
write(d)
| |||
elif mode == "check":
| |||
ok = check(d)
| |||
ok = check_journal(d) and ok if (d / "journal_out.txt").exists() else ok
| |||
sys.exit(0 if ok else 1)
| |||
elif mode == "journal":
| |||
sys.exit(0 if check_journal(d) else 1)
|